diff --git a/.changeset/bounded-agent-coordination.md b/.changeset/bounded-agent-coordination.md index 2ff2c0af..62724678 100644 --- a/.changeset/bounded-agent-coordination.md +++ b/.changeset/bounded-agent-coordination.md @@ -2,7 +2,7 @@ "aicodeman": minor --- -Add experimental bounded agent inspection with expiring scoped cursors and explicit pane snapshots, plus staged local Codex input and single-use hash-confirmed submission receipts. Verify the exact composer, caret, lifecycle and input revision before sending Enter, and serialize dispatch against managed input and auxiliary key operations. +Add experimental bounded agent inspection with expiring scoped cursors (the newest 8 kept per caller and session) and explicit pane snapshots, plus staged local Codex input and single-use hash-confirmed submission receipts. Verify the exact composer, caret, lifecycle and input revision before sending Enter, and serialize dispatch against managed input and auxiliary key operations. The three new CLI verbs (`codeman agent inspect`, `stage`, `submit`) and their HTTP routes (`/api/v1` and `/api`) are labeled experimental, so the versioning policy's rule for experimental surfaces applies: their shape and behavior may change in minor or patch releases. They are opt-in behind a new synced setting, `agentCoordinationEnabled` (App Settings, Cross-session features, "Experimental agent coordination", default OFF, translated for zh-CN): while it is off all three routes answer `403 FORBIDDEN` with a message naming it, and the CLI verbs print that refusal verbatim. @@ -10,7 +10,7 @@ Codex rollout attribution remembers the attributed file per session and re-verif The guarded composer read takes its glyph and footer from the CLI registry: codex's `workDetect.promptGlyph` and a new `workDetect.composerFooterLine` capability (config-supplied regex, compiled through the ReDoS guard). Only a CLI declaring it takes guarded input. Codex's footer effort words are now one shared list for this and `modelDetect.screenLine`. -Stage waits for the pane to show what it typed: it re-reads the composer every 50 ms for up to one second and issues a receipt only for a frame showing exactly the message (Codex repaints 30 to 60 ms after the keys, so the first capture was often stale and stage refused with the text already typed). Every refusal after typing now says the message may be in the composer and points at the Ctrl+U recovery. A visible-frame pane capture reads the screen and the caret in one tmux invocation. Stage refuses messages that would change what Enter does (codex: a leading `/` or `!`, or a last word starting with `@` or `$`, which open a command or completion popup) and frames with a suggestion popup above the composer, from three new registry fields (`workDetect.composerPopupLine`, `composerCommandPrefixes`, `composerCompletionSigils`). Submit waits until 250 ms after the staged keys before sending Enter, since codex turns an Enter within 120 ms of typed text into a newline. While a guarded submit is in flight, `POST /api/sessions/:id/input` answers `409 CONFLICT` with its seq returned instead of a `200` for input that was never written. +Stage waits for the pane to show what it typed: it re-reads the composer every 50 ms for up to one second and issues a receipt only for a frame showing exactly the message (Codex repaints 30 to 60 ms after the keys, so the first capture was often stale and stage refused with the text already typed). Every refusal after typing now says the message may be in the composer and points at the Ctrl+U recovery. A visible-frame pane capture reads the screen and the caret in one tmux invocation, falling back to the two separate calls only for output it cannot split; a call that fails is not repeated. Stage refuses messages that would change what Enter does (codex: a leading `/` or `!`, or a last word starting with `@` or `$`, which open a command or completion popup) and frames with a suggestion popup above the composer, from three new registry fields (`workDetect.composerPopupLine`, `composerCommandPrefixes`, `composerCompletionSigils`). Submit waits until 250 ms after the staged keys before sending Enter, since codex turns an Enter within 120 ms of typed text into a newline. Input that arrives while a guarded submit holds its input lease (the few milliseconds of its Enter) is held and written in arrival order once the lease is released, so a browser keystroke, a Shift+Enter or Ctrl+Enter, or an approval answer typed in that window is no longer lost; a WebSocket frame is acknowledged only once it is written. `POST /api/sessions/:id/input` answers `409 CONFLICT` with its seq returned for that span instead of a `200` for input that was never written, and a lease that would hold more than 256 inputs or 1 MiB closes the terminal WebSocket (code 1013) so the browser resends in order, and answers `send-key` with `409 CONFLICT`. Document the initial local Codex support and conversation-attribution limits in the API, CLI and packaged skill. Other modes retain labeled snapshots or refuse unverifiable guarded input. diff --git a/docs/api-reference.md b/docs/api-reference.md index d99d4ef2..2476f983 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -427,8 +427,9 @@ turn that never started. Verified live; this is the most common silent failure o this endpoint. Any input, waiting or not, answers `409 CONFLICT` with nothing written while an -experimental guarded submit holds the session's input lease (a few milliseconds; -see [guarded input](#experimental-bounded-agent-inspection-and-guarded-input)). +experimental guarded submit holds the session's input lease, or input held behind +it is still being written (a few milliseconds; see +[guarded input](#experimental-bounded-agent-inspection-and-guarded-input)). Retry it. A **plain prompt** (printable text followed by exactly one `\r`, nothing else) is @@ -659,9 +660,10 @@ thread, task completion or authorization. Opaque server-side cursors bind caller, session, pane generation, conversation, opened-file identity, byte offset and bounded content anchors. Terminal resizes -do not change this inspection scope. At most 1024 live -cursors are retained for 15 minutes; restart, eviction or expiry requires a fresh -tail baseline. Replacement, truncation, detected rewrites, reassociation, foreign +do not change this inspection scope. Each owner and session keeps its newest 8 +cursors (a retry of the last call still works; older ones are retired), at most +1024 live cursors are retained in all, each for 15 minutes; restart, eviction or +expiry requires a fresh tail baseline. Replacement, truncation, detected rewrites, reassociation, foreign cursors or loss of the attributed source return `409 CONFLICT`. No transcript filesystem path is returned as cursor/identity metadata; evidence retains the owner's native record content, which can itself contain paths. @@ -708,13 +710,21 @@ may already be in the composer: inspect the pane and follow the stranded-draft recovery (Ctrl+U) in the skill reference before staging again. While a guarded submit holds the input lease (the few milliseconds of its -conditional Enter), every other Codeman write path refuses, and -`POST /api/v1/sessions/:id/input` answers `409 CONFLICT` ("A guarded submit is in -progress on this session; nothing was written") instead of a `200` for bytes -that were never written. A tagged `clientId`/`seq` frame keeps its seq, so the -browser's durable queue, which keeps a frame on a `409`, delivers it on the next -retry. WebSocket keystrokes in that window are left unacknowledged and redelivered -the same way. +conditional Enter), input that reaches the session is held, not refused, and is +written in arrival order as soon as the lease is released, whatever the submit's +outcome. That covers terminal WebSocket keystrokes (each frame is acknowledged +only once it is written, so a frame typed during the lease lands before the one +typed after it), Shift+Enter and Ctrl+Enter from `POST .../send-key`, approval +answers and Codeman's own managed writes. A held write that is still being sent +holds everything behind it. `POST /api/v1/sessions/:id/input` instead answers +`409 CONFLICT` ("A guarded submit is in progress on this session; nothing was +written") for that span, the lease plus the held input's drain, rather than a +`200` for bytes that were never written; a tagged `clientId`/`seq` frame keeps +its seq, so the browser's durable queue, whose POST fallback sends one frame at a +time and keeps it on a `409`, delivers it on the next retry. A lease holds at +most 256 inputs or 1 MiB; past that a WebSocket is closed with code `1013` and +nothing later from it is applied, so the browser reconnects and resends every +unacknowledged frame in order, and `send-key` answers `409 CONFLICT`. Direct auxiliary key operations participate in the same pending-write and input lease checks. Complete allowlisted focus/mouse reports and empty writes preserve diff --git a/docs/architecture-invariants.md b/docs/architecture-invariants.md index d0fd2444..2e2495eb 100644 --- a/docs/architecture-invariants.md +++ b/docs/architecture-invariants.md @@ -1281,6 +1281,9 @@ candidate-metadata budgets, bounded path probes before canonical reads, and complete UTF8 JSONL records. A full page advances its cursor without claiming a blocked record. Opaque cursors bind caller/session/lifecycle/file identity; pane geometry is excluded so changing devices does not invalidate transcript paging. +Cursors are capped per owner and session (`INSPECT_CURSORS_PER_SCOPE`, the newest 8, +under the global 1,024), so one polling agent retires its own oldest cursors instead +of evicting every other caller's from the shared FIFO. ⚠️ Codex never prunes `~/.codex/sessions`, so attribution must not cost a store walk per call. The first call walks (negative-filter prefixes read 1 KiB first, @@ -1317,9 +1320,18 @@ and the first capture is often from before them; another writer or a lifecycle c ends the wait at once, and every refusal after typing says the message may be in the composer. `capturePaneBuffer()` reads the visible frame and its caret in ONE tmux call (`capture-pane \; display-message`, split by `splitFrameAndCursor()`), falling back to -the two calls on any doubt; keep the poll anyway. While a guarded submit holds the -lease, `POST .../input` answers `409` (seq returned) instead of a `200` for unwritten -bytes. ⚠️ Submit waits `SUBMIT_QUIET_MS` (250) after the staged keys before its +the two calls only for output it cannot split; a combined call that THREW goes to the +capture's catch and is never rerun (a hung tmux would block the event loop for a second +timeout on every `GET .../terminal`). Keep the poll anyway. ⚠️ Input that arrives +while a guarded submit holds the lease is HELD, never refused (`holdForGuardedLease()`, +`_inGuardedTurn()`), and written in arrival order when the lease is released: the +browser's WebSocket path does not wait for ACKs, so a refused seq N was overtaken by +N+1 and its redelivery dropped as a duplicate. A WS frame is ACKed only once written; +a held async write (mux write, named key) holds everything behind it until it settles +(`guardedInputHeld`); a full hold queue (256 inputs / 1 MiB) closes the socket with +`1013` and applies nothing later from it, so the browser resends in order. `POST +.../input` alone answers `409` (seq returned) for the lease plus the drain, since its +browser drainer is serialized and retries. ⚠️ Submit waits `SUBMIT_QUIET_MS` (250) after the staged keys before its checks: stage types plain keys, codex's paste-burst detector reads them as a paste, and an Enter within 120 ms of one becomes a newline while submit would report it delivered. Stage cancels older verifier retries and sends no Enter. Receipts contain hashes and @@ -1330,7 +1342,8 @@ Submission additionally binds input and geometry revisions and rechecks the exact composer/caret at the conditional Enter boundary. The route must retain the Session snapshot's staging generation rather than replacing it with the inspection generation. Managed input and auxiliary named keys participate in the -same pending/lease boundary. Complete focus/mouse reports and empty writes are +same pending/lease boundary: they wait for the lease, and only a full hold queue +refuses them (`GuardedInputBusyError`, which `send-key` answers as `409 CONFLICT`). Complete focus/mouse reports and empty writes are not edits. Never retry ambiguous Enter delivery or clear an unverified composer. The ordinary SubmitVerifier uses the shared write boundary, so each re-press updates `lastSubmitAt` just like any other Enter. diff --git a/src/session.ts b/src/session.ts index 1b1410d1..dc43c4ff 100644 --- a/src/session.ts +++ b/src/session.ts @@ -465,6 +465,17 @@ export interface ClaudeMessage { duration_ms?: number; } +/** + * A write that a guarded submit's lease could not hold, because the per-lease hold + * queue (`Session.MAX_GUARDED_HELD_INPUTS`) was full. Nothing was written; retry it. + */ +export class GuardedInputBusyError extends Error { + constructor() { + super('A guarded submit is in progress on this session; nothing was written. Retry the input.'); + this.name = 'GuardedInputBusyError'; + } +} + /** * Whether a programmatic write leaves the keep-alive turn running: the nudge itself, or a bare * Enter (which submits nothing new). Anything else carrying an Enter starts a turn of its own. @@ -4218,7 +4229,7 @@ export class Session extends EventEmitter { prompt: string, options?: { model?: string; onProgress?: (info: { tokens?: number; status?: string }) => void } ): Promise<{ result: string; cost: number }> { - if (this._guardedSubmitInFlight) { + if (this.guardedInputHeld) { this._guardedInputRevision++; throw new Error('Guarded submission in progress'); } @@ -4738,6 +4749,19 @@ export class Session extends EventEmitter { private _guardedInputRevision = 0; private _guardedPendingWrites = 0; private _guardedSubmitInFlight = false; + /** + * Input that arrived while a guarded submit held the lease, in arrival order. Each + * entry runs when the lease is released; an entry that returns a promise (a mux + * write, a named key) is waited for before the next runs, so nothing overtakes it. + */ + private _guardedHeld: Array<{ run: () => void | Promise; bytes: number }> = []; + private _guardedHeldBytes = 0; + /** True while the drain waits for a held asynchronous write to finish. */ + private _guardedHeldDraining = false; + /** Held entries per lease. A lease lasts milliseconds; past this the caller backs off. */ + static readonly MAX_GUARDED_HELD_INPUTS = 256; + /** Held bytes per lease (16 browser frames at the 64 KiB frame cap). */ + static readonly MAX_GUARDED_HELD_BYTES = 16 * 64 * 1024; /** Ephemeral identity, never persisted or based on a potentially reused pid alone. */ get inspectionGeneration(): string { return `${this._inspectionInstance}:${this._guardedPaneRevision}:${this.owner ?? ''}`; @@ -4750,13 +4774,89 @@ export class Session extends EventEmitter { return this._guardedPendingWrites; } /** - * True while a guarded submit holds the input lease. Every write path refuses for - * that span (a few milliseconds), so a route that can report a refusal checks this - * first instead of answering as if the bytes went out. + * True while a guarded submit holds the input lease (a few milliseconds). Input that + * arrives in that span is held and written after it (see {@link guardedInputHeld}). */ get guardedSubmitInFlight(): boolean { return this._guardedSubmitInFlight; } + /** + * True while input is being held behind a guarded submit: the lease itself, or held + * input not yet written after it. A write that is not routed through + * {@link holdForGuardedLease} must refuse in this window, or it overtakes held input. + */ + get guardedInputHeld(): boolean { + return this._guardedSubmitInFlight || this._guardedHeld.length > 0 || this._guardedHeldDraining; + } + /** + * Hold input behind a guarded submit's lease instead of refusing it. + * + * `free`: nothing is held, so the caller writes inline as it always did. `held`: `run` + * is queued and runs, in arrival order with every other held input, once the lease is + * released (whatever the submit's outcome). `full`: the per-lease cap is reached and + * nothing was queued; the caller must stop delivering from that source until it can + * resend in order. Refusing instead lost input: the browser's WebSocket path does not + * wait for ACKs, so a refused seq N was overtaken by N+1 and its redelivery was then + * rejected as a duplicate. `data`, when given, is observed as input on arrival, so a + * keystroke held during the lease still moves the input revision at once. + */ + holdForGuardedLease(run: () => void | Promise, bytes = 0, data?: string): 'free' | 'held' | 'full' { + if (!this.guardedInputHeld) return 'free'; + if (data !== undefined) this._observeGuardedKeyboardInput(data); + if ( + this._guardedHeld.length >= Session.MAX_GUARDED_HELD_INPUTS || + this._guardedHeldBytes + bytes > Session.MAX_GUARDED_HELD_BYTES + ) + return 'full'; + this._guardedHeld.push({ run, bytes }); + this._guardedHeldBytes += bytes; + return 'held'; + } + /** + * Run an asynchronous write now, or in its turn behind a guarded submit's lease. The + * turn lasts until the write settles, so later held input waits for it. Resolves + * `whenFull` without running anything when the hold queue is full. + */ + private _inGuardedTurn(write: () => Promise, whenFull: () => T): Promise { + if (!this.guardedInputHeld) return write(); + return new Promise((resolve, reject) => { + const held = this.holdForGuardedLease(() => { + const running = write(); + running.then(resolve, reject); + return running; + }); + if (held === 'full') { + try { + resolve(whenFull()); + } catch (err) { + reject(err); + } + } + }); + } + /** Run held input in arrival order until it is all written or a new lease is taken. */ + private _drainGuardedHeld(): void { + while (this._guardedHeld.length > 0 && !this._guardedSubmitInFlight && !this._guardedHeldDraining) { + const entry = this._guardedHeld.shift()!; + this._guardedHeldBytes -= entry.bytes; + let running: void | Promise; + try { + running = entry.run(); + } catch (err) { + console.error(`[Session ${this.id}] held input failed after a guarded submit:`, getErrorMessage(err)); + continue; + } + if (running && typeof running.then === 'function') { + this._guardedHeldDraining = true; + void running + .catch(() => {}) + .then(() => { + this._guardedHeldDraining = false; + this._drainGuardedHeld(); + }); + } + } + } /** Expose only the path selection needed for attributed Codex evidence, never credentials. */ getCodexHomeOverride(): string | undefined { return this._envOverrides?.CODEX_HOME || undefined; @@ -4773,7 +4873,7 @@ export class Session extends EventEmitter { this._closing || this.paneLifecycleInFlight || this._guardedPendingWrites || - (!allowLease && this._guardedSubmitInFlight) || + (!allowLease && this.guardedInputHeld) || this.remote || this.docker || this.paneExit || @@ -4808,7 +4908,7 @@ export class Session extends EventEmitter { /** Stage through ordinary delivery, after cancelling any older Enter retry timer. */ async stageGuardedInput(message: string): Promise { if (/[\r\n]/.test(message)) return false; - if (this._guardedPendingWrites || this._guardedSubmitInFlight) { + if (this._guardedPendingWrites || this.guardedInputHeld) { this._guardedInputRevision++; return false; } @@ -4839,18 +4939,29 @@ export class Session extends EventEmitter { return sent && dispatched; } finally { this._guardedSubmitInFlight = false; + this._drainGuardedHeld(); } } - /** Lease/pending boundary for named keys whose bytes bypass the ordinary write methods. */ + /** + * Lease/pending boundary for named keys whose bytes bypass the ordinary write methods. + * During a guarded submit the operation waits for the lease like any other write; + * only a full hold queue refuses it, with {@link GuardedInputBusyError}. + */ async performAuxiliaryInput(operation: () => Promise): Promise { this._guardedInputRevision++; - if (this._guardedSubmitInFlight) throw new Error('Guarded submission in progress'); - this._guardedPendingWrites++; - try { - await operation(); - } finally { - this._guardedPendingWrites--; - } + return this._inGuardedTurn( + async () => { + this._guardedPendingWrites++; + try { + await operation(); + } finally { + this._guardedPendingWrites--; + } + }, + () => { + throw new GuardedInputBusyError(); + } + ); } /** * Sends input directly to the PTY process. @@ -5040,7 +5151,17 @@ export class Session extends EventEmitter { */ async writeViaMux(data: string, options: SessionWriteOptions = {}): Promise { this._observeGuardedKeyboardInput(data); - if (this._guardedSubmitInFlight) return false; + // A guarded submit's lease lasts for its Enter only: wait for it, in arrival order, + // rather than refusing, so an approval answer or a programmatic prompt is not + // dropped. Without a lease the write starts synchronously, so the pending count + // rises before any await and no lease can be taken while it is in flight. + return this._inGuardedTurn( + () => this._writeViaMuxNow(data, options), + () => false + ); + } + + private async _writeViaMuxNow(data: string, options: SessionWriteOptions): Promise { this._guardedPendingWrites++; try { const submittedPrompt = this._trackSubmit(data, options); @@ -5090,7 +5211,13 @@ export class Session extends EventEmitter { ? null : this._mux.capturePaneText?.(this._muxSession.muxName), sendEnter: () => { - if (this._guardedSubmitInFlight) this._submitVerifier?.cancel(); + // A retry Enter is never held behind a guarded submit: once that submit's + // Enter went out, a re-press would submit whatever is on the composer next. + if (this.guardedInputHeld) { + this._guardedInputRevision++; + this._submitVerifier?.cancel(); + return Promise.resolve(false); + } return this.writeViaMux('\r'); }, // ⚠ NO fallback glyph here, unlike the screen-reading probe elsewhere in this file. @@ -5261,7 +5388,7 @@ export class Session extends EventEmitter { // Legacy method for sending input - wraps runPrompt async sendInput(input: string): Promise { - if (this._guardedSubmitInFlight) { + if (this.guardedInputHeld) { this._guardedInputRevision++; throw new Error('Guarded submission in progress'); } diff --git a/src/web/routes/session-routes.ts b/src/web/routes/session-routes.ts index 369ea2dc..ae4a523f 100644 --- a/src/web/routes/session-routes.ts +++ b/src/web/routes/session-routes.ts @@ -35,6 +35,7 @@ import { type RemoteHost, } from '../../types.js'; import { + GuardedInputBusyError, Session, cliTakesSessionModel, isAltScreenStripMode, @@ -2078,12 +2079,14 @@ export function registerSessionRoutes( } // A guarded submit (agent-staging.ts) holds the input lease for the few - // milliseconds of its conditional Enter, and every write path refuses while it - // does. Say so instead of answering 200 for bytes that were never written, and - // give the seq back so the browser's durable queue (which keeps a frame on any - // 409) delivers it once the lease is released. Nothing from here to the write - // below awaits, so the lease cannot be taken in between. - if (!duplicate && session.guardedSubmitInFlight === true) { + // milliseconds of its conditional Enter, and input that arrives meanwhile is held + // and written in arrival order after it. This route refuses for that span (the + // lease plus the held input's drain) rather than overtake held input. Say so + // instead of answering 200 for bytes that were never written, and give the seq + // back so the browser's durable queue (which keeps a frame on any 409, and whose + // POST drainer sends one frame at a time) delivers it on its next retry. Nothing + // from here to the write below awaits, so the lease cannot be taken in between. + if (!duplicate && session.guardedInputHeld === true) { if (tagged) session.forgetInputSeq(clientId as string, seq as number); return reply .code(409) @@ -2416,7 +2419,7 @@ export function registerSessionRoutes( // own `capabilities.newline`. Claude Code's Ink input recognizes 0x0a as "insert newline" vs // 0x0d (carriage return = submit). - app.post('/api/sessions/:id/send-key', async (req) => { + app.post('/api/sessions/:id/send-key', async (req, reply) => { const { id } = req.params as { id: string }; const body = req.body as Record; const key = typeof body?.key === 'string' ? body.key : ''; @@ -2461,6 +2464,11 @@ export function registerSessionRoutes( }) ); } catch (err) { + // A guarded submit's lease holds the key until its Enter is out; only a full + // hold queue refuses it, and that is a retryable conflict, not a tmux failure. + if (err instanceof GuardedInputBusyError) { + return reply.code(409).send(createErrorResponse(ApiErrorCode.CONFLICT, err.message)); + } console.error('[Server] send-key failed:', err); return createErrorResponse(ApiErrorCode.INTERNAL_ERROR, 'tmux send-keys failed'); } diff --git a/src/web/routes/ws-routes.ts b/src/web/routes/ws-routes.ts index 19c2ddca..99a86691 100644 --- a/src/web/routes/ws-routes.ts +++ b/src/web/routes/ws-routes.ts @@ -29,6 +29,11 @@ * {"t":"i","d":"...","seq":N,"cid":"..."} — input (keystroke or paste). seq+cid are * optional reliable-delivery tags: the server applies each * (cid,seq) at-most-once and ACKs with {"t":"ia","seq":N}. + * A frame that arrives while a guarded submit holds the + * session's input lease is held, then applied and ACKed in + * arrival order once it is released; when that hold queue + * is full the socket is closed with 1013 and nothing later + * from it is applied, so the client resends in order. * {"t":"z","c":N,"r":N,"f":bool} — resize terminal (f=true forces SIGWINCH even if dims unchanged) */ @@ -169,12 +174,69 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost const sizingToken = Symbol('ws-desktop-sizing'); let holdsDesktopClaim = false; + // Set once an input frame could not be held behind a guarded submit's lease + // (the hold queue was full). Nothing later from this socket is applied: it is + // closed, and the browser reconnects and resends everything unACKed, in order. + let inputStalled = false; + + // Apply one input frame and ACK it. Runs inline, or later and in arrival order + // when a guarded submit held the input lease as the frame arrived. + const applyInputFrame = (msg: { d: string; seq?: unknown; cid?: unknown }) => { + // Reliable delivery: when the frame carries a clientId + seq, apply it + // exactly once (skip a duplicate redelivery) but ACK it regardless so + // the client can drop it from its durable queue. Frames without seq + // (legacy/other tools) are applied as-is — no behavior change. + const cid = typeof msg.cid === 'string' ? msg.cid : null; + const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null; + const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true; + let delivered = true; + if (apply) { + // Typed input from a claim-holding desktop keeps the claim "hot" + // and re-asserts the desktop layout after a mobile override. + if (holdsDesktopClaim) session.noteDesktopActivity(); + // Browser keystrokes are the user's own, so they may name the tab. + delivered = session.write(msg.d, { fromUser: true }); + // A session whose PTY is gone swallows the write. ACKing anyway told + // the client to drop the frame from its durable queue and left the seq + // burnt, so the retry that reliable delivery exists for was rejected as + // a duplicate: the input was lost for good. + if (!delivered && cid && seq !== null) session.forgetInputSeq(cid, seq); + } + if (seq !== null && socket.readyState === 1) { + if (apply) { + if (delivered) socket.send(`{"t":"ia","seq":${seq}}`); + } else { + // REJECTED as a duplicate. ACK it — the client must still drop it + // from its durable queue — but say so, and hand back our watermark. + // + // A plain ACK here is indistinguishable from "applied", which is + // what made a client with a rolled-back counter unrecoverable: its + // seqs persist to localStorage on a DEBOUNCED write, so a tab killed + // between a send and that write comes back with a counter BELOW this + // watermark, every later keystroke lands at or under it, and each one + // is dropped-but-ACKed. The UI stays clean, nothing is delivered, and + // a reload restores the same stale counter. `last` is what lets the + // client lift itself out. + // ⚠️ Defensive: the session arrives through a structural port, and an + // implementation without this method must not take the whole input + // path down with it — a throw here aborts the message handler and the + // frame is never ACKed at all, which strands it in the client's queue. + const watermark = + typeof (session as { lastInputSeq?: (c: string) => number }).lastInputSeq === 'function' + ? (session as { lastInputSeq: (c: string) => number }).lastInputSeq(cid as string) + : seq; + socket.send(`{"t":"ia","seq":${seq},"dup":true,"last":${watermark}}`); + } + } + }; + // Attach message handler synchronously BEFORE any async work // (@fastify/websocket requirement to avoid dropped messages). socket.on('message', (raw) => { try { const msg = JSON.parse(String(raw)); if (msg.t === 'i' && typeof msg.d === 'string') { + if (inputStalled) return; if (msg.d.length > MAX_INPUT_LENGTH) { // Refused for good, so say so: a silent return left the frame // unACKed and the client redelivered it every few seconds forever @@ -185,51 +247,27 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost } return; } - // Reliable delivery: when the frame carries a clientId + seq, apply it - // exactly once (skip a duplicate redelivery) but ACK it regardless so - // the client can drop it from its durable queue. Frames without seq - // (legacy/other tools) are applied as-is — no behavior change. - const cid = typeof msg.cid === 'string' ? msg.cid : null; - const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null; - const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true; - let delivered = true; - if (apply) { - // Typed input from a claim-holding desktop keeps the claim "hot" - // and re-asserts the desktop layout after a mobile override. - if (holdsDesktopClaim) session.noteDesktopActivity(); - // Browser keystrokes are the user's own, so they may name the tab. - delivered = session.write(msg.d, { fromUser: true }); - // A session whose PTY is gone swallows the write. ACKing anyway told - // the client to drop the frame from its durable queue and left the seq - // burnt, so the retry that reliable delivery exists for was rejected as - // a duplicate: the input was lost for good. - if (!delivered && cid && seq !== null) session.forgetInputSeq(cid, seq); - } - if (seq !== null && socket.readyState === 1) { - if (apply) { - if (delivered) socket.send(`{"t":"ia","seq":${seq}}`); - } else { - // REJECTED as a duplicate. ACK it — the client must still drop it - // from its durable queue — but say so, and hand back our watermark. - // - // A plain ACK here is indistinguishable from "applied", which is - // what made a client with a rolled-back counter unrecoverable: its - // seqs persist to localStorage on a DEBOUNCED write, so a tab killed - // between a send and that write comes back with a counter BELOW this - // watermark, every later keystroke lands at or under it, and each one - // is dropped-but-ACKed. The UI stays clean, nothing is delivered, and - // a reload restores the same stale counter. `last` is what lets the - // client lift itself out. - // ⚠️ Defensive: the session arrives through a structural port, and an - // implementation without this method must not take the whole input - // path down with it — a throw here aborts the message handler and the - // frame is never ACKed at all, which strands it in the client's queue. - const watermark = - typeof (session as { lastInputSeq?: (c: string) => number }).lastInputSeq === 'function' - ? (session as { lastInputSeq: (c: string) => number }).lastInputSeq(cid as string) - : seq; - socket.send(`{"t":"ia","seq":${seq},"dup":true,"last":${watermark}}`); - } + // A guarded submit (agent-staging.ts) holds the input lease across its + // awaited Enter, so frames keep arriving while it is held. Refusing one + // lost it: this path does not wait for ACKs, so the next seq landed after + // the lease, moved the watermark past the refused one, and its redelivery + // was then dropped as a duplicate. Hold the frame instead; the session runs + // it, ACK included, in arrival order the moment the lease is released. + // ⚠️ Defensive, like `lastInputSeq` below: the session arrives through a + // structural port, and one without this method must still deliver input + // (a throw here would drop the frame unACKed). + const held = + typeof session.holdForGuardedLease === 'function' + ? session.holdForGuardedLease(() => applyInputFrame(msg), msg.d.length, msg.d) + : 'free'; + if (held === 'free') { + applyInputFrame(msg); + } else if (held === 'full') { + // Not held, not applied, not ACKed, and no later frame from this socket + // is applied either, so nothing can overtake it. + inputStalled = true; + console.warn('[ws] input backlog during a guarded submit; closing for redelivery', { sessionId: id }); + socket.close(1013, 'Input backlog during guarded submit'); } } else if ( msg.t === 'z' && diff --git a/test/agent-staging-session.test.ts b/test/agent-staging-session.test.ts index ff106fdf..38d2d852 100644 --- a/test/agent-staging-session.test.ts +++ b/test/agent-staging-session.test.ts @@ -50,7 +50,7 @@ it('fences writes and reused-pid pane lifecycles without persisting prompt conte expect(session.guardedInputSnapshot()).toBeNull(); }); -it('checks conditional Enter at the dispatch boundary and rejects concurrent browser input', async () => { +it('checks conditional Enter at the dispatch boundary and holds concurrent input until after it', async () => { let dispatch: (() => void) | undefined; const writes: string[] = []; const mux = { @@ -84,32 +84,41 @@ it('checks conditional Enter at the dispatch boundary and rejects concurrent bro muxSession: { muxName: 'codeman-test', sessionId: 'test', createdAt: 1, pid: 1 }, } as unknown as ConstructorParameters[0]); try { + const settle = () => new Promise((resolve) => setImmediate(resolve)); const expected = session.guardedInputSnapshot()!; const sending = session.submitGuardedInput(expected); - expect(await session.writeViaMux('other')).toBe(false); + // Held behind the lease, not refused: it still invalidates the pending Enter, and + // it goes out once the lease is released. + const other = session.writeViaMux('other'); dispatch!(); expect(await sending).toBe(false); expect(writes).toEqual([]); + await settle(); + dispatch!(); + expect(await other).toBe(true); + expect(writes).toEqual(['other']); const next = session.submitGuardedInput(session.guardedInputSnapshot()!); dispatch!(); expect(await next).toBe(true); - expect(writes).toEqual(['\r']); + expect(writes).toEqual(['other', '\r']); const restarting = session.submitGuardedInput(session.guardedInputSnapshot()!); await (session as unknown as { _withPaneLifecycle(op: () => Promise): Promise })._withPaneLifecycle( async () => {} ); dispatch!(); expect(await restarting).toBe(false); - expect(writes).toEqual(['\r']); + expect(writes).toEqual(['other', '\r']); const auxiliary = session.submitGuardedInput(session.guardedInputSnapshot()!); - await expect( - session.performAuxiliaryInput(async () => { - throw new Error('must not execute'); - }) - ).rejects.toThrow('Guarded submission'); + const key = session.performAuxiliaryInput(async () => { + writes.push('aux'); + }); + await settle(); + // The named key waits for the lease instead of failing, and never runs inside it. + expect(writes).toEqual(['other', '\r']); dispatch!(); expect(await auxiliary).toBe(false); - expect(writes).toEqual(['\r']); + await key; + expect(writes).toEqual(['other', '\r', 'aux']); } finally { await session.stop(false); } diff --git a/test/guarded-lease-held-input.test.ts b/test/guarded-lease-held-input.test.ts new file mode 100644 index 00000000..ffeb0a99 --- /dev/null +++ b/test/guarded-lease-held-input.test.ts @@ -0,0 +1,395 @@ +/** + * @fileoverview Input that lands while a guarded submit holds the input lease is held + * and applied in arrival order once the lease is released, never refused and lost. + * + * Shaped like the review repro: a real `Session`, a mux that runs `beforeDispatch` and + * then awaits the Enter the way `TmuxManager.sendInput` does, and the real WebSocket + * route driven through a fake socket. The browser's WebSocket path does not wait for + * ACKs, so the client model below sends every frame at once and redelivers whatever + * is still unACKed afterwards, in seq order, the way `_redeliverSweep` does. + */ +import { EventEmitter } from 'node:events'; +import { afterEach, describe, expect, it } from 'vitest'; +import type { FastifyInstance } from 'fastify'; +import { formatPaneSnapshot } from '../src/tmux-manager.js'; +import { GuardedInputBusyError, Session } from '../src/session.js'; +import { registerWsRoutes } from '../src/web/routes/ws-routes.js'; +import type { SessionPort } from '../src/web/ports/session-port.js'; + +type Frame = { t: 'i'; d: string; seq?: number; cid?: string }; + +class FakeSocket extends EventEmitter { + readyState = 1; + sent: Array> = []; + closedWith: number | null = null; + send(data: string): void { + this.sent.push(JSON.parse(data)); + } + close(code?: number): void { + this.closedWith = code ?? 1000; + this.readyState = 3; + } + ping(): void {} + terminate(): void { + this.readyState = 3; + } + frame(msg: Frame): void { + this.emit('message', Buffer.from(JSON.stringify(msg))); + } + acks(): number[] { + return this.sent.filter((m) => m.t === 'ia').map((m) => m.seq as number); + } +} + +interface Harness { + session: Session; + pty: string[]; + /** Resolve the in-flight Enter (true = tmux accepted it, false = it failed). */ + finishEnter(ok?: boolean | Error): void; + enterDispatched(): boolean; + connect(cid: string): FakeSocket; + stop(): Promise; +} + +const sockets: FakeSocket[] = []; + +function harness(awaitBeforeGuard = false): Harness { + const pty: string[] = []; + let finish: ((ok: boolean | Error) => void) | undefined; + let dispatched = false; + const mux = { + isAvailable: () => true, + capturePaneBuffer: ( + _name: string, + _pane?: string, + opts?: { capturedGeometry?: { cols: number; rows: number } } + ) => { + if (opts) opts.capturedGeometry = { cols: 100, rows: 20 }; + return formatPaneSnapshot(['› hello', ' GPT-6-Luna default · ~/case'], { + cols: 100, + rows: 20, + cursorX: 7, + cursorY: 0, + }); + }, + // Like TmuxManager.sendInput's Enter-only branch: the guard runs synchronously, + // then the `send-keys Enter` child process is awaited with the lease still held. + sendInput: async (_id: string, text: string, options?: { beforeDispatch?: () => boolean }) => { + if (options?.beforeDispatch && text !== '\r') return false; + const settle = () => + new Promise((resolve) => { + finish = resolve; + }); + // A transport that awaits before its guard leaves a window in which held + // keystrokes must still invalidate the Enter. + const early = awaitBeforeGuard ? await settle() : undefined; + if (options?.beforeDispatch && !options.beforeDispatch()) return false; + dispatched = true; + const outcome = early ?? (await settle()); + if (outcome instanceof Error) throw outcome; + if (outcome) pty.push(text); + return outcome; + }, + }; + const session = new Session({ + workingDir: '/tmp', + mode: 'codex', + mux, + muxSession: { muxName: 'codeman-test', sessionId: 'test', createdAt: 1, pid: 1 }, + } as unknown as ConstructorParameters[0]); + const access = session as unknown as { ptyProcess: { write(data: string): void } | null }; + access.ptyProcess = { write: (data: string) => pty.push(data) }; + + let handler: ((socket: FakeSocket, req: unknown) => void) | undefined; + const app = { + get: (_path: string, _opts: unknown, h: (socket: FakeSocket, req: unknown) => void) => { + handler = h; + }, + } as unknown as FastifyInstance; + const ctx = { sessions: new Map([[session.id, session]]) } as unknown as SessionPort; + registerWsRoutes(app, ctx, () => ({ bindHost: '127.0.0.1', allowedHosts: [] })); + + return { + session, + pty, + finishEnter: (ok = true) => finish!(ok), + enterDispatched: () => dispatched, + connect: (cid: string) => { + const socket = new FakeSocket(); + sockets.push(socket); + handler!(socket, { headers: { host: 'localhost:3000' }, params: { id: session.id }, query: { cid } }); + return socket; + }, + stop: async () => { + access.ptyProcess = null; + await session.stop(false); + }, + }; +} + +/** Redeliver every still-unACKed frame in seq order, as the browser's sweep does. */ +function redeliverUnacked(socket: FakeSocket, frames: Frame[]): void { + const acked = new Set(socket.acks()); + for (const f of frames) if (f.seq !== undefined && !acked.has(f.seq)) socket.frame(f); +} + +afterEach(() => { + for (const s of sockets.splice(0)) s.emit('close', 1000, Buffer.from('')); +}); + +describe('input during a guarded submit lease', () => { + it('delivers a keystroke that landed during the lease before the one typed after it (review repro)', async () => { + const h = harness(); + try { + const socket = h.connect('c1'); + const a: Frame = { t: 'i', d: 'a', seq: 1, cid: 'c1' }; + const b: Frame = { t: 'i', d: 'b', seq: 2, cid: 'c1' }; + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + expect(h.enterDispatched()).toBe(true); + expect(h.session.guardedSubmitInFlight).toBe(true); + socket.frame(a); + // Held, not refused: no ACK yet and nothing written while the lease is held. + expect(socket.acks()).toEqual([]); + expect(h.pty).toEqual([]); + h.finishEnter(true); + await expect(submit).resolves.toBe(true); + socket.frame(b); + redeliverUnacked(socket, [a, b]); + expect(h.pty).toEqual(['\r', 'a', 'b']); + expect(socket.acks()).toEqual([1, 2]); + expect(socket.sent.filter((m) => m.dup)).toEqual([]); + } finally { + await h.stop(); + } + }); + + it('keeps arrival order for several frames during the lease and one after it', async () => { + const h = harness(); + try { + const socket = h.connect('c1'); + const frames: Frame[] = ['a', 'b', 'c', 'd'].map((d, i) => ({ t: 'i', d, seq: i + 1, cid: 'c1' })); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + for (const f of frames.slice(0, 3)) socket.frame(f); + expect(h.pty).toEqual([]); + h.finishEnter(true); + await submit; + socket.frame(frames[3]); + redeliverUnacked(socket, frames); + expect(h.pty).toEqual(['\r', 'a', 'b', 'c', 'd']); + expect(socket.acks()).toEqual([1, 2, 3, 4]); + } finally { + await h.stop(); + } + }); + + it('interleaves a second client by arrival, each with its own seq space', async () => { + const h = harness(); + try { + const one = h.connect('c1'); + const two = h.connect('c2'); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + one.frame({ t: 'i', d: 'a', seq: 1, cid: 'c1' }); + two.frame({ t: 'i', d: 'x', seq: 1, cid: 'c2' }); + one.frame({ t: 'i', d: 'b', seq: 2, cid: 'c1' }); + h.finishEnter(true); + await submit; + two.frame({ t: 'i', d: 'y', seq: 2, cid: 'c2' }); + expect(h.pty).toEqual(['\r', 'a', 'x', 'b', 'y']); + expect(one.acks()).toEqual([1, 2]); + expect(two.acks()).toEqual([1, 2]); + } finally { + await h.stop(); + } + }); + + it('flushes held input when the lease ends in failure, refused Enter or thrown', async () => { + for (const outcome of [false, new Error('tmux send-keys failed')]) { + const h = harness(); + try { + const socket = h.connect('c1'); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + socket.frame({ t: 'i', d: 'a', seq: 1, cid: 'c1' }); + h.finishEnter(outcome); + if (outcome instanceof Error) await expect(submit).rejects.toThrow('tmux send-keys failed'); + else await expect(submit).resolves.toBe(false); + socket.frame({ t: 'i', d: 'b', seq: 2, cid: 'c1' }); + expect(h.pty).toEqual(['a', 'b']); + expect(socket.acks()).toEqual([1, 2]); + expect(h.session.guardedSubmitInFlight).toBe(false); + } finally { + await h.stop(); + } + } + }); + + it('holds untagged frames too, which have no redelivery to fall back on', async () => { + const h = harness(); + try { + const socket = h.connect('c1'); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + socket.frame({ t: 'i', d: 'a' }); + h.finishEnter(true); + await submit; + socket.frame({ t: 'i', d: 'b' }); + expect(h.pty).toEqual(['\r', 'a', 'b']); + } finally { + await h.stop(); + } + }); + + it('a frame held for a socket that closed is still applied once; its redelivery is a duplicate', async () => { + const h = harness(); + try { + const first = h.connect('c1'); + const a: Frame = { t: 'i', d: 'a', seq: 1, cid: 'c1' }; + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + first.frame(a); + first.close(); + first.emit('close', 1006, Buffer.from('')); + h.finishEnter(true); + await submit; + expect(first.sent).toEqual([]); + const second = h.connect('c1'); + second.frame(a); + second.frame({ t: 'i', d: 'b', seq: 2, cid: 'c1' }); + expect(h.pty).toEqual(['\r', 'a', 'b']); + expect(second.sent.filter((m) => m.t === 'ia')).toEqual([ + { t: 'ia', seq: 1, dup: true, last: 1 }, + { t: 'ia', seq: 2 }, + ]); + } finally { + await h.stop(); + } + }); + + it('closes the socket instead of holding past the cap, and the reconnect resends in order', async () => { + const h = harness(); + try { + const socket = h.connect('c1'); + const cap = Session.MAX_GUARDED_HELD_INPUTS; + const frames: Frame[] = Array.from({ length: cap + 3 }, (_, i) => ({ + t: 'i', + d: `k${i}`, + seq: i + 1, + cid: 'c1', + })); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + for (const f of frames) socket.frame(f); + // Frames past the cap are not held, not applied and not ACKed; the socket is + // closed so the browser reconnects and resends everything still unACKed. + expect(socket.closedWith).toBe(1013); + h.finishEnter(true); + await submit; + expect(h.pty.length).toBe(1 + cap); + expect(socket.acks()).toEqual([]); + socket.emit('close', 1013, Buffer.from('')); + const again = h.connect('c1'); + for (const f of frames) again.frame(f); + expect(h.pty).toEqual(['\r', ...frames.map((f) => f.d)]); + expect(again.acks()).toEqual(frames.map((f) => f.seq)); + } finally { + await h.stop(); + } + }); + + it('holds managed writes and named keys behind the lease, and refuses them only when the hold queue is full', async () => { + const h = harness(); + try { + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + const order: string[] = []; + const key = h.session.performAuxiliaryInput(async () => { + order.push('key'); + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(order).toEqual([]); + for (let i = 1; i < Session.MAX_GUARDED_HELD_INPUTS; i++) { + expect(h.session.holdForGuardedLease(() => order.push('frame'))).toBe('held'); + } + expect(h.session.holdForGuardedLease(() => order.push('lost'))).toBe('full'); + await expect(h.session.performAuxiliaryInput(async () => order.push('lost'))).rejects.toBeInstanceOf( + GuardedInputBusyError + ); + await expect(h.session.writeViaMux('lost')).resolves.toBe(false); + h.finishEnter(true); + await submit; + await key; + expect(order[0]).toBe('key'); + expect(order.filter((o) => o === 'frame')).toHaveLength(Session.MAX_GUARDED_HELD_INPUTS - 1); + expect(order).not.toContain('lost'); + } finally { + await h.stop(); + } + }); + + it('caps held bytes as well as held frames', async () => { + const h = harness(); + try { + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + expect(h.session.holdForGuardedLease(() => {}, Session.MAX_GUARDED_HELD_BYTES)).toBe('held'); + expect(h.session.holdForGuardedLease(() => {}, 1)).toBe('full'); + h.finishEnter(true); + await submit; + expect(h.session.holdForGuardedLease(() => {}, Session.MAX_GUARDED_HELD_BYTES + 1)).toBe('free'); + } finally { + await h.stop(); + } + }); + + it('a frame arriving while a held named key is still being sent waits for it', async () => { + const h = harness(); + try { + const socket = h.connect('c1'); + let keySent!: () => void; + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + const key = h.session.performAuxiliaryInput(async () => { + await new Promise((resolve) => { + keySent = resolve; + }); + h.pty.push(''); + }); + socket.frame({ t: 'i', d: 'a', seq: 1, cid: 'c1' }); + h.finishEnter(true); + await submit; + // The lease is over, but the held key is still in flight: input stays held. + expect(h.session.guardedSubmitInFlight).toBe(false); + expect(h.session.guardedInputHeld).toBe(true); + socket.frame({ t: 'i', d: 'b', seq: 2, cid: 'c1' }); + expect(h.pty).toEqual(['\r']); + keySent(); + await key; + await new Promise((resolve) => setImmediate(resolve)); + expect(h.pty).toEqual(['\r', '', 'a', 'b']); + expect(socket.acks()).toEqual([1, 2]); + expect(h.session.guardedInputHeld).toBe(false); + } finally { + await h.stop(); + } + }); + + it('a keystroke held before the guard runs still invalidates the Enter, then lands', async () => { + const h = harness(true); + try { + const socket = h.connect('c1'); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + socket.frame({ t: 'i', d: 'a', seq: 1, cid: 'c1' }); + h.finishEnter(true); + await expect(submit).resolves.toBe(false); + expect(h.enterDispatched()).toBe(false); + expect(h.pty).toEqual(['a']); + expect(socket.acks()).toEqual([1]); + } finally { + await h.stop(); + } + }); +}); diff --git a/test/routes/session-routes.test.ts b/test/routes/session-routes.test.ts index f770a76c..cb9d4891 100644 --- a/test/routes/session-routes.test.ts +++ b/test/routes/session-routes.test.ts @@ -25,7 +25,7 @@ import { tmpdir } from 'node:os'; import { createMockRouteContext, type MockRouteContext } from '../mocks/index.js'; import { installRouteErrorHandler } from '../../src/web/route-error-handler.js'; import { ApiErrorCode, httpStatusForErrorCode } from '../../src/types.js'; -import { Session } from '../../src/session.js'; +import { GuardedInputBusyError, Session } from '../../src/session.js'; import { agentInbox } from '../../src/web/agent-inbox.js'; // Mock execFile so the send-key route's `tmux` invocation is observable (not run for real). @@ -265,6 +265,33 @@ describe('session-routes', () => { }); }); + it('answers 409 CONFLICT, without an error log, when a guarded submit cannot hold the key', async () => { + execFile.mockReset(); + const session = harness.ctx._session as unknown as { + performAuxiliaryInput: (op: () => Promise) => Promise; + }; + const original = session.performAuxiliaryInput; + session.performAuxiliaryInput = async () => { + throw new GuardedInputBusyError(); + }; + const errors = vi.spyOn(console, 'error').mockImplementation(() => {}); + try { + const res = await harness.app.inject({ + method: 'POST', + url: '/api/sessions/test-session-1/send-key', + payload: { key: 'S-Enter' }, + }); + expect(res.statusCode).toBe(409); + expect(res.json()).toMatchObject({ success: false, errorCode: 'CONFLICT' }); + expect(res.json().error).toMatch(/guarded submit is in progress.*nothing was written/); + expect(errors).not.toHaveBeenCalled(); + expect(execFile).not.toHaveBeenCalled(); + } finally { + errors.mockRestore(); + session.performAuxiliaryInput = original; + } + }); + it('rejects keys outside the hex allowlist without invoking tmux', async () => { execFile.mockReset(); const res = await harness.app.inject({ @@ -1203,7 +1230,7 @@ describe('session-routes', () => { const session = harness.ctx.sessions.get(harness.ctx._sessionId) as any; session.writeBuffer.length = 0; const post = (payload: unknown) => harness.app.inject({ method: 'POST', url, payload }); - session.guardedSubmitInFlight = true; + session.guardedInputHeld = true; for (const payload of [ { input: 'x' }, { input: 'x', useMux: true }, @@ -1218,7 +1245,7 @@ describe('session-routes', () => { expect(session.writeBuffer).toEqual([]); // Once the lease is released the browser's retry of the same seq is applied, not // rejected as a duplicate. - session.guardedSubmitInFlight = false; + session.guardedInputHeld = false; expect((await post({ input: 'tagged', seq: 7, clientId: 'cid-guard' })).statusCode).toBe(200); expect(session.writeBuffer).toEqual(['tagged']); });