BitSquare/Codeman
2
0 Fork 0
mirror of https://github.com/Ark0N/Codeman.git synced 2026-11-14 10:39:32 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): hold input during a guarded submit instead of refusing it

Browse Source 
The input lease is held across the awaited tmux Enter, so WebSocket frames
keep arriving while it is held. A tagged frame (seq N) was refused and its
seq rolled back, but the browser's WebSocket path does not wait for ACKs:
seq N+1 landed after the lease and moved the watermark past N, and the
redelivered N was then dropped as a duplicate. The keystroke was lost.

Input that arrives during the lease is now held and written in arrival order
when the lease is released, whatever the submit's outcome. A WebSocket frame
is applied and ACKed only once written; managed writes (writeViaMux) and
named keys (send-key) wait for their turn the same way, and a held write that
is still being sent holds everything behind it. A keystroke held before the
guard runs still moves the input revision, so it still invalidates the Enter.
The hold is capped per lease (256 inputs, 1 MiB): past it the WebSocket is
closed with 1013 and nothing later from it is applied, so the browser resends
every unACKed frame in order, and send-key answers 409 CONFLICT instead of
logging a tmux failure as INTERNAL_ERROR. POST .../input keeps its 409 for the
lease plus the held input's drain; its browser drainer is serialized.

Docs: correct the api-reference paragraph that said these keystrokes are
redelivered, and describe the capture fallback and the per-scope cursor cap
from the previous two commits in api-reference, architecture-invariants and
the changeset.
This commit is contained in:
Aamer Akhter committed 2026-10-26 09:19:05 -04:00
1 parent ecaddb40ae
commit 2aad93914e
9 files changed
+727 -100
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/web/routes/session-routes.ts
+15 -7
View File
@@ -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<string, unknown>;
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');
}
src/web/routes/ws-routes.ts
+83 -45
View File
@@ -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' &&