Merge pull request #226 from christianhaberl/fix/input-loss-on-failed-delivery

fix(api,ws): an input whose delivery fails can be retried instead of being lost

.changeset/input-delivery-retryable.md

@@ -0,0 +1,26 @@
+---
+'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.
+
+Response codes are unchanged: a session can legitimately have no PTY yet (created
+but not started), so turning that into a failure status would be a contract change
+of its own.
+
+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.

src/session.ts

@@ -2620,20 +2620,23 @@ export class Session extends EventEmitter {
    * For interactive sessions, this is how you send user input to Claude.
    * Remember to include `\r` (carriage return) to simulate pressing Enter.
    *
-   * @param data - The input data to send (text, escape sequences, etc.)
-   *
    * @example
    * ```typescript
    * session.write('hello world');  // Text only, no Enter
    * session.write('\r');           // Enter key
    * session.write('ls -la\r');     // Command with Enter
    * ```
+   *
+   * @param data - The input data to send (text, escape sequences, etc.)
+   * @returns true if the data reached a PTY. A session whose PTY is gone still
+   * discards the data, but it used to do so with no signal at all — which is how
+   * input could disappear while the caller believed it had been delivered.
    */
-  write(data: string): void {
+  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 ─────────────────────────────────────────────
@@ -2691,6 +2694,23 @@ export class Session extends EventEmitter {
     return true;
   }
 
+  /**
+   * 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);
+    }
+  }
+
   /**
    * Sends input via the terminal multiplexer's direct input mechanism.
    *

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 {};
   });

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 (

test/mocks/mock-session.ts

@@ -30,13 +30,22 @@ export class MockSession extends EventEmitter {
     this._muxName = `codeman-test-${id.slice(0, 8)}`;
   }
 
-  /** 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<boolean> {
+    if (this.failWrites) return false;
     this.writeBuffer.push(data);
     return true;
   }
@@ -44,6 +53,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<string, number>();
+  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;

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<string, number> };
+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<string, unknown>, 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);
+  });
+});

test/routes/ws-routes.test.ts

@@ -208,6 +208,55 @@ describe('ws-routes', () => {
       }
     });
 
+    it('ACKs a delivered input and burns its seq', async () => {
+      const ws = await connectWs('/ws/sessions/ws-test-session/terminal');
+      try {
+        const session = ctx._session;
+        ws.send(JSON.stringify({ t: 'i', d: 'ok\r', cid: 'c1', seq: 1 }));
+
+        expect(await nextMessage(ws)).toEqual({ t: 'ia', seq: 1 });
+        expect(session.shouldApplyInput('c1', 1)).toBe(false);
+      } finally {
+        ws.close();
+      }
+    });
+
+    it('withholds the ACK and re-opens the seq when the write did not land', async () => {
+      // A session whose PTY is gone swallows the write. ACKing anyway told the
+      // client to drop the frame from its durable queue while the seq stayed
+      // burnt, so the retry that reliable delivery exists for was rejected as a
+      // duplicate — the input was lost for good.
+      const ws = await connectWs('/ws/sessions/ws-test-session/terminal');
+      try {
+        const session = ctx._session;
+        session.failWrites = true;
+
+        ws.send(JSON.stringify({ t: 'i', d: 'lost\r', cid: 'c1', seq: 1 }));
+
+        await expect(nextMessage(ws, 600)).rejects.toThrow(/timeout/);
+        expect(session.shouldApplyInput('c1', 1)).toBe(true);
+      } finally {
+        ws.close();
+      }
+    });
+
+    it('still ACKs a duplicate frame the server deliberately skipped', async () => {
+      // Dedup must stay silent-but-acknowledged: the client has to be able to
+      // drop a frame it already delivered once.
+      const ws = await connectWs('/ws/sessions/ws-test-session/terminal');
+      try {
+        const session = ctx._session;
+        session.shouldApplyInput('c1', 7); // pretend seq 7 already landed
+
+        ws.send(JSON.stringify({ t: 'i', d: 'again\r', cid: 'c1', seq: 7 }));
+
+        expect(await nextMessage(ws)).toEqual({ t: 'ia', seq: 7 });
+        expect(session.writeBuffer).not.toContain('again\r');
+      } finally {
+        ws.close();
+      }
+    });
+
     it('ignores input exceeding MAX_INPUT_LENGTH', async () => {
       const ws = await connectWs('/ws/sessions/ws-test-session/terminal');
       try {