From 98876d3368d51997852935f6cb2da0e5e85754a6 Mon Sep 17 00:00:00 2001 From: Luke Craig Date: Wed, 8 Jul 2026 18:34:51 -0400 Subject: [PATCH 1/9] feat(api): round-5 data contracts (models, TS types, client surface) Shared shapes the round-5 cockpit features build on, landed together since they cross the wire as one contract: - QueuedReply/QueueView, SlashCommand, Runway* (session/window/response), Hooks status, Profile/ProfileParam, layout Snapshot* models - TmuxPane gains window_id, last_activity, queued, and the split preview/preview_tail (heavy full screen ships only on ?previews=1) - matching TS interfaces + api client methods --- backend/muse/models.py | 81 +++++++++++++++++++++- frontend/src/api/client.ts | 134 +++++++++++++++++++++++++++++++++++-- frontend/src/api/types.ts | 122 ++++++++++++++++++++++++++++++++- 3 files changed, 331 insertions(+), 6 deletions(-) diff --git a/backend/muse/models.py b/backend/muse/models.py index f6c16e7..05b06aa 100644 --- a/backend/muse/models.py +++ b/backend/muse/models.py @@ -642,6 +642,7 @@ class TmuxPane(BaseModel): pane_id: str session_name: str window_index: int + window_id: str = "" # stable window handle (@) — the target for move-window window_name: str window_active: bool pane_index: int @@ -650,15 +651,21 @@ class TmuxPane(BaseModel): cwd: str title: str = "" session_attached: bool = True # False => detached session (scratch/background) + last_activity: int = 0 # epoch secs of last window activity (most-recent sort) muse_session_id: Optional[str] = None # set if this pane runs a tracked session context_pct: Optional[float] = None # context-window occupancy (tracked sessions) + queued: int = 0 # replies queued for delivery when this session's turn ends # Attention status for grouping the mobile task list. # responded = a live session whose turn ended (a response is ready for you). status: Literal["needs_you", "responded", "working", "idle"] = "idle" attention: str = "" # short reason, e.g. "permission prompt", "working" # Claude Code's permission mode (Shift+Tab cycles it); None for non-Claude panes. mode: Optional[Literal["default", "acceptEdits", "plan", "bypass"]] = None - preview: str = "" # recent captured lines (plain text) + # Full screen text is heavy (hundreds of KB across a fleet) — it ships only + # when the client asks (?previews=1); the one-line tail always ships for + # task-list subtitles. The deck fetches live screens per-pane instead. + preview: str = "" # visible screen (ANSI), only when previews=1 + preview_tail: str = "" # last visible line (ANSI stripped, short) options: list[PendingOption] = Field(default_factory=list) # menu detected in buffer @@ -668,6 +675,78 @@ class TmuxLayout(BaseModel): reason: Optional[str] = None +class SlashCommand(BaseModel): + """One entry in the composer's "/" autocomplete: a Claude Code slash command + available to the pane's session (built-in, user-global, or project-local).""" + + name: str # invoked as "/{name}" — subdir commands are namespaced with ":" + description: str = "" + source: Literal["builtin", "user", "project"] = "builtin" + + +class QueuedReply(BaseModel): + """A user-authored message waiting to be typed into a session's pane the next + time that session is genuinely idle (turn ended, no menu pending).""" + + id: int + session_id: str + text: str + created_at: Optional[datetime] = None + status: Literal["pending", "sent", "cancelled", "failed"] = "pending" + # turn = deliver alone and let it run a full turn; append = glue onto the + # previous queued item so both go in one message. + mode: Literal["turn", "append"] = "turn" + sent_at: Optional[datetime] = None + error: Optional[str] = None + + +class QueueView(BaseModel): + """A session's queue plus why it isn't delivering right now (so the UI can + explain a held reply instead of leaving it silently pending).""" + + items: list[QueuedReply] = [] + hold_reason: Optional[str] = None + + +class RunwaySession(BaseModel): + session_id: str + title: str = "" + cost_usd: float = 0.0 + + +class RunwayWindow(BaseModel): + """Spend vs budget for one rate-limit window (5h or weekly).""" + + label: str + window_seconds: int + anchor: Optional[datetime] = None # window start + anchor_source: str = "estimated" # "reset" when anchored to an observed reset + elapsed_seconds: int = 0 + remaining_seconds: int = 0 + cost_usd: float = 0.0 + budget_usd: Optional[float] = None + # Where the budget came from: "configured" (MUSE_LIMIT_*_USD), "observed" + # (spend at the last real limit hit), or "none" (subscription plan with no + # calibration yet — show spend without a ceiling, never a made-up estimate). + budget_source: str = "none" + pct_used: Optional[float] = None # cost/budget, None without a budget + pct_elapsed: float = 0.0 # time progress through the window + + +class RunwayResponse(BaseModel): + """How much headroom is left before hitting the plan's usage limits — the + fleet-driving question ('can I keep all these sessions running?').""" + + generated_at: datetime + plan_label: Optional[str] = None + five_hour: RunwayWindow + week: RunwayWindow + burn_usd_per_hour: float = 0.0 # trailing burn rate (last 30 min, annualized to 1h) + projected_exhaust_at: Optional[datetime] = None # when the 5h budget runs out at this burn + exhaust_before_reset: bool = False # True => you'll hit the limit before the window resets + top_sessions: list[RunwaySession] = Field(default_factory=list) # burners this 5h window + + class Bookmark(BaseModel): message_uuid: str note: str = "" diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 0652a78..f27a1af 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -24,11 +24,19 @@ import type { NotifyConfig, OpenLoop, NotifyResult, + PaneScreen, + SlashCommand, PendingOptions, PushSubscriptionInfo, + QueuedReply, + QueueView, + RunwayResponse, + HooksStatus, TmuxLayout, Pack, + LayoutSnapshot, PersistedOutput, + Profile, ReentryBrief, RelatedSession, SearchResponse, @@ -48,11 +56,23 @@ function notifyAuthRequired(status: number): void { if (status === 401) window.dispatchEvent(new Event("muse:auth-required")); } +// FastAPI reports errors as `{ "detail": "..." }`; prefer that human message over the +// bare status line so config/validation errors (e.g. a broken profiles.toml) surface. +async function errorMessage(res: Response, url: string): Promise { + try { + const body = (await res.clone().json()) as { detail?: unknown }; + if (typeof body?.detail === "string" && body.detail) return body.detail; + } catch { + /* not JSON — fall through to the status line */ + } + return `${res.status} ${res.statusText} for ${url}`; +} + async function getJSON(url: string, signal?: AbortSignal): Promise { const res = await fetch(url, { signal }); if (!res.ok) { notifyAuthRequired(res.status); - throw new Error(`${res.status} ${res.statusText} for ${url}`); + throw new Error(await errorMessage(res, url)); } return res.json() as Promise; } @@ -65,7 +85,7 @@ async function sendJSON(method: string, url: string, body?: unknown): Promise }); if (!res.ok) { notifyAuthRequired(res.status); - throw new Error(`${res.status} ${res.statusText} for ${url}`); + throw new Error(await errorMessage(res, url)); } return res.json() as Promise; } @@ -307,6 +327,32 @@ export const api = { sendSessionKey: (sessionId: string, key: "escape" | "enter" | "accept") => sendJSON<{ ok: boolean }>("POST", `/api/sessions/${sessionId}/keys`, { key }), + // --- queued replies (deliver when the turn actually ends) --- + getQueue: (sessionId: string, signal?: AbortSignal) => + getJSON(`/api/sessions/${sessionId}/queue`, signal), + + queueReply: (sessionId: string, text: string, mode: "turn" | "append" = "turn") => + sendJSON("POST", `/api/sessions/${sessionId}/queue`, { text, mode }), + + // User override: deliver the next queued batch now (skips wait-for-idle). + sendQueuedNow: (sessionId: string) => + sendJSON<{ ok: boolean; sent: number[] }>( + "POST", + `/api/sessions/${sessionId}/queue/send-now`, + {}, + ), + + setQueuedReplyMode: (sessionId: string, qid: number, mode: "turn" | "append") => + sendJSON<{ ok: boolean }>("POST", `/api/sessions/${sessionId}/queue/${qid}/mode`, { mode }), + + cancelQueuedReply: (sessionId: string, qid: number) => + sendJSON<{ ok: boolean }>("DELETE", `/api/sessions/${sessionId}/queue/${qid}`), + + // --- budget runway + hook telemetry --- + getRunway: (signal?: AbortSignal) => getJSON("/api/runway", signal), + + getHooksStatus: () => getJSON("/api/hooks/status"), + getSessionSends: (sessionId: string, limit = 20) => getJSON<{ ts: string; action: string; detail: string }[]>( `/api/sessions/${sessionId}/sends?limit=${limit}`, @@ -318,11 +364,26 @@ export const api = { ), // --- tmux topology (mobile panes view) --- - getTmuxLayout: (signal?: AbortSignal) => - getJSON("/api/tmux/layout", signal), + // previews=false omits per-pane screen text (~250KB across a fleet) — the + // task list polls that shape; the deck uses getPaneScreen for live terminals. + getTmuxLayout: (previews = true, signal?: AbortSignal) => + getJSON(`/api/tmux/layout?previews=${previews ? 1 : 0}`, signal), + + getPaneScreen: (paneId: string, signal?: AbortSignal) => + getJSON( + `/api/tmux/panes/${encodeURIComponent(paneId)}/screen`, + signal, + ), // Pane ids look like "%12" — the leading % is URL-encoding's escape char, so it // MUST be encoded (→ "%2512") or the server decodes it into a different/invalid id. + // Slash commands available to a pane's session, for the composer's "/" menu. + getPaneCommands: (paneId: string, signal?: AbortSignal) => + getJSON( + `/api/tmux/panes/${encodeURIComponent(paneId)}/commands`, + signal, + ), + sendToPane: (paneId: string, text: string, submit = true) => sendJSON<{ ok: boolean }>( "POST", @@ -348,6 +409,71 @@ export const api = { newPaneSession: (cwd?: string) => sendJSON<{ ok: boolean; pane_id: string }>("POST", "/api/tmux/new", { cwd: cwd ?? null }), + // --- launch profiles (~/.muse/profiles.toml) --- + listProfiles: () => getJSON("/api/tmux/profiles"), + + launchProfile: (name: string, values: Record, session?: string | null) => + sendJSON<{ ok: boolean; pane_id: string }>( + "POST", + `/api/tmux/profiles/${encodeURIComponent(name)}/launch`, + { values, session: session ?? null }, + ), + + // --- session restore (rebuild the tmux layout after a reboot) --- + getTmuxSnapshot: () => getJSON("/api/tmux/snapshot"), + + restoreTmuxLayout: (groups: string[]) => + sendJSON<{ ok: boolean; restored: number; skipped: number; groups: string[] }>( + "POST", + "/api/tmux/restore", + { groups }, + ), + + // --- groups (tmux sessions) --- + // Organize panes by moving whole windows between tmux sessions. Window ids look + // like "@12" — @ is encoding-safe, but encode anyway to match the pane-id pattern. + createTmuxSession: (name: string) => + sendJSON<{ ok: boolean; session: string }>("POST", "/api/tmux/sessions", { name }), + + renameTmuxSession: (name: string, next: string) => + sendJSON<{ ok: boolean; session: string }>( + "POST", + `/api/tmux/sessions/${encodeURIComponent(name)}/rename`, + { name: next }, + ), + + deleteTmuxSession: (name: string) => + sendJSON<{ ok: boolean }>("DELETE", `/api/tmux/sessions/${encodeURIComponent(name)}`), + + moveTmuxWindow: (windowId: string, session: string) => + sendJSON<{ ok: boolean; window_id: string; session: string }>( + "POST", + `/api/tmux/windows/${encodeURIComponent(windowId)}/move`, + { session }, + ), + + renameTmuxWindow: (windowId: string, name: string) => + sendJSON<{ ok: boolean; window_id: string; name: string }>( + "POST", + `/api/tmux/windows/${encodeURIComponent(windowId)}/rename`, + { name }, + ), + + // Removing a session: preview the profile cleanup (if any), then close the window. + getWindowCleanup: (windowId: string) => + getJSON<{ + window_name: string; + cwd: string; + cleanup: { profile: string; command: string } | null; + }>(`/api/tmux/windows/${encodeURIComponent(windowId)}/cleanup`), + + closeTmuxWindow: (windowId: string, cleanup: boolean) => + sendJSON<{ ok: boolean; cleanup_ran: boolean; cleanup_ok: boolean; cleanup_output: string }>( + "POST", + `/api/tmux/windows/${encodeURIComponent(windowId)}/close`, + { cleanup }, + ), + // --- web push notifications --- getVapidKey: () => getJSON<{ public_key: string }>("/api/notify/vapid-key"), diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 4e51e2d..09a3d5a 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -774,6 +774,7 @@ export interface TmuxPane { pane_id: string; session_name: string; window_index: number; + window_id: string; // stable window handle (@) — the target for move-window window_name: string; window_active: boolean; pane_index: number; @@ -782,12 +783,23 @@ export interface TmuxPane { cwd: string; title: string; session_attached: boolean; + last_activity: number; muse_session_id: string | null; context_pct: number | null; + queued: number; status: "needs_you" | "responded" | "working" | "idle"; attention: string; mode: "default" | "acceptEdits" | "plan" | "bypass" | null; - preview: string; + preview: string; // full visible screen (ANSI); empty when polled with previews=0 + preview_tail: string; // last visible line — task-list subtitle + options: PendingOption[]; +} + +/** One pane's live visible screen (from /api/tmux/panes/{id}/screen). */ +export interface PaneScreen { + ok: boolean; + text: string; + mode: "default" | "acceptEdits" | "plan" | "bypass" | null; options: PendingOption[]; } @@ -797,6 +809,114 @@ export interface TmuxLayout { reason: string | null; } +export interface SlashCommand { + name: string; // invoked as "/{name}"; subdir commands are namespaced with ":" + description: string; + source: "builtin" | "user" | "project"; +} + +/** A declared input a launch profile prompts for before running (substituted as {key}). */ +export interface ProfileParam { + key: string; + prompt: string; + default: string; +} + +/** A launch profile from ~/.muse/profiles.toml: a template for opening a new window. */ +export interface Profile { + name: string; + cwd: string; + command: string; + params: ProfileParam[]; + builtin: boolean; // the always-present "Claude" default +} + +/** One window in a saved layout snapshot (for session-restore after a reboot). */ +export interface SnapshotWindow { + window_name: string; + cwd: string; + command: string; + kind: "claude" | "shell"; + session_id: string | null; // resumable Claude id, when captured + live: boolean; // already running in the current tmux (skipped on restore) +} + +export interface SnapshotGroup { + name: string; + windows: SnapshotWindow[]; +} + +/** The latest tmux topology snapshot, annotated for the restore UI. */ +export interface LayoutSnapshot { + ts: string | null; // null when nothing captured yet + groups: SnapshotGroup[]; + offer: boolean; // post-reboot signal: has Claude windows, none currently live + restorable_count: number; +} + +/** A message queued for delivery when its session's turn ends. */ +export interface QueuedReply { + id: number; + session_id: string; + text: string; + created_at: string | null; + status: "pending" | "sent" | "cancelled" | "failed"; + /** turn = deliver alone (own turn); append = glue onto the previous item. */ + mode: "turn" | "append"; + sent_at: string | null; + error: string | null; +} + +export interface QueueView { + items: QueuedReply[]; + /** Why the pending queue isn't delivering right now (null = would deliver). */ + hold_reason: string | null; +} + +export interface RunwaySession { + session_id: string; + title: string; + cost_usd: number; +} + +export interface RunwayWindow { + label: string; + window_seconds: number; + anchor: string | null; + anchor_source: string; // "reset" when anchored to an observed usage-limit reset + elapsed_seconds: number; + remaining_seconds: number; + cost_usd: number; + budget_usd: number | null; + budget_source: "configured" | "observed" | "none"; + pct_used: number | null; + pct_elapsed: number; +} + +/** Budget headroom before the plan's 5h/weekly limits bite. */ +export interface RunwayResponse { + generated_at: string; + plan_label: string | null; + five_hour: RunwayWindow; + week: RunwayWindow; + burn_usd_per_hour: number; + projected_exhaust_at: string | null; + exhaust_before_reset: boolean; + top_sessions: RunwaySession[]; +} + +/** Claude Code hook relay install/ingest state. */ +export interface HooksStatus { + script_exists: boolean; + endpoint_file_exists: boolean; + installed_events: string[]; + expected_events: string[]; + events_seen: number; + by_event: Record; + last_event_at: string | null; + recent: { ts: string; event: string; session_id: string; message: string }[]; +} + export interface PendingOptions { session_id: string; source: "permission" | "tool_question" | "none"; From e025d8080a7be3828349fb74805b1384f9c24267 Mon Sep 17 00:00:00 2001 From: Luke Craig Date: Wed, 8 Jul 2026 18:35:04 -0400 Subject: [PATCH 2/9] feat(runway): budget cockpit (5h/weekly headroom + observed ceiling) /api/runway reports spend vs budget in each rate-limit window, trailing burn rate, and projected exhaust time. Subscription plans publish no $ ceiling, so it's learned by observation: a limit-hit records the window's spend at that moment (usage_history.limit_hit), and the recent max becomes the effective cap. Context-window occupancy is inferred per model from the largest prompt seen. TTL-cached over the mtime-cached usage scan so the cockpit can poll it cheaply. --- backend/muse/main.py | 13 +++ backend/muse/routers/insights.py | 10 +- backend/muse/runway.py | 157 ++++++++++++++++++++++++++ backend/muse/usage_cache.py | 43 ++++--- backend/muse/usage_history.py | 51 ++++++++- backend/tests/test_context_pct.py | 73 ++++++++++++ backend/tests/test_limit_ceiling.py | 35 ++++++ backend/tests/test_runway.py | 125 ++++++++++++++++++++ frontend/src/components/RunwayBar.tsx | 88 +++++++++++++++ 9 files changed, 580 insertions(+), 15 deletions(-) create mode 100644 backend/muse/runway.py create mode 100644 backend/tests/test_context_pct.py create mode 100644 backend/tests/test_limit_ceiling.py create mode 100644 backend/tests/test_runway.py create mode 100644 frontend/src/components/RunwayBar.tsx diff --git a/backend/muse/main.py b/backend/muse/main.py index 60d18a0..0c1873d 100644 --- a/backend/muse/main.py +++ b/backend/muse/main.py @@ -66,6 +66,19 @@ async def lifespan(app: FastAPI): app.state.autopilot = AutopilotController() # Parsed usage-limit resets anchor stats' 5h window (observed > estimated). app.state.autopilot.on_reset = app.state.service.usage_history.record_reset + + # Limit-hit sightings calibrate the runway's observed ceiling: the window's + # spend at the moment of the hit is the plan's effective cap (subscription + # plans publish no $ numbers, so the wall is only learnable by hitting it). + def _on_limit(kind: str) -> None: + from . import runway as runway_mod + + rw = runway_mod.compute_runway(app.state.service.usage_history) + cost = rw.five_hour.cost_usd if kind == "5h" else rw.week.cost_usd + if cost > 0: + app.state.service.usage_history.record_limit_hit(kind, cost) + + app.state.autopilot.on_limit = _on_limit # AI idle mode: the controller requests drafts and reads results through # these callables (it never imports the service or touches the AI worker). app.state.autopilot.enqueue_draft = app.state.service.enqueue_draft_reply diff --git a/backend/muse/routers/insights.py b/backend/muse/routers/insights.py index d743e52..756c17a 100644 --- a/backend/muse/routers/insights.py +++ b/backend/muse/routers/insights.py @@ -5,7 +5,8 @@ from fastapi import APIRouter, HTTPException, Query, Request -from ..models import OutcomesResponse, TimelineResponse +from .. import runway as runway_mod +from ..models import OutcomesResponse, RunwayResponse, TimelineResponse router = APIRouter(prefix="/api", tags=["insights"]) @@ -14,6 +15,13 @@ def _service(request: Request): return request.app.state.service +@router.get("/runway", response_model=RunwayResponse) +def runway(request: Request) -> RunwayResponse: + """Budget headroom in the current 5h/weekly windows — cheap enough for the + cockpit to poll (TTL-cached over the mtime-cached usage scan).""" + return runway_mod.get_runway(_service(request).usage_history) + + @router.get("/insights", response_model=OutcomesResponse) def insights(request: Request, days: int = 30) -> OutcomesResponse: if days not in (0, 7, 30, 90): diff --git a/backend/muse/runway.py b/backend/muse/runway.py new file mode 100644 index 0000000..788fd44 --- /dev/null +++ b/backend/muse/runway.py @@ -0,0 +1,157 @@ +"""Budget runway: how much headroom is left in the current 5-hour / weekly +rate-limit windows, who's burning it, and when you'll hit the wall. + +This is the fleet-driving question ("can I keep all these sessions running, or +do I need to pause one?") answered cheaply enough for the phone cockpit to poll: +one pass over the (mtime-cached) usage scan, TTL-cached for _TTL seconds. + +The window math mirrors stats.compute_stats: the 5h window anchors to an +OBSERVED reset when autopilot ever parsed one (reset + k*5h), else falls back to +the trailing 5 hours; the weekly window is trailing 7 days. + +Budgets are honest or absent. Subscription plans (Team/Max) have no published $ +caps, so unlike the stats page this deliberately does NOT use plan.py's rough +tier estimates: a budget is shown only when configured (MUSE_LIMIT_*_USD) or +OBSERVED — the window's spend the last time a limit banner was actually seen +(recorded via usage_history.record_limit_hit). Otherwise the runway reports +spend, burn, and reset time with no made-up ceiling. +""" + +from __future__ import annotations + +import time +from datetime import datetime, timedelta, timezone +from typing import Optional + +from . import discovery +from .config import get_settings +from .models import RunwayResponse, RunwaySession, RunwayWindow +from .plan import detect_plan +from .pricing import cost_usd +from .usage_cache import scan_all + +HOUR5_SECONDS = 5 * 3600 +WEEK_SECONDS = 7 * 86400 +_BURN_LOOKBACK_SECONDS = 30 * 60 # trailing burn-rate sample +_TTL = 10.0 + +_cache: tuple[float, Optional[RunwayResponse]] = (0.0, None) + + +def five_hour_window_start(now: datetime, history) -> tuple[datetime, str]: + """Window start + anchor source ("reset" when derived from an observed + usage-limit reset, else "estimated" trailing-5h).""" + if history is not None: + try: + reset = history.latest_reset() + except Exception: + reset = None + if reset is not None: + k = (now - reset).total_seconds() // HOUR5_SECONDS + start = reset + timedelta(seconds=k * HOUR5_SECONDS) + if start <= now: + return start, "reset" + return now - timedelta(seconds=HOUR5_SECONDS), "estimated" + + +def _resolve_budget(env_value: Optional[float], history, kind: str + ) -> tuple[Optional[float], str]: + """(budget, source): configured env override > observed ceiling > none.""" + if env_value is not None: + return env_value, "configured" + ceiling = None + if history is not None: + try: + ceiling = history.observed_ceiling(kind) + except Exception: + ceiling = None + if ceiling: + return ceiling, "observed" + return None, "none" + + +def compute_runway(history=None, now: Optional[datetime] = None) -> RunwayResponse: + now = now or datetime.now(timezone.utc) + settings = get_settings() + plan = detect_plan(settings.limit_5h_usd, settings.limit_week_usd) + budget_5h, src_5h = _resolve_budget(settings.limit_5h_usd, history, "5h") + budget_week, src_week = _resolve_budget(settings.limit_week_usd, history, "week") + + start_5h, anchor_source = five_hour_window_start(now, history) + end_5h = start_5h + timedelta(seconds=HOUR5_SECONDS) + start_week = now - timedelta(seconds=WEEK_SECONDS) + burn_cut = now - timedelta(seconds=_BURN_LOOKBACK_SECONDS) + + spent_5h = spent_week = burn_sample = 0.0 + per_session: dict[str, float] = {} + for e in scan_all().events: + if e.ts is None or e.total <= 0 or e.ts < start_week: + continue + cost = cost_usd(e.model, e.input, e.output, e.cc, e.cr) + spent_week += cost + if e.ts >= start_5h: + spent_5h += cost + per_session[e.sid] = per_session.get(e.sid, 0.0) + cost + if e.ts >= burn_cut: + burn_sample += cost + + burn_per_hour = burn_sample * (3600 / _BURN_LOOKBACK_SECONDS) + + projected: Optional[datetime] = None + exhaust_before_reset = False + if budget_5h: + remaining_usd = budget_5h - spent_5h + if remaining_usd <= 0: + projected = now + exhaust_before_reset = True + elif burn_per_hour > 0: + projected = now + timedelta(hours=remaining_usd / burn_per_hour) + exhaust_before_reset = projected < end_5h + + titles = {s.session_id: s.title for s in discovery.list_sessions()} + top = sorted(per_session.items(), key=lambda kv: kv[1], reverse=True)[:5] + + def window(label: str, seconds: int, start: datetime, spent: float, + budget: Optional[float], budget_source: str, source: str) -> RunwayWindow: + elapsed = max(0, min(seconds, int((now - start).total_seconds()))) + return RunwayWindow( + label=label, + window_seconds=seconds, + anchor=start, + anchor_source=source, + elapsed_seconds=elapsed, + remaining_seconds=seconds - elapsed, + cost_usd=round(spent, 4), + budget_usd=round(budget, 2) if budget else None, + budget_source=budget_source, + pct_used=round(spent / budget, 4) if budget else None, + pct_elapsed=round(elapsed / seconds, 4), + ) + + return RunwayResponse( + generated_at=now, + plan_label=plan.label if plan else None, + five_hour=window("5-hour window", HOUR5_SECONDS, start_5h, spent_5h, + budget_5h, src_5h, anchor_source), + week=window("Weekly window", WEEK_SECONDS, start_week, spent_week, + budget_week, src_week, "estimated"), + burn_usd_per_hour=round(burn_per_hour, 4), + projected_exhaust_at=projected, + exhaust_before_reset=exhaust_before_reset, + top_sessions=[ + RunwaySession(session_id=sid, title=titles.get(sid, sid[:8]), + cost_usd=round(c, 4)) + for sid, c in top + ], + ) + + +def get_runway(history=None) -> RunwayResponse: + """TTL-cached wrapper — safe for the cockpit's poll cadence.""" + global _cache + ts, cached = _cache + if cached is not None and time.monotonic() - ts < _TTL: + return cached + fresh = compute_runway(history) + _cache = (time.monotonic(), fresh) + return fresh diff --git a/backend/muse/usage_cache.py b/backend/muse/usage_cache.py index 83fe3c8..05c4d4d 100644 --- a/backend/muse/usage_cache.py +++ b/backend/muse/usage_cache.py @@ -143,21 +143,33 @@ def _agent_type(sub_path: Path) -> str: return at +def _window_for_peak(peak: int) -> int: + """Context window inferred from the largest single-request prompt seen for a + model: any prompt over 200k proves the 1M-token window is in use (you can't + send 250k tokens to a 200k model), so that model runs on 1M.""" + return 1_000_000 if peak > 200_000 else 200_000 + + def context_pcts(scan: Scan) -> dict[str, float]: """Latest main-thread context size as a % of the model window, per session. + The window is decided PER MODEL (not per session): the largest prompt any + session sent to a given model tells us that model's window, so a session that + simply hasn't crossed 200k yet is still measured against the right window + instead of reading ~5x too high against an assumed 200k. (Lifted out of AutopilotController so the board ticker shares it.)""" - latest: dict[str, tuple[datetime, int]] = {} - peak: dict[str, int] = {} + latest: dict[str, tuple[datetime, int, str]] = {} + model_peak: dict[str, int] = {} for e in scan.events: if e.is_subagent or e.ts is None or e.context <= 0: continue cur = latest.get(e.sid) if cur is None or e.ts > cur[0]: - latest[e.sid] = (e.ts, e.context) - peak[e.sid] = max(peak.get(e.sid, 0), e.context) + latest[e.sid] = (e.ts, e.context, e.model or "") + m = e.model or "" + model_peak[m] = max(model_peak.get(m, 0), e.context) out = {} - for sid, (_ts, ctx) in latest.items(): - window = 1_000_000 if peak[sid] > 200_000 else 200_000 + for sid, (_ts, ctx, model) in latest.items(): + window = _window_for_peak(model_peak.get(model, 0)) out[sid] = 100.0 * ctx / window return out @@ -186,15 +198,16 @@ def _file_aggregate(path: Path, events: list[Event]) -> tuple: cost = 0.0 latest_ts = None latest_ctx = 0 + latest_model = "" peak_ctx = 0 for e in events: tokens += e.input + e.output + e.cc # real work (excl. cache reads) cost += cost_usd(e.model, e.input, e.output, e.cc, e.cr) if not e.is_subagent and e.ts is not None and e.context > 0: if latest_ts is None or e.ts > latest_ts: - latest_ts, latest_ctx = e.ts, e.context + latest_ts, latest_ctx, latest_model = e.ts, e.context, e.model or "" peak_ctx = max(peak_ctx, e.context) - agg = (mtime, sid, tokens, cost, latest_ts, latest_ctx, peak_ctx) + agg = (mtime, sid, tokens, cost, latest_ts, latest_ctx, peak_ctx, latest_model) _file_agg_cache[path] = agg return agg @@ -204,7 +217,7 @@ def board_rollup() -> tuple[dict[str, tuple[int, float]], dict[str, float]]: O(#files) per call thanks to the per-file mtime-keyed aggregate cache.""" projects = get_settings().projects_dir aggs: dict[str, tuple[int, float]] = {} - latest: dict[str, tuple] = {} # sid -> (ts, ctx) + latest: dict[str, tuple] = {} # sid -> (ts, ctx, model) peak: dict[str, int] = {} if not projects.is_dir(): return aggs, {} @@ -216,7 +229,7 @@ def feed(path: Path, sid: str, project: str, is_sub: bool, atype: str, aid: str if a[4] is not None: cur = latest.get(sid) if cur is None or a[4] > cur[0]: - latest[sid] = (a[4], a[5]) + latest[sid] = (a[4], a[5], a[7]) peak[sid] = max(peak.get(sid, 0), a[6]) for project_dir in projects.iterdir(): @@ -228,10 +241,14 @@ def feed(path: Path, sid: str, project: str, is_sub: bool, atype: str, aid: str feed(sub, sub.parent.parent.name, project_dir.name, True, _agent_type(sub), sub.stem) + # Window per MODEL: the biggest prompt any session sent to a model reveals + # that model's window, so low-usage sessions aren't measured against 200k. + model_peak: dict[str, int] = {} + for sid, (_ts, _ctx, model) in latest.items(): + model_peak[model] = max(model_peak.get(model, 0), peak.get(sid, 0)) pcts = {} - for sid, (_ts, ctx) in latest.items(): - window = 1_000_000 if peak.get(sid, 0) > 200_000 else 200_000 - pcts[sid] = 100.0 * ctx / window + for sid, (_ts, ctx, model) in latest.items(): + pcts[sid] = 100.0 * ctx / _window_for_peak(model_peak.get(model, 0)) return aggs, pcts diff --git a/backend/muse/usage_history.py b/backend/muse/usage_history.py index 28f2c04..842084a 100644 --- a/backend/muse/usage_history.py +++ b/backend/muse/usage_history.py @@ -15,7 +15,7 @@ import sqlite3 import threading -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path from typing import Callable, Iterable, Optional @@ -28,6 +28,11 @@ resets_at TEXT PRIMARY KEY, -- UTC ISO; observed 5h-window reset boundary observed_at TEXT ); +CREATE TABLE IF NOT EXISTS limit_hit ( + observed_at TEXT PRIMARY KEY, -- local ISO; when a usage-limit banner was seen + kind TEXT NOT NULL DEFAULT '5h', -- '5h' | 'week' + window_cost REAL NOT NULL -- spend in that window at the moment of the hit +); CREATE TABLE IF NOT EXISTS usage_daily ( day TEXT NOT NULL, -- local YYYY-MM-DD model TEXT NOT NULL, @@ -147,6 +152,50 @@ def latest_reset(self) -> Optional[datetime]: except ValueError: return None + # --- observed limit ceilings ------------------------------------------------- + # Subscription plans have no published $ caps, so the honest budget is the + # spend level at which the wall was actually hit. Each banner sighting records + # the window's spend; the recent MAX is the effective ceiling. + + def record_limit_hit(self, kind: str, window_cost: float, + dedupe_minutes: int = 30) -> bool: + """Record one observed limit hit. Banners persist across ticks, so hits of + the same kind within `dedupe_minutes` are one sighting. Returns True if + recorded.""" + now = datetime.now().astimezone() + with self._lock: + r = self._conn.execute( + "SELECT MAX(observed_at) AS m FROM limit_hit WHERE kind=?", (kind,) + ).fetchone() + if r and r["m"]: + try: + last = datetime.fromisoformat(r["m"]) + if (now - last).total_seconds() < dedupe_minutes * 60: + return False + except ValueError: + pass + + def work(): + self._conn.execute( + "INSERT OR IGNORE INTO limit_hit(observed_at, kind, window_cost) " + "VALUES(?,?,?)", + (now.isoformat(), kind, float(window_cost)), + ) + self._write(work) + return True + + def observed_ceiling(self, kind: str = "5h", days: int = 30) -> Optional[float]: + """Best estimate of the window's real cap: the highest spend that ever hit + the wall recently (the cap is at least that high).""" + cutoff = (datetime.now().astimezone() - timedelta(days=days)).isoformat() + with self._lock: + r = self._conn.execute( + "SELECT MAX(window_cost) AS m FROM limit_hit " + "WHERE kind=? AND observed_at >= ?", + (kind, cutoff), + ).fetchone() + return float(r["m"]) if r and r["m"] is not None else None + def day_count(self) -> int: with self._lock: r = self._conn.execute("SELECT COUNT(DISTINCT day) AS n FROM usage_daily").fetchone() diff --git a/backend/tests/test_context_pct.py b/backend/tests/test_context_pct.py new file mode 100644 index 0000000..7c3a32f --- /dev/null +++ b/backend/tests/test_context_pct.py @@ -0,0 +1,73 @@ +"""Context-window occupancy %. The window is inferred PER MODEL from the largest +prompt seen for that model, so two sessions on the same model share one window — +a low-usage session must not read ~5x too high against an assumed 200k.""" + +from datetime import datetime, timezone + +from muse.usage_cache import Event, Scan, _window_for_peak, context_pcts + + +def _ev(sid, ctx, model, secs): + # context = input + cc + cr; put it all in cache_read for simplicity. + return Event( + sid=sid, + project_dir="p", + ts=datetime(2026, 7, 4, 0, 0, secs, tzinfo=timezone.utc), + input=0, + output=10, + cc=0, + cr=ctx, + model=model, + is_subagent=False, + agent_type="", + ) + + +def test_window_for_peak_buckets_at_200k(): + assert _window_for_peak(150_000) == 200_000 + assert _window_for_peak(200_000) == 200_000 + assert _window_for_peak(200_001) == 1_000_000 + + +def test_same_model_shares_a_window(): + # One opus session crossed 200k (proving a 1M window); another opus session + # is only at 100k. Both must be measured against 1M — the small one ~10%, + # NOT 50% against a wrongly-assumed 200k. + scan = Scan( + events=[ + _ev("big", 300_000, "opus", 1), + _ev("small", 100_000, "opus", 2), + ], + sessions=2, + sessions_by_project={}, + ) + pct = context_pcts(scan) + assert round(pct["big"]) == 30 + assert round(pct["small"]) == 10 + + +def test_distinct_models_get_distinct_windows(): + # A model that never exceeded 200k keeps the 200k window. + scan = Scan( + events=[ + _ev("a", 300_000, "opus", 1), # opus → 1M + _ev("b", 100_000, "haiku", 2), # haiku stayed small → 200k + ], + sessions=2, + sessions_by_project={}, + ) + pct = context_pcts(scan) + assert round(pct["a"]) == 30 # 300k / 1M + assert round(pct["b"]) == 50 # 100k / 200k + + +def test_latest_event_wins_per_session(): + scan = Scan( + events=[ + _ev("s", 50_000, "haiku", 1), + _ev("s", 120_000, "haiku", 5), # later → this is the current context + ], + sessions=1, + sessions_by_project={}, + ) + assert round(context_pcts(scan)["s"]) == 60 # 120k / 200k diff --git a/backend/tests/test_limit_ceiling.py b/backend/tests/test_limit_ceiling.py new file mode 100644 index 0000000..0cbc906 --- /dev/null +++ b/backend/tests/test_limit_ceiling.py @@ -0,0 +1,35 @@ +"""Observed limit ceilings: record/dedupe/read, and the controller's on_limit hook.""" + +import pytest + +from muse.usage_history import UsageHistoryStore + + +@pytest.fixture +def store(tmp_path): + s = UsageHistoryStore(tmp_path / "h.db") + yield s + s.close() + + +def test_record_and_ceiling(store): + assert store.observed_ceiling("5h") is None + assert store.record_limit_hit("5h", 62.5) is True + assert store.observed_ceiling("5h") == 62.5 + assert store.observed_ceiling("week") is None # kinds are independent + + +def test_dedupe_within_window(store): + assert store.record_limit_hit("5h", 60.0) is True + # The banner persists across ticks — same sighting, not a new data point. + assert store.record_limit_hit("5h", 61.0, dedupe_minutes=30) is False + assert store.observed_ceiling("5h") == 60.0 + # A different kind is a separate sighting. + assert store.record_limit_hit("week", 700.0) is True + + +def test_ceiling_is_recent_max(store): + store.record_limit_hit("5h", 55.0, dedupe_minutes=0) + store.record_limit_hit("5h", 71.0, dedupe_minutes=0) + store.record_limit_hit("5h", 63.0, dedupe_minutes=0) + assert store.observed_ceiling("5h") == 71.0 diff --git a/backend/tests/test_runway.py b/backend/tests/test_runway.py new file mode 100644 index 0000000..c9b8725 --- /dev/null +++ b/backend/tests/test_runway.py @@ -0,0 +1,125 @@ +"""Runway computation: window anchoring, spend sums, burn projection, and the +honest budget policy (configured > observed ceiling > none — never estimates).""" + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from muse import runway +from muse.usage_cache import Event, Scan + +NOW = datetime(2026, 7, 1, 12, 0, tzinfo=timezone.utc) + + +def _event(sid, minutes_ago, output=100_000): + return Event( + sid=sid, project_dir="-p", ts=NOW - timedelta(minutes=minutes_ago), + input=1000, output=output, cc=0, cr=0, + model="claude-opus-4-8", is_subagent=False, agent_type="", + ) + + +class FakeHistory: + def __init__(self, reset=None, ceilings=None): + self._reset = reset + self._ceilings = ceilings or {} + + def latest_reset(self): + return self._reset + + def observed_ceiling(self, kind="5h", days=30): + return self._ceilings.get(kind) + + +def _settings(limit_5h=None, limit_week=None): + return SimpleNamespace(limit_5h_usd=limit_5h, limit_week_usd=limit_week) + + +@pytest.fixture(autouse=True) +def env(monkeypatch, tmp_path): + events = [ + _event("hot", 10), # in 5h window + burn sample + _event("hot", 20), + _event("warm", 60), # in 5h window only + _event("old", 60 * 26), # weekly only + _event("ancient", 60 * 24 * 8), # outside every window + ] + monkeypatch.setattr(runway, "scan_all", lambda: Scan(events, 5, {"-p": 5})) + monkeypatch.setattr(runway.discovery, "list_sessions", lambda: []) + monkeypatch.setattr(runway, "get_settings", _settings) + monkeypatch.setattr( + runway, "detect_plan", + lambda a, b: type("P", (), {"label": "Claude Team · Max 5×"})(), + ) + runway._cache = (0.0, None) + + +def test_window_sums_and_top_sessions(): + r = runway.compute_runway(history=None, now=NOW) + per_event = r.five_hour.cost_usd / 3 # 3 events in the 5h window, equal cost + assert per_event > 0 + assert r.week.cost_usd == pytest.approx(per_event * 4) + assert [s.session_id for s in r.top_sessions] == ["hot", "warm"] + assert r.five_hour.anchor_source == "estimated" + assert r.plan_label == "Claude Team · Max 5×" + + +def test_subscription_without_calibration_has_no_budget(): + # No env override, no observed limit hit → no made-up ceiling, no projection. + r = runway.compute_runway(history=FakeHistory(), now=NOW) + for w in (r.five_hour, r.week): + assert w.budget_usd is None and w.pct_used is None and w.budget_source == "none" + assert r.projected_exhaust_at is None and r.exhaust_before_reset is False + assert r.burn_usd_per_hour > 0 # burn is always reported + + +def test_observed_ceiling_becomes_budget(): + hist = FakeHistory(ceilings={"5h": 80.0, "week": 500.0}) + r = runway.compute_runway(history=hist, now=NOW) + assert r.five_hour.budget_usd == 80.0 and r.five_hour.budget_source == "observed" + assert r.week.budget_usd == 500.0 and r.week.budget_source == "observed" + assert r.five_hour.pct_used == pytest.approx(r.five_hour.cost_usd / 80.0, abs=1e-3) + assert r.projected_exhaust_at is not None + + +def test_configured_env_beats_observed(monkeypatch): + monkeypatch.setattr(runway, "get_settings", lambda: _settings(limit_5h=42.0)) + r = runway.compute_runway(history=FakeHistory(ceilings={"5h": 80.0}), now=NOW) + assert r.five_hour.budget_usd == 42.0 and r.five_hour.budget_source == "configured" + + +def test_observed_reset_anchors_window(): + # Reset observed 7h ago → current window began 2h ago (reset + 1*5h). + r = runway.compute_runway(history=FakeHistory(reset=NOW - timedelta(hours=7)), now=NOW) + assert r.five_hour.anchor_source == "reset" + assert r.five_hour.anchor == NOW - timedelta(hours=2) + assert r.five_hour.elapsed_seconds == 2 * 3600 + assert r.five_hour.cost_usd > 0 + + +def test_burn_projection_with_budget(monkeypatch): + monkeypatch.setattr(runway, "get_settings", lambda: _settings(limit_5h=150.0)) + r = runway.compute_runway(history=None, now=NOW) + # 2 events in the last 30 min → burn = 2*cost/0.5h + assert r.burn_usd_per_hour == pytest.approx((r.five_hour.cost_usd / 3) * 4, rel=1e-6) + assert r.projected_exhaust_at is not None + # cheap events: exhaustion lands far beyond this window + assert r.exhaust_before_reset is False + + +def test_exhausted_budget_projects_now(monkeypatch): + monkeypatch.setattr(runway, "get_settings", lambda: _settings(limit_5h=0.01)) + r = runway.compute_runway(history=None, now=NOW) + assert r.exhaust_before_reset is True + assert r.projected_exhaust_at == NOW + + +def test_ttl_cache(monkeypatch): + calls = [] + orig = runway.compute_runway + monkeypatch.setattr(runway, "compute_runway", + lambda history=None, now=None: calls.append(1) or orig(history, now)) + runway.get_runway() + runway.get_runway() + assert len(calls) == 1 diff --git a/frontend/src/components/RunwayBar.tsx b/frontend/src/components/RunwayBar.tsx new file mode 100644 index 0000000..d319f49 --- /dev/null +++ b/frontend/src/components/RunwayBar.tsx @@ -0,0 +1,88 @@ +import { useCallback, useState } from "react"; +import { Link } from "react-router-dom"; +import { api } from "../api/client"; +import type { RunwayResponse, RunwayWindow } from "../api/types"; +import { usePolling } from "../hooks/usePolling"; + +function fmtDur(secs: number): string { + const h = Math.floor(secs / 3600); + const m = Math.round((secs % 3600) / 60); + return h > 0 ? `${h}h${String(m).padStart(2, "0")}` : `${m}m`; +} + +function level(w: RunwayWindow, exhaust: boolean): "ok" | "warn" | "over" { + if (w.pct_used != null && w.pct_used >= 1) return "over"; + // Ahead of pace or projected to hit the wall before the reset → warn. + if (exhaust || (w.pct_used != null && w.pct_used > w.pct_elapsed + 0.15)) return "warn"; + return "ok"; +} + +const SOURCE_HINT: Record = { + observed: "ceiling calibrated from your last observed limit hit", + configured: "ceiling from MUSE_LIMIT_*_USD", + none: "no $ cap on a subscription plan — muse learns the ceiling when a limit banner appears", +}; + +function Seg({ w, exhaust, name }: { w: RunwayWindow; exhaust: boolean; name: string }) { + const title = `${w.label} — ${SOURCE_HINT[w.budget_source] ?? ""}`; + // Subscription plans have no published $ caps: without a configured/observed + // ceiling there is no bar to fill — just report spend honestly. + if (w.budget_usd == null) { + return ( +
+ {name} + ${w.cost_usd.toFixed(0)} + ↻{fmtDur(w.remaining_seconds)} +
+ ); + } + const lvl = level(w, exhaust); + const pct = Math.min(1, w.pct_used ?? 0); + return ( +
+ {name} + + + {/* time-progress notch: spend left of the notch = under pace */} + + + + ${w.cost_usd.toFixed(0)}/{w.budget_source === "observed" ? "~" : ""}$ + {w.budget_usd.toFixed(0)} + + ↻{fmtDur(w.remaining_seconds)} +
+ ); +} + +/** One-line budget cockpit: spend vs the plan's 5h + weekly windows, with a + * pace notch and a projected time-to-limit warning. Polls every 30s (the server + * side is TTL-cached over the mtime-cached usage scan). */ +export default function RunwayBar() { + const [runway, setRunway] = useState(null); + const refresh = useCallback(async () => setRunway(await api.getRunway()), []); + usePolling(refresh, 30000); + + if (!runway) return null; + const exhaust = runway.exhaust_before_reset; + const eta = + exhaust && runway.projected_exhaust_at + ? Math.max(0, (new Date(runway.projected_exhaust_at).getTime() - Date.now()) / 1000) + : null; + + return ( + + + + {runway.burn_usd_per_hour > 0 && ( + + {exhaust && eta != null + ? eta <= 0 + ? "⚠ limit hit" + : `⚠ limit in ~${fmtDur(eta)}` + : `$${runway.burn_usd_per_hour.toFixed(0)}/h`} + + )} + + ); +} From 6271ec5aea1f15ac8e6086ec1e83a53b5d98dd4d Mon Sep 17 00:00:00 2001 From: Luke Craig Date: Wed, 8 Jul 2026 18:35:19 -0400 Subject: [PATCH 3/9] feat(hooks): Claude Code hook relay for instant session events A relay script posts Stop/Notification/etc. hook events straight to muse, so session-state transitions surface immediately instead of waiting for the polling tick. `muse hooks install` writes the relay + wires settings.json; the server refreshes the endpoint file on start (port/token can change) and keeps a recent-events ring. Hook-driven alerts suppress the tick's duplicate for a short window so the same transition doesn't fire twice. --- backend/muse/alerts.py | 103 ++++++++++++- backend/muse/cli.py | 24 +++ backend/muse/hooksetup.py | 228 +++++++++++++++++++++++++++++ backend/muse/main.py | 11 +- backend/muse/routers/hooks.py | 108 ++++++++++++++ backend/tests/test_alerts_hooks.py | 96 ++++++++++++ backend/tests/test_hooks_api.py | 100 +++++++++++++ backend/tests/test_hooksetup.py | 52 +++++++ 8 files changed, 719 insertions(+), 3 deletions(-) create mode 100644 backend/muse/hooksetup.py create mode 100644 backend/muse/routers/hooks.py create mode 100644 backend/tests/test_alerts_hooks.py create mode 100644 backend/tests/test_hooks_api.py create mode 100644 backend/tests/test_hooksetup.py diff --git a/backend/muse/alerts.py b/backend/muse/alerts.py index 6a8b4e2..2d30133 100644 --- a/backend/muse/alerts.py +++ b/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.""" diff --git a/backend/muse/cli.py b/backend/muse/cli.py index d8cd166..edad8f6 100644 --- a/backend/muse/cli.py +++ b/backend/muse/cli.py @@ -170,12 +170,36 @@ def _status() -> int: return 0 +def _hooks(action: str) -> int: + from . import hooksetup + + if action == "install": + print(hooksetup.install()) + return 0 + if action == "uninstall": + print(hooksetup.uninstall()) + return 0 + st = hooksetup.status() + missing = [e for e in st["expected_events"] if e not in st["installed_events"]] + print(f"relay script: {'ok' if st['script_exists'] else 'MISSING'} ({hooksetup.script_path()})") + print(f"endpoint file: {'ok' if st['endpoint_file_exists'] else 'MISSING (written at muse start)'}") + print(f"settings.json: {', '.join(st['installed_events']) or 'no muse hooks'}" + + (f" (missing: {', '.join(missing)})" if missing else "")) + return 0 if (st["script_exists"] and not missing) else 1 + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="muse", description="Manage the muse server.") sub = parser.add_subparsers(dest="cmd", required=True) for name in ("start", "stop", "restart", "status"): sub.add_parser(name) + hooks = sub.add_parser( + "hooks", help="install/uninstall the Claude Code hook relay (instant session events)" + ) + hooks.add_argument("action", choices=["install", "uninstall", "status"]) args = parser.parse_args(argv) + if args.cmd == "hooks": + return _hooks(args.action) return {"start": _start, "stop": _stop, "restart": _restart, "status": _status}[args.cmd]() diff --git a/backend/muse/hooksetup.py b/backend/muse/hooksetup.py new file mode 100644 index 0000000..b95ac7c --- /dev/null +++ b/backend/muse/hooksetup.py @@ -0,0 +1,228 @@ +"""Claude Code hook integration: install/uninstall the relay, and the endpoint +file the relay reads. + +Claude Code can run a command on lifecycle events (Stop, Notification, …) and +pipe the event JSON to its stdin. We install ONE tiny relay script that POSTs +that payload to muse's /api/hooks/claude — giving muse instant, authoritative +"turn ended / needs input" signals instead of 2.5–15s polling heuristics. + +Design constraints: +- The relay must NEVER block or break Claude Code: 3s curl timeout, always exit 0. +- No secrets in ~/.claude/settings.json: the script reads ~/.muse/hook_endpoint.json + (0600), which the SERVER rewrites at every startup with its live port + token — + so token rotation or a port change never needs a reinstall. +- settings.json is merged conservatively: only our entries are added/removed; + everything else (rtk hooks, statusline, …) is preserved byte-for-byte in spirit. +""" + +from __future__ import annotations + +import json +import os +import stat +import time +from pathlib import Path +from typing import Optional + +from .config import get_settings + +# Events muse cares about. Stop = turn ended (deliver queued reply / alert); +# Notification = needs permission or sat idle; UserPromptSubmit = user replied +# (clears alert state instantly); SessionEnd = process exited. +HOOK_EVENTS = ["Stop", "Notification", "UserPromptSubmit", "SessionEnd"] + +_SCRIPT = """#!/bin/sh +# muse hook relay (installed by `muse hooks install`). +# POSTs the Claude Code hook payload (stdin) to the local muse server. +# Always exits 0 — a muse outage must never block Claude Code. +f="$HOME/.muse/hook_endpoint.json" +[ -r "$f" ] || exit 0 +url=$(sed -n 's/.*"url": *"\\([^"]*\\)".*/\\1/p' "$f") +tok=$(sed -n 's/.*"token": *"\\([^"]*\\)".*/\\1/p' "$f") +[ -n "$url" ] || exit 0 +if [ -n "$tok" ]; then + curl -s -m 3 -X POST "$url/api/hooks/claude" -H "Content-Type: application/json" \\ + -H "Authorization: Bearer $tok" --data-binary @- >/dev/null 2>&1 +else + curl -s -m 3 -X POST "$url/api/hooks/claude" -H "Content-Type: application/json" \\ + --data-binary @- >/dev/null 2>&1 +fi +exit 0 +""" + + +def state_dir() -> Path: + return get_settings().db_path.parent + + +def script_path() -> Path: + return state_dir() / "hook.sh" + + +def endpoint_path() -> Path: + return state_dir() / "hook_endpoint.json" + + +def claude_settings_path() -> Path: + return get_settings().claude_dir / "settings.json" + + +def write_endpoint_file() -> None: + """(Re)write ~/.muse/hook_endpoint.json with the server's live URL + token. + Called at every server startup so the relay always has current credentials.""" + s = get_settings() + path = endpoint_path() + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "url": f"http://127.0.0.1:{s.port}", + "token": s.resolve_auth_token() or "", + } + path.write_text(json.dumps(payload, indent=1) + "\n", encoding="utf-8") + path.chmod(0o600) + + +def write_script() -> Path: + path = script_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(_SCRIPT, encoding="utf-8") + path.chmod(path.stat().st_mode | stat.S_IXUSR) + return path + + +# --- settings.json merge (pure, unit-testable) -------------------------------- + + +def _is_ours(hook: dict, command: str) -> bool: + return hook.get("type") == "command" and hook.get("command") == command + + +def add_muse_hooks(settings: dict, command: str) -> dict: + """Return a copy of `settings` with the muse relay registered for each of + HOOK_EVENTS. Idempotent; never touches other hooks.""" + out = json.loads(json.dumps(settings)) # deep copy, JSON-safe by construction + hooks = out.setdefault("hooks", {}) + for event in HOOK_EVENTS: + matchers = hooks.setdefault(event, []) + present = any( + _is_ours(h, command) + for m in matchers + if isinstance(m, dict) + for h in m.get("hooks", []) + if isinstance(h, dict) + ) + if not present: + matchers.append({"hooks": [{"type": "command", "command": command}]}) + return out + + +def remove_muse_hooks(settings: dict, command: Optional[str] = None) -> dict: + """Return a copy of `settings` with every muse relay entry stripped (matched + by exact command, or by the hook.sh basename when command is None). Empty + matcher groups / event lists left behind are pruned.""" + marker = command or str(script_path()) + out = json.loads(json.dumps(settings)) + hooks = out.get("hooks") + if not isinstance(hooks, dict): + return out + for event in list(hooks): + matchers = hooks[event] + if not isinstance(matchers, list): + continue + for m in matchers: + if isinstance(m, dict) and isinstance(m.get("hooks"), list): + m["hooks"] = [ + h for h in m["hooks"] + if not (isinstance(h, dict) and h.get("type") == "command" + and marker in str(h.get("command", ""))) + ] + hooks[event] = [ + m for m in matchers + if not (isinstance(m, dict) and m.get("hooks") == [] and set(m) <= {"hooks", "matcher"}) + ] + if hooks[event] == []: + del hooks[event] + if hooks == {}: + del out["hooks"] + return out + + +# --- install / uninstall / status (used by the CLI) --------------------------- + + +def _read_settings(path: Path) -> dict: + if not path.is_file(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +def _write_settings(path: Path, settings: dict) -> None: + # Keep a timestamped backup under ~/.muse (never clobber the user's config + # without a way back). + if path.is_file(): + backups = state_dir() / "backups" + backups.mkdir(parents=True, exist_ok=True) + (backups / f"settings.json.{int(time.time())}").write_text( + path.read_text(encoding="utf-8"), encoding="utf-8" + ) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.muse-tmp") + tmp.write_text(json.dumps(settings, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, path) + + +def install() -> str: + script = write_script() + path = claude_settings_path() + settings = _read_settings(path) + merged = add_muse_hooks(settings, str(script)) + if merged != settings: + _write_settings(path, merged) + if not endpoint_path().is_file(): + # The server rewrites this at startup; seed it now so hooks work + # immediately if muse is already running on the default port. + try: + write_endpoint_file() + except OSError: + pass + return ( + f"Installed muse hooks ({', '.join(HOOK_EVENTS)}) → {script}\n" + f"Relay target: {endpoint_path()} (rewritten at every muse start).\n" + f"New Claude Code sessions pick this up automatically; running ones on next launch." + ) + + +def uninstall() -> str: + path = claude_settings_path() + settings = _read_settings(path) + stripped = remove_muse_hooks(settings) + if stripped != settings: + _write_settings(path, stripped) + try: + script_path().unlink() + except OSError: + pass + return "Removed muse hooks from Claude Code settings." + + +def status() -> dict: + """Local install state (no server round-trip).""" + script = script_path() + settings = _read_settings(claude_settings_path()) if claude_settings_path().is_file() else {} + installed_events = [ + event + for event, matchers in (settings.get("hooks") or {}).items() + if isinstance(matchers, list) + and any( + str(script) in str(h.get("command", "")) + for m in matchers + if isinstance(m, dict) + for h in m.get("hooks", []) + if isinstance(h, dict) + ) + ] + return { + "script_exists": script.is_file(), + "endpoint_file_exists": endpoint_path().is_file(), + "installed_events": sorted(installed_events), + "expected_events": HOOK_EVENTS, + } diff --git a/backend/muse/main.py b/backend/muse/main.py index 0c1873d..02252bb 100644 --- a/backend/muse/main.py +++ b/backend/muse/main.py @@ -16,7 +16,7 @@ from fastapi.responses import FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles -from . import db, lifecycle +from . import db, hooksetup, lifecycle from .auth import AuthMiddleware from .compression import GzipBufferedMiddleware from .config import get_settings @@ -87,6 +87,15 @@ def _on_limit(kind: str) -> None: app.state.autopilot.start() app.state.alerts = AlertsWatcher(app.state.service) app.state.alerts.start() + # Claude Code hook ingestion: refresh the relay's endpoint file (port+token + # can change between runs) and prime the recent-events ring. + try: + hooksetup.write_endpoint_file() + except OSError: + pass + from collections import deque as _deque + + app.state.hook_events = _deque(maxlen=200) # AI worker: single daemon thread executing headless `claude -p` jobs. get_settings().ai_workdir.mkdir(parents=True, exist_ok=True) app.state.service.ai_worker.start() diff --git a/backend/muse/routers/hooks.py b/backend/muse/routers/hooks.py new file mode 100644 index 0000000..57bd34a --- /dev/null +++ b/backend/muse/routers/hooks.py @@ -0,0 +1,108 @@ +"""Ingest Claude Code hook events (the relay installed by `muse hooks install`). + +Claude Code pipes each lifecycle event's JSON to ~/.muse/hook.sh, which POSTs it +here. This turns muse's picture of the fleet from "polled every few seconds" +into "told the instant something happens": + +- Stop → deliver the next queued reply for that session immediately; + if nothing was queued, push a (settled) "ready for you" alert. +- Notification → push "needs your permission / input" immediately. +- UserPromptSubmit→ the user replied — cancel any in-flight turn-ended alert. +- SessionEnd → push "session ended" (when that alert rule is on). + +The handler itself must return FAST (Claude Code waits on the relay, which waits +on curl): all real work runs as a background task on the event loop. The polling +watcher stays as the fallback for sessions without hooks; suppression windows in +AlertsWatcher keep the two sources from double-firing. +""" + +from __future__ import annotations + +import asyncio +import re +from collections import deque +from datetime import datetime, timezone + +from fastapi import APIRouter, Request + +from .. import hooksetup + +router = APIRouter(prefix="/api", tags=["hooks"]) + +_SID_RE = re.compile(r"^[0-9a-fA-F-]{8,64}$") +# Small pause after Stop before typing a queued reply: the TUI needs a beat to +# repaint its prompt after the turn ends. +_QUEUE_DELIVER_DELAY = 1.0 + + +def _ring(request: Request) -> deque: + ring = getattr(request.app.state, "hook_events", None) + if ring is None: + ring = request.app.state.hook_events = deque(maxlen=200) + return ring + + +async def _dispatch(state, event: str, sid: str, payload: dict) -> None: + alerts = state.alerts + if event == "Stop": + await asyncio.sleep(_QUEUE_DELIVER_DELAY) + delivered = await asyncio.to_thread(state.autopilot.deliver_queued, sid) + if delivered: + # The queued reply is the user's answer — no "ready for you" push, + # and any in-flight settle alert is superseded. + alerts.hook_user_replied(sid) + else: + await alerts.hook_turn_ended(sid) + elif event == "Notification": + await alerts.hook_needs_you(sid, str(payload.get("message") or "")) + elif event == "UserPromptSubmit": + alerts.hook_user_replied(sid) + elif event == "SessionEnd": + await alerts.hook_session_end(sid) + # Other events (SubagentStop, PreCompact, …) are recorded but need no action. + + +@router.post("/hooks/claude") +async def claude_hook(request: Request) -> dict: + """Accept one Claude Code hook payload. Never errors on bad input — the + relay must always succeed from Claude Code's point of view.""" + try: + payload = await request.json() + except Exception: + return {"ok": False, "detail": "unparseable payload"} + if not isinstance(payload, dict): + return {"ok": False, "detail": "not an object"} + event = str(payload.get("hook_event_name") or "") + sid = str(payload.get("session_id") or "") + _ring(request).appendleft( + { + "ts": datetime.now(timezone.utc).isoformat(), + "event": event, + "session_id": sid, + "message": str(payload.get("message") or "")[:200], + } + ) + if event and _SID_RE.match(sid): + # Fire-and-forget: the relay's curl has a 3s timeout and Claude Code is + # waiting on it — the settle delays alone exceed that. + asyncio.get_running_loop().create_task( + _dispatch(request.app.state, event, sid, payload) + ) + return {"ok": True} + + +@router.get("/hooks/status") +def hooks_status(request: Request) -> dict: + """Install state + live ingest counters (the AlertsPage 'instant telemetry' + card and `muse hooks status` both read this).""" + ring = _ring(request) + by_event: dict[str, int] = {} + for e in ring: + by_event[e["event"]] = by_event.get(e["event"], 0) + 1 + return { + **hooksetup.status(), + "events_seen": len(ring), + "by_event": by_event, + "last_event_at": ring[0]["ts"] if ring else None, + "recent": list(ring)[:20], + } diff --git a/backend/tests/test_alerts_hooks.py b/backend/tests/test_alerts_hooks.py new file mode 100644 index 0000000..8221239 --- /dev/null +++ b/backend/tests/test_alerts_hooks.py @@ -0,0 +1,96 @@ +"""Hook-driven alert paths on AlertsWatcher: instant fire, settle-and-recheck, +reply cancellation, and tick suppression.""" + +import asyncio +from datetime import datetime, timezone + +import pytest + +from muse import alerts +from muse.models import AlertRules, LiveSession, NotifyConfig, NotifyResult, SessionSummary + + +class FakeService: + def __init__(self): + self.sent = [] + self.rules = AlertRules() + self.cfg = NotifyConfig(enabled=True, topic="t") + + def get_alert_rules(self): + return self.rules + + def get_notify_config(self): + return self.cfg + + def list_sessions(self): + return [SessionSummary(session_id="s", project_dir="-p", title="My sess", + mtime=datetime.now(timezone.utc), state="waiting")] + + def send_notification(self, message, **kw): + self.sent.append((message, kw)) + return NotifyResult(ok=True, detail="HTTP 200") + + +@pytest.fixture +def watcher(monkeypatch): + monkeypatch.setattr(alerts, "_TURN_END_SETTLE_SECONDS", 0.0) + monkeypatch.setattr( + alerts.live_discovery, "discover", + lambda: [LiveSession(session_id="s", pid=1, status="idle", pane_id="%1")], + ) + return alerts.AlertsWatcher(FakeService()) + + +def test_needs_you_fires_once_per_pause(watcher): + asyncio.run(watcher.hook_needs_you("s", "Claude needs your permission to use Bash")) + asyncio.run(watcher.hook_needs_you("s", "Claude needs your permission to use Bash")) + assert len(watcher.service.sent) == 1 + title, kw = watcher.service.sent[0] + assert "needs you" in kw["title"] and "/drive/s" in kw["click"] + # the polling tick is suppressed for the same transition + assert watcher._suppressed("s", "waiting") is True + + +def test_turn_ended_fires_when_still_idle(watcher): + asyncio.run(watcher.hook_turn_ended("s")) + assert len(watcher.service.sent) == 1 + assert "ready for you" in watcher.service.sent[0][1]["title"] + + +def test_turn_ended_cancelled_by_user_reply(watcher, monkeypatch): + monkeypatch.setattr(alerts, "_TURN_END_SETTLE_SECONDS", 0.05) + + async def scenario(): + t = asyncio.create_task(watcher.hook_turn_ended("s")) + await asyncio.sleep(0.01) + watcher.hook_user_replied("s") # user answered during the settle window + await t + + asyncio.run(scenario()) + assert watcher.service.sent == [] + + +def test_turn_ended_skipped_when_session_moved_on(watcher, monkeypatch): + monkeypatch.setattr( + alerts.live_discovery, "discover", + lambda: [LiveSession(session_id="s", pid=1, status="busy", pane_id="%1")], + ) + asyncio.run(watcher.hook_turn_ended("s")) + assert watcher.service.sent == [] + + +def test_session_end_respects_rule(watcher): + asyncio.run(watcher.hook_session_end("s")) # on_stopped defaults False + assert watcher.service.sent == [] + watcher.service.rules = AlertRules(on_stopped=True) + asyncio.run(watcher.hook_session_end("s")) + assert len(watcher.service.sent) == 1 + asyncio.run(watcher.hook_session_end("s")) # deduped + assert len(watcher.service.sent) == 1 + + +def test_rules_off_means_silent(watcher): + watcher.service.rules = AlertRules(on_waiting=False) + asyncio.run(watcher.hook_needs_you("s", "m")) + asyncio.run(watcher.hook_turn_ended("s")) + assert watcher.service.sent == [] diff --git a/backend/tests/test_hooks_api.py b/backend/tests/test_hooks_api.py new file mode 100644 index 0000000..fdc223e --- /dev/null +++ b/backend/tests/test_hooks_api.py @@ -0,0 +1,100 @@ +"""Hook ingestion endpoint + event dispatch (fast: all delays zeroed).""" + +import asyncio +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from muse.routers import hooks as hooks_router + + +class FakeAlerts: + def __init__(self): + self.calls = [] + + def hook_user_replied(self, sid): + self.calls.append(("user_replied", sid)) + + async def hook_needs_you(self, sid, message): + self.calls.append(("needs_you", sid, message)) + + async def hook_turn_ended(self, sid): + self.calls.append(("turn_ended", sid)) + + async def hook_session_end(self, sid): + self.calls.append(("session_end", sid)) + + +def _state(delivered): + return SimpleNamespace( + alerts=FakeAlerts(), + autopilot=SimpleNamespace(deliver_queued=lambda sid: delivered), + hook_events=None, + ) + + +@pytest.fixture(autouse=True) +def fast(monkeypatch): + monkeypatch.setattr(hooks_router, "_QUEUE_DELIVER_DELAY", 0.0) + + +def _dispatch(state, event, sid="a" * 36, payload=None): + asyncio.run(hooks_router._dispatch(state, event, sid, payload or {})) + + +def test_stop_delivers_queue_and_skips_alert(): + state = _state(delivered=[7]) + _dispatch(state, "Stop") + assert state.alerts.calls == [("user_replied", "a" * 36)] + + +def test_stop_with_empty_queue_alerts(): + state = _state(delivered=[]) + _dispatch(state, "Stop") + assert state.alerts.calls == [("turn_ended", "a" * 36)] + + +def test_notification_and_prompt_and_end(): + state = _state(delivered=[]) + _dispatch(state, "Notification", payload={"message": "needs permission for Bash"}) + _dispatch(state, "UserPromptSubmit") + _dispatch(state, "SessionEnd") + _dispatch(state, "SubagentStop") # recorded upstream, no action + assert state.alerts.calls == [ + ("needs_you", "a" * 36, "needs permission for Bash"), + ("user_replied", "a" * 36), + ("session_end", "a" * 36), + ] + + +def test_endpoint_records_and_never_errors(monkeypatch): + app = FastAPI() + app.include_router(hooks_router.router) + dispatched = [] + + async def fake_dispatch(state, event, sid, payload): + dispatched.append((event, sid)) + + monkeypatch.setattr(hooks_router, "_dispatch", fake_dispatch) + app.state.alerts = FakeAlerts() + app.state.autopilot = SimpleNamespace(deliver_queued=lambda sid: []) + c = TestClient(app) + + sid = "0f0e0d0c-1111-2222-3333-444455556666" + r = c.post("/api/hooks/claude", json={"hook_event_name": "Stop", "session_id": sid}) + assert r.status_code == 200 and r.json()["ok"] is True + + # malformed payloads are acknowledged, not errored (the relay must never break) + r = c.post("/api/hooks/claude", content=b"not json", + headers={"Content-Type": "application/json"}) + assert r.status_code == 200 and r.json()["ok"] is False + r = c.post("/api/hooks/claude", json={"session_id": "../../etc/passwd", + "hook_event_name": "Stop"}) + assert r.status_code == 200 # bad sid: recorded but not dispatched + + st = c.get("/api/hooks/status").json() + assert st["events_seen"] == 2 # the two parseable payloads + assert st["last_event_at"] is not None + assert dispatched == [("Stop", sid)] diff --git a/backend/tests/test_hooksetup.py b/backend/tests/test_hooksetup.py new file mode 100644 index 0000000..b65e312 --- /dev/null +++ b/backend/tests/test_hooksetup.py @@ -0,0 +1,52 @@ +"""Hook installer: conservative settings.json merge, idempotence, uninstall.""" + +from muse import hooksetup + +CMD = "/home/u/.muse/hook.sh" + +RTK = {"matcher": "Bash", "hooks": [{"type": "command", "command": "rtk hook claude"}]} + + +def test_add_registers_all_events_and_preserves_existing(): + settings = { + "model": "claude-fable-5", + "hooks": {"PreToolUse": [RTK], "Stop": [{"hooks": [{"type": "command", "command": "other"}]}]}, + } + out = hooksetup.add_muse_hooks(settings, CMD) + # untouched keys and foreign hooks survive + assert out["model"] == "claude-fable-5" + assert out["hooks"]["PreToolUse"] == [RTK] + assert out["hooks"]["Stop"][0]["hooks"][0]["command"] == "other" + # ours added for every event + for event in hooksetup.HOOK_EVENTS: + cmds = [ + h["command"] + for m in out["hooks"][event] + for h in m.get("hooks", []) + ] + assert CMD in cmds + # input not mutated + assert "Notification" not in settings["hooks"] + + +def test_add_is_idempotent(): + once = hooksetup.add_muse_hooks({}, CMD) + twice = hooksetup.add_muse_hooks(once, CMD) + assert once == twice + + +def test_remove_strips_only_ours(): + merged = hooksetup.add_muse_hooks( + {"hooks": {"PreToolUse": [RTK]}, "theme": "dark"}, CMD + ) + out = hooksetup.remove_muse_hooks(merged, CMD) + assert out["hooks"] == {"PreToolUse": [RTK]} + assert out["theme"] == "dark" + # removing from a settings dict with no hooks is a no-op + assert hooksetup.remove_muse_hooks({"a": 1}, CMD) == {"a": 1} + + +def test_remove_then_add_roundtrip(): + base = {"hooks": {"SessionEnd": [{"hooks": [{"type": "command", "command": "notify.cjs"}]}]}} + merged = hooksetup.add_muse_hooks(base, CMD) + assert hooksetup.remove_muse_hooks(merged, CMD) == base From 46d10e549fad6b7ab58290bf8ea838c39374248b Mon Sep 17 00:00:00 2001 From: Luke Craig Date: Wed, 8 Jul 2026 18:37:14 -0400 Subject: [PATCH 4/9] feat(autopilot): end-of-turn reply queue, layout snapshots, limit observer Three controller-loop behaviors that share the autopilot store & tick: - Reply queue: messages queue and deliver only when a session is genuinely idle (turn ended, no pending menu), one batch per cooldown; /api/queue + QueueChips/ReplyBox drive it, hold_reason explains a held reply. - Layout snapshots: the controller captures tmux topology on change (pruned history ring) so a fleet can be restored after a reboot. - Limit observer: when a rate-limit banner appears, _observe_limit anchors the 5h window (on_reset) and reports the hit (on_limit) so the runway can calibrate its observed ceiling. --- backend/muse/autopilot/controller.py | 176 ++++++++++- backend/muse/autopilot/snapshot.py | 178 ++++++++++++ backend/muse/autopilot/store.py | 173 ++++++++++- backend/muse/main.py | 4 + backend/muse/routers/queue.py | 86 ++++++ backend/tests/test_reply_queue.py | 258 ++++++++++++++++ backend/tests/test_snapshot.py | 211 ++++++++++++++ .../src/components/board/QueueChips.test.tsx | 28 ++ frontend/src/components/board/QueueChips.tsx | 110 +++++++ .../src/components/board/ReplyBox.test.tsx | 113 +++++++ frontend/src/components/board/ReplyBox.tsx | 275 ++++++++++++++++-- 11 files changed, 1584 insertions(+), 28 deletions(-) create mode 100644 backend/muse/autopilot/snapshot.py create mode 100644 backend/muse/routers/queue.py create mode 100644 backend/tests/test_reply_queue.py create mode 100644 backend/tests/test_snapshot.py create mode 100644 frontend/src/components/board/QueueChips.test.tsx create mode 100644 frontend/src/components/board/QueueChips.tsx create mode 100644 frontend/src/components/board/ReplyBox.test.tsx diff --git a/backend/muse/autopilot/controller.py b/backend/muse/autopilot/controller.py index a950e1b..d7996d1 100644 --- a/backend/muse/autopilot/controller.py +++ b/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/ diff --git a/backend/muse/autopilot/snapshot.py b/backend/muse/autopilot/snapshot.py new file mode 100644 index 0000000..0d7e5b4 --- /dev/null +++ b/backend/muse/autopilot/snapshot.py @@ -0,0 +1,178 @@ +"""Snapshot the tmux topology so it can be rebuilt after a reboot ("session restore"). + +tmux (sessions/windows/panes) is wiped on reboot, but Claude Code conversations persist on +disk (~/.claude/projects//.jsonl) and resume with +`claude --resume `. We periodically capture the structure — which Claude sessions exist, +how they're grouped, their cwds/names — and on demand recreate the windows, resuming each +Claude where it left off. Window granularity only (v1 doesn't restore intra-window splits). +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shlex +from datetime import datetime, timezone + +from . import sessions as live_discovery +from . import tmux + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _rep_pane(panes: list[dict], pane_sid: dict[str, str]) -> dict: + """The pane that best represents a window: the Claude pane (one with a resolved + session id) if any, else the active pane, else the lowest-index pane.""" + return ( + next((p for p in panes if pane_sid.get(p["pane_id"])), None) + or next((p for p in panes if p["pane_active"]), None) + or min(panes, key=lambda p: p["pane_index"]) + ) + + +def build_snapshot() -> dict | None: + """Capture the current topology, or None when there's nothing worth keeping (tmux + down, or no Claude windows) — so we never clobber a good last-known-good with an + empty capture around a reboot.""" + if not tmux.available(): + return None + panes = tmux.list_layout() + if not panes: + return None + pane_sid = {s.pane_id: s.session_id for s in live_discovery.discover() if s.pane_id} + + # session_name -> window_index -> [panes], preserving tmux's natural order. + grouped: dict[str, dict[int, list[dict]]] = {} + for p in panes: + grouped.setdefault(p["session_name"], {}).setdefault(p["window_index"], []).append(p) + + out_groups: list[dict] = [] + claude_windows = 0 + for sname, windows in grouped.items(): + wins: list[dict] = [] + for widx in sorted(windows): + rep = _rep_pane(windows[widx], pane_sid) + sid = pane_sid.get(rep["pane_id"]) + is_claude = sid is not None or rep["command"] == "claude" + if is_claude: + claude_windows += 1 + wins.append( + { + "window_name": rep["window_name"], + "cwd": rep["cwd"], + "command": rep["command"], + "kind": "claude" if is_claude else "shell", + "session_id": sid, + } + ) + out_groups.append({"name": sname, "windows": wins}) + + if claude_windows == 0: + return None + return {"ts": _now(), "groups": out_groups} + + +def topology_sig(snapshot: dict) -> str: + """Stable hash of the structure (NOT the timestamp) for cheap change detection.""" + parts = [ + f"{g['name']}|{w['window_name']}|{w['cwd']}|{w.get('session_id') or ''}|{w['kind']}" + for g in snapshot["groups"] + for w in g["windows"] + ] + return hashlib.sha1("\n".join(parts).encode("utf-8")).hexdigest() + + +def _live_index() -> tuple[set[str], set[tuple[str, str, str]], set[str]]: + """(live Claude session ids, live (session,window,cwd) triples, live session names) + from the current tmux + Claude discovery — used to skip windows already running.""" + live_sids = {s.session_id for s in live_discovery.discover() if s.session_id} + layout = tmux.list_layout() + live_wins = {(p["session_name"], p["window_name"], p["cwd"]) for p in layout} + session_names = {p["session_name"] for p in layout} + return live_sids, live_wins, session_names + + +def _is_live(gname: str, w: dict, live_sids: set[str], live_wins: set[tuple[str, str, str]]) -> bool: + """Whether a snapshot window is already running. Claude windows key on their session + id (strong); shell windows fall back to (group, name, cwd).""" + sid = w.get("session_id") + if sid: + return sid in live_sids + return (gname, w["window_name"], w["cwd"]) in live_wins + + +def annotate_liveness(snapshot: dict) -> dict: + """Return the snapshot with a per-window `live` flag plus `offer`/`restorable_count`. + `offer` is the clean-reboot signal that drives the UI banner: the snapshot has ≥2 + Claude windows and none of them are currently running.""" + live_sids, live_wins, _ = _live_index() + groups_out: list[dict] = [] + claude_total = claude_live = restorable = 0 + for g in snapshot["groups"]: + wins = [] + for w in g["windows"]: + live = _is_live(g["name"], w, live_sids, live_wins) + if w["kind"] == "claude": + claude_total += 1 + claude_live += 1 if live else 0 + if not live: + restorable += 1 + wins.append({**w, "live": live}) + groups_out.append({"name": g["name"], "windows": wins}) + return { + "ts": snapshot["ts"], + "groups": groups_out, + "offer": claude_total >= 2 and claude_live == 0, + "restorable_count": restorable, + } + + +def _command_for(w: dict) -> str: + """The shell command to launch a restored window. Claude windows resume the exact + session (falling back to the latest conversation in the cwd if that fails); a Claude + window whose id we couldn't capture continues the latest; shells open bare.""" + if w["kind"] != "claude": + return "" # empty → tmux opens the default shell + sid = w.get("session_id") + if sid: + return f"claude --resume {shlex.quote(sid)} || claude --continue" + return "claude --continue" + + +def restore(snapshot: dict, group_names: list[str], only_missing: bool = True) -> dict: + """Rebuild the selected groups from a snapshot. Creates each group's tmux session if + absent and (re)creates its windows in order, skipping any already running so partial + or repeated restores never duplicate. Returns {restored, skipped, groups}.""" + live_sids, live_wins, session_names = _live_index() + wanted = set(group_names) + restored = skipped = 0 + touched: list[str] = [] + + for g in snapshot["groups"]: + if g["name"] not in wanted: + continue + to_make = [ + w + for w in g["windows"] + if not (only_missing and _is_live(g["name"], w, live_sids, live_wins)) + ] + skipped += len(g["windows"]) - len(to_make) + if not to_make: + continue + touched.append(g["name"]) + session_exists = g["name"] in session_names + for w in to_make: + cwd = w["cwd"] if w["cwd"] and os.path.isdir(w["cwd"]) else os.path.expanduser("~") + cmd = _command_for(w) + if not session_exists: + ok, _ = tmux.new_session( + g["name"], cwd=cwd, command=cmd or None, window_name=w["window_name"] + ) + session_exists = True + else: + ok, _ = tmux.new_window(cwd, cmd, session=g["name"], name=w["window_name"]) + restored += 1 if ok else 0 + return {"restored": restored, "skipped": skipped, "groups": touched} diff --git a/backend/muse/autopilot/store.py b/backend/muse/autopilot/store.py index 78ec1c9..6138d41 100644 --- a/backend/muse/autopilot/store.py +++ b/backend/muse/autopilot/store.py @@ -12,7 +12,7 @@ from typing import Optional from .. import db -from ..models import AutopilotConfig, AutopilotLogEntry +from ..models import AutopilotConfig, AutopilotLogEntry, QueuedReply _SCHEMA = """ CREATE TABLE IF NOT EXISTS autopilot_config ( @@ -35,6 +35,24 @@ CREATE TABLE IF NOT EXISTS autopilot_log ( ts TEXT, session_id TEXT, action TEXT, detail TEXT ); +CREATE TABLE IF NOT EXISTS reply_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + text TEXT NOT NULL, + created_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', -- pending | sent | cancelled | failed + mode TEXT NOT NULL DEFAULT 'turn', -- turn (own turn) | append (glue to previous) + sent_at TEXT, + error TEXT +); +CREATE INDEX IF NOT EXISTS reply_queue_pending + ON reply_queue(session_id) WHERE status = 'pending'; +CREATE TABLE IF NOT EXISTS tmux_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, -- when captured (ISO) + sig TEXT NOT NULL, -- topology signature (change detection) + snapshot_json TEXT NOT NULL -- serialized {ts, groups:[...]} +); """ # Columns added after the initial release (migrate existing DBs). @@ -49,6 +67,7 @@ # requested against (Phase B discards the draft if the session moved on). "ALTER TABLE autopilot_config ADD COLUMN ai_pending_job_id TEXT", "ALTER TABLE autopilot_config ADD COLUMN ai_requested_updated_at TEXT", + "ALTER TABLE reply_queue ADD COLUMN mode TEXT NOT NULL DEFAULT 'turn'", ] @@ -112,6 +131,33 @@ def _do() -> None: db.retry_locked(_do) + # --- tmux layout snapshots (session restore) --------------------------- + def save_snapshot(self, snapshot_json: str, sig: str) -> None: + """Append a topology snapshot and prune to the last ~20 rows (a small history + ring so a good last-known-good survives even if a newer capture is thin).""" + + def _do() -> None: + with self._lock: + self._conn.execute( + "INSERT INTO tmux_snapshots(ts, sig, snapshot_json) VALUES(?,?,?)", + (_now(), sig, snapshot_json), + ) + self._conn.execute( + "DELETE FROM tmux_snapshots WHERE id NOT IN " + "(SELECT id FROM tmux_snapshots ORDER BY id DESC LIMIT 20)" + ) + self._conn.commit() + + db.retry_locked(_do) + + def latest_snapshot(self) -> Optional[tuple[str, str]]: + """The most recent (sig, snapshot_json), or None if nothing captured yet.""" + with self._lock: + row = self._conn.execute( + "SELECT sig, snapshot_json FROM tmux_snapshots ORDER BY id DESC LIMIT 1" + ).fetchone() + return (row["sig"], row["snapshot_json"]) if row else None + # --- active-hours schedule --------------------------------------------- def get_schedule(self) -> tuple[bool, int, int]: enabled = self._get_kv("sched_enabled") == "1" @@ -243,6 +289,131 @@ def get_ai_pending(self, sid: str) -> tuple[Optional[str], Optional[datetime]]: return None, None return row["ai_pending_job_id"], _dt(row["ai_requested_updated_at"]) + # --- reply queue ---------------------------------------------------------- + # User-authored messages waiting for their session's turn to end. The + # controller delivers them (it owns automated tmux writes); these methods are + # just FIFO bookkeeping. Delivered/cancelled rows are kept briefly as history. + + def queue_add(self, sid: str, text: str, mode: str = "turn") -> QueuedReply: + def _do() -> int: + with self._lock: + cur = self._conn.execute( + "INSERT INTO reply_queue(session_id, text, created_at, mode) " + "VALUES(?,?,?,?)", + (sid, text, _now(), mode), + ) + self._conn.commit() + return int(cur.lastrowid) + + qid = db.retry_locked(_do) + return QueuedReply(id=qid, session_id=sid, text=text, mode=mode, + created_at=datetime.now(timezone.utc)) + + def queue_for(self, sid: str, limit: int = 20) -> list[QueuedReply]: + """This session's queue: all pending items (FIFO) + a few recent outcomes.""" + with self._lock: + rows = self._conn.execute( + "SELECT * FROM (SELECT * FROM reply_queue WHERE session_id=? " + "ORDER BY (status='pending') DESC, id DESC LIMIT ?) ORDER BY id ASC", + (sid, limit), + ).fetchall() + return [self._row_to_queued(r) for r in rows] + + def queue_next(self, sid: str) -> Optional[QueuedReply]: + with self._lock: + row = self._conn.execute( + "SELECT * FROM reply_queue WHERE session_id=? AND status='pending' " + "ORDER BY id ASC LIMIT 1", + (sid,), + ).fetchone() + return self._row_to_queued(row) if row else None + + def queue_next_batch(self, sid: str) -> list[QueuedReply]: + """What one delivery should type: the oldest pending item plus any items + immediately after it whose mode is 'append' (they glue on rather than + waiting for their own turn).""" + with self._lock: + rows = self._conn.execute( + "SELECT * FROM reply_queue WHERE session_id=? AND status='pending' " + "ORDER BY id ASC", + (sid,), + ).fetchall() + batch: list[QueuedReply] = [] + for r in rows: + item = self._row_to_queued(r) + if batch and item.mode != "append": + break + batch.append(item) + return batch + + def queue_set_mode(self, sid: str, qid: int, mode: str) -> bool: + """Flip a still-pending item between 'turn' and 'append'.""" + def _do() -> bool: + with self._lock: + cur = self._conn.execute( + "UPDATE reply_queue SET mode=? " + "WHERE id=? AND session_id=? AND status='pending'", + (mode, qid, sid), + ) + self._conn.commit() + return cur.rowcount > 0 + + return db.retry_locked(_do) + + def queue_counts(self) -> dict[str, int]: + """Pending count per session — one cheap query for board/panes badges.""" + with self._lock: + rows = self._conn.execute( + "SELECT session_id, COUNT(*) AS n FROM reply_queue " + "WHERE status='pending' GROUP BY session_id" + ).fetchall() + return {r["session_id"]: r["n"] for r in rows} + + def queue_cancel(self, sid: str, qid: int) -> bool: + """Cancel one still-pending item. False if it's gone or already delivered.""" + def _do() -> bool: + with self._lock: + cur = self._conn.execute( + "UPDATE reply_queue SET status='cancelled' " + "WHERE id=? AND session_id=? AND status='pending'", + (qid, sid), + ) + self._conn.commit() + return cur.rowcount > 0 + + return db.retry_locked(_do) + + def queue_mark(self, qid: int, status: str, error: Optional[str] = None) -> None: + def _do() -> None: + with self._lock: + self._conn.execute( + "UPDATE reply_queue SET status=?, sent_at=?, error=? WHERE id=?", + (status, _now() if status == "sent" else None, error, qid), + ) + # Bound history: keep the most recent 200 non-pending rows. + self._conn.execute( + "DELETE FROM reply_queue WHERE status != 'pending' AND id NOT IN " + "(SELECT id FROM reply_queue WHERE status != 'pending' " + "ORDER BY id DESC LIMIT 200)" + ) + self._conn.commit() + + db.retry_locked(_do) + + @staticmethod + def _row_to_queued(r: sqlite3.Row) -> QueuedReply: + keys = r.keys() + return QueuedReply( + id=r["id"], + session_id=r["session_id"], + text=r["text"], + created_at=_dt(r["created_at"]), + status=r["status"], + mode=(r["mode"] if "mode" in keys else "turn") or "turn", + sent_at=_dt(r["sent_at"]), + error=r["error"], + ) + # --- log ---------------------------------------------------------------- def log(self, sid: str, action: str, detail: str = "") -> None: def _do() -> None: diff --git a/backend/muse/main.py b/backend/muse/main.py index 02252bb..8ee3779 100644 --- a/backend/muse/main.py +++ b/backend/muse/main.py @@ -29,12 +29,14 @@ auth, autopilot, board, + hooks, insights, interact, investigations, launch, notify, options, + queue, sessions, stream, tmux, @@ -167,6 +169,8 @@ def create_app() -> FastAPI: app.include_router(insights.router) app.include_router(interact.router) app.include_router(options.router) + app.include_router(queue.router) + app.include_router(hooks.router) app.include_router(tmux.router) app.include_router(auth.router) diff --git a/backend/muse/routers/queue.py b/backend/muse/routers/queue.py new file mode 100644 index 0000000..e10fd1d --- /dev/null +++ b/backend/muse/routers/queue.py @@ -0,0 +1,86 @@ +"""Queued replies: leave a message for a BUSY session; muse types it into the +session's pane when the turn actually ends (idle, no menu, no rate-limit banner). + +This is the "send when ready" counterpart to interact.py's immediate send. The +delivery itself lives in the autopilot controller (the one disciplined automated +tmux-write site) — these endpoints are just the user-facing queue bookkeeping. +""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel + +from ..models import QueuedReply, QueueView + +router = APIRouter(prefix="/api", tags=["queue"]) + +_MAX_TEXT = 4000 + + +class QueueRequest(BaseModel): + text: str + mode: str = "turn" # turn = own turn; append = glue onto the previous item + + +class QueueModeRequest(BaseModel): + mode: str # turn | append + + +@router.get("/sessions/{session_id}/queue") +def get_queue(session_id: str, request: Request) -> QueueView: + ap = request.app.state.autopilot + return QueueView( + items=ap.store.queue_for(session_id), + hold_reason=ap.queue_hold_reason(session_id), + ) + + +@router.post("/sessions/{session_id}/queue") +def add_to_queue(session_id: str, body: QueueRequest, request: Request) -> QueuedReply: + text = body.text.strip() + if not text: + raise HTTPException(status_code=400, detail="text is empty") + if len(text) > _MAX_TEXT: + raise HTTPException(status_code=400, detail=f"text too long (max {_MAX_TEXT})") + if body.mode not in ("turn", "append"): + raise HTTPException(status_code=400, detail=f"unknown mode: {body.mode!r}") + ap = request.app.state.autopilot + item = ap.store.queue_add(session_id, text, mode=body.mode) + ap.store.log(session_id, "queue_add", f"#{item.id} {text[:80]}") + return item + + +@router.post("/sessions/{session_id}/queue/{qid}/mode") +def set_queued_mode( + session_id: str, qid: int, body: QueueModeRequest, request: Request +) -> dict: + """Flip a pending item between 'turn' (its own turn) and 'append' (glue onto + the previous item so both go in one message).""" + if body.mode not in ("turn", "append"): + raise HTTPException(status_code=400, detail=f"unknown mode: {body.mode!r}") + ap = request.app.state.autopilot + if not ap.store.queue_set_mode(session_id, qid, body.mode): + raise HTTPException(status_code=404, detail="not pending (already sent or cancelled)") + return {"ok": True, "mode": body.mode} + + +@router.post("/sessions/{session_id}/queue/send-now") +def send_queued_now(session_id: str, request: Request) -> dict: + """User override: deliver the next queued batch immediately, skipping the + wait-for-idle gate. Still refuses to type over an open menu / limit banner.""" + ap = request.app.state.autopilot + sent, reason = ap.deliver_now(session_id) + if not sent: + raise HTTPException(status_code=409, detail=reason or "nothing to send") + ap.store.log(session_id, "queue_send_now", f"{sent}") + return {"ok": True, "sent": sent} + + +@router.delete("/sessions/{session_id}/queue/{qid}") +def cancel_queued(session_id: str, qid: int, request: Request) -> dict: + ap = request.app.state.autopilot + if not ap.store.queue_cancel(session_id, qid): + raise HTTPException(status_code=404, detail="not pending (already sent or cancelled)") + ap.store.log(session_id, "queue_cancel", f"#{qid}") + return {"ok": True} diff --git a/backend/tests/test_reply_queue.py b/backend/tests/test_reply_queue.py new file mode 100644 index 0000000..4be5088 --- /dev/null +++ b/backend/tests/test_reply_queue.py @@ -0,0 +1,258 @@ +"""Queued replies: store FIFO bookkeeping, guarded controller delivery, and the +REST endpoints.""" + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from muse.autopilot import controller as ctl_mod +from muse.autopilot.controller import AutopilotController +from muse.autopilot.store import AutopilotStore +from muse.models import LiveSession +from muse.routers import queue as queue_router + + +@pytest.fixture +def store(tmp_path): + s = AutopilotStore(tmp_path / "test.db") + yield s + s.close() + + +# --- store --------------------------------------------------------------------- + + +def test_queue_fifo_and_counts(store): + a = store.queue_add("s1", "first") + b = store.queue_add("s1", "second") + store.queue_add("s2", "other") + assert store.queue_counts() == {"s1": 2, "s2": 1} + assert store.queue_next("s1").id == a.id + store.queue_mark(a.id, "sent") + assert store.queue_next("s1").id == b.id + assert store.queue_counts()["s1"] == 1 + + +def test_queue_cancel_only_pending(store): + item = store.queue_add("s1", "x") + assert store.queue_cancel("s1", item.id) is True + assert store.queue_cancel("s1", item.id) is False # already cancelled + sent = store.queue_add("s1", "y") + store.queue_mark(sent.id, "sent") + assert store.queue_cancel("s1", sent.id) is False # already delivered + assert store.queue_next("s1") is None + + +def test_queue_for_lists_pending_and_history(store): + a = store.queue_add("s1", "done") + store.queue_mark(a.id, "sent") + b = store.queue_add("s1", "waiting") + rows = store.queue_for("s1") + assert [(r.id, r.status) for r in rows] == [(a.id, "sent"), (b.id, "pending")] + + +def test_queue_batch_and_mode(store): + a = store.queue_add("s1", "first") + b = store.queue_add("s1", "and also this", mode="append") + c = store.queue_add("s1", "separate turn") + # first + its trailing appends batch together; the 'turn' item waits. + assert [i.id for i in store.queue_next_batch("s1")] == [a.id, b.id] + # flip c to append → it joins the batch + assert store.queue_set_mode("s1", c.id, "append") is True + assert [i.id for i in store.queue_next_batch("s1")] == [a.id, b.id, c.id] + # a sent item can't change mode + store.queue_mark(a.id, "sent") + assert store.queue_set_mode("s1", a.id, "append") is False + # after 'a' is gone, the batch restarts at 'b' (append glues to nothing → leads) + assert [i.id for i in store.queue_next_batch("s1")] == [b.id, c.id] + + +# --- controller delivery --------------------------------------------------------- + + +@pytest.fixture +def controller(tmp_path, monkeypatch): + monkeypatch.setattr( + ctl_mod, "get_settings", lambda: type("S", (), {"db_path": tmp_path / "t.db", + "ai_daily_budget_usd": 0.0})() + ) + c = AutopilotController() + c._sent: list[tuple] = [] + monkeypatch.setattr( + ctl_mod.tmux, "send_text", + lambda p, t, submit=True: (c._sent.append((p, t)) or (True, "")), + ) + yield c + c.store.close() + + +def _live(monkeypatch, status="idle", waiting_for=None, pane="%1"): + ls = LiveSession(session_id="s1", pid=1, status=status, + waiting_for=waiting_for, pane_id=pane) + monkeypatch.setattr(ctl_mod.live_discovery, "discover", lambda: [ls]) + + +def test_deliver_when_idle(controller, monkeypatch): + _live(monkeypatch) + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "❯ plain prompt") + item = controller.store.queue_add("s1", "go on") + assert controller.deliver_queued() == [item.id] + assert controller._sent == [("%1", "go on")] + assert controller.store.queue_next("s1") is None + row = controller.store.queue_for("s1")[0] + assert row.status == "sent" and row.sent_at is not None + + +def test_no_delivery_while_busy_or_waiting(controller, monkeypatch): + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "") + controller.store.queue_add("s1", "later") + _live(monkeypatch, status="busy") + assert controller.deliver_queued() == [] + _live(monkeypatch, status="idle", waiting_for="permission") + assert controller.deliver_queued() == [] + _live(monkeypatch, status="idle", pane=None) + assert controller.deliver_queued() == [] + assert controller._sent == [] + + +def test_no_delivery_over_visible_menu(controller, monkeypatch): + _live(monkeypatch) + menu = ( + "Do you want to proceed?\n" + "❯ 1. Yes\n" + " 2. Yes, and don't ask again\n" + " 3. No\n" + ) + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: menu) + controller.store.queue_add("s1", "2 looks fine") # digits would SELECT — must not send + assert controller.deliver_queued() == [] + assert controller._sent == [] + + +def test_no_delivery_into_rate_limit_banner(controller, monkeypatch): + _live(monkeypatch) + monkeypatch.setattr( + ctl_mod.tmux, "capture_pane", + lambda p, n: "5-hour limit reached ∙ resets 3pm", + ) + resets = [] + hits = [] + controller.on_reset = resets.append + controller.on_limit = hits.append + controller.store.queue_add("s1", "keep going") + assert controller.deliver_queued() == [] + assert controller._sent == [] + assert len(resets) == 1 # the observed reset still anchors stats + assert hits == ["5h"] # …and the sighting calibrates the observed ceiling + + +def test_append_items_deliver_as_one_paste(controller, monkeypatch): + _live(monkeypatch) + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "") + pasted = [] + monkeypatch.setattr( + ctl_mod.tmux, "paste_text", + lambda p, t, submit=True: (pasted.append((p, t)) or (True, "")), + ) + a = controller.store.queue_add("s1", "run the tests") + b = controller.store.queue_add("s1", "then update the changelog", mode="append") + assert controller.deliver_queued() == [a.id, b.id] + # multiline combined → bracketed paste, not send-keys (raw LF would submit early) + assert pasted == [("%1", "run the tests\nthen update the changelog")] + assert controller._sent == [] + assert controller.store.queue_next("s1") is None + + +def test_cooldown_blocks_double_fire(controller, monkeypatch): + _live(monkeypatch) + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "") + controller.store.queue_add("s1", "one") + controller.store.queue_add("s1", "two") + assert len(controller.deliver_queued()) == 1 + assert len(controller.deliver_queued()) == 0 # within cooldown + controller._queue_sent_at["s1"] = 0.0 # cooldown elapsed + assert len(controller.deliver_queued()) == 1 + assert [t for _, t in controller._sent] == ["one", "two"] + + +def test_hold_reason_explains_why_not_delivering(controller, monkeypatch): + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "❯ plain") + controller.store.queue_add("s1", "later") + _live(monkeypatch, status="busy") + assert controller.queue_hold_reason("s1") == "session is busy, not idle" + _live(monkeypatch, status="shell") + assert controller.queue_hold_reason("s1") == "session is shell, not idle" + _live(monkeypatch, status="idle", waiting_for="permission") + assert controller.queue_hold_reason("s1") == "waiting for permission" + _live(monkeypatch, status="idle", pane=None) + assert controller.queue_hold_reason("s1") == "session isn’t running in tmux" + _live(monkeypatch, status="idle") + assert controller.queue_hold_reason("s1") is None # would deliver + + +def test_hold_reason_none_when_queue_empty(controller, monkeypatch): + _live(monkeypatch, status="busy") + assert controller.queue_hold_reason("s1") is None + + +def test_deliver_now_overrides_idle_gate(controller, monkeypatch): + _live(monkeypatch, status="shell") # would never auto-deliver + monkeypatch.setattr(ctl_mod.tmux, "capture_pane", lambda p, n: "❯ plain") + item = controller.store.queue_add("s1", "go anyway") + assert controller.deliver_queued() == [] # auto path holds + sent, reason = controller.deliver_now("s1") + assert sent == [item.id] and reason is None + assert controller._sent == [("%1", "go anyway")] + + +def test_deliver_now_still_refuses_a_menu(controller, monkeypatch): + _live(monkeypatch, status="shell") + monkeypatch.setattr( + ctl_mod.tmux, "capture_pane", lambda p, n: "Proceed?\n❯ 1. Yes\n 2. No\n" + ) + controller.store.queue_add("s1", "1") + sent, reason = controller.deliver_now("s1") + assert sent == [] and reason and "menu" in reason + assert controller._sent == [] + + +def test_deliver_now_nothing_queued(controller, monkeypatch): + _live(monkeypatch) + assert controller.deliver_now("s1") == ([], "nothing queued") + + +# --- REST ------------------------------------------------------------------------ + + +@pytest.fixture +def client(store): + app = FastAPI() + app.include_router(queue_router.router) + app.state.autopilot = type( + "AP", + (), + { + "store": store, + "queue_hold_reason": lambda self, sid: None, + "deliver_now": lambda self, sid: ([1], None), + }, + )() + return TestClient(app) + + +def test_queue_endpoints_roundtrip(client): + r = client.post("/api/sessions/s9/queue", json={"text": " after this, run tests "}) + assert r.status_code == 200 + item = r.json() + assert item["text"] == "after this, run tests" and item["status"] == "pending" + view = client.get("/api/sessions/s9/queue").json() + assert [q["id"] for q in view["items"]] == [item["id"]] + assert view["hold_reason"] is None + assert client.delete(f"/api/sessions/s9/queue/{item['id']}").status_code == 200 + assert client.delete(f"/api/sessions/s9/queue/{item['id']}").status_code == 404 + assert client.post("/api/sessions/s9/queue", json={"text": " "}).status_code == 400 + + +def test_send_now_endpoint(client, store): + store.queue_add("s9", "x") + assert client.post("/api/sessions/s9/queue/send-now").json() == {"ok": True, "sent": [1]} diff --git a/backend/tests/test_snapshot.py b/backend/tests/test_snapshot.py new file mode 100644 index 0000000..7d94ef4 --- /dev/null +++ b/backend/tests/test_snapshot.py @@ -0,0 +1,211 @@ +"""Session restore: snapshot the tmux topology (tagging each window with its resumable +Claude session id), and rebuild selected groups after a reboot — recreating windows that +run `claude --resume `, skipping anything already live so restores never duplicate.""" + +import json + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from muse.autopilot import snapshot as snap +from muse.models import LiveSession +from muse.routers import tmux as tmux_router + + +def pane(session, widx, pane_id, *, cmd="claude", cwd="/w", name="win", active=True, pidx=0): + return { + "session_name": session, + "window_index": widx, + "window_name": name, + "window_active": active, + "pane_id": pane_id, + "pane_index": pidx, + "pane_active": active, + "command": cmd, + "cwd": cwd, + "title": "", + "session_attached": True, + "last_activity": 0, + "window_id": "@" + pane_id[1:], + } + + +# --- build_snapshot --------------------------------------------------------------- + + +def test_build_tags_windows_with_resumable_session_id(monkeypatch): + panes = [ + pane("work", 0, "%1", cwd="/a", name="api"), + pane("work", 1, "%2", cwd="/b", name="ui"), + pane("exp", 0, "%3", cmd="bash", cwd="/c", name="shell"), + ] + live = [ + LiveSession(session_id="sid-a", pid=10, pane_id="%1", cwd="/a"), + LiveSession(session_id="sid-b", pid=11, pane_id="%2", cwd="/b"), + ] + monkeypatch.setattr(snap.tmux, "available", lambda: True) + monkeypatch.setattr(snap.tmux, "list_layout", lambda: panes) + monkeypatch.setattr(snap.live_discovery, "discover", lambda: live) + + s = snap.build_snapshot() + work = next(g for g in s["groups"] if g["name"] == "work") + assert [w["session_id"] for w in work["windows"]] == ["sid-a", "sid-b"] + assert all(w["kind"] == "claude" for w in work["windows"]) + exp = next(g for g in s["groups"] if g["name"] == "exp") + assert exp["windows"][0]["kind"] == "shell" and exp["windows"][0]["session_id"] is None + + +def test_build_returns_none_without_claude(monkeypatch): + monkeypatch.setattr(snap.tmux, "available", lambda: True) + monkeypatch.setattr(snap.tmux, "list_layout", lambda: [pane("s", 0, "%1", cmd="bash")]) + monkeypatch.setattr(snap.live_discovery, "discover", lambda: []) + assert snap.build_snapshot() is None + + +def test_topology_sig_ignores_ts_and_tracks_structure(): + a = {"ts": "T1", "groups": [{"name": "g", "windows": [ + {"window_name": "w", "cwd": "/x", "session_id": "s", "kind": "claude"}]}]} + b = {**a, "ts": "T2"} # only ts differs + assert snap.topology_sig(a) == snap.topology_sig(b) + c = json.loads(json.dumps(a)) + c["groups"][0]["windows"][0]["cwd"] = "/y" + assert snap.topology_sig(c) != snap.topology_sig(a) + + +# --- annotate_liveness ------------------------------------------------------------ + + +def _snap(*groups): + return {"ts": "T", "groups": list(groups)} + + +def _g(name, *windows): + return {"name": name, "windows": list(windows)} + + +def _w(name, sid, kind="claude", cwd="/w"): + return {"window_name": name, "cwd": cwd, "command": "claude", "kind": kind, "session_id": sid} + + +def test_offer_true_when_claude_windows_none_live(monkeypatch): + monkeypatch.setattr(snap, "_live_index", lambda: (set(), set(), set())) + out = snap.annotate_liveness(_snap(_g("a", _w("x", "s1"), _w("y", "s2")))) + assert out["offer"] is True and out["restorable_count"] == 2 + assert all(not w["live"] for g in out["groups"] for w in g["windows"]) + + +def test_offer_false_when_a_session_is_live(monkeypatch): + monkeypatch.setattr(snap, "_live_index", lambda: ({"s1"}, set(), set())) + out = snap.annotate_liveness(_snap(_g("a", _w("x", "s1"), _w("y", "s2")))) + assert out["offer"] is False # not a clean reboot — something's still running + live = {w["window_name"]: w["live"] for g in out["groups"] for w in g["windows"]} + assert live == {"x": True, "y": False} + + +# --- restore ---------------------------------------------------------------------- + + +@pytest.fixture +def restore_calls(monkeypatch, tmp_path): + """Fresh tmux with nothing live; record new_session/new_window calls.""" + calls = {"session": [], "window": []} + monkeypatch.setattr(snap, "_live_index", lambda: (set(), set(), set())) + monkeypatch.setattr( + snap.tmux, "new_session", + lambda name, cwd=None, command=None, window_name=None: + calls["session"].append((name, cwd, command, window_name)) or (True, ""), + ) + monkeypatch.setattr( + snap.tmux, "new_window", + lambda cwd, command, session=None, name=None: + calls["window"].append((cwd, command, session, name)) or (True, "%9"), + ) + monkeypatch.setattr(snap.os.path, "isdir", lambda p: True) # cwds exist + return calls + + +def test_restore_creates_session_then_windows_resuming_claude(restore_calls): + s = _snap(_g("work", _w("api", "sid-a", cwd="/a"), _w("ui", "sid-b", cwd="/b"))) + res = snap.restore(s, ["work"]) + assert res["restored"] == 2 and res["groups"] == ["work"] + # First missing window creates the session (no leftover placeholder shell)... + assert restore_calls["session"] == [("work", "/a", "claude --resume sid-a || claude --continue", "api")] + # ...the rest append as windows. + assert restore_calls["window"] == [("/b", "claude --resume sid-b || claude --continue", "work", "ui")] + + +def test_restore_skips_live_windows(monkeypatch, restore_calls): + monkeypatch.setattr(snap, "_live_index", lambda: ({"sid-a"}, set(), {"work"})) + s = _snap(_g("work", _w("api", "sid-a"), _w("ui", "sid-b"))) + res = snap.restore(s, ["work"]) + assert res["restored"] == 1 and res["skipped"] == 1 + # 'work' already exists (live), so the surviving window appends — no new session. + assert restore_calls["session"] == [] + assert restore_calls["window"] == [("/w", "claude --resume sid-b || claude --continue", "work", "ui")] + + +def test_restore_shell_and_unknown_id_and_missing_cwd(monkeypatch, restore_calls): + monkeypatch.setattr(snap.os.path, "isdir", lambda p: p == "/ok") + s = _snap(_g( + "g", + _w("editor", None, kind="shell", cwd="/gone"), # shell + missing cwd → ~ + bare shell + _w("chat", None, kind="claude", cwd="/ok"), # claude, no id → --continue + )) + snap.restore(s, ["g"]) + home = snap.os.path.expanduser("~") + assert restore_calls["session"][0] == ("g", home, None, "editor") # command None → default shell + assert restore_calls["window"][0] == ("/ok", "claude --continue", "g", "chat") + + +# --- endpoints -------------------------------------------------------------------- + + +class FakeStore: + def __init__(self, snapshot=None): + self._latest = (snap.topology_sig(snapshot), json.dumps(snapshot)) if snapshot else None + self.entries = [] + + def latest_snapshot(self): + return self._latest + + def log(self, sid, action, detail=""): + self.entries.append((sid, action, detail)) + + +def _client(store, monkeypatch): + app = FastAPI() + app.include_router(tmux_router.router) + app.state.autopilot = type("AP", (), {"store": store})() + monkeypatch.setattr(tmux_router.tmux, "available", lambda: True) + return TestClient(app) + + +def test_get_snapshot_endpoint_annotates(monkeypatch): + monkeypatch.setattr(tmux_router.layout_snapshot, "_live_index", lambda: (set(), set(), set())) + store = FakeStore(_snap(_g("a", _w("x", "s1"), _w("y", "s2")))) + r = _client(store, monkeypatch).get("/api/tmux/snapshot") + assert r.status_code == 200 and r.json()["offer"] is True + + empty = _client(FakeStore(None), monkeypatch).get("/api/tmux/snapshot") + assert empty.json() == {"ts": None, "groups": [], "offer": False, "restorable_count": 0} + + +def test_restore_endpoint_rebuilds_selected(monkeypatch): + monkeypatch.setattr(tmux_router.layout_snapshot, "_live_index", lambda: (set(), set(), set())) + monkeypatch.setattr(tmux_router.layout_snapshot.os.path, "isdir", lambda p: True) + made = [] + monkeypatch.setattr( + tmux_router.layout_snapshot.tmux, "new_session", + lambda name, cwd=None, command=None, window_name=None: made.append(name) or (True, ""), + ) + store = FakeStore(_snap(_g("a", _w("x", "s1")), _g("b", _w("y", "s2")))) + r = _client(store, monkeypatch).post("/api/tmux/restore", json={"groups": ["a"]}) + assert r.status_code == 200 and r.json()["restored"] == 1 + assert made == ["a"] # only the selected group + assert ("tmux", "layout_restore", "1 windows in ['a']") in store.entries + + +def test_restore_endpoint_400_without_snapshot(monkeypatch): + r = _client(FakeStore(None), monkeypatch).post("/api/tmux/restore", json={"groups": ["a"]}) + assert r.status_code == 400 diff --git a/frontend/src/components/board/QueueChips.test.tsx b/frontend/src/components/board/QueueChips.test.tsx new file mode 100644 index 0000000..7ddc804 --- /dev/null +++ b/frontend/src/components/board/QueueChips.test.tsx @@ -0,0 +1,28 @@ +/** QueueChips gates polling on `enabled` so 8+ deck cards don't each poll the + * queue every 5s. Disabled = no fetches; flipping on fetches immediately. */ +import { render, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import QueueChips from "./QueueChips"; + +vi.mock("../../api/client", () => ({ api: { getQueue: vi.fn() } })); +import { api } from "../../api/client"; + +const getQueue = vi.mocked(api.getQueue); + +afterEach(() => getQueue.mockReset()); + +describe("QueueChips polling gate", () => { + it("does not fetch while disabled", () => { + getQueue.mockResolvedValue({ items: [], hold_reason: null }); + render(); + expect(getQueue).not.toHaveBeenCalled(); + }); + + it("fetches immediately once enabled", async () => { + getQueue.mockResolvedValue({ items: [], hold_reason: null }); + const { rerender } = render(); + expect(getQueue).not.toHaveBeenCalled(); + rerender(); + await waitFor(() => expect(getQueue).toHaveBeenCalledWith("s")); + }); +}); diff --git a/frontend/src/components/board/QueueChips.tsx b/frontend/src/components/board/QueueChips.tsx new file mode 100644 index 0000000..ffc9e29 --- /dev/null +++ b/frontend/src/components/board/QueueChips.tsx @@ -0,0 +1,110 @@ +import { useCallback, useEffect, useState } from "react"; +import { api } from "../../api/client"; +import type { QueuedReply } from "../../api/types"; +import { usePolling } from "../../hooks/usePolling"; + +/** Pending queued replies for one session: what muse will type in when the turn + * ends, each cancellable until the moment it's delivered. Recently delivered + * items flash through briefly (status flips to "sent" and then ages out). */ +export default function QueueChips({ + sessionId, + refreshKey = 0, + enabled = true, +}: { + sessionId: string; + refreshKey?: number; // parent bumps this after queueing to refresh instantly + enabled?: boolean; // gate polling — deck cards enable only while on screen +}) { + const [items, setItems] = useState([]); + const [holdReason, setHoldReason] = useState(null); + const [sendErr, setSendErr] = useState(null); + const [sendingNow, setSendingNow] = useState(false); + + const refresh = useCallback(async () => { + const view = await api.getQueue(sessionId); + setItems(view.items); + setHoldReason(view.hold_reason); + }, [sessionId]); + + usePolling(refresh, 5000, enabled); + useEffect(() => { + if (!enabled) return; + refresh().catch(() => {}); + }, [refresh, refreshKey, enabled]); + + const pending = items.filter((q) => q.status === "pending"); + if (pending.length === 0) return null; + + const sendNow = async () => { + setSendingNow(true); + setSendErr(null); + try { + await api.sendQueuedNow(sessionId); + } catch (e) { + // 409 with the block reason (menu open, no pane, …) — show it. + setSendErr(e instanceof Error ? e.message.replace(/^\d+:\s*/, "") : "couldn’t send"); + } finally { + setSendingNow(false); + refresh().catch(() => {}); + } + }; + + const cancel = async (qid: number) => { + try { + await api.cancelQueuedReply(sessionId, qid); + } catch { + /* already delivered — the refresh below shows the truth */ + } + refresh().catch(() => {}); + }; + + const toggleMode = async (q: QueuedReply) => { + try { + await api.setQueuedReplyMode(sessionId, q.id, q.mode === "turn" ? "append" : "turn"); + } catch { + /* already delivered */ + } + refresh().catch(() => {}); + }; + + return ( +
+ {pending.map((q, i) => ( + + + {q.text} + + + ))} + {/* Why it hasn't delivered yet + an explicit override. Without this a held + reply just sits there silently and reads as "the queue is broken". */} + {(holdReason || sendErr) && ( +
+ + {sendErr ? `⚠ ${sendErr}` : `held — ${holdReason}`} + + +
+ )} +
+ ); +} diff --git a/frontend/src/components/board/ReplyBox.test.tsx b/frontend/src/components/board/ReplyBox.test.tsx new file mode 100644 index 0000000..e3d746a --- /dev/null +++ b/frontend/src/components/board/ReplyBox.test.tsx @@ -0,0 +1,113 @@ +/** Long-pressing the send button opens a menu of send modes; "Compact first" + * runs /compact then queues the reply. */ +import { act, fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import ReplyBox from "./ReplyBox"; + +vi.mock("../../api/client", () => ({ + api: { + respondToSession: vi.fn().mockResolvedValue({}), + queueReply: vi.fn().mockResolvedValue({}), + getPaneCommands: vi.fn().mockResolvedValue([]), + }, +})); +import { api } from "../../api/client"; + +const respondToSession = vi.mocked(api.respondToSession); +const queueReply = vi.mocked(api.queueReply); + +afterEach(() => { + respondToSession.mockClear(); + queueReply.mockClear(); +}); + +function renderBox() { + render(); + fireEvent.change(screen.getByRole("textbox"), { target: { value: "do the thing" } }); + return screen.getByLabelText("Send"); +} + +// Long-press = pointerDown, then let the 450ms timer fire. +function longPress(btn: HTMLElement) { + vi.useFakeTimers(); + fireEvent.pointerDown(btn); + act(() => vi.advanceTimersByTime(460)); + vi.useRealTimers(); +} + +// Pretend we're on (or off) a fine-pointer desktop for useIsDesktop(). +function setDesktop(on: boolean) { + window.matchMedia = ((q: string) => ({ + matches: on, + media: q, + onchange: null, + addEventListener() {}, + removeEventListener() {}, + addListener() {}, + removeListener() {}, + dispatchEvent() { + return false; + }, + })) as unknown as typeof window.matchMedia; +} + +describe("ReplyBox desktop keyboard", () => { + afterEach(() => { + delete (window as { matchMedia?: unknown }).matchMedia; + }); + + it("Enter sends; Shift+Enter is a newline", async () => { + setDesktop(true); + render(); + const ta = screen.getByRole("textbox"); + fireEvent.change(ta, { target: { value: "hi" } }); + fireEvent.keyDown(ta, { key: "Enter", shiftKey: true }); + expect(respondToSession).not.toHaveBeenCalled(); + fireEvent.keyDown(ta, { key: "Enter" }); + await waitFor(() => expect(respondToSession).toHaveBeenCalledWith("s1", "hi")); + }); + + it("exposes the send-options caret on desktop but not on touch", () => { + setDesktop(false); + const { unmount } = render( + , + ); + expect(screen.queryByLabelText("Send options")).toBeNull(); + unmount(); + setDesktop(true); + render(); + fireEvent.click(screen.getByLabelText("Send options")); + expect(screen.getByRole("menu")).toBeTruthy(); + }); +}); + +describe("ReplyBox send-options menu", () => { + it("long-press opens the menu; a plain tap does not", () => { + const btn = renderBox(); + expect(screen.queryByRole("menu")).toBeNull(); + fireEvent.click(btn); // plain tap + expect(screen.queryByRole("menu")).toBeNull(); + longPress(btn); + expect(screen.getByRole("menu")).toBeTruthy(); + expect(screen.getByText("Compact first")).toBeTruthy(); + }); + + it("'Compact first' sends /compact then the reply right behind it", async () => { + const btn = renderBox(); + longPress(btn); + fireEvent.click(screen.getByText("Compact first")); + await waitFor(() => expect(respondToSession).toHaveBeenCalledWith("s1", "do the thing")); + expect(respondToSession).toHaveBeenNthCalledWith(1, "s1", "/compact"); + expect(respondToSession).toHaveBeenNthCalledWith(2, "s1", "do the thing"); + expect(queueReply).not.toHaveBeenCalled(); + expect(screen.queryByRole("menu")).toBeNull(); // closed after picking + }); + + it("'Send now' from the menu sends immediately", async () => { + const btn = renderBox(); + longPress(btn); + fireEvent.click(screen.getByText("Send now")); + await waitFor(() => expect(respondToSession).toHaveBeenCalledWith("s1", "do the thing")); + expect(queueReply).not.toHaveBeenCalled(); + }); +}); diff --git a/frontend/src/components/board/ReplyBox.tsx b/frontend/src/components/board/ReplyBox.tsx index b6b4c76..0b7b1ea 100644 --- a/frontend/src/components/board/ReplyBox.tsx +++ b/frontend/src/components/board/ReplyBox.tsx @@ -1,6 +1,9 @@ -import { useEffect, useRef, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { api } from "../../api/client"; +import { useIsDesktop } from "../../util/useIsDesktop"; +import { useTypeToFocus } from "../../util/typeToFocus"; import AiActionButton from "../AiActionButton"; +import { useSlashMenu } from "../SlashMenu"; /** Reply to a live Claude session from the board: text → its tmux pane. * Disabled (with the reason) when the session has no matched pane. */ @@ -10,19 +13,44 @@ export default function ReplyBox({ busy, variant = "board", draft, + onQueued, + onSent, + showInterrupt = true, + slashPaneId, + captureTyping = false, }: { sessionId: string; hasPane: boolean; - busy: boolean; // live_status === "busy": sending mid-turn queues the message + busy: boolean; // live_status === "busy": sending mid-turn steers the current turn variant?: "board" | "cockpit"; // cockpit = phone-first multiline textarea draft?: string; // when set, prefills the box (AI suggestion) — you edit and send + onQueued?: () => void; // notify the parent so its queue chips refresh immediately + onSent?: () => void; // notify the parent so it can refresh sooner than its poll + showInterrupt?: boolean; // hide the Esc button where the parent already has one + slashPaneId?: string; // pane id to source "/" autocomplete from (enables the menu) + captureTyping?: boolean; // desktop: start-typing-anywhere focuses this composer }) { + const desktop = useIsDesktop(); const [text, setText] = useState(""); - const [state, setState] = useState<"idle" | "sending" | "sent" | "error">("idle"); + const [state, setState] = useState<"idle" | "sending" | "sent" | "queued" | "error">("idle"); const [error, setError] = useState(null); const sentTimer = useRef(undefined); + // What a tap on the primary button does; the last option picked from the menu + // sticks, so a plain tap repeats it. Long-pressing opens the menu of send modes. + const [mode, setMode] = useState<"send" | "queue">("send"); + const [menuOpen, setMenuOpen] = useState(false); + const pressTimer = useRef(undefined); + const longFired = useRef(false); + const taRef = useRef(null); + const menuRef = useRef(null); - useEffect(() => () => window.clearTimeout(sentTimer.current), []); + useEffect( + () => () => { + window.clearTimeout(sentTimer.current); + window.clearTimeout(pressTimer.current); + }, + [], + ); // A parent-supplied draft is a DEFAULT, not an override: apply it only when the // box is empty or still holds the previous default (i.e. the user hasn't typed // their own text). This lets the suggestion refresh each turn without ever @@ -44,6 +72,47 @@ export default function ReplyBox({ await api.respondToSession(sessionId, t); setText(""); setState("sent"); + onSent?.(); + sentTimer.current = window.setTimeout(() => setState("idle"), 2500); + } catch (e) { + setState("error"); + setError(e instanceof Error ? e.message : String(e)); + } + }; + + // "Queue" hands the message to muse to type in when the turn ACTUALLY ends + // (idle, no permission menu) — vs Send-while-busy, which steers the current turn. + const queue = async () => { + const t = text.trim(); + if (!t || state === "sending") return; + setState("sending"); + setError(null); + try { + await api.queueReply(sessionId, t); + setText(""); + setState("queued"); + onQueued?.(); + sentTimer.current = window.setTimeout(() => setState("idle"), 2500); + } catch (e) { + setState("error"); + setError(e instanceof Error ? e.message : String(e)); + } + }; + + // "Compact first": send /compact, then send this reply right behind it. No need + // to wait for idle — Claude Code's own input queue holds the reply until the + // compaction turn finishes, so it lands right after the fresh summary. + const compactThenSend = async () => { + if (state === "sending") return; + const t = text.trim(); + setState("sending"); + setError(null); + try { + await api.respondToSession(sessionId, "/compact"); + if (t) await api.respondToSession(sessionId, t); + setText(""); + setState("sent"); + onSent?.(); sentTimer.current = window.setTimeout(() => setState("idle"), 2500); } catch (e) { setState("error"); @@ -61,22 +130,108 @@ export default function ReplyBox({ } }; + // Primary button: tap runs the current mode; a long press (no drag needed) + // opens the send-options menu instead of firing, so a stray click after it is + // swallowed. + const startPress = () => { + longFired.current = false; + pressTimer.current = window.setTimeout(() => { + longFired.current = true; + setMenuOpen(true); + navigator.vibrate?.(15); + }, 450); + }; + const endPress = () => window.clearTimeout(pressTimer.current); + const primary = () => { + if (longFired.current) { + longFired.current = false; // this "click" was the end of a long press — ignore + return; + } + if (mode === "queue") queue(); + else send(); + }; + // Pick a way to send from the long-press menu: set the sticky mode (so the next + // plain tap repeats it) and fire it now. + const pick = (choice: "send" | "queue" | "compact") => { + setMenuOpen(false); + if (choice === "compact") { + compactThenSend(); + return; + } + setMode(choice); + (choice === "queue" ? queue : send)(); + }; + + const slash = useSlashMenu({ paneId: slashPaneId, text, setText }); + + // Desktop: begin typing anywhere on the card and land in this composer. + const append = useCallback((ch: string) => setText((t) => t + ch), []); + useTypeToFocus(taRef, append, desktop && captureTyping && hasPane); + + // Multi-line: grow the textarea with its content (cockpit only), capped so a + // long paste scrolls internally instead of eating the whole card. + useEffect(() => { + const el = taRef.current; + if (!el) return; + el.style.height = "auto"; + el.style.height = `${Math.min(el.scrollHeight, 160)}px`; + }, [text]); + + // Keyboard-drive the send-options menu: focus the first row when it opens. + useEffect(() => { + if (menuOpen) menuRef.current?.querySelector(".send-menu-row")?.focus(); + }, [menuOpen]); + const onMenuKey = (e: React.KeyboardEvent) => { + const rows = Array.from( + menuRef.current?.querySelectorAll(".send-menu-row") ?? [], + ); + const at = rows.indexOf(document.activeElement as HTMLButtonElement); + if (e.key === "Escape") { + e.preventDefault(); + setMenuOpen(false); + taRef.current?.focus(); + } else if (e.key === "ArrowDown") { + e.preventDefault(); + rows[(at + 1) % rows.length]?.focus(); + } else if (e.key === "ArrowUp") { + e.preventDefault(); + rows[(at - 1 + rows.length) % rows.length]?.focus(); + } + }; + const placeholder = hasPane ? busy - ? "queue a message for when this turn ends…" + ? "⏳ queue for when this turn ends, or send to steer now…" : "reply to this session…" : "no tmux pane matched — open it in your terminal"; return (
+ {slash.menu} {variant === "cockpit" ? (