diff --git a/docs/api-reference.md b/docs/api-reference.md index 2476f983..1be7dc6f 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -426,11 +426,12 @@ worker's prompt but never submitted, and the wait then runs its full timeout on turn that never started. Verified live; this is the most common silent failure on this endpoint. -Any input, waiting or not, answers `409 CONFLICT` with nothing written while an -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. +Input tagged with `clientId` and `seq`, waiting or not, answers `409 CONFLICT` +with nothing written while an 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. Untagged input is held instead and written in arrival order once the +lease is released. A **plain prompt** (printable text followed by exactly one `\r`, nothing else) is delivered through tmux even without `useMux`: the text is typed, Enter is pressed as @@ -716,15 +717,18 @@ 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 +holds everything behind it. A tagged `POST /api/v1/sessions/:id/input` (one +carrying `clientId` and `seq`) 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; it 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. +Untagged input to that route is held like a WebSocket frame, because its senders +(the phone accessory keys, the Stop button) never read the answer. 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`. +unacknowledged frame in order, and `send-key` and an untagged raw input answer +`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 2e2495eb..34016ed2 100644 --- a/docs/architecture-invariants.md +++ b/docs/architecture-invariants.md @@ -1330,8 +1330,15 @@ N+1 and its redelivery dropped as a duplicate. A WS frame is ACKed only once wri 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 +.../input` answers `409` (seq returned) to TAGGED input for the lease plus the drain, +since its browser drainer is serialized and retries; untagged input there (the phone +accessory keys and the Stop button's Ctrl+C, fire-and-forget POSTs that never read the +answer) is held instead, the raw write through `holdForGuardedLease()` and a mux write +by `writeViaMux`'s own turn (never wrap `writeViaMux` in a hold: the drain would queue +it again behind later input and wait on itself). `write()` refuses during the lease +only, since the drain writes held input through it, so a direct `write()` caller such +as cron's `paste` mode (`cron-service.ts`) can land ahead of input still draining. +⚠️ 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 diff --git a/src/session.ts b/src/session.ts index dc43c4ff..08dd83d3 100644 --- a/src/session.ts +++ b/src/session.ts @@ -4784,6 +4784,9 @@ export class Session extends EventEmitter { * 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. + * `write()` is the known exception: it refuses during the lease only, since the drain + * writes held input through it, so a direct `write()` caller (cron's `paste` mode in + * `cron/cron-service.ts`) can still land ahead of input that is draining. */ get guardedInputHeld(): boolean { return this._guardedSubmitInFlight || this._guardedHeld.length > 0 || this._guardedHeldDraining; @@ -4983,6 +4986,7 @@ export class Session extends EventEmitter { */ write(data: string, options: SessionWriteOptions = {}): boolean { this._observeGuardedKeyboardInput(data); + // The lease only, not the drain, which writes held input through here (see guardedInputHeld). if (this._guardedSubmitInFlight) return false; const submittedPrompt = this._trackSubmit(data, options); if (!this.ptyProcess) return false; diff --git a/src/web/routes/session-routes.ts b/src/web/routes/session-routes.ts index ae4a523f..1d72f986 100644 --- a/src/web/routes/session-routes.ts +++ b/src/web/routes/session-routes.ts @@ -2080,15 +2080,18 @@ export function registerSessionRoutes( // A guarded submit (agent-staging.ts) holds the input lease for the few // 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 + // and written in arrival order after it. A TAGGED request is refused 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 + // POST drainer sends one frame at a time) delivers it on its next retry. An + // UNTAGGED request has no retry behind it (the phone accessory keys and the Stop + // button's Ctrl+C are fire-and-forget POSTs that never read the answer), so it is + // held instead, like a WebSocket frame: `writeViaMux` takes its own turn behind + // the lease, and the raw write below is held explicitly. Nothing from here to the + // write below awaits, so the lease cannot be taken in between. + const refuseGuarded = () => + reply .code(409) .send( createErrorResponse( @@ -2096,6 +2099,9 @@ export function registerSessionRoutes( 'A guarded submit is in progress on this session; nothing was written. Retry the input.' ) ); + if (!duplicate && tagged && session.guardedInputHeld === true) { + session.forgetInputSeq(clientId as string, seq as number); + return refuseGuarded(); } // Only a waiting request pays for the tmux probe: the browser's plain input path @@ -2208,7 +2214,38 @@ export function registerSessionRoutes( // Same rollback. NOT an error response, deliberately: a session can // legitimately have no PTY yet (created but not started), and callers have // always been able to write to one without a 4xx. - delivered = session.write(inputStr, { fromUser: true }); + // + // During a guarded submit (untagged only: a tagged one was refused above) the + // write is held and runs in arrival order once the lease is released. The answer + // is a 200 once held; a send-and-wait reads `delivered` from the held write. Only + // a full hold queue refuses, after releasing the waiter registered above. + let heldWrite: Promise | null = null; + if (session.guardedInputHeld === true) { + let settle!: (written: boolean) => void; + const written = new Promise((resolve) => { + settle = resolve; + }); + const held = session.holdForGuardedLease( + () => { + let ok = false; + try { + ok = session.write(inputStr, { fromUser: true }); + } finally { + settle(ok); + } + }, + inputStr.length, + inputStr + ); + if (held === 'full') { + abort.abort(); + stopDeathWatch(); + return refuseGuarded(); + } + if (held === 'held') heldWrite = written; + } + if (heldWrite && !waitPromise) return {}; + delivered = heldWrite ? await heldWrite : session.write(inputStr, { fromUser: true }); if (!delivered && tagged) { session.forgetInputSeq(clientId as string, seq as number); } diff --git a/test/guarded-lease-held-input.test.ts b/test/guarded-lease-held-input.test.ts index ffeb0a99..31191fcb 100644 --- a/test/guarded-lease-held-input.test.ts +++ b/test/guarded-lease-held-input.test.ts @@ -9,12 +9,14 @@ * is still unACKed afterwards, in seq order, the way `_redeliverSweep` does. */ import { EventEmitter } from 'node:events'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } 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 { registerSessionRoutes } from '../src/web/routes/session-routes.js'; import type { SessionPort } from '../src/web/ports/session-port.js'; +import { createRouteTestHarness } from './routes/_route-test-utils.js'; type Frame = { t: 'i'; d: string; seq?: number; cid?: string }; @@ -376,6 +378,59 @@ describe('input during a guarded submit lease', () => { } }); + it('holds an untagged raw POST .../input during the lease and writes it after the Enter', async () => { + const h = harness(); + const route = await createRouteTestHarness(registerSessionRoutes); + try { + (route.ctx.sessions as unknown as Map).set(h.session.id, h.session); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + expect(h.session.guardedSubmitInFlight).toBe(true); + // The Stop button's Ctrl+C: an untagged POST whose answer nobody reads. + const res = await route.app.inject({ + method: 'POST', + url: `/api/sessions/${h.session.id}/input`, + payload: { input: '\x03' }, + }); + expect(res.statusCode).toBe(200); + expect(h.pty).toEqual([]); + h.finishEnter(true); + await expect(submit).resolves.toBe(true); + expect(h.pty).toEqual(['\r', '\x03']); + } finally { + await route.app.close(); + await h.stop(); + } + }); + + it('a held untagged raw send-and-wait reports the held write as delivered', async () => { + const h = harness(); + const route = await createRouteTestHarness(registerSessionRoutes); + try { + (route.ctx.sessions as unknown as Map).set(h.session.id, h.session); + const submit = h.session.submitGuardedInput(h.session.guardedInputSnapshot()!); + await Promise.resolve(); + const answer = route.app.inject({ + method: 'POST', + url: `/api/sessions/${h.session.id}/input`, + payload: { input: '\x03', useMux: false, wait: 'exit', waitTimeout: 1000 }, + }); + // Private, but the only proof the route reached the hold before the lease ends. + const queue = (h.session as unknown as { _guardedHeld: unknown[] })._guardedHeld; + await vi.waitFor(() => expect(queue).toHaveLength(1)); + expect(h.pty).toEqual([]); + h.finishEnter(true); + await submit; + const res = await answer; + expect(res.statusCode).toBe(200); + expect(res.json().data).toMatchObject({ delivered: true, duplicate: false }); + expect(h.pty).toEqual(['\r', '\x03']); + } finally { + await route.app.close(); + await h.stop(); + } + }); + it('a keystroke held before the guard runs still invalidates the Enter, then lands', async () => { const h = harness(true); try { diff --git a/test/mocks/mock-session.ts b/test/mocks/mock-session.ts index fa2b8c5e..d5df6104 100644 --- a/test/mocks/mock-session.ts +++ b/test/mocks/mock-session.ts @@ -111,6 +111,22 @@ export class MockSession extends EventEmitter { await operation(); } + /** Mirrors Session.guardedInputHeld; a route test sets it to stand in for a guarded submit's lease. */ + guardedInputHeld = false; + /** Input held behind that lease, in arrival order (see releaseGuardedLease). */ + heldGuardedInput: Array<() => void | Promise> = []; + /** Mirrors Session.holdForGuardedLease, without the cap (a test replaces it to model a full queue). */ + holdForGuardedLease(run: () => void | Promise, _bytes = 0, _data?: string): 'free' | 'held' | 'full' { + if (!this.guardedInputHeld) return 'free'; + this.heldGuardedInput.push(run); + return 'held'; + } + /** Ends the stand-in lease and runs the held input in arrival order, as Session does. */ + releaseGuardedLease(): void { + this.guardedInputHeld = false; + for (const run of this.heldGuardedInput.splice(0)) void run(); + } + private _muxName: string | null = null; constructor(id: string = 'mock-session-id') { @@ -134,9 +150,14 @@ export class MockSession extends EventEmitter { /** Write via mux (used by respawn controller) */ async writeViaMux(data: string): Promise { - if (this.failWrites) return false; - this.writeBuffer.push(data); - return true; + const write = (): boolean => { + if (this.failWrites) return false; + this.writeBuffer.push(data); + return true; + }; + // Like Session.writeViaMux, it waits its turn behind a stand-in guarded lease. + if (!this.guardedInputHeld) return write(); + return new Promise((resolve) => this.heldGuardedInput.push(() => resolve(write()))); } /** diff --git a/test/routes/session-routes.test.ts b/test/routes/session-routes.test.ts index cb9d4891..39940a96 100644 --- a/test/routes/session-routes.test.ts +++ b/test/routes/session-routes.test.ts @@ -1224,7 +1224,7 @@ describe('session-routes', () => { expect(session.writeBuffer).toEqual(['prompt', '\r']); }); - it('refuses with 409 and gives the seq back while a guarded submit holds the input lease', async () => { + it('refuses tagged input with 409 and gives the seq back while a guarded submit holds the input lease', async () => { const url = `/api/sessions/${harness.ctx._sessionId}/input`; // eslint-disable-next-line @typescript-eslint/no-explicit-any const session = harness.ctx.sessions.get(harness.ctx._sessionId) as any; @@ -1232,10 +1232,8 @@ describe('session-routes', () => { const post = (payload: unknown) => harness.app.inject({ method: 'POST', url, payload }); session.guardedInputHeld = true; for (const payload of [ - { input: 'x' }, - { input: 'x', useMux: true }, - { input: 'prompt\r' }, { input: 'tagged', seq: 7, clientId: 'cid-guard' }, + { input: 'prompt\r', useMux: true, seq: 1, clientId: 'cid-guard-mux' }, ]) { const refused = await post(payload); expect(refused.statusCode, JSON.stringify(payload)).toBe(409); @@ -1243,13 +1241,76 @@ describe('session-routes', () => { expect(refused.json().error).toMatch(/guarded submit is in progress.*nothing was written/); } expect(session.writeBuffer).toEqual([]); + expect(session.heldGuardedInput).toEqual([]); // Once the lease is released the browser's retry of the same seq is applied, not // rejected as a duplicate. - session.guardedInputHeld = false; + session.releaseGuardedLease(); expect((await post({ input: 'tagged', seq: 7, clientId: 'cid-guard' })).statusCode).toBe(200); expect(session.writeBuffer).toEqual(['tagged']); }); + it('holds an untagged raw POST during a guarded submit and writes it once the lease is released', async () => { + // The phone accessory keys and the Stop button's Ctrl+C are untagged, fire-and-forget + // POSTs that never read the answer, so a 409 here silently lost the tap. + const url = `/api/sessions/${harness.ctx._sessionId}/input`; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const session = harness.ctx.sessions.get(harness.ctx._sessionId) as any; + session.writeBuffer.length = 0; + session.guardedInputHeld = true; + try { + for (const input of ['\x03', '\x1b[A']) { + const res = await harness.app.inject({ method: 'POST', url, payload: { input } }); + expect(res.statusCode, JSON.stringify(input)).toBe(200); + expect(res.json()).toMatchObject({ success: true }); + } + expect(session.writeBuffer).toEqual([]); + session.releaseGuardedLease(); + expect(session.writeBuffer).toEqual(['\x03', '\x1b[A']); + } finally { + session.releaseGuardedLease(); + } + }); + + it('lets untagged mux input wait its own turn behind the lease instead of refusing it', async () => { + const url = `/api/sessions/${harness.ctx._sessionId}/input`; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + 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.guardedInputHeld = true; + try { + // Fire-and-forget `useMux`: answered at once, written in its turn. + expect((await post({ input: 'x', useMux: true })).statusCode).toBe(200); + // A plain prompt's mux write is awaited, so it answers once written. + const prompt = post({ input: 'prompt\r' }); + await vi.waitFor(() => expect(session.heldGuardedInput).toHaveLength(2)); + expect(session.writeBuffer).toEqual([]); + session.releaseGuardedLease(); + expect((await prompt).statusCode).toBe(200); + expect(session.writeBuffer).toEqual(['x', 'prompt\r']); + } finally { + session.releaseGuardedLease(); + } + }); + + it('refuses an untagged raw POST with 409 only when the hold queue is full', async () => { + const url = `/api/sessions/${harness.ctx._sessionId}/input`; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const session = harness.ctx.sessions.get(harness.ctx._sessionId) as any; + session.writeBuffer.length = 0; + session.guardedInputHeld = true; + session.holdForGuardedLease = () => 'full'; + try { + const res = await harness.app.inject({ method: 'POST', url, payload: { input: '\x03' } }); + expect(res.statusCode).toBe(409); + expect(res.json()).toMatchObject({ success: false, errorCode: 'CONFLICT' }); + expect(session.writeBuffer).toEqual([]); + } finally { + delete session.holdForGuardedLease; + session.releaseGuardedLease(); + } + }); + it('always applies untagged input (curl/legacy, no dedup)', async () => { const url = `/api/sessions/${harness.ctx._sessionId}/input`; // eslint-disable-next-line @typescript-eslint/no-explicit-any