diff --git a/public/queue.ts b/public/queue.ts new file mode 100644 index 0000000..ccbabac --- /dev/null +++ b/public/queue.ts @@ -0,0 +1,70 @@ +/** + * public/queue.ts — W2: client helper for the server-side PTY-inject queue. + * + * Enqueue a follow-up prompt that the server fires into a session's PTY the next + * time Claude goes idle (so "run tests → open a PR" advances itself while you're + * away). Enqueue is a same-origin HTTP POST (Origin-guarded server-side), NOT a + * WS frame — it must work with zero tabs open and target any session (e.g. from + * the manage page), not just the WS-bound one. + * + * Discipline (mirrors quick-reply): every call NEVER throws — a failed fetch or + * non-2xx response returns a falsy result the caller can ignore. Byte-shuttle: + * `text` is sent verbatim; `appendEnter` tells the server to append \r. + */ + +/** Result of an enqueue attempt. `length` = the new queue depth on success. */ +export interface EnqueueResult { + ok: boolean + /** New queue depth (present on success). */ + length?: number + /** HTTP status on a non-2xx response (absent on network error). */ + status?: number +} + +/** POST a follow-up prompt to a session's inject queue. Never throws. */ +export async function enqueueFollowup( + sessionId: string, + text: string, + appendEnter: boolean, +): Promise { + try { + if (typeof fetch === 'undefined') return { ok: false } + const res = await fetch(`/live-sessions/${encodeURIComponent(sessionId)}/queue`, { + method: 'POST', + credentials: 'same-origin', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ text, appendEnter }), + }) + if (!res.ok) return { ok: false, status: res.status } + const data = (await res.json()) as { length?: unknown } + return { ok: true, length: typeof data.length === 'number' ? data.length : undefined } + } catch { + return { ok: false } + } +} + +/** DELETE all pending entries for a session (cancel-all). Returns success. Never throws. */ +export async function clearQueue(sessionId: string): Promise { + try { + if (typeof fetch === 'undefined') return false + const res = await fetch(`/live-sessions/${encodeURIComponent(sessionId)}/queue`, { + method: 'DELETE', + credentials: 'same-origin', + }) + return res.ok + } catch { + return false + } +} + +/** + * Extract the queue depth from an incoming server frame. Returns the length for a + * `{type:'queue',length}` frame, else null (so callers can filter unrelated + * frames). Defensive: tolerates any unknown input shape. + */ +export function queueLengthFromFrame(msg: unknown): number | null { + if (msg === null || typeof msg !== 'object') return null + const o = msg as Record + if (o['type'] !== 'queue') return null + return typeof o['length'] === 'number' ? o['length'] : null +} diff --git a/public/quick-reply.ts b/public/quick-reply.ts index 998e1b7..ae16ab0 100644 --- a/public/quick-reply.ts +++ b/public/quick-reply.ts @@ -147,6 +147,10 @@ function clearChildren(parent: HTMLElement): void { export interface QuickReplyOpts { /** Called with the full byte string to inject into the terminal. */ onSend: (data: string) => void + /** W2: optional. When provided, the snippet editor gains a "Queue" button that + * enqueues the typed text as a follow-up prompt (fired on next idle) instead of + * sending it now. `appendEnter` mirrors the editor checkbox. */ + onQueue?: (text: string, appendEnter: boolean) => void } /** Returned by mountQuickReply to allow cleanup. */ @@ -165,7 +169,7 @@ export interface QuickReplyHandle { * Labels are set via textContent — never innerHTML (SEC-L3). */ export function mountQuickReply(container: HTMLElement, opts: QuickReplyOpts): QuickReplyHandle { - const { onSend } = opts + const { onSend, onQueue } = opts // Track disposers for all event listeners so dispose() is clean. const disposers: Array<() => void> = [] @@ -233,6 +237,22 @@ export function mountQuickReply(container: HTMLElement, opts: QuickReplyOpts): Q editor.appendChild(labelInput) editor.appendChild(enterWrap) editor.appendChild(saveBtn) + + // W2: "Queue" defers the typed text as a follow-up prompt (fires on next + // idle) rather than saving it as a chip. Shown only when a handler is wired. + if (onQueue !== undefined) { + const queueBtn = el('button', 'qr-editor-queue', 'Queue') + queueBtn.title = 'Queue as a follow-up (runs when Claude next goes idle)' + function onQueueClick() { + const text = textInput.value.trim() + if (!text) return // empty text — no-op, editor stays open + onQueue?.(text, enterCheck.checked) + editor.remove() + } + queueBtn.addEventListener('click', onQueueClick) + editor.appendChild(queueBtn) + } + editor.appendChild(cancelBtn) container.appendChild(editor) diff --git a/public/tabs.ts b/public/tabs.ts index 1c41178..d8cca91 100644 --- a/public/tabs.ts +++ b/public/tabs.ts @@ -31,6 +31,7 @@ import { renderDiffFile } from './diff.js' import { renderTelemetryGauge } from './preview-grid.js' import { mountPushToggle } from './push.js' import { mountQuickReply } from './quick-reply.js' +import { enqueueFollowup } from './queue.js' import { mountTimeline, type TimelineHandle } from './timeline.js' import { createVoiceInput, type VoiceInput } from './voice.js' import { matchCommand, type VoiceMatchContext } from './voice-commands.js' @@ -216,7 +217,21 @@ export class TabApp { const keybar = document.getElementById('keybar') if (keybar?.parentElement) keybar.parentElement.insertBefore(host, keybar) else document.body.appendChild(host) - mountQuickReply(host, { onSend: (data) => this.sendToActive(data) }) + mountQuickReply(host, { + onSend: (data) => this.sendToActive(data), + // W2: "Queue" in the snippet editor enqueues a follow-up for the active + // session instead of sending it now (fires when Claude next goes idle). + onQueue: (text, appendEnter) => this.enqueueToActive(text, appendEnter), + }) + } + + /** W2: enqueue a follow-up prompt for the active session (best-effort — the + * POST is Origin-guarded server-side; failures are surfaced only via the badge + * not updating). No-op when no tab has attached yet. */ + enqueueToActive(text: string, appendEnter: boolean): void { + const id = this.activeSessionId() + if (id === null) return + void enqueueFollowup(id, text, appendEnter) } /** A4: hidden activity-timeline panel; toggled per active session. */ @@ -725,6 +740,8 @@ export class TabApp { }, // B2: telemetry is the single source of truth on the session; just re-render. onTelemetry: () => this.refreshTab(entry), + // W2: pending inject-queue depth changed — re-render the "N queued" badge. + onQueue: () => this.refreshTab(entry), // Split-grid: clicking anywhere in this pane makes it the focused quadrant. onFocus: () => this.setFocused(this.tabs.indexOf(entry)), }) @@ -1417,6 +1434,12 @@ export class TabApp { if (label) label.textContent = title const claude = el.querySelector('.tab-claude') if (claude) claude.textContent = claudeIcon(cs) + // W2: "⧗N" when the session has queued follow-ups, else empty (badge hidden). + const queue = el.querySelector('.tab-queue') + if (queue) { + const n = entry.session.queueLength + queue.textContent = n > 0 ? `⧗${n}` : '' + } const gauge = el.querySelector('.tab-gauge') if (gauge) renderTelemetryGauge(gauge, entry.session.telemetry, STATUSLINE_TTL_MS) // B2 } @@ -1512,6 +1535,11 @@ export class TabApp { claude.className = 'tab-claude' tabEl.appendChild(claude) + // W2: pending inject-queue depth badge (filled in by refreshTab). + const queue = document.createElement('span') + queue.className = 'tab-queue' + tabEl.appendChild(queue) + // B2: per-tab telemetry gauge container (filled in by refreshTab). const gauge = document.createElement('span') gauge.className = 'tab-gauge' diff --git a/public/terminal-session.ts b/public/terminal-session.ts index 979b317..210dd78 100644 --- a/public/terminal-session.ts +++ b/public/terminal-session.ts @@ -115,6 +115,9 @@ export interface TerminalSessionOpts { /** Optional: fired when new statusLine telemetry arrives (B2). Single source of * truth — T-tabs reads session.telemetry via the getter, not its own copy. */ onTelemetry?: (telemetry: StatusTelemetry) => void + /** Optional: fired when the pending inject-queue depth changes (W2), so the tab + * can render an "N queued" badge. Broadcast to every mirrored device. */ + onQueue?: (length: number) => void /** Optional: fired on a pointerdown anywhere in this pane (split-grid focus). * Lets TabApp move the focused-pane (activeIndex) to the clicked quadrant. */ onFocus?: () => void @@ -137,6 +140,7 @@ export class TerminalSession { private readonly onStatus: ((status: SessionStatus) => void) | undefined private readonly onClaudeStatus: ((status: ClaudeStatus, detail?: string) => void) | undefined private readonly onTelemetry: ((telemetry: StatusTelemetry) => void) | undefined + private readonly onQueue: ((length: number) => void) | undefined private readonly onFocus: (() => void) | undefined private readonly spawnCwd: string | undefined private readonly initialInput: string | undefined @@ -147,6 +151,7 @@ export class TerminalSession { private pendingApprovalValue = false private pendingToolValue: string | undefined = undefined private telemetryValue: StatusTelemetry | null = null + private queueLengthValue = 0 private pendingGateValue: PermissionGate | null = null // W1: bounded command/diff preview of the held tool, from the last pending // status frame. Null when no approval is held or the tool wasn't previewable. @@ -179,6 +184,7 @@ export class TerminalSession { this.onStatus = opts.onStatus this.onClaudeStatus = opts.onClaudeStatus this.onTelemetry = opts.onTelemetry + this.onQueue = opts.onQueue this.onFocus = opts.onFocus this.spawnCwd = opts.cwd this.initialInput = opts.initialInput @@ -255,6 +261,12 @@ export class TerminalSession { return this.telemetryValue } + /** W2: current pending inject-queue depth (0 when empty), from the last + * `queue` frame. Every mirrored device sees the same value (broadcast). */ + get queueLength(): number { + return this.queueLengthValue + } + /** The gate kind from the last pending status frame (B4): 'plan' or 'tool'. * Null when no approval is held or the last status had no gate. */ get pendingGate(): PermissionGate | null { @@ -393,6 +405,13 @@ export class TerminalSession { this.onTelemetry?.(msg.telemetry) break } + case 'queue': { + // W2: pending inject-queue depth changed — update the badge on every + // mirrored device (shared session). + this.queueLengthValue = msg.length + this.onQueue?.(msg.length) + break + } case 'exit': { this.setStatus('exited') const reason = msg.reason ? ` (${msg.reason})` : '' diff --git a/src/config.ts b/src/config.ts index aa4f10f..6b517f5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -63,6 +63,10 @@ const DEFAULT_DIFF_MAX_FILES = 300 const DEFAULT_STATUSLINE_TTL_MS = 30_000 // B3 worktrees const DEFAULT_WORKTREE_TIMEOUT_MS = 10_000 +// W2 inject queue +const DEFAULT_QUEUE_MAX_ITEMS = 10 +const DEFAULT_QUEUE_ITEM_MAX_BYTES = 4096 +const DEFAULT_QUEUE_SETTLE_MS = 1500 /** Valid --permission-mode values (R0-confirmed). Whitelist for validation. */ const PERMISSION_MODES: readonly PermissionMode[] = ['default', 'acceptEdits', 'plan', 'auto'] @@ -384,6 +388,16 @@ export function loadConfig(env: EnvLike): Config { ) const allowAutoMode = parseBool(env['ALLOW_AUTO_MODE'], false) // default off (SEC-M5) + // W2 server-side PTY-inject follow-up queue + const queueEnabled = parseBool(env['QUEUE_ENABLED'], true) + const queueMaxItems = parseNonNegativeInt(env['QUEUE_MAX_ITEMS'], 'QUEUE_MAX_ITEMS', DEFAULT_QUEUE_MAX_ITEMS) + const queueItemMaxBytes = parseNonNegativeInt( + env['QUEUE_ITEM_MAX_BYTES'], + 'QUEUE_ITEM_MAX_BYTES', + DEFAULT_QUEUE_ITEM_MAX_BYTES, + ) + const queueSettleMs = parseNonNegativeInt(env['QUEUE_SETTLE_MS'], 'QUEUE_SETTLE_MS', DEFAULT_QUEUE_SETTLE_MS) + // ── assemble + freeze ─────────────────────────────────────────────────────── // `satisfies Config` on the base object verifies all existing Config fields // are present without triggering excess-property checks on the v0.7 additions. @@ -435,5 +449,10 @@ export function loadConfig(env: EnvLike): Config { worktreeTimeoutMs, defaultPermissionMode, allowAutoMode, + // W2 inject queue + queueEnabled, + queueMaxItems, + queueItemMaxBytes, + queueSettleMs, }) } diff --git a/src/server.ts b/src/server.ts index fe90388..eec054f 100644 --- a/src/server.ts +++ b/src/server.ts @@ -32,7 +32,7 @@ import type { WebSocket as WsWebSocket } from 'ws' import type { IncomingMessage } from 'node:http' import { loadConfig } from './config.js' -import { parseClientMessage, serialize } from './protocol.js' +import { parseClientMessage, serialize, SESSION_ID_RE } from './protocol.js' import { isOriginAllowed } from './http/origin.js' import { parseHookEvent } from './http/hook.js' import { deriveApprovalPreview } from './http/approval-preview.js' @@ -76,6 +76,7 @@ const LOG_FIELD_MAX = 200 const RATE_LIMIT_WINDOW_MS = 60_000 const DECISION_RATE_MAX = 10 // POST /hook/decision ≤ 10/min/IP const SUBSCRIBE_RATE_MAX = 5 // POST/DELETE /push/subscribe ≤ 5/min/IP +const QUEUE_RATE_MAX = 20 // W2: POST/DELETE /live-sessions/:id/queue ≤ 20/min/IP /** The four permission modes the WS approve relay accepts (B4); else undefined. */ const PERMISSION_MODES: ReadonlySet = new Set([ @@ -219,6 +220,47 @@ export function startServer(cfg: Config): { close(): Promise } { // Per-IP rate limiters for the new state-changing routes (SEC-H9). const decisionLimiter = createRateLimiter(DECISION_RATE_MAX, RATE_LIMIT_WINDOW_MS) const subscribeLimiter = createRateLimiter(SUBSCRIBE_RATE_MAX, RATE_LIMIT_WINDOW_MS) + const queueLimiter = createRateLimiter(QUEUE_RATE_MAX, RATE_LIMIT_WINDOW_MS) // W2 + + // W2: per-session settle timers for the idle-drain. A Stop/SessionEnd hook + // schedules a debounced drain of one queue entry after cfg.queueSettleMs, so + // the injected text lands once the prompt is ready. Cleared on shutdown; the + // fired callback re-checks idle + a stable lastOutputAt cursor before draining. + const drainTimers = new Map>() + + /** + * W2: schedule (or reschedule) a single idle-drain for `sessionId`. Debounced + * per session — each Stop clears any pending timer and restarts the window, so + * flapping Stop events fire at most one drain. On fire we drain ONE entry only + * if the session is still idle, unexited, and its lastOutputAt has not moved + * since scheduling (no output mid-settle = Claude really finished, not + * mid-render). One entry per idle → natural pacing (the injected prompt makes + * Claude work again, and its next Stop drains the next entry). + */ + function scheduleDrain(sessionId: string): void { + if (!cfg.queueEnabled) return + const session = manager.get(sessionId) + if (session === undefined || session.exitedAt !== null) return + if (session.queue.length === 0) return + + const existing = drainTimers.get(sessionId) + if (existing !== undefined) clearTimeout(existing) + + const outputCursor = session.lastOutputAt + const timer = setTimeout(() => { + drainTimers.delete(sessionId) + const s = manager.get(sessionId) + if (s === undefined || s.exitedAt !== null) return + // Settle guard: new output since scheduling → Claude is still active; skip + // (a later genuine Stop reschedules). Also require the idle status to hold. + if (s.lastOutputAt !== outputCursor) return + if (s.claudeStatus !== 'idle') return + manager.drainOne(sessionId) + }, cfg.queueSettleMs) + // Don't let a pending drain keep the process alive; a stale fire is harmless. + timer.unref() + drainTimers.set(sessionId, timer) + } // H3/A1: held PermissionRequest responses, keyed by sessionId. `res` is parked // until approve/reject or timeout. `token`+`expiresAt` = the per-decision @@ -350,6 +392,70 @@ export function startServer(cfg: Config): { close(): Promise } { }) }) + // ── W2 inject queue: enqueue a follow-up prompt fired on next idle ──────── + // State-changing (causes shell input on idle) → Origin guard (CSRF) + per-IP + // rate limit. Body: { text: string, appendEnter?: boolean }. Bytes are stored + // and later injected VERBATIM (byte-shuttle) — bounded in size and count, never + // parsed as a shell command. `text` + optional trailing \r is capped by + // queueItemMaxBytes; the queue depth is capped by queueMaxItems. + app.post('/live-sessions/:id/queue', express.json({ limit: '16kb' }), (req, res) => { + if (!requireAllowedOrigin(req, res)) return + if (!cfg.queueEnabled) { + res.status(503).json({ error: 'queue disabled' }) + return + } + if (!queueLimiter(req.socket.remoteAddress ?? '', Date.now())) { + res.status(429).end() + return + } + const id = req.params.id + if (!SESSION_ID_RE.test(id)) { + res.status(400).json({ error: 'invalid session id' }) + return + } + const body = (req.body ?? {}) as Record + const rawText = body['text'] + if (typeof rawText !== 'string' || rawText.length === 0) { + res.status(400).json({ error: 'text must be a non-empty string' }) + return + } + // FE decides Enter (mirrors quick-reply's appendEnter): materialize the exact + // bytes to inject here so the stored entry is byte-identical to a keystroke. + const text = body['appendEnter'] === true ? rawText + '\r' : rawText + if (Buffer.byteLength(text, 'utf8') > cfg.queueItemMaxBytes) { + res.status(413).json({ error: 'text too large' }) + return + } + const result = manager.enqueueFollowup(id, text) + if (!result.ok) { + // full → 409 (never silently drop); unknown/exited → 404. + res.status(result.reason === 'full' ? 409 : 404).json({ error: result.reason }) + return + } + res.json({ length: result.length }) + }) + + // Read-only queue view (depth + the user's own queued text). No Origin guard — + // same threat model as GET /live-sessions. + app.get('/live-sessions/:id/queue', (req, res) => { + const session = manager.get(req.params.id) + if (session === undefined) { + res.status(404).end() + return + } + res.json({ length: session.queue.length, items: [...session.queue] }) + }) + + // Cancel all pending entries (escape hatch). State-changing → Origin guard. + app.delete('/live-sessions/:id/queue', (req, res) => { + if (!requireAllowedOrigin(req, res)) return + if (!manager.clearQueue(req.params.id)) { + res.status(404).end() + return + } + res.json({ length: 0 }) + }) + // CSRF guard for the state-changing DELETE routes (Arch 5b / Sec H2): the WS // upgrade checks Origin, but plain HTTP routes don't — without this, a foreign // page could fire a no-preflight DELETE and Kill-All sessions. Reuse the same @@ -426,6 +532,9 @@ export function startServer(cfg: Config): { close(): Promise } { if (ev.eventClass === 'Stop' || ev.eventClass === 'SessionEnd') { const session = manager.get(ev.sessionId) if (session !== undefined) void pushService.notify(session, 'done') + // W2: Claude just went idle — schedule a debounced drain of the next queued + // follow-up (fires after the settle delay iff still idle + output stable). + scheduleDrain(ev.sessionId) } res.status(204).end() }) @@ -962,6 +1071,9 @@ export function startServer(cfg: Config): { close(): Promise } { // ── Graceful shutdown helper ────────────────────────────────────────────── function doShutdown(): void { clearInterval(reapTimer) + // W2: cancel any pending idle-drain timers so they can't fire post-shutdown. + for (const t of drainTimers.values()) clearTimeout(t) + drainTimers.clear() manager.shutdown() } diff --git a/src/session/manager.ts b/src/session/manager.ts index c7cbf50..ed7594c 100644 --- a/src/session/manager.ts +++ b/src/session/manager.ts @@ -30,6 +30,7 @@ import type { ClaudeStatus, Config, Dims, + EnqueueResult, LiveSessionInfo, NotifyService, PermissionGate, @@ -41,7 +42,7 @@ import type { } from '../types.js'; import { WS_OPEN } from '../types.js'; import { serialize } from '../protocol.js'; -import { createSession, attachWs, broadcast, kill } from './session.js'; +import { createSession, attachWs, broadcast, kill, writeInput } from './session.js'; import { appendEvent, makeTimelineEvent } from './timeline.js'; import { hasSession, tmuxName } from './tmux.js'; @@ -194,6 +195,7 @@ export function createSessionManager( rows: s.pty.rows, telemetry: s.telemetry, // B2: latest telemetry for the thumbnail wall lastOutputAt: s.lastOutputAt, // T-iOS-37: unread watermark (M3 record) + queueLength: s.queue.length, // W2: pending inject-queue depth })) .sort((a, b) => b.createdAt - a.createdAt); } @@ -323,6 +325,57 @@ export function createSessionManager( return count; } + /** + * W2: append a verbatim byte string to a session's inject queue. + * + * The queue is a bounded FIFO (cfg.queueMaxItems). The HTTP route validates and + * size-caps `text` before calling; here we only enforce depth + existence. + * Immutable update: replace the frozen array wholesale, never mutate in place. + * Broadcasts the new depth so every mirrored device updates its badge. + */ + function enqueueFollowup(id: string, text: string): EnqueueResult { + const session = sessions.get(id); + if (session === undefined) return { ok: false, reason: 'unknown' }; + if (session.exitedAt !== null) return { ok: false, reason: 'exited' }; + if (session.queue.length >= cfg.queueMaxItems) return { ok: false, reason: 'full' }; + + session.queue = Object.freeze([...session.queue, text]); + const length = session.queue.length; + broadcast(session, { type: 'queue', length }); + return { ok: true, length }; + } + + /** + * W2: pop the head entry and write it to the PTY (byte-identical to a keystroke; + * writeInput broadcasts the resulting output to all mirrors). Returns the + * injected string, or null when there is nothing to drain / the session is + * unknown or already exited (double-guards L4 — writeInput is itself a no-op + * after exit). Broadcasts the new depth. + */ + function drainOne(id: string): string | null { + const session = sessions.get(id); + if (session === undefined || session.exitedAt !== null) return null; + if (session.queue.length === 0) return null; + + const head = session.queue[0] as string; + session.queue = Object.freeze(session.queue.slice(1)); + writeInput(session, head); + broadcast(session, { type: 'queue', length: session.queue.length }); + return head; + } + + /** + * W2: clear all pending entries (cancel-all escape hatch). Broadcasts depth 0. + * Returns false for an unknown session. + */ + function clearQueue(id: string): boolean { + const session = sessions.get(id); + if (session === undefined) return false; + session.queue = Object.freeze([] as string[]); + broadcast(session, { type: 'queue', length: 0 }); + return true; + } + /** * Called on SIGINT/SIGTERM/close. For non-tmux sessions, kill the PTY. For * tmux sessions (H1), kill only the client pty — the tmux server keeps the @@ -348,6 +401,9 @@ export function createSessionManager( handleStatusLine, sweepStuck, reapIdle, + enqueueFollowup, + drainOne, + clearQueue, shutdown, }; } diff --git a/src/session/session.ts b/src/session/session.ts index 228ed71..8e5371d 100644 --- a/src/session/session.ts +++ b/src/session/session.ts @@ -137,6 +137,8 @@ export function createSession( timeline: Object.freeze([] as TimelineEvent[]), stuckNotified: false, // A5: re-armed to false by each pty output telemetry: null, // B2: updated by manager.handleStatusLine + // W2: inject follow-up queue — empty at spawn; replaced wholesale by manager. + queue: Object.freeze([] as string[]), }; // onData: persist to scrollback, refresh liveness, broadcast to all clients. diff --git a/src/types.ts b/src/types.ts index 3e5cf03..bdb4c35 100644 --- a/src/types.ts +++ b/src/types.ts @@ -70,6 +70,11 @@ export interface Config { // B4 permission-mode relay readonly defaultPermissionMode: PermissionMode; // DEFAULT_PERMISSION_MODE, default 'default' readonly allowAutoMode: boolean; // ALLOW_AUTO_MODE, default false (SEC-M5) + // W2 server-side PTY-inject follow-up queue + readonly queueEnabled: boolean; // QUEUE_ENABLED, default true (routes 503 when off) + readonly queueMaxItems: number; // QUEUE_MAX_ITEMS, default 10 (bounded depth, DoS guard) + readonly queueItemMaxBytes: number; // QUEUE_ITEM_MAX_BYTES, default 4096 (bounded size) + readonly queueSettleMs: number; // QUEUE_SETTLE_MS, default 1500 (drain-after-idle delay) } /** process.env is structurally assignable to this; keeps the file free of `NodeJS.*`. */ @@ -136,7 +141,11 @@ export type ServerMessage = * ignore it (additive + optional). */ preview?: ApprovalPreview; } - | { type: 'telemetry'; telemetry: StatusTelemetry }; + | { type: 'telemetry'; telemetry: StatusTelemetry } + /** W2: the current pending-inject queue depth for this session. Broadcast to + * every attached device so all mirrors show the same "N queued" badge. Older + * clients ignore it (additive variant). */ + | { type: 'queue'; length: number }; /** parseClientMessage result — never throws; errors flow here (§5.3). */ export type ParseResult = @@ -247,6 +256,11 @@ export interface Session { stuckNotified: boolean; /** B2: latest statusLine telemetry for this session; null until first report. */ telemetry: StatusTelemetry | null; + /** W2: bounded FIFO of verbatim byte strings to inject when Claude next goes + * idle. Head fires first. Mutable runtime handle on the immutable meta (like + * timeline): replaced wholesale (Object.freeze of a new array), never mutated + * in place. Capped by cfg.queueMaxItems; each entry capped by queueItemMaxBytes. */ + queue: readonly string[]; readonly pty: IPty; } @@ -278,6 +292,9 @@ export interface LiveSessionInfo { * (lastOutputAt > local last-seen). Additive OPTIONAL: older consumers * that build or read LiveSessionInfo without it stay valid. */ readonly lastOutputAt?: number; + /** W2: number of pending inject-queue entries (0 when empty). Additive OPTIONAL + * so the manage grid and /live-sessions can surface queue depth per session. */ + readonly queueLength?: number; } /* ───────────────── project manager (v0.6, §4.3 FEATURE doc) ──────────────── */ @@ -330,6 +347,13 @@ export interface ProjectDetail { claudeMd?: string; // its content (truncated for display) when present } +/** W2: outcome of SessionManager.enqueueFollowup. Success carries the new depth; + * failure names the reason so the HTTP route can pick the right status code + * (unknown/exited → 404, full → 409). */ +export type EnqueueResult = + | { ok: true; length: number } + | { ok: false; reason: 'unknown' | 'full' | 'exited' }; + export interface SessionManager { handleAttach( ws: WebSocketLike, @@ -366,6 +390,18 @@ export interface SessionManager { sweepStuck(now: number): void; /** reclaim when now - max(detachedAt, lastOutputAt) > idleTtlMs (M3). Returns count. */ reapIdle(now: number): number; + /** W2: append verbatim `text` to a session's inject queue (bounded by + * queueMaxItems); broadcasts the new depth. The route validates/sizes `text` + * before calling. Never spawns; only mutates the queue handle. */ + enqueueFollowup(id: string, text: string): EnqueueResult; + /** W2: pop the head entry and write it to the PTY (byte-identical to a + * keystroke, broadcast to all mirrors); broadcasts the new depth. Returns the + * injected string, or null when the queue is empty / session is unknown or + * exited (double-guards L4). */ + drainOne(id: string): string | null; + /** W2: clear all pending entries (cancel-all escape hatch); broadcasts depth 0. + * Returns false for an unknown session. */ + clearQueue(id: string): boolean; shutdown(): void; } diff --git a/test/config.test.ts b/test/config.test.ts index cfe6202..21ce127 100644 --- a/test/config.test.ts +++ b/test/config.test.ts @@ -730,3 +730,40 @@ describe('loadConfig — v0.7 all new fields present in frozen result', () => { expect(cfg).toHaveProperty('allowAutoMode') }) }) + +// ── W2 inject-queue config ──────────────────────────────────────────────────── +describe('loadConfig — W2 inject queue', () => { + beforeEach(() => { + mockNetworkInterfaces.mockReturnValue({}) + mockHomedir.mockReturnValue('/home/testuser') + }) + + it('defaults: queueEnabled true, queueMaxItems 10, queueItemMaxBytes 4096, queueSettleMs 1500', () => { + const cfg = loadConfig({}) + expect(cfg.queueEnabled).toBe(true) + expect(cfg.queueMaxItems).toBe(10) + expect(cfg.queueItemMaxBytes).toBe(4096) + expect(cfg.queueSettleMs).toBe(1500) + }) + + it('reads env overrides', () => { + const cfg = loadConfig({ + QUEUE_ENABLED: '0', + QUEUE_MAX_ITEMS: '3', + QUEUE_ITEM_MAX_BYTES: '256', + QUEUE_SETTLE_MS: '500', + }) + expect(cfg.queueEnabled).toBe(false) + expect(cfg.queueMaxItems).toBe(3) + expect(cfg.queueItemMaxBytes).toBe(256) + expect(cfg.queueSettleMs).toBe(500) + }) + + it('throws for a negative QUEUE_MAX_ITEMS (fail-fast)', () => { + expect(() => loadConfig({ QUEUE_MAX_ITEMS: '-1' })).toThrow() + }) + + it('throws for a non-integer QUEUE_SETTLE_MS', () => { + expect(() => loadConfig({ QUEUE_SETTLE_MS: 'soon' })).toThrow() + }) +}) diff --git a/test/integration/queue.test.ts b/test/integration/queue.test.ts new file mode 100644 index 0000000..55ac1de --- /dev/null +++ b/test/integration/queue.test.ts @@ -0,0 +1,395 @@ +/** + * W2 — integration tests for the server-side PTY-inject queue routes and the + * idle-drain wiring, against a real startServer. + * + * Sandbox-runnable (no PTY): Origin/CSRF guard, id/text validation, size + depth + * caps, rate limit, QUEUE_ENABLED=0 → 503, unknown-session 404. + * + * itPty-gated (needs a real shell — auto-skips where posix_spawn is blocked): + * - happy path: POST queue → 200 {length:1}; GET /live-sessions shows + * queueLength; GET/DELETE .../queue. + * - idle-drain: enqueue → POST Stop → the marker echoes through the PTY. + * - one-per-idle pacing: two entries, one Stop drains only the first. + * - settle guard: a non-idle hook during the window cancels the drain. + */ + +import net from 'node:net' + +import { afterEach, describe, expect, it } from 'vitest' +import WebSocket from 'ws' +import * as nodePty from 'node-pty' + +import { loadConfig } from '../../src/config.js' +import { startServer } from '../../src/server.js' + +const PTY_AVAILABLE = (() => { + try { + const p = nodePty.spawn(process.env['SHELL'] ?? '/bin/sh', [], { cols: 80, rows: 24 }) + p.kill() + return true + } catch { + return false + } +})() +const itPty = PTY_AVAILABLE ? it : it.skip + +function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const srv = net.createServer() + srv.listen(0, '127.0.0.1', () => { + const addr = srv.address() + if (addr === null || typeof addr === 'string') { + srv.close() + reject(new Error('bad addr')) + return + } + const port = addr.port + srv.close(() => resolve(port)) + }) + srv.on('error', reject) + }) +} + +const handles: { close(): Promise }[] = [] + +async function spawnServer( + overrides: Record = {}, +): Promise<{ port: number; origin: string }> { + const port = await getFreePort() + const cfg = loadConfig({ + PORT: String(port), + BIND_HOST: '127.0.0.1', + SHELL_PATH: process.env['SHELL'] ?? '/bin/zsh', + ALLOWED_ORIGINS: `http://127.0.0.1:${port}`, + USE_TMUX: '0', + IDLE_TTL: '86400', + ...overrides, + }) + const handle = startServer(cfg) + handles.push(handle) + await new Promise((r) => setTimeout(r, 80)) + return { port, origin: `http://127.0.0.1:${port}` } +} + +function waitForOpen(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error('open timeout')), 3000) + ws.once('open', () => { + clearTimeout(t) + resolve() + }) + ws.once('error', (e) => { + clearTimeout(t) + reject(e) + }) + }) +} + +function waitForMessage( + ws: WebSocket, + pred: (m: Record) => boolean, + timeoutMs = 6000, +): Promise> { + return new Promise((resolve, reject) => { + const t = setTimeout(() => { + ws.off('message', onMsg) + reject(new Error('message timeout')) + }, timeoutMs) + const onMsg = (raw: WebSocket.RawData): void => { + let m: Record + try { + m = JSON.parse(raw.toString('utf8')) as Record + } catch { + return + } + if (pred(m)) { + clearTimeout(t) + ws.off('message', onMsg) + resolve(m) + } + } + ws.on('message', onMsg) + }) +} + +/** Resolve once no WS message has arrived for `quietMs` (the shell prompt settled). */ +function waitForQuiet(ws: WebSocket, quietMs = 300, maxMs = 4000): Promise { + return new Promise((resolve) => { + let quietTimer: ReturnType + const done = (): void => { + clearTimeout(quietTimer) + clearTimeout(hardStop) + ws.off('message', onMsg) + resolve() + } + const arm = (): void => { + clearTimeout(quietTimer) + quietTimer = setTimeout(done, quietMs) + } + const onMsg = (): void => arm() + const hardStop = setTimeout(done, maxMs) + ws.on('message', onMsg) + arm() + }) +} + +/** Collect every output-frame's data string for `windowMs`. */ +function collectOutput(ws: WebSocket, windowMs: number): Promise { + return new Promise((resolve) => { + let acc = '' + const onMsg = (raw: WebSocket.RawData): void => { + try { + const m = JSON.parse(raw.toString('utf8')) as Record + if (m['type'] === 'output') acc += String(m['data']) + } catch { + /* ignore */ + } + } + ws.on('message', onMsg) + setTimeout(() => { + ws.off('message', onMsg) + resolve(acc) + }, windowMs) + }) +} + +const isAttached = (m: Record): boolean => m['type'] === 'attached' + +async function attachSession(port: number, origin: string): Promise<{ ws: WebSocket; sessionId: string }> { + const ws = new WebSocket(`ws://127.0.0.1:${port}/term`, { headers: { Origin: origin } }) + await waitForOpen(ws) + ws.send(JSON.stringify({ type: 'attach', sessionId: null })) + const attached = await waitForMessage(ws, isAttached) + return { ws, sessionId: attached['sessionId'] as string } +} + +function postQueue( + port: number, + id: string, + body: unknown, + headers: Record = {}, +): Promise { + return fetch(`http://127.0.0.1:${port}/live-sessions/${id}/queue`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...headers }, + body: JSON.stringify(body), + }) +} + +function postStop(port: number, sessionId: string): Promise { + return fetch(`http://127.0.0.1:${port}/hook`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Webterm-Session': sessionId }, + body: JSON.stringify({ hook_event_name: 'Stop' }), + }) +} + +afterEach(async () => { + while (handles.length > 0) await handles.pop()?.close() + await new Promise((r) => setTimeout(r, 30)) +}) + +const UNKNOWN = '00000000-0000-4000-8000-000000000000' + +// ── Route guards + validation (no PTY needed) ───────────────────────────────── + +describe('POST /live-sessions/:id/queue — guards & validation', () => { + it('rejects a foreign Origin with 403 (CSRF)', async () => { + const { port } = await spawnServer() + const res = await postQueue(port, UNKNOWN, { text: 'x' }, { Origin: 'http://evil.example' }) + expect(res.status).toBe(403) + }) + + it('rejects a missing Origin with 403 (default-deny)', async () => { + const { port } = await spawnServer() + const res = await postQueue(port, UNKNOWN, { text: 'x' }) // no Origin header + expect(res.status).toBe(403) + }) + + it('rejects a malformed session id with 400', async () => { + const { port, origin } = await spawnServer() + const res = await postQueue(port, 'not-a-uuid', { text: 'x' }, { Origin: origin }) + expect(res.status).toBe(400) + }) + + it('rejects empty / non-string text with 400', async () => { + const { port, origin } = await spawnServer() + expect((await postQueue(port, UNKNOWN, { text: '' }, { Origin: origin })).status).toBe(400) + expect((await postQueue(port, UNKNOWN, { text: 42 }, { Origin: origin })).status).toBe(400) + expect((await postQueue(port, UNKNOWN, {}, { Origin: origin })).status).toBe(400) + }) + + it('rejects oversized text with 413', async () => { + const { port, origin } = await spawnServer({ QUEUE_ITEM_MAX_BYTES: '16' }) + const res = await postQueue(port, UNKNOWN, { text: 'x'.repeat(17) }, { Origin: origin }) + expect(res.status).toBe(413) + }) + + it('returns 404 for a well-formed but unknown session id', async () => { + const { port, origin } = await spawnServer() + const res = await postQueue(port, UNKNOWN, { text: 'x' }, { Origin: origin }) + expect(res.status).toBe(404) + }) + + it('returns 503 when QUEUE_ENABLED=0', async () => { + const { port, origin } = await spawnServer({ QUEUE_ENABLED: '0' }) + const res = await postQueue(port, UNKNOWN, { text: 'x' }, { Origin: origin }) + expect(res.status).toBe(503) + }) + + it('rate-limits after QUEUE_RATE_MAX requests with 429', async () => { + const { port, origin } = await spawnServer() + // 20 allowed (each 400 on empty text — rate is checked before validation), + // the 21st exceeds the per-IP window and returns 429. + let sawRate = false + for (let i = 0; i < 21; i++) { + const res = await postQueue(port, UNKNOWN, { text: '' }, { Origin: origin }) + if (res.status === 429) sawRate = true + } + expect(sawRate).toBe(true) + }) +}) + +describe('GET/DELETE /live-sessions/:id/queue — guards', () => { + it('GET returns 404 for an unknown session', async () => { + const { port } = await spawnServer() + const res = await fetch(`http://127.0.0.1:${port}/live-sessions/${UNKNOWN}/queue`) + expect(res.status).toBe(404) + }) + + it('DELETE rejects a foreign Origin with 403', async () => { + const { port } = await spawnServer() + const res = await fetch(`http://127.0.0.1:${port}/live-sessions/${UNKNOWN}/queue`, { + method: 'DELETE', + headers: { Origin: 'http://evil.example' }, + }) + expect(res.status).toBe(403) + }) + + it('DELETE returns 404 for an unknown session', async () => { + const { port, origin } = await spawnServer() + const res = await fetch(`http://127.0.0.1:${port}/live-sessions/${UNKNOWN}/queue`, { + method: 'DELETE', + headers: { Origin: origin }, + }) + expect(res.status).toBe(404) + }) +}) + +// ── Happy path + drain (real PTY) ───────────────────────────────────────────── + +describe('inject queue — real session', () => { + itPty('enqueues, surfaces queueLength, lists items, and clears', async () => { + const { port, origin } = await spawnServer() + const { ws, sessionId } = await attachSession(port, origin) + await waitForQuiet(ws) + + const enq = await postQueue(port, sessionId, { text: 'echo hi', appendEnter: true }, { Origin: origin }) + expect(enq.status).toBe(200) + expect(await enq.json()).toEqual({ length: 1 }) + + // /live-sessions surfaces the depth. + const live = (await (await fetch(`http://127.0.0.1:${port}/live-sessions`)).json()) as Array<{ + id: string + queueLength?: number + }> + expect(live.find((s) => s.id === sessionId)?.queueLength).toBe(1) + + // GET .../queue returns the stored (verbatim) items. + const view = (await ( + await fetch(`http://127.0.0.1:${port}/live-sessions/${sessionId}/queue`) + ).json()) as { length: number; items: string[] } + expect(view.length).toBe(1) + expect(view.items).toEqual(['echo hi\r']) + + // DELETE clears the queue. + const del = await fetch(`http://127.0.0.1:${port}/live-sessions/${sessionId}/queue`, { + method: 'DELETE', + headers: { Origin: origin }, + }) + expect(del.status).toBe(200) + const view2 = (await ( + await fetch(`http://127.0.0.1:${port}/live-sessions/${sessionId}/queue`) + ).json()) as { length: number } + expect(view2.length).toBe(0) + + ws.close() + }) + + itPty('drains the head into the PTY after a Stop hook (idle-drain)', async () => { + const { port, origin } = await spawnServer({ QUEUE_SETTLE_MS: '50' }) + const { ws, sessionId } = await attachSession(port, origin) + await waitForQuiet(ws) + + const enq = await postQueue( + port, + sessionId, + { text: 'echo QUEUED_MARKER_A', appendEnter: true }, + { Origin: origin }, + ) + expect(enq.status).toBe(200) + + const marker = waitForMessage( + ws, + (m) => m['type'] === 'output' && String(m['data']).includes('QUEUED_MARKER_A'), + ) + const stop = await postStop(port, sessionId) + expect(stop.status).toBe(204) + + await marker // the shell echoed the injected keystrokes → marker present + ws.close() + }) + + itPty('fires only one entry per idle (Stop) — the next waits for the next Stop', async () => { + const { port, origin } = await spawnServer({ QUEUE_SETTLE_MS: '50' }) + const { ws, sessionId } = await attachSession(port, origin) + await waitForQuiet(ws) + + await postQueue(port, sessionId, { text: 'echo MARK_ONE', appendEnter: true }, { Origin: origin }) + await postQueue(port, sessionId, { text: 'echo MARK_TWO', appendEnter: true }, { Origin: origin }) + + // First Stop → only MARK_ONE drains. + const one = waitForMessage(ws, (m) => m['type'] === 'output' && String(m['data']).includes('MARK_ONE')) + await postStop(port, sessionId) + await one + + // Without another Stop, MARK_TWO must NOT be injected. + const idleWindow = await collectOutput(ws, 500) + expect(idleWindow.includes('MARK_TWO')).toBe(false) + + // A second Stop drains MARK_TWO. + await waitForQuiet(ws) + const two = waitForMessage(ws, (m) => m['type'] === 'output' && String(m['data']).includes('MARK_TWO')) + await postStop(port, sessionId) + await two + + ws.close() + }) + + itPty('settle guard: a non-idle hook during the window cancels the drain', async () => { + const { port, origin } = await spawnServer({ QUEUE_SETTLE_MS: '300' }) + const { ws, sessionId } = await attachSession(port, origin) + await waitForQuiet(ws) + + await postQueue( + port, + sessionId, + { text: 'echo SHOULD_NOT_RUN', appendEnter: true }, + { Origin: origin }, + ) + + // Stop schedules a drain at +300ms; a PreToolUse immediately flips the status + // back to 'working', so the settle-time idle re-check skips the drain. + await postStop(port, sessionId) + await fetch(`http://127.0.0.1:${port}/hook`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Webterm-Session': sessionId }, + body: JSON.stringify({ hook_event_name: 'PreToolUse', tool_name: 'Bash' }), + }) + + const window = await collectOutput(ws, 600) // past the 300ms settle + expect(window.includes('SHOULD_NOT_RUN')).toBe(false) + + ws.close() + }) +}) diff --git a/test/manager.test.ts b/test/manager.test.ts index f167782..021c126 100644 --- a/test/manager.test.ts +++ b/test/manager.test.ts @@ -88,6 +88,11 @@ const CFG: Config = { worktreeTimeoutMs: 10_000, defaultPermissionMode: 'default', allowAutoMode: false, + // W2 inject queue + queueEnabled: true, + queueMaxItems: 10, + queueItemMaxBytes: 4096, + queueSettleMs: 1500, }; const DIMS: Dims = { cols: 80, rows: 24 }; @@ -1145,3 +1150,137 @@ describe('list — lastOutputAt field (T-iOS-37)', () => { }); }); }); + +// ── W2: inject-queue methods (enqueueFollowup / drainOne / clearQueue) ──────── +describe('W2 inject queue — enqueueFollowup', () => { + it('appends and returns {ok,length}, broadcasting {type:queue,length} to clients', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const ws = createMockWs(); + const s = mgr.handleAttach(ws, null, DIMS, 1_000); + ws.sent.length = 0; + + const r1 = mgr.enqueueFollowup(s.meta.id, 'echo one\r'); + expect(r1).toEqual({ ok: true, length: 1 }); + expect(s.queue).toEqual(['echo one\r']); + expect(parseSent(ws).some((m) => m.type === 'queue' && m.length === 1)).toBe(true); + + const r2 = mgr.enqueueFollowup(s.meta.id, 'echo two\r'); + expect(r2).toEqual({ ok: true, length: 2 }); + expect(s.queue).toEqual(['echo one\r', 'echo two\r']); + }); + + it('caps at queueMaxItems → {ok:false,reason:full}, no broadcast, queue unchanged', () => { + const mgr = createSessionManager({ ...CFG, queueMaxItems: 2 }); + nextPty = createMockPty(); + const ws = createMockWs(); + const s = mgr.handleAttach(ws, null, DIMS, 1_000); + mgr.enqueueFollowup(s.meta.id, 'a'); + mgr.enqueueFollowup(s.meta.id, 'b'); + ws.sent.length = 0; + + const r = mgr.enqueueFollowup(s.meta.id, 'c'); + expect(r).toEqual({ ok: false, reason: 'full' }); + expect(s.queue).toEqual(['a', 'b']); // immutable: unchanged + expect(parseSent(ws).some((m) => m.type === 'queue')).toBe(false); // no broadcast + }); + + it('unknown id → {ok:false,reason:unknown}', () => { + const mgr = createSessionManager(CFG); + expect(mgr.enqueueFollowup('11111111-1111-4111-8111-111111111111', 'x')).toEqual({ + ok: false, + reason: 'unknown', + }); + }); + + it('exited session → {ok:false,reason:exited}', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const s = mgr.handleAttach(createMockWs(), null, DIMS, 1_000); + s.exitedAt = 2_000; + expect(mgr.enqueueFollowup(s.meta.id, 'x')).toEqual({ ok: false, reason: 'exited' }); + }); +}); + +describe('W2 inject queue — drainOne', () => { + it('pops the head, writes it to the PTY, shrinks the queue, and broadcasts depth', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const ws = createMockWs(); + const s = mgr.handleAttach(ws, null, DIMS, 1_000); + mgr.enqueueFollowup(s.meta.id, 'HEAD\r'); + mgr.enqueueFollowup(s.meta.id, 'TAIL\r'); + ws.sent.length = 0; + const pty = s.pty as MockIPty; + pty.writes.length = 0; + + const drained = mgr.drainOne(s.meta.id); + expect(drained).toBe('HEAD\r'); + expect(pty.writes).toEqual(['HEAD\r']); // byte-identical inject + expect(s.queue).toEqual(['TAIL\r']); // length now 1 + expect(parseSent(ws).some((m) => m.type === 'queue' && m.length === 1)).toBe(true); + }); + + it('empty queue → null, no PTY write', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const s = mgr.handleAttach(createMockWs(), null, DIMS, 1_000); + const pty = s.pty as MockIPty; + pty.writes.length = 0; + + expect(mgr.drainOne(s.meta.id)).toBeNull(); + expect(pty.writes).toEqual([]); + }); + + it('exited session with items queued → null (double-guards L4), no write', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const s = mgr.handleAttach(createMockWs(), null, DIMS, 1_000); + mgr.enqueueFollowup(s.meta.id, 'x'); + s.exitedAt = 2_000; + const pty = s.pty as MockIPty; + pty.writes.length = 0; + + expect(mgr.drainOne(s.meta.id)).toBeNull(); + expect(pty.writes).toEqual([]); + }); + + it('unknown id → null', () => { + const mgr = createSessionManager(CFG); + expect(mgr.drainOne('11111111-1111-4111-8111-111111111111')).toBeNull(); + }); +}); + +describe('W2 inject queue — clearQueue', () => { + it('empties the queue, broadcasts length 0, returns true', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const ws = createMockWs(); + const s = mgr.handleAttach(ws, null, DIMS, 1_000); + mgr.enqueueFollowup(s.meta.id, 'a'); + mgr.enqueueFollowup(s.meta.id, 'b'); + ws.sent.length = 0; + + expect(mgr.clearQueue(s.meta.id)).toBe(true); + expect(s.queue).toEqual([]); + expect(parseSent(ws).some((m) => m.type === 'queue' && m.length === 0)).toBe(true); + }); + + it('unknown id → false', () => { + const mgr = createSessionManager(CFG); + expect(mgr.clearQueue('11111111-1111-4111-8111-111111111111')).toBe(false); + }); +}); + +describe('W2 inject queue — list() surfaces queueLength', () => { + it('reports the pending depth per session', () => { + const mgr = createSessionManager(CFG); + nextPty = createMockPty(); + const s = mgr.handleAttach(createMockWs(), null, DIMS, 1_000); + mgr.enqueueFollowup(s.meta.id, 'a'); + mgr.enqueueFollowup(s.meta.id, 'b'); + + const entry = mgr.list().find((e) => e.id === s.meta.id); + expect(entry?.queueLength).toBe(2); + }); +}); diff --git a/test/queue.test.ts b/test/queue.test.ts new file mode 100644 index 0000000..88ff2e9 --- /dev/null +++ b/test/queue.test.ts @@ -0,0 +1,97 @@ +// @vitest-environment jsdom +/** + * test/queue.test.ts — W2 frontend inject-queue helper (public/queue.ts). + * + * fetch is mocked; every helper must NEVER throw (mirrors quick-reply's + * never-throw discipline). Covers: POST url/body + parsed result, non-2xx → + * {ok:false,status}, network reject → {ok:false}, DELETE, and the frame parser. + */ + +import { describe, it, expect, vi, afterEach } from 'vitest' + +import { enqueueFollowup, clearQueue, queueLengthFromFrame } from '../public/queue.js' + +afterEach(() => { + vi.restoreAllMocks() + // @ts-expect-error — reset the mocked global between tests + delete globalThis.fetch +}) + +function mockFetch(impl: (url: string, init?: RequestInit) => Promise | Response): void { + globalThis.fetch = vi.fn(impl as unknown as typeof fetch) +} + +describe('enqueueFollowup', () => { + it('POSTs to the session queue URL with a JSON {text,appendEnter} body', async () => { + const calls: Array<{ url: string; init?: RequestInit }> = [] + mockFetch((url, init) => { + calls.push({ url, init }) + return new Response(JSON.stringify({ length: 1 }), { status: 200 }) + }) + + const result = await enqueueFollowup('sess-1', 'run tests', true) + + expect(result).toEqual({ ok: true, length: 1 }) + expect(calls).toHaveLength(1) + expect(calls[0]!.url).toBe('/live-sessions/sess-1/queue') + expect(calls[0]!.init?.method).toBe('POST') + expect(JSON.parse(String(calls[0]!.init?.body))).toEqual({ + text: 'run tests', + appendEnter: true, + }) + }) + + it('returns {ok:false,status} on a non-2xx response and never throws', async () => { + mockFetch(() => new Response('nope', { status: 409 })) + const result = await enqueueFollowup('sess-1', 'x', false) + expect(result).toEqual({ ok: false, status: 409 }) + }) + + it('returns {ok:false} on a network reject and never throws', async () => { + mockFetch(() => Promise.reject(new Error('offline'))) + const result = await enqueueFollowup('sess-1', 'x', false) + expect(result).toEqual({ ok: false }) + }) + + it('returns {ok:true} with undefined length when the body omits length', async () => { + mockFetch(() => new Response(JSON.stringify({}), { status: 200 })) + const result = await enqueueFollowup('sess-1', 'x', false) + expect(result).toEqual({ ok: true, length: undefined }) + }) +}) + +describe('clearQueue', () => { + it('DELETEs the session queue URL and returns true on success', async () => { + const calls: Array<{ url: string; init?: RequestInit }> = [] + mockFetch((url, init) => { + calls.push({ url, init }) + return new Response(JSON.stringify({ length: 0 }), { status: 200 }) + }) + + const ok = await clearQueue('sess-2') + expect(ok).toBe(true) + expect(calls[0]!.url).toBe('/live-sessions/sess-2/queue') + expect(calls[0]!.init?.method).toBe('DELETE') + }) + + it('returns false on a non-2xx and on a network reject (never throws)', async () => { + mockFetch(() => new Response('', { status: 404 })) + expect(await clearQueue('sess-2')).toBe(false) + + mockFetch(() => Promise.reject(new Error('offline'))) + expect(await clearQueue('sess-2')).toBe(false) + }) +}) + +describe('queueLengthFromFrame', () => { + it('returns the length for a {type:queue,length} frame', () => { + expect(queueLengthFromFrame({ type: 'queue', length: 3 })).toBe(3) + }) + + it('returns null for unrelated or malformed frames', () => { + expect(queueLengthFromFrame({ type: 'status', status: 'idle' })).toBeNull() + expect(queueLengthFromFrame({ type: 'queue' })).toBeNull() + expect(queueLengthFromFrame(null)).toBeNull() + expect(queueLengthFromFrame('nope')).toBeNull() + }) +}) diff --git a/test/quick-reply.test.ts b/test/quick-reply.test.ts index 4bc132c..2b454da 100644 --- a/test/quick-reply.test.ts +++ b/test/quick-reply.test.ts @@ -549,4 +549,45 @@ describe('mountQuickReply — DOM', () => { expect(countAfter).toBe(countBefore) handle.dispose() }) + + // ── W2: the editor's "Queue" button (only shown when onQueue is wired) ──────── + it('shows no Queue button when onQueue is not provided', () => { + const container = document.createElement('div') + const handle = mountQuickReply(container, { onSend: vi.fn() }) + ;(container.querySelector('.qr-add-btn') as HTMLButtonElement).click() + expect(container.querySelector('.qr-editor-queue')).toBeNull() + handle.dispose() + }) + + it('Queue button calls onQueue with the typed text + appendEnter and closes the editor', () => { + const container = document.createElement('div') + const onQueue = vi.fn() + const handle = mountQuickReply(container, { onSend: vi.fn(), onQueue }) + ;(container.querySelector('.qr-add-btn') as HTMLButtonElement).click() + + const textInput = container.querySelector('.qr-editor-text') as HTMLInputElement + textInput.value = 'run tests then open a PR' + const enterCheck = container.querySelector('.qr-editor-enter') as HTMLInputElement + enterCheck.checked = true + + ;(container.querySelector('.qr-editor-queue') as HTMLButtonElement).click() + + expect(onQueue).toHaveBeenCalledWith('run tests then open a PR', true) + // Enqueue must NOT persist a chip nor send immediately. + expect(loadPalette().some((c) => c.text === 'run tests then open a PR')).toBe(false) + expect(container.querySelector('.qr-editor')).toBeNull() + handle.dispose() + }) + + it('Queue button with empty text is a no-op (onQueue not called, editor stays)', () => { + const container = document.createElement('div') + const onQueue = vi.fn() + const handle = mountQuickReply(container, { onSend: vi.fn(), onQueue }) + ;(container.querySelector('.qr-add-btn') as HTMLButtonElement).click() + ;(container.querySelector('.qr-editor-queue') as HTMLButtonElement).click() + + expect(onQueue).not.toHaveBeenCalled() + expect(container.querySelector('.qr-editor')).not.toBeNull() + handle.dispose() + }) }) diff --git a/test/tabs.test.ts b/test/tabs.test.ts index dba92db..ddd2946 100644 --- a/test/tabs.test.ts +++ b/test/tabs.test.ts @@ -29,6 +29,8 @@ class FakeTerminalSession { pendingEpoch = 0 /** B2: latest statusLine telemetry (single source of truth). */ telemetry: unknown = null + /** W2: pending inject-queue depth (tests set it before driving onQueue). */ + queueLength = 0 /** Captured so tests can assert initialInput ends with \r (Finding 1). */ initialInput: string | undefined = undefined connect = vi.fn() @@ -50,6 +52,7 @@ class FakeTerminalSession { onActivity?: () => void onTitle?: (t: string) => void onTelemetry?: (t: unknown) => void + onQueue?: (n: number) => void } static instances: FakeTerminalSession[] = [] constructor(opts: { @@ -60,6 +63,7 @@ class FakeTerminalSession { onActivity?: () => void onTitle?: (t: string) => void onTelemetry?: (t: unknown) => void + onQueue?: (n: number) => void [key: string]: unknown }) { constructed += 1 @@ -89,13 +93,28 @@ vi.mock('../public/cell-monitor.js', () => ({ mountCellMonitor })) const mountPushToggle = vi.fn() vi.mock('../public/push.js', () => ({ mountPushToggle })) -const quickReply: { onSend?: (d: string) => void } = {} -const mountQuickReply = vi.fn((_c: HTMLElement, opts: { onSend: (d: string) => void }) => { - quickReply.onSend = opts.onSend - return { dispose: vi.fn() } -}) +const quickReply: { + onSend?: (d: string) => void + onQueue?: (text: string, appendEnter: boolean) => void +} = {} +const mountQuickReply = vi.fn( + ( + _c: HTMLElement, + opts: { onSend: (d: string) => void; onQueue?: (text: string, appendEnter: boolean) => void }, + ) => { + quickReply.onSend = opts.onSend + quickReply.onQueue = opts.onQueue + return { dispose: vi.fn() } + }, +) vi.mock('../public/quick-reply.js', () => ({ mountQuickReply })) +// ── Mock the W2 queue helper (assert enqueue routing without real fetch) ────── +const enqueueFollowupMock = vi.fn(async () => ({ ok: true, length: 1 })) +vi.mock('../public/queue.js', () => ({ + enqueueFollowup: (...args: unknown[]) => enqueueFollowupMock(...(args as [])), +})) + const timelineDispose = vi.fn() const mountTimeline = vi.fn(() => ({ dispose: timelineDispose })) vi.mock('../public/timeline.js', () => ({ mountTimeline })) @@ -179,6 +198,8 @@ beforeEach(() => { voiceStart.mockClear() voiceStop.mockClear() quickReply.onSend = undefined + quickReply.onQueue = undefined + enqueueFollowupMock.mockClear() voice.onTranscript = undefined voice.onInterim = undefined // Default: server forbids 'auto' (loadUiConfig resolves to allowAutoMode=false). @@ -1636,3 +1657,64 @@ describe('TabApp — split-grid watch board (v1)', () => { expect(mountCellMonitor.mock.lastCall?.[1]).toBe('sess-1') }) }) + +// ── W2: inject-queue badge + enqueue affordance ────────────────────────────── +describe('TabApp — W2 inject queue', () => { + it('a queue frame (onQueue) updates the active tab’s ⧗N badge in place', () => { + const { paneHost, tabBar } = makeHosts() + const app = new TabApp(paneHost, tabBar) + app.newTab() + const inst = FakeTerminalSession.instances[0]! + + // No queued items → empty badge. + expect((tabBar.querySelector('.tab-queue') as HTMLElement).textContent).toBe('') + + // Server broadcasts depth 3 → badge shows ⧗3. + inst.queueLength = 3 + inst.cbs.onQueue?.(3) + expect((tabBar.querySelector('.tab-queue') as HTMLElement).textContent).toBe('⧗3') + + // Drains back to 0 → badge clears. + inst.queueLength = 0 + inst.cbs.onQueue?.(0) + expect((tabBar.querySelector('.tab-queue') as HTMLElement).textContent).toBe('') + }) + + it('enqueueToActive routes to enqueueFollowup with the active session id', () => { + const { paneHost, tabBar } = makeHosts() + const app = new TabApp(paneHost, tabBar) + app.openSession('11111111-1111-4111-8111-111111111111') + + app.enqueueToActive('run tests', true) + + expect(enqueueFollowupMock).toHaveBeenCalledWith( + '11111111-1111-4111-8111-111111111111', + 'run tests', + true, + ) + }) + + it('the quick-reply Queue affordance enqueues for the active session', () => { + const { paneHost, tabBar } = makeHosts() + const app = new TabApp(paneHost, tabBar) + app.openSession('22222222-2222-4222-8222-222222222222') + + // The onQueue handler wired into mountQuickReply routes through enqueueToActive. + quickReply.onQueue?.('open a PR', true) + + expect(enqueueFollowupMock).toHaveBeenCalledWith( + '22222222-2222-4222-8222-222222222222', + 'open a PR', + true, + ) + }) + + it('enqueueToActive is a no-op when no session has attached (id null)', () => { + const { paneHost, tabBar } = makeHosts() + const app = new TabApp(paneHost, tabBar) + app.newTab() // fresh session → id null until server assigns + + app.enqueueToActive('x', false) + expect(enqueueFollowupMock).not.toHaveBeenCalled() + }) +}) diff --git a/vitest.config.ts b/vitest.config.ts index 37d23f8..804a0f9 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -24,6 +24,7 @@ export default defineConfig({ 'src/**/*.ts', 'public/terminal-session.ts', 'public/link-paths.ts', + 'public/queue.ts', 'public/tabs.ts', 'public/grid-layout.ts', 'public/grid-presets.ts',