mattapperson · LukasParke · Aug 13, 2026
@@ -544,15 +544,25 @@ export class AgentIpcClient {
     if (frame.type === 'error') {
       // Error frames are non-fatal: a single bad frame from a peer or a
       // transient handler exception should not tear down the chat
-      // connection. Surface to any in-flight `send` ack waiter so that
-      // call rejects, and stash the error so callers can inspect it via
+      // connection. Stash the error so callers can inspect it via
       // `getLastError()`. Streaming continues.
       const err = new Error(`server error (${frame.error.kind}): ${frame.error.message}`);
       this.lastError = err;
-      // Server only emits acks against `send`; the only deferred we can
-      // reasonably correlate is one of the pending send waiters. Without
-      // a messageId on the error frame we can't pinpoint which — fail
-      // them all so a chained `await client.send(...)` doesn't hang.
+      const correlatedId = frame.error.messageId;
+      if (correlatedId !== undefined) {
+        // The server told us exactly which send failed — reject only that
+        // waiter. Other in-flight sends may have executed fine; failing
+        // them too made callers retry messages that already ran.
+        const pending = this.pendingAcks.get(correlatedId);
+        if (pending) {
+          this.pendingAcks.delete(correlatedId);
+          pending.reject(err);
+        }
+        return;
+      }
+      // Uncorrelated error (unparseable frame, non-send handler): we can't
+      // pinpoint a waiter, so fail them all rather than hang a chained
+      // `await client.send(...)`.
       for (const pending of this.pendingAcks.values()) {
         pending.reject(err);
       }

@@ -164,6 +164,14 @@ const ErrorFrameSchema = z.object({
   error: z.object({
     kind: z.string().min(1),
     message: z.string().min(1),
+    /**
+     * When the failed frame was a `send`, the message id it carried — so
+     * the client can reject exactly the matching ack waiter instead of
+     * failing every in-flight send (which made unrelated messages that DID
+     * execute look failed, inviting duplicate retries). Optional: errors
+     * from unparseable or non-send frames have nothing to correlate.
+     */
+    messageId: z.string().min(1).optional(),
   }),
 });
 

@@ -175,6 +175,13 @@ interface ClientState {
   buffer: string;
   /** Whether this client has issued `subscribe`. */
   subscribed: boolean;
+  /**
+   * Per-client dispatch chain. Frames from one client execute strictly in
+   * arrival order — a `send` and an `abort` in one TCP chunk must not race,
+   * and two `send`s must reach `harness.execute` in the order sent. A fire-
+   * and-forget `void dispatchLine(...)` per line gave no such guarantee.
+   */
+  dispatchChain: Promise<void>;
 }
 
 interface FrameContext {
@@ -195,6 +202,17 @@ type FrameHandler = (frame: ClientFrame, ctx: FrameContext) => Promise<void>;
  */
 const MAX_UNIX_SOCKET_PATH_BYTES = 104;
 
+/**
+ * Disconnect a client whose kernel-side socket buffer backs up past this
+ * many bytes. A stalled consumer (TUI suspended with ^Z, dead SSH hop)
+ * otherwise makes Node buffer every broadcast frame in process memory,
+ * unbounded, while both stream pumps keep producing. Disconnecting is the
+ * safe move in both modes: a durable client resumes by seq via
+ * `durableResume` replay and loses nothing; a non-durable client was
+ * always best-effort streaming.
+ */
+const MAX_SOCKET_BUFFERED_BYTES = 4 * 1024 * 1024;
+
 function errorMessage(err: unknown): string {
   return err instanceof Error ? err.message : String(err);
 }
@@ -243,6 +261,18 @@ function writeFrame(socket: Socket, frame: ServerFrame): void {
   }
   try {
     socket.write(encodeFrame(frame));
+    /* Backpressure guard: `write()` returning false just means "buffering";
+     * that's fine transiently. What is NOT fine is a consumer that never
+     * drains — cap the buffered bytes and cut the connection. The client's
+     * `close` handler removes it from the set; a durable client reconnects
+     * and resumes by seq. */
+    if (socket.writableLength > MAX_SOCKET_BUFFERED_BYTES) {
+      socket.destroy(
+        new Error(
+          `agent-ipc-server: client write buffer exceeded ${MAX_SOCKET_BUFFERED_BYTES} bytes (stalled consumer)`,
+        ),
+      );
+    }
   } catch {
     // Synchronous write throw is rare but possible (e.g. socket entered
     // an erroring state between the `destroyed` check and the call).
@@ -673,6 +703,7 @@ export class AgentIpcServer {
       socket,
       buffer: '',
       subscribed: false,
+      dispatchChain: Promise.resolve(),
     };
     this.clients.add(client);
     socket.setEncoding('utf8');
@@ -702,7 +733,13 @@ export class AgentIpcServer {
       const line = client.buffer.slice(0, nl);
       client.buffer = client.buffer.slice(nl + 1);
       if (line.length > 0) {
-        void this.dispatchLine(client, line);
+        /* Serialize per client: each frame's handler runs after the previous
+         * one settles, so frames arriving in one chunk keep their order
+         * (send-then-abort, send-then-send). dispatchLine never rejects
+         * (its own catch writes an error frame), but chain through catch
+         * anyway so an unexpected throw can't wedge the chain. */
+        const next = client.dispatchChain.then(() => this.dispatchLine(client, line));
+        client.dispatchChain = next.catch(() => undefined);
       }
       nl = client.buffer.indexOf('\n');
     }
@@ -745,6 +782,13 @@ export class AgentIpcServer {
         error: {
           kind: 'handler_error',
           message: errorMessage(err),
+          // Correlate send failures so the client rejects only the matching
+          // ack waiter — other in-flight sends may have executed fine.
+          ...(frame.type === 'send'
+            ? {
+                messageId: frame.messageId,
+              }
+            : {}),
         },
       });
     }

@@ -82,9 +82,15 @@ export interface DurableOutboundQueue {
    * frames as acknowledged.
    */
   ackUpTo(throughSeq: number): Promise<void>;
-  /** Remove every persisted frame + reset meta — used on a clean shutdown. */
+  /**
+   * Remove every persisted frame and settle the ack watermark — used on a
+   * clean shutdown. `headSeq` is NOT reset (invariant 1: a seq is never
+   * reused), so a queue that appends again after `clear()` continues the
+   * sequence rather than restarting at 1 under clients that still hold a
+   * pre-clear delivery watermark.
+   */
   clear(): Promise<void>;
-  /** Number of frames currently persisted. Exposed for test assertions. */
+  /** Number of frames currently un-acked (`headSeq - lastAckedSeq`). */
   queueSize(): Promise<number>;
   /** Monotonic head seq — every appended frame gets head+1, head+2, …. */
   getHeadSeq(): number;
@@ -242,16 +248,14 @@ export async function createDurableOutboundQueue(
     const effective = Math.min(throughSeq, headSeq);
     const previous = lastAckedSeq;
     lastAckedSeq = effective;
-    // Delete frames in `(previous, effective]`.
-    const keys = await storage.list(framePrefix);
-    for (const key of keys) {
-      const seq = parseSeqFromFrameKey(key, socketId);
-      if (seq === null) {
-        continue;
-      }
-      if (seq > previous && seq <= effective) {
-        await storage.delete(key);
-      }
+    /* Frame keys are computable from their seq — delete `(previous,
+     * effective]` directly instead of a storage.list() scan. Acks fire on
+     * every client watermark push, and a per-ack scan over a namespace that
+     * shares its backing directory with checkpoints and ledger shards made
+     * each ack O(total keys). `storage.delete` on an already-gone key is a
+     * no-op by contract, so gaps (from clear/compaction) cost nothing. */
+    for (let seq = previous + 1; seq <= effective; seq++) {
+      await storage.delete(frameKey(socketId, seq));
     }
     await flushMeta();
   }
@@ -261,21 +265,22 @@ export async function createDurableOutboundQueue(
     for (const key of keys) {
       await storage.delete(key);
     }
-    await storage.delete(metaKey(socketId));
-    headSeq = 0;
-    lastAckedSeq = 0;
+    /* headSeq stays MONOTONIC through a clear — invariant 1 says a seq is
+     * never reused. Resetting to 0 here meant a queue cleared and then
+     * appended-to restarted at seq 1, and any client still holding a
+     * pre-clear `highestDeliveredDurableSeq` watermark would silently drop
+     * every new frame up to it (client dedupe is `seq <= highestDelivered`).
+     * Frames are gone; the counter is not. `lastAckedSeq` advances to
+     * headSeq (everything persisted is deleted ⇒ everything is settled),
+     * and the meta doc records both so a restart doesn't resurrect 0. */
+    lastAckedSeq = headSeq;
+    await flushMeta();
   }
 
   async function queueSize(): Promise<number> {
-    const keys = await storage.list(framePrefix);
-    let count = 0;
-    for (const key of keys) {
-      const seq = parseSeqFromFrameKey(key, socketId);
-      if (seq !== null && seq > lastAckedSeq) {
-        count += 1;
-      }
-    }
-    return count;
+    /* In-memory bounds are authoritative: appends only persist inside
+     * `(lastAckedSeq, headSeq]` and acks delete below the watermark. */
+    return headSeq - lastAckedSeq;
   }
 
   return {

@@ -1,4 +1,5 @@
 import { execFileSync, spawn } from 'node:child_process';
+import { readFileSync } from 'node:fs';
 import path from 'node:path';
 import type {
   ProcessSubprocessRequest,
@@ -68,7 +69,36 @@ function isErrnoException(err: unknown): err is NodeJS.ErrnoException {
   return typeof err === 'object' && err !== null && 'code' in err;
 }
 
+/**
+ * Linux fast path: field 22 of `/proc/<pid>/stat` is the process start time
+ * in clock ticks since boot — a stable pid-recycle discriminator readable in
+ * microseconds, vs ~10ms to fork `ps`. The value only needs to be a stable
+ * identity token (compared for equality against the persisted manifest),
+ * not a human-readable date, so the raw tick count is fine. Fields are
+ * located from the LAST `)` because field 2 (comm) can itself contain
+ * spaces and parens.
+ */
+function readProcStatStartTime(pid: number): string | null {
+  try {
+    const stat = readFileSync(`/proc/${pid}/stat`, 'utf8');
+    const afterComm = stat.slice(stat.lastIndexOf(')') + 2);
+    const fields = afterComm.split(' ');
+    // afterComm starts at field 3 ("state"), so starttime (field 22) is index 19.
+    const starttime = fields[19];
+    return starttime && starttime.length > 0 ? `proc:${starttime}` : null;
+  } catch {
+    return null;
+  }
+}
+
 function readPidStartTime(pid: number): string | null {
+  if (process.platform === 'linux') {
+    const viaProc = readProcStatStartTime(pid);
+    if (viaProc !== null) {
+      return viaProc;
+    }
+    // Fall through to ps — /proc may be restricted (containers, hardening).
+  }
   try {
     const out = execFileSync(
       'ps',
@@ -234,7 +264,7 @@ export function createLocalSubprocessAdapter(
 
     const now = nowIso();
     const pid = child.pid;
-    const pidStarttime = signaller.startTime(pid);
+    const pidStarttime = await signaller.startTime(pid);
     const handle = await save({
       id: `subprocess-${crypto.randomUUID()}`,
       status: 'running',
@@ -328,7 +358,7 @@ export function createLocalSubprocessAdapter(
       );
     }
 
-    const spawned = spawnStepChild({
+    const spawned = await spawnStepChild({
       spawnFn,
       signaller,
       request,
@@ -470,7 +500,7 @@ export function createLocalSubprocessAdapter(
       if (!manifest) {
         return null;
       }
-      const handle = hydrateFromManifest(manifest, signaller);
+      const handle = await hydrateFromManifest(manifest, signaller);
       if (!handle) {
         // Pid drift or process gone — clear the stale manifest so
         // subsequent listLive calls don't re-surface it forever.
@@ -492,11 +522,18 @@ export function createLocalSubprocessAdapter(
       }
       if (storage) {
         const manifests = await listLocalManifests(storage);
-        for (const manifest of manifests) {
-          if (live.has(manifest.handleId)) {
-            continue;
-          }
-          const hydrated = hydrateFromManifest(manifest, signaller);
+        const candidates = manifests.filter((m) => !live.has(m.handleId));
+        /* Hydrate in parallel: each check may read /proc or fork `ps`, and a
+         * restart with N persisted manifests paid them serially before. Each
+         * result carries its own manifest so the pairing survives the reorder
+         * a bare `Promise.all` over handles would invite. */
+        const hydratedAll = await Promise.all(
+          candidates.map(async (manifest) => ({
+            manifest,
+            hydrated: await hydrateFromManifest(manifest, signaller),
+          })),
+        );
+        for (const { manifest, hydrated } of hydratedAll) {
           if (!hydrated) {
             await clearIfDurable(manifest.handleId);
             continue;

@@ -230,14 +230,14 @@ function nowIso(): string {
  * in either case the process we recorded is no longer the one running
  * under that pid, so rebinding would be unsafe.
  */
-export function hydrateFromManifest(
+export async function hydrateFromManifest(
   manifest: LocalManifest,
   signaller: ProcessSignaller,
-): SubprocessHandle | null {
+): Promise<SubprocessHandle | null> {
   if (!signaller.isAlive(manifest.pid)) {
     return null;
   }
-  const currentStart = signaller.startTime(manifest.pid);
+  const currentStart = await signaller.startTime(manifest.pid);
   if (manifest.pidStarttime !== null && currentStart !== null) {
     if (currentStart !== manifest.pidStarttime) {
       return null;

@@ -72,7 +72,7 @@ export interface AttachCompletionArgs {
  * once the handle is registered (so the `on('close')` callback can
  * close over the factory's handle map + save/clearIfDurable closures).
  */
-export function spawnStepChild(args: SpawnStepChildArgs): SpawnStepChildResult {
+export async function spawnStepChild(args: SpawnStepChildArgs): Promise<SpawnStepChildResult> {
   let asyncSpawnError: unknown = null;
   const child = args.spawnFn(args.bootstrapCommand, args.bootstrapArgs, {
     cwd: args.request.overrides.cwdInit,
@@ -100,7 +100,7 @@ export function spawnStepChild(args: SpawnStepChildArgs): SpawnStepChildResult {
   }
 
   const pid = child.pid;
-  const pidStarttime = args.signaller.startTime(pid);
+  const pidStarttime = await args.signaller.startTime(pid);
 
   // Write the request envelope to stdin as a single newline-terminated JSON
   // frame. The child parses one frame on boot.

@@ -10,5 +10,11 @@ export type SubprocessSignal = 'SIGTERM' | 'SIGSTOP' | 'SIGCONT';
 export interface ProcessSignaller {
   kill(target: number, signal: SubprocessSignal): void;
   isAlive(pid: number): boolean;
-  startTime(pid: number): string | null;
+  /**
+   * Stable start-time identity token for `pid`, or null when unreadable.
+   * May be sync or async: the default signaller reads `/proc` (sync, µs) on
+   * Linux and shells out to `ps` elsewhere; an async implementation lets a
+   * host avoid blocking the event loop on that fork. Callers must await.
+   */
+  startTime(pid: number): string | null | Promise<string | null>;
 }