From ebfcac6ad125e7f100377caebf10b6566dce98d4 Mon Sep 17 00:00:00 2001 From: Claudia Date: Fri, 7 Aug 2026 01:36:33 +0200 Subject: [PATCH] fix(api,ws): an input whose delivery fails can be retried instead of being lost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both input paths recorded the (clientId, seq) pair as applied and acknowledged the frame BEFORE knowing whether the write had landed: the POST route because its mux write is fire-and-forget so the response never waits on a tmux child, the WebSocket handler because it ACKed unconditionally. When the write then failed, the client dropped the frame from its durable queue and the server rejected the retry as a duplicate. The reliable-delivery layer was guaranteeing exactly-once delivery of something that had never been delivered — and `Session.write()` returned void, so a session whose PTY was gone swallowed the data with no signal at all. - `forgetInputSeq()` rolls the bookkeeping back on failure, but only when that seq is still the newest one; a later input has superseded it and must not re-open. - The WebSocket handler withholds its ACK when the write did not land, so the client redelivers. - `Session.write()` reports whether it reached a PTY. Response codes are unchanged, deliberately: a session can legitimately have no PTY yet, and turning that into a failure status would be a contract change of its own. What this does NOT do: remove the root cause. The POST still answers 200 before the mux write is attempted, so a client that treats any 2xx as final cannot learn about that failure. What closes is the narrower window — the write failed AND the ACK never reached the client — plus the whole WebSocket path. Closing the rest would mean awaiting the tmux child inside the request. 9 tests. They drive the HTTP route, not only the Session primitives: with the rollback removed from the route, 2 of them fail. Co-Authored-By: Claude Opus 5 (1M context) --- .changeset/input-delivery-retryable.md | 23 +++ src/session.ts | 30 +++- src/web/routes/session-routes.ts | 31 ++-- src/web/routes/ws-routes.ts | 10 +- test/mocks/mock-session.ts | 16 ++- test/routes/session-input-delivery.test.ts | 157 +++++++++++++++++++++ 6 files changed, 251 insertions(+), 16 deletions(-) create mode 100644 .changeset/input-delivery-retryable.md create mode 100644 test/routes/session-input-delivery.test.ts diff --git a/.changeset/input-delivery-retryable.md b/.changeset/input-delivery-retryable.md new file mode 100644 index 00000000..c9ec5357 --- /dev/null +++ b/.changeset/input-delivery-retryable.md @@ -0,0 +1,23 @@ +--- +'aicodeman': patch +--- + +An input whose delivery fails can be retried instead of being lost for good. + +Both input paths recorded the `(clientId, seq)` pair as applied and acknowledged +the frame _before_ knowing whether the write had landed — the POST route because +its mux write is fire-and-forget, the WebSocket handler because it ACKed +unconditionally. When the write then failed, the client dropped the frame from its +durable queue and the server rejected the retry as a duplicate: the reliable +delivery layer was guaranteeing exactly-once delivery of something that had never +been delivered. + +The bookkeeping is now rolled back on failure and the WebSocket ACK withheld, so +the client redelivers. `Session.write()` reports whether it reached a PTY at all +instead of silently swallowing the data, and the non-mux POST branch — whose +response has not gone out yet — answers `OPERATION_FAILED` rather than a cheerful 200. + +Note this does not remove the root cause: the POST still answers 200 before the +mux write is attempted, so a client that treats any 2xx as final still cannot +learn about that failure. Closing that would mean awaiting the tmux child in the +request path. diff --git a/src/session.ts b/src/session.ts index f86139f5..b8d86e7b 100644 --- a/src/session.ts +++ b/src/session.ts @@ -2551,11 +2551,16 @@ export class Session extends EventEmitter { * session.write('ls -la\r'); // Command with Enter * ``` */ - write(data: string): void { + /** + * @returns true if the data reached a PTY. A session whose PTY is gone silently + * swallowed every write before this signal existed, which is how input could + * disappear with the caller believing it had been delivered. + */ + write(data: string): boolean { this._trackSubmit(data); - if (this.ptyProcess) { - this.ptyProcess.write(data); - } + if (!this.ptyProcess) return false; + this.ptyProcess.write(data); + return true; } // ── Conversation tracking ───────────────────────────────────────────── @@ -2600,6 +2605,23 @@ export class Session extends EventEmitter { * half-open socket silently drops frames with no error) would type a prompt * twice whenever an ACK is lost after the write landed. */ + /** + * Undo the bookkeeping of {@link shouldApplyInput} for a delivery that failed. + * + * Without this, the reliable-delivery layer guarantees exactly-once delivery of + * something that may never have been delivered: the seq is recorded as applied + * BEFORE the write is attempted, so a client retry — the very mechanism the seq + * exists for — is rejected as a duplicate and the input is lost for good. + * + * Only rolls back if `seq` is still the newest recorded one; a later input has + * already superseded it and must not be re-opened. + */ + forgetInputSeq(clientId: string, seq: number): void { + if (this._appliedInputSeq.get(clientId) === seq) { + this._appliedInputSeq.set(clientId, seq - 1); + } + } + shouldApplyInput(clientId: string, seq: number): boolean { const last = this._appliedInputSeq.get(clientId); if (last !== undefined && seq <= last) return false; diff --git a/src/web/routes/session-routes.ts b/src/web/routes/session-routes.ts index 5bf58a7a..c7a959a3 100644 --- a/src/web/routes/session-routes.ts +++ b/src/web/routes/session-routes.ts @@ -879,28 +879,41 @@ export function registerSessionRoutes( // Reliable delivery (POST fallback when the WebSocket is down): a 2xx IS the // client's ACK, so a tagged duplicate redelivery must still return 200 but // skip the write. Untagged requests (curl/legacy) always apply. - if (typeof clientId === 'string' && typeof seq === 'number' && !session.shouldApplyInput(clientId, seq)) { + const tagged = typeof clientId === 'string' && typeof seq === 'number'; + if (tagged && !session.shouldApplyInput(clientId as string, seq as number)) { return {}; } // Write input to PTY. Direct write is synchronous; writeViaMux // (tmux send-keys) is fire-and-forget to avoid blocking the HTTP response. if (useMux) { - // Fire-and-forget: don't block HTTP response on tmux child process. - // Fallback to direct write on failure. + // Fire-and-forget: don't block the HTTP response on a tmux child process. + // Fallback to a direct write on failure. + // + // Because the response has already been sent by then, a failure here is the + // one case the caller can never learn about — so the dedup bookkeeping is + // rolled back. Otherwise the seq stays recorded as applied and a retry, the + // very mechanism reliable delivery exists for, is rejected as a duplicate. + const undoOnFailure = () => { + if (tagged) session.forgetInputSeq(clientId as string, seq as number); + }; session .writeViaMux(inputStr) .then((ok) => { - if (!ok) { - console.warn(`[Server] writeViaMux failed for session ${id}, falling back to direct write`); - session.write(inputStr); - } + if (ok) return; + console.warn(`[Server] writeViaMux failed for session ${id}, falling back to direct write`); + if (!session.write(inputStr)) undoOnFailure(); }) .catch(() => { - session.write(inputStr); + if (!session.write(inputStr)) undoOnFailure(); }); } else { - session.write(inputStr); + // 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. + if (!session.write(inputStr) && tagged) { + session.forgetInputSeq(clientId as string, seq as number); + } } return {}; }); diff --git a/src/web/routes/ws-routes.ts b/src/web/routes/ws-routes.ts index 1b59d6eb..fc6e4c61 100644 --- a/src/web/routes/ws-routes.ts +++ b/src/web/routes/ws-routes.ts @@ -180,13 +180,19 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost 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(); - session.write(msg.d); + delivered = session.write(msg.d); + // 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 (delivered && seq !== null && socket.readyState === 1) { socket.send(`{"t":"ia","seq":${seq}}`); } } else if ( diff --git a/test/mocks/mock-session.ts b/test/mocks/mock-session.ts index 342ba18c..9e886bf5 100644 --- a/test/mocks/mock-session.ts +++ b/test/mocks/mock-session.ts @@ -31,12 +31,22 @@ export class MockSession extends EventEmitter { } /** Direct PTY write (used by session.write()) */ - write(data: string): void { + /** + * Set to simulate a session whose PTY is gone: both write paths report failure, + * which is the state in which input used to disappear silently. + */ + failWrites = false; + + /** Direct PTY write (used by session.write()). Mirrors the real boolean return. */ + write(data: string): boolean { + if (this.failWrites) return false; this.writeBuffer.push(data); + return true; } /** Write via mux (used by respawn controller) */ async writeViaMux(data: string): Promise { + if (this.failWrites) return false; this.writeBuffer.push(data); return true; } @@ -44,6 +54,10 @@ export class MockSession extends EventEmitter { /** Exactly-once input dedup — mirrors Session.shouldApplyInput so route tests * exercising the reliable-delivery path behave like production. */ private _appliedInputSeq = new Map(); + forgetInputSeq(clientId: string, seq: number): void { + if (this._appliedInputSeq.get(clientId) === seq) this._appliedInputSeq.set(clientId, seq - 1); + } + shouldApplyInput(clientId: string, seq: number): boolean { const last = this._appliedInputSeq.get(clientId); if (last !== undefined && seq <= last) return false; diff --git a/test/routes/session-input-delivery.test.ts b/test/routes/session-input-delivery.test.ts new file mode 100644 index 00000000..2d655ffd --- /dev/null +++ b/test/routes/session-input-delivery.test.ts @@ -0,0 +1,157 @@ +/** + * @fileoverview A lost input must stay retryable. + * + * POST /api/sessions/:id/input answers 200 BEFORE the write is attempted — the mux + * write is fire-and-forget so the HTTP response never waits on a tmux child. The + * dedup bookkeeping, however, recorded the (clientId, seq) pair as applied at that + * same moment. A write that then failed left the client with a 200, no message in + * the pane, and a seq the server would reject as a duplicate on retry: the input was + * unrecoverable by the very mechanism meant to make delivery reliable. + * + * Observed in the wild: a prompt shown as sent in a chat client, a 200 in the proxy + * log, and an empty prompt line in the pane. + */ + +import fastifyCookie from '@fastify/cookie'; +import Fastify, { type FastifyInstance } from 'fastify'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { Session } from '../../src/session.js'; +import { ApiErrorCode, httpStatusForErrorCode } from '../../src/types.js'; +import { installRouteErrorHandler } from '../../src/web/route-error-handler.js'; +import { registerSessionRoutes } from '../../src/web/routes/session-routes.js'; +import { createMockRouteContext, type MockRouteContext } from '../mocks/index.js'; + +async function createEnvelopeHarness(): Promise<{ app: FastifyInstance; ctx: MockRouteContext }> { + const app = Fastify({ logger: false }); + await app.register(fastifyCookie); + const ctx = createMockRouteContext(); + registerSessionRoutes(app, ctx as never); + + app.addHook('preSerialization', (req, reply, payload: unknown, done) => { + if (!req.url.startsWith('/api')) return done(null, payload); + if (payload === null || typeof payload !== 'object') return done(null, payload); + const p = payload as { success?: unknown; errorCode?: unknown }; + if (p.success === false) { + if (reply.statusCode === 200 && typeof p.errorCode === 'string') { + reply.code(httpStatusForErrorCode(p.errorCode as ApiErrorCode)); + } + return done(null, payload); + } + if (p.success === true) return done(null, payload); + return done(null, { success: true, data: payload }); + }); + + installRouteErrorHandler(app); + await app.ready(); + return { app, ctx }; +} + +type Internals = { _appliedInputSeq: Map }; +const seqOf = (s: Session, client: string) => (s as unknown as Internals)._appliedInputSeq.get(client); + +describe('input dedup bookkeeping', () => { + const make = () => new Session({ workingDir: '/tmp', mode: 'claude' }); + + it('accepts an increasing seq once and rejects the replay', () => { + const s = make(); + expect(s.shouldApplyInput('c1', 1)).toBe(true); + expect(s.shouldApplyInput('c1', 1)).toBe(false); + expect(s.shouldApplyInput('c1', 2)).toBe(true); + }); + + it('forgetInputSeq re-opens a failed delivery for retry', () => { + const s = make(); + expect(s.shouldApplyInput('c1', 7)).toBe(true); + + s.forgetInputSeq('c1', 7); // the write failed after the 200 went out + + expect(s.shouldApplyInput('c1', 7)).toBe(true); + }); + + it('does not re-open a seq that a later input has superseded', () => { + // Rolling back blindly would let an old, already-superseded message replay. + const s = make(); + s.shouldApplyInput('c1', 7); + s.shouldApplyInput('c1', 8); + + s.forgetInputSeq('c1', 7); + + expect(s.shouldApplyInput('c1', 8)).toBe(false); + expect(seqOf(s, 'c1')).toBe(8); + }); + + it('is scoped per client', () => { + const s = make(); + s.shouldApplyInput('c1', 5); + s.forgetInputSeq('c2', 5); + expect(s.shouldApplyInput('c1', 5)).toBe(false); + }); + + it('tolerates a rollback for a client that was never seen', () => { + const s = make(); + expect(() => s.forgetInputSeq('ghost', 3)).not.toThrow(); + }); +}); + +describe('Session.write delivery signal', () => { + it('reports false when there is no PTY instead of swallowing the data', () => { + // The silent swallow was the third way input could vanish: no PTY, no error, + // no return value — the caller had no way to know. + const s = new Session({ workingDir: '/tmp', mode: 'claude' }); + expect(s.write('hello\r')).toBe(false); + }); +}); + +/** + * Wiring, not primitives. + * + * The first version of this file tested Session directly and nothing else: reverting + * the route to master — deleting the rollback call, the load-bearing half of the fix — + * left all six tests green. These drive the actual HTTP route. + */ +describe('POST /api/sessions/:id/input rollback wiring', () => { + let harness: { app: FastifyInstance; ctx: MockRouteContext }; + + beforeEach(async () => { + harness = await createEnvelopeHarness(); + }); + afterEach(async () => { + await harness.app.close(); + }); + + const post = (body: Record, id = 'test-session-1') => + harness.app.inject({ method: 'POST', url: `/api/sessions/${id}/input`, payload: body }); + + it('rolls the seq back when both the mux write and the direct write fail', async () => { + const session = harness.ctx.sessions.get('test-session-1')!; + session.failWrites = true; // writeViaMux false AND write() false + + await post({ input: 'lost\r', useMux: true, clientId: 'c1', seq: 1 }); + await new Promise((r) => setTimeout(r, 20)); // the mux write is fire-and-forget + + // The retry the client would make must be accepted, not swallowed as a duplicate. + expect(session.shouldApplyInput('c1', 1)).toBe(true); + }); + + it('keeps the seq burnt when delivery succeeded', async () => { + const session = harness.ctx.sessions.get('test-session-1')!; + + await post({ input: 'fine\r', useMux: true, clientId: 'c1', seq: 1 }); + await new Promise((r) => setTimeout(r, 20)); + + expect(session.shouldApplyInput('c1', 1)).toBe(false); + }); + + it('rolls the seq back on the non-mux path too', async () => { + // Still a 200: a session may legitimately have no PTY yet, and turning that + // into a failure status would be a contract change. Re-opening the seq is not. + const session = harness.ctx.sessions.get('test-session-1')!; + session.failWrites = true; + + const res = await post({ input: 'x\r', useMux: false, clientId: 'c2', seq: 5 }); + + expect(res.statusCode).toBe(200); + expect(session.shouldApplyInput('c2', 5)).toBe(true); + }); +});