diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c1f8d0d5..edcc4df8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -75,5 +75,8 @@ jobs: npm run i18n:extract || true git diff --exit-code src/locales/en.json || (echo "::error::New i18n keys found — run 'npm run i18n:sync' and commit the result" && exit 1) + - name: Test frontend + run: npm test + - name: Build frontend run: npm run build diff --git a/frontend-svelte/package.json b/frontend-svelte/package.json index f02b61a2..36725922 100644 --- a/frontend-svelte/package.json +++ b/frontend-svelte/package.json @@ -6,6 +6,7 @@ "scripts": { "dev": "vite", "build": "vite build", + "test": "node --test src/**/*.test.js", "preview": "vite preview", "i18n:extract": "node scripts/i18n-extract.js" }, diff --git a/frontend-svelte/src/components/TmuxPaneModal.svelte b/frontend-svelte/src/components/TmuxPaneModal.svelte index ffc973ab..2e429812 100644 --- a/frontend-svelte/src/components/TmuxPaneModal.svelte +++ b/frontend-svelte/src/components/TmuxPaneModal.svelte @@ -11,8 +11,8 @@ passthrough — xterm onData chunks are mapped to the backend's `/tmux/pane/keys` endpoint (named keys for control sequences, literal text otherwise), so an operator can answer first-run dialogs or interrupt a - wedged REPL without SSH + `tmux attach`. Sends are serialized through a - promise chain to preserve keystroke order. + wedged REPL without SSH + `tmux attach`. Sends start immediately and repeat + unacknowledged events; the daemon orders and de-duplicates them by sequence. Props: show: bool — modal open state (bind from parent) @@ -25,6 +25,7 @@ import { onDestroy } from 'svelte'; import Modal from './Modal.svelte'; import { api, sse } from '../lib/api.js'; + import { createPaneInputSender } from '../lib/paneInputSender.js'; export let show = false; export let agent = ''; @@ -55,9 +56,7 @@ let inputEnabled = false; let inputError = ''; let onDataDisposable = null; - // Serialize sends so fast typing can't reorder keystrokes (fetches - // racing each other would scramble e.g. "ls" into "sl"). - let sendChain = Promise.resolve(); + let paneInputSender = null; // xterm onData control sequences → backend named keys (tmux names). const KEY_SEQUENCES = { @@ -81,22 +80,8 @@ '\x1b[6~': 'NPage', }; - // Bumped on teardown (close/switch) so queued keystrokes from a previous - // pane are dropped instead of landing on the newly-selected agent/label. - let sendEpoch = 0; function queueSend(payload) { - // Snapshot the target at enqueue time — a queued link of the chain must - // hit the pane the keystroke was typed for, not the live agent/label. - const targetAgent = agent; - const targetLabel = label; - const epoch = sendEpoch; - sendChain = sendChain - .then(() => { - if (epoch !== sendEpoch) return; // pane torn down/switched — drop - return api('POST', `/agents/${encodeURIComponent(targetAgent)}/tmux/pane/keys`, { ...payload, label: targetLabel }); - }) - .then(() => { if (epoch === sendEpoch && inputError) inputError = ''; }) - .catch((e) => { if (epoch === sendEpoch) inputError = `send failed: ${e?.message || e}`; }); + paneInputSender?.enqueue(payload); } function handleTerminalData(data) { @@ -222,6 +207,20 @@ async function mount() { statusMessage = 'Loading terminal…'; + // Capture this mount's target. Requests initiated before a close/switch + // must finish against the old pane, never the newly selected one. + const targetAgent = agent; + const targetLabel = label; + paneInputSender = createPaneInputSender({ + send: (body, options) => api( + 'POST', + `/agents/${encodeURIComponent(targetAgent)}/tmux/pane/keys`, + { ...body, label: targetLabel }, + options, + ), + onSuccess: () => { if (inputError) inputError = ''; }, + onError: (e) => { inputError = `send failed: ${e?.message || e}`; }, + }); try { const [{ Terminal }, { FitAddon }] = await Promise.all([ import('@xterm/xterm'), @@ -294,6 +293,7 @@ if (sseSource) { try { sseSource.close(); } catch {} sseSource = null; } if (resizeObserver) { try { resizeObserver.disconnect(); } catch {} resizeObserver = null; } if (onDataDisposable) { try { onDataDisposable.dispose(); } catch {} onDataDisposable = null; } + if (paneInputSender) { paneInputSender.dispose(); paneInputSender = null; } if (terminal) { try { terminal.dispose(); } catch {} terminal = null; } fitAddon = null; statusMessage = ''; @@ -302,8 +302,6 @@ lastReqRows = 0; inputEnabled = false; inputError = ''; - sendEpoch++; // invalidate any keystrokes still queued for the old pane - sendChain = Promise.resolve(); } onDestroy(teardown); diff --git a/frontend-svelte/src/lib/api.js b/frontend-svelte/src/lib/api.js index 98beef1a..622f81d5 100644 --- a/frontend-svelte/src/lib/api.js +++ b/frontend-svelte/src/lib/api.js @@ -6,11 +6,12 @@ function authRedirectTarget(payload) { return `${path}?next=${encodeURIComponent(next || '/')}`; } -export async function api(method, path, body) { +export async function api(method, path, body, { keepalive = false } = {}) { const opts = { method, headers: { 'Content-Type': 'application/json' }, credentials: 'same-origin', + keepalive, }; if (body) opts.body = JSON.stringify(body); const resp = await fetch(`${API}${path}`, opts); diff --git a/frontend-svelte/src/lib/paneInputSender.js b/frontend-svelte/src/lib/paneInputSender.js new file mode 100644 index 00000000..06692e69 --- /dev/null +++ b/frontend-svelte/src/lib/paneInputSender.js @@ -0,0 +1,80 @@ +const SUBMIT_KEY = 'Enter'; + +function newClientId() { + if (globalThis.crypto?.randomUUID) return globalThis.crypto.randomUUID(); + return `pane-${Date.now()}-${Math.random().toString(16).slice(2)}`; +} + +/** + * Deliver pane input immediately while preserving order at the daemon. + * + * Every request repeats all events that have not yet been acknowledged. This + * makes a later request (especially Enter) a cumulative retry of any slower + * text requests ahead of it. The daemon de-duplicates events by client/seq. + * Enter requests use fetch keepalive so a submission already initiated by the + * keypress may finish after the dashboard tab closes. + */ +export function createPaneInputSender({ send, onError = () => {}, onSuccess = () => {} }) { + const clientId = newClientId(); + let nextSeq = 0; + let ackedSeq = 0; + let pending = []; + let uiActive = true; + + function acceptAck(result) { + const ack = Number(result?.acked_seq || 0); + if (Number.isInteger(ack) && ack > ackedSeq) { + ackedSeq = ack; + pending = pending.filter((event) => event.seq > ackedSeq); + } + // A stale prefix acknowledgement must not clear a failure for a newer + // event (notably an exhausted Enter) that is still pending. + if (uiActive && pending.length === 0) onSuccess(); + } + + function transmit(events, keepalive, retriesLeft = 0) { + const requestMaxSeq = events.at(-1)?.seq || 0; + let request; + try { + // Call send synchronously: Enter must start its keepalive request in + // the key event handler, not from a promise queued behind text. + request = send({ client_id: clientId, events }, { keepalive }); + } catch (error) { + request = Promise.reject(error); + } + return Promise.resolve(request) + .then(acceptAck) + .catch((error) => { + if (retriesLeft > 0) { + const retryEvents = pending.map((event) => ({ ...event })); + if (retryEvents.length) return transmit(retryEvents, true, retriesLeft - 1); + } + // A newer cumulative request may already have applied every + // event in this failed request. Its late rejection is stale, + // not a current send failure. + if (requestMaxSeq <= ackedSeq) return undefined; + if (uiActive) onError(error); + return undefined; + }); + } + + function enqueue(payload) { + const event = { + seq: ++nextSeq, + text: payload.text || '', + key: payload.key || '', + }; + pending.push(event); + const events = pending.map((item) => ({ ...item })); + const submitting = event.key === SUBMIT_KEY; + void transmit(events, submitting, submitting ? 2 : 0); + } + + function dispose() { + // Requests were started synchronously and target the pane captured by + // the component's send callback. Let them finish; only silence stale UI. + uiActive = false; + } + + return { enqueue, dispose }; +} diff --git a/frontend-svelte/src/lib/paneInputSender.test.js b/frontend-svelte/src/lib/paneInputSender.test.js new file mode 100644 index 00000000..4d17023e --- /dev/null +++ b/frontend-svelte/src/lib/paneInputSender.test.js @@ -0,0 +1,147 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { createPaneInputSender } from './paneInputSender.js'; + +function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +test('rapid text then Enter starts a cumulative keepalive submit immediately', async () => { + const calls = []; + const requests = []; + const sender = createPaneInputSender({ + send(body, options) { + const request = deferred(); + calls.push({ body, options }); + requests.push(request); + return request.promise; + }, + }); + + sender.enqueue({ text: 'rapid' }); + sender.enqueue({ key: 'Enter' }); + + assert.equal(calls.length, 2, 'Enter is not promise-chained behind text'); + assert.equal(calls[0].options.keepalive, false); + assert.equal(calls[1].options.keepalive, true); + assert.deepEqual(calls[1].body.events, [ + { seq: 1, text: 'rapid', key: '' }, + { seq: 2, text: '', key: 'Enter' }, + ]); + + sender.dispose(); + requests[1].resolve({ acked_seq: 2 }); + requests[0].resolve({ acked_seq: 1 }); + await Promise.all(requests.map((request) => request.promise)); +}); + +test('an out-of-order acknowledgement prunes cumulative retries monotonically', async () => { + const calls = []; + const requests = []; + const sender = createPaneInputSender({ + send(body, options) { + const request = deferred(); + calls.push({ body, options }); + requests.push(request); + return request.promise; + }, + }); + + sender.enqueue({ text: 'a' }); + sender.enqueue({ text: 'b' }); + requests[1].resolve({ acked_seq: 2 }); + await requests[1].promise; + await Promise.resolve(); + + sender.enqueue({ key: 'Enter' }); + assert.deepEqual(calls[2].body.events, [ + { seq: 3, text: '', key: 'Enter' }, + ]); + + requests[0].resolve({ acked_seq: 1 }); + requests[2].resolve({ acked_seq: 3 }); + await Promise.all(requests.map((request) => request.promise)); +}); + +test('a failed final submit retries its unacknowledged cumulative batch', async () => { + const calls = []; + const sender = createPaneInputSender({ + send(body, options) { + calls.push({ body, options }); + if (body.events.at(-1).key === 'Enter' && calls.length < 4) { + return Promise.reject(new Error('transient network failure')); + } + return Promise.resolve({ acked_seq: body.events.at(-1).seq }); + }, + }); + + sender.enqueue({ text: 'answer' }); + await Promise.resolve(); + sender.enqueue({ key: 'Enter' }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const submits = calls.filter((call) => call.body.events.at(-1).key === 'Enter'); + assert.equal(submits.length, 3); + assert.ok(submits.every((call) => call.options.keepalive)); +}); + +test('a late old success cannot clear an exhausted Enter failure', async () => { + const oldText = deferred(); + const callbacks = []; + let callCount = 0; + const sender = createPaneInputSender({ + send(body) { + callCount += 1; + if (callCount === 1) return oldText.promise; + assert.equal(body.events.at(-1).key, 'Enter'); + return Promise.reject(new Error('submit exhausted')); + }, + onError: () => callbacks.push('error'), + onSuccess: () => callbacks.push('success'), + }); + + sender.enqueue({ text: 'visible' }); + sender.enqueue({ key: 'Enter' }); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(callCount, 4, 'original Enter plus two bounded retries'); + assert.deepEqual(callbacks, ['error']); + + oldText.resolve({ acked_seq: 1 }); + await oldText.promise; + await Promise.resolve(); + assert.deepEqual(callbacks, ['error']); +}); + +test('a late old failure is suppressed after a newer Enter succeeds', async () => { + const oldText = deferred(); + const submit = deferred(); + const callbacks = []; + let callCount = 0; + const sender = createPaneInputSender({ + send() { + callCount += 1; + return callCount === 1 ? oldText.promise : submit.promise; + }, + onError: () => callbacks.push('error'), + onSuccess: () => callbacks.push('success'), + }); + + sender.enqueue({ text: 'answer' }); + sender.enqueue({ key: 'Enter' }); + submit.resolve({ acked_seq: 2 }); + await submit.promise; + await Promise.resolve(); + assert.deepEqual(callbacks, ['success']); + + oldText.reject(new Error('stale prefix failure')); + await Promise.resolve(); + await Promise.resolve(); + assert.deepEqual(callbacks, ['success']); +}); diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 0de2b6a0..43e313ee 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -9611,12 +9611,11 @@ async def send_tmux_pane_keys(agent_name: str, req: TmuxPaneKeysRequest): from lera's container rollout, #735 — three dialogs, three round trips to a shell). - Exactly one of ``text`` / ``key`` per request; ``key`` must be in - ``TmuxSession.PANE_KEY_WHITELIST``. Bounded to 1024 chars — the - modal sends keystrokes, not documents. Every send is logged with - the agent name for auditability (input content included: this is - an operator-facing admin surface, and "what did I type into the - wedged dialog" is exactly what the log needs to answer). + Legacy callers provide exactly one of ``text`` / ``key``. Dashboard + clients provide ``client_id`` plus cumulative sequenced ``events``; + the session applies each sequence once, so Enter can start immediately + and safely repeat any unacknowledged text ahead of it. Input remains + bounded to 1024 events / literal chars per request. 404 unknown agent · 409 not a tmux session · 400 bad input. """ @@ -9625,32 +9624,80 @@ async def send_tmux_pane_keys(agent_name: str, req: TmuxPaneKeysRequest): agent = agents.get(agent_name) if not agent: raise HTTPException(404, f"Agent '{agent_name}' not found") - if bool(req.text) == bool(req.key): - raise HTTPException(400, "provide exactly one of 'text' or 'key'") - if len(req.text) > 1024: - raise HTTPException(400, "text too long (max 1024 chars)") - if req.text and any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in req.text): - # Control bytes in 'text' would bypass the named-key whitelist - # ("\x04" is C-d either way) — controls only via whitelisted 'key'. + + batch_mode = bool(req.client_id or req.events) + if batch_mode: + if req.text or req.key: + raise HTTPException(400, "sequenced events cannot mix with text/key") + if not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", req.client_id): + raise HTTPException(400, "invalid terminal client_id") + if not req.events or len(req.events) > 1024: + raise HTTPException(400, "events must contain 1..1024 items") + seqs = [event.seq for event in req.events] + if any(seq < 1 for seq in seqs) or seqs != sorted(set(seqs)): + raise HTTPException(400, "event seq values must be positive and increasing") + if sum(len(event.text) for event in req.events) > 1024: + raise HTTPException(400, "event text too long (max 1024 chars total)") + inputs = [(event.text, event.key) for event in req.events] + else: + if bool(req.text) == bool(req.key): + raise HTTPException(400, "provide exactly one of 'text' or 'key'") + if len(req.text) > 1024: + raise HTTPException(400, "text too long (max 1024 chars)") + inputs = [(req.text, req.key)] + + for text, key in inputs: + if bool(text) == bool(key): + raise HTTPException(400, "each input must provide exactly one text/key") + if text and any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in text): + # Control bytes in 'text' would bypass the named-key whitelist + # ("\x04" is C-d either way) — controls only as named keys. + raise HTTPException( + 400, + "text must not contain control characters; " + "use a whitelisted 'key' for control sequences", + ) + if key and key not in TmuxSession.PANE_KEY_WHITELIST: + raise HTTPException( + 400, + f"key {key!r} not allowed; whitelist: " + f"{sorted(TmuxSession.PANE_KEY_WHITELIST)}", + ) + + session = broker.get_streaming_session(agent_name, label=req.label) + if session is None: raise HTTPException( - 400, - "text must not contain control characters; " - "use a whitelisted 'key' for control sequences", + 409, f"Agent '{agent_name}' has no live tmux session" ) - if req.key and req.key not in TmuxSession.PANE_KEY_WHITELIST: - raise HTTPException( - 400, - f"key {req.key!r} not allowed; whitelist: " - f"{sorted(TmuxSession.PANE_KEY_WHITELIST)}", + + if batch_mode: + batch_sender = getattr(session, "send_pane_key_events", None) + if not callable(batch_sender): + raise HTTPException( + 409, f"Agent '{agent_name}' has no live tmux session" + ) + first_seq, last_seq = req.events[0].seq, req.events[-1].seq + _log( + f"api: pane-events → {agent_name} " + f"(client={req.client_id!r}, seq={first_seq}..{last_seq})" + ) + acked_seq = await batch_sender( + client_id=req.client_id, + events=[(event.seq, event.text, event.key) for event in req.events], ) + if acked_seq < last_seq: + raise HTTPException(502, "tmux send-keys failed (see daemon log)") + return { + "sent": True, + "agent": agent_name, + "acked_seq": acked_seq, + } - session = broker.get_streaming_session(agent_name, label=req.label) sender = getattr(session, "send_pane_keys", None) - if session is None or not callable(sender): + if not callable(sender): raise HTTPException( 409, f"Agent '{agent_name}' has no live tmux session" ) - summary = f"text={req.text!r}" if req.text else f"key={req.key}" _log(f"api: pane-keys → {agent_name} ({summary})") ok = await sender(text=req.text, key=req.key) diff --git a/src/pinky_daemon/api_models.py b/src/pinky_daemon/api_models.py index c50570d9..ddaac96e 100644 --- a/src/pinky_daemon/api_models.py +++ b/src/pinky_daemon/api_models.py @@ -1100,14 +1100,24 @@ class PeerFleetAclSetRequest(BaseModel): selectors: list[PeerFleetAclEntryRequest] = [] +class TmuxPaneKeyEvent(BaseModel): + """One sequenced literal/key event from a terminal modal client.""" + + seq: int + text: str = "" + key: str = "" + + class TmuxPaneKeysRequest(BaseModel): """Operator keystrokes for the typeable pane view (terminal modal). - Exactly one of ``text`` (literal characters, no tmux keyname - interpretation) or ``key`` (named tmux key — Enter, Up, C-c, ... — - validated against ``TmuxSession.PANE_KEY_WHITELIST``) per request. + Legacy callers provide exactly one of ``text`` / ``key``. Dashboard + terminal clients provide ``client_id`` plus a cumulative list of sequenced + ``events`` so an Enter request can retry slower text safely and atomically. """ text: str = "" key: str = "" label: str = "main" + client_id: str = "" + events: list[TmuxPaneKeyEvent] = Field(default_factory=list) diff --git a/src/pinky_daemon/tmux_session.py b/src/pinky_daemon/tmux_session.py index 0dc0b150..abfd287a 100644 --- a/src/pinky_daemon/tmux_session.py +++ b/src/pinky_daemon/tmux_session.py @@ -64,7 +64,7 @@ import shlex import threading import time -from collections import deque +from collections import OrderedDict, deque from dataclasses import dataclass, field, replace from pathlib import Path @@ -1201,6 +1201,11 @@ def __init__( self._scheduler_pending_turns: list[_QueuedTurn] = [] self._pane_queue_operations: deque[_QueuedTurn | None] = deque() self._pane_dequeued_turns: deque[_QueuedTurn | None] = deque() + # Dashboard terminal requests start immediately and may arrive out of + # order. Serialize pane input and remember each client's acknowledged + # sequence so cumulative retries never duplicate text or Enter. + self._pane_input_lock = asyncio.Lock() + self._pane_input_acked: OrderedDict[str, int] = OrderedDict() # Background watchdog that ages the deque head against # ``_TURN_DONE_TIMEOUT_SEC`` and triggers ``force_restart`` when # a stop hook fails to land. Issue #560 replaces the per-iter @@ -3881,6 +3886,7 @@ async def resize_pane(self, *, cols: int, rows: int) -> bool: "Up", "Down", "Left", "Right", "Home", "End", "PPage", "NPage", "C-c", "C-u", }) + _PANE_INPUT_CLIENT_LIMIT = 1024 async def send_pane_keys(self, *, text: str = "", key: str = "") -> bool: """Operator keystrokes from the pane-view modal (typeable terminal). @@ -3899,6 +3905,13 @@ async def send_pane_keys(self, *, text: str = "", key: str = "") -> bool: an operator can resolve first-run dialogs / wedged prompts from the web UI without SSH + ``tmux attach``. """ + async with self._pane_input_lock: + return await self._send_pane_keys_unlocked(text=text, key=key) + + async def _send_pane_keys_unlocked( + self, *, text: str = "", key: str = "" + ) -> bool: + """Validated single-event implementation; caller holds pane-input lock.""" if bool(text) == bool(key): return False # exactly one input mode per call if key and key not in self.PANE_KEY_WHITELIST: @@ -3932,6 +3945,55 @@ async def send_pane_keys(self, *, text: str = "", key: str = "") -> bool: return False return True + async def send_pane_key_events( + self, + *, + client_id: str, + events: list[tuple[int, str, str]], + ) -> int: + """Apply a cumulative sequenced input batch exactly once. + + Concurrent dashboard fetches can complete in any order. Each request + repeats its unacknowledged prefix, so whichever request reaches this + lock first can fill every sequence through its final event. Later + requests skip the already-applied prefix and return the same receipt. + """ + if not client_id or not events: + return 0 + seqs = [seq for seq, _, _ in events] + if any(seq < 1 for seq in seqs) or seqs != sorted(set(seqs)): + return 0 + + async with self._pane_input_lock: + if ( + client_id not in self._pane_input_acked + and len(self._pane_input_acked) >= self._PANE_INPUT_CLIENT_LIMIT + ): + self._pane_input_acked.popitem(last=False) + acked = self._pane_input_acked.get(client_id, 0) + for seq, text, key in events: + if seq <= acked: + continue + if seq != acked + 1: + _log( + f"tmux[{self.agent_name}]: pane-input gap for " + f"client={client_id!r} (acked={acked}, got={seq})" + ) + return acked + if not await self._send_pane_keys_unlocked(text=text, key=key): + return acked + acked = seq + # Persist partial progress before the next tmux operation: if + # Enter fails after text lands, its cumulative retry must not + # type that text a second time. + self._pane_input_acked[client_id] = acked + + self._pane_input_acked[client_id] = acked + self._pane_input_acked.move_to_end(client_id) + while len(self._pane_input_acked) > self._PANE_INPUT_CLIENT_LIMIT: + self._pane_input_acked.popitem(last=False) + return acked + # Claude Code reserves a buffer below the raw model cap so the # ``/compact`` autocompact step fires before the API rejects the # next turn for context exhaustion. Empirically 33K on the 200K diff --git a/tests/test_agent_status.py b/tests/test_agent_status.py index 3669635c..4cbf3d7d 100644 --- a/tests/test_agent_status.py +++ b/tests/test_agent_status.py @@ -699,6 +699,75 @@ async def send_pane_keys(self, *, text="", key=""): {"text": "", "key": "Enter"}, ] + def test_cumulative_sequenced_events_reach_batch_sender(self): + """The Enter request may safely repeat slower text in one batch.""" + app, client = self._client() + calls = [] + + class _PaneSession: + async def send_pane_key_events(self, *, client_id, events): + calls.append((client_id, events)) + return events[-1][0] + + app.state.broker.register_streaming("dymok", _PaneSession(), label="main") + resp = self._post(client, { + "client_id": "dashboard-123", + "events": [ + {"seq": 1, "text": "rapid"}, + {"seq": 2, "key": "Enter"}, + ], + }) + assert resp.status_code == 200 + assert resp.json() == { + "sent": True, + "agent": "dymok", + "acked_seq": 2, + } + assert calls == [( + "dashboard-123", + [(1, "rapid", ""), (2, "", "Enter")], + )] + + def test_400_on_invalid_sequenced_events(self): + app, client = self._client() + + class _PaneSession: + async def send_pane_key_events(self, *, client_id, events): + return events[-1][0] + + app.state.broker.register_streaming("dymok", _PaneSession(), label="main") + bad_bodies = [ + {"client_id": "bad id", "events": [{"seq": 1, "text": "x"}]}, + {"client_id": "ok", "events": []}, + {"client_id": "ok", "events": [{"seq": 0, "text": "x"}]}, + {"client_id": "ok", "events": [ + {"seq": 2, "text": "x"}, {"seq": 1, "key": "Enter"}, + ]}, + {"client_id": "ok", "events": [{"seq": 1, "text": "x", "key": "Enter"}]}, + {"client_id": "ok", "events": [{"seq": 1, "text": "\x1b"}]}, + {"client_id": "ok", "events": [{"seq": 1, "key": "C-d"}]}, + {"client_id": "ok", "events": [{"seq": 1, "text": "x"}], "key": "Enter"}, + ] + for body in bad_bodies: + assert self._post(client, body).status_code == 400, body + + def test_502_when_batch_receipt_stops_before_final_event(self): + app, client = self._client() + + class _PaneSession: + async def send_pane_key_events(self, *, client_id, events): + return 1 + + app.state.broker.register_streaming("dymok", _PaneSession(), label="main") + resp = self._post(client, { + "client_id": "dashboard-123", + "events": [ + {"seq": 1, "text": "answer"}, + {"seq": 2, "key": "Enter"}, + ], + }) + assert resp.status_code == 502 + def test_502_when_session_send_fails(self): app, client = self._client() diff --git a/tests/test_tmux_session.py b/tests/test_tmux_session.py index d220a4fb..9a207b1a 100644 --- a/tests/test_tmux_session.py +++ b/tests/test_tmux_session.py @@ -9602,6 +9602,83 @@ async def test_send_pane_keys_false_on_tmux_failure() -> None: assert await session.send_pane_keys(key="Enter") is False +@pytest.mark.asyncio +async def test_send_pane_key_events_cumulative_retry_is_idempotent() -> None: + """A keepalive Enter batch can repeat an in-flight text event safely.""" + session, tmux = _make_session() + tmux.send_literal = AsyncMock(return_value=_ok()) + tmux.send_keys = AsyncMock(return_value=_ok()) + + ack = await session.send_pane_key_events( + client_id="dashboard-1", + events=[(1, "answer", ""), (2, "", "Enter")], + ) + stale_ack = await session.send_pane_key_events( + client_id="dashboard-1", + events=[(1, "answer", "")], + ) + + assert ack == stale_ack == 2 + tmux.send_literal.assert_awaited_once_with("answer") + tmux.send_keys.assert_awaited_once_with("Enter", enter=False) + + +@pytest.mark.asyncio +async def test_send_pane_key_events_later_cumulative_request_fills_gap() -> None: + """A later request may arrive first and carry the complete ordered prefix.""" + session, tmux = _make_session() + tmux.send_literal = AsyncMock(return_value=_ok()) + tmux.send_keys = AsyncMock(return_value=_ok()) + + ack = await session.send_pane_key_events( + client_id="dashboard-2", + events=[(1, "a", ""), (2, "b", ""), (3, "", "Enter")], + ) + old_request_ack = await session.send_pane_key_events( + client_id="dashboard-2", + events=[(1, "a", ""), (2, "b", "")], + ) + + assert ack == old_request_ack == 3 + assert [call.args[0] for call in tmux.send_literal.await_args_list] == ["a", "b"] + tmux.send_keys.assert_awaited_once_with("Enter", enter=False) + + +@pytest.mark.asyncio +async def test_send_pane_key_events_refuses_missing_sequence() -> None: + session, tmux = _make_session() + tmux.send_literal = AsyncMock(return_value=_ok()) + + ack = await session.send_pane_key_events( + client_id="dashboard-3", + events=[(2, "missing-one", "")], + ) + + assert ack == 0 + tmux.send_literal.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_send_pane_key_events_persists_partial_receipt_before_retry() -> None: + """A failed Enter retry must not duplicate text that already landed.""" + session, tmux = _make_session() + tmux.send_literal = AsyncMock(return_value=_ok()) + tmux.send_keys = AsyncMock(side_effect=[_fail("transient"), _ok()]) + events = [(1, "answer", ""), (2, "", "Enter")] + + partial_ack = await session.send_pane_key_events( + client_id="dashboard-4", events=events, + ) + final_ack = await session.send_pane_key_events( + client_id="dashboard-4", events=events, + ) + + assert partial_ack == 1 + assert final_ack == 2 + tmux.send_literal.assert_awaited_once_with("answer") + assert tmux.send_keys.await_count == 2 + + # ────────────────────────────────────────────────────────────────────────── # #230 — _watchdog_liveness: live carve-out signal for the OUTER watchdogs # (daemon SessionWatchdog warn/recover + scheduler idle-sleep). Active ONLY