test: let the I/O wait of the yielder test pass a check phase

On a slow machine a 1 ms timer can resume within the same timers phase,
before the check phase that ends the slice, so the yielder rightly
yielded and the test failed in CI.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/acp-agent.ts

@@ -261,6 +261,7 @@ import { ClientCapabilities as ToolCallClientCapabilities } from "./tool-calls/c
 import { AcpToolCallRenderer, type ToolUpdateMeta } from "./tool-calls/renderer.js";
 import { resolveSkillPath } from "./tool-calls/reporters/interaction.js";
 import {
+  messageLoopYielder,
   nodeToWebReadable,
   nodeToWebWritable,
   Pushable,
@@ -4794,6 +4795,14 @@ export class ClaudeAcpAgent {
 
     try {
       while (true) {
+        // A backlog of buffered SDK messages drains without a macrotask
+        // boundary (each await resolves in a microtask), which would starve
+        // timers and incoming requests, `session/cancel` among them, until
+        // the backlog is gone. Give the event loop a turn when this stretch
+        // has run past its budget. Messages keep their order: the next one is
+        // pulled only after the pause.
+        const pause = messageLoopYielder.maybeYield();
+        if (pause) await pause;
         pendingNext ??= session.query
           .next()
           .then((result) => ({ kind: "message" as const, result }));
@@ -7983,6 +7992,10 @@ export class ClaudeAcpAgent {
     };
 
     const replayMessage = async (message: SessionMessage): Promise<void> => {
+      // A long history is sent in order, but with event-loop turns in between,
+      // so other sessions and incoming requests are not starved meanwhile.
+      const pause = messageLoopYielder.maybeYield();
+      if (pause) await pause;
       if (pending?.stopped || isReplayHiddenMetaMessage(message)) {
         return;
       }

src/diff.ts

@@ -569,11 +569,52 @@ function textLines(text: string): string[] {
  * its time budget.
  */
 function filePatch(oldText: string | null, newText: string): FilePatch | undefined {
-  const hunks = diffHunks(oldText, newText);
+  const hunks = memoizedDiffHunks(oldText, newText);
   if (!hunks || hunks.length === 0) return undefined;
   return { change: oldText === null ? "create" : "update", hunks };
 }
 
+/** The number of recent line diffs that {@link memoizedDiffHunks} keeps. */
+const DIFF_MEMO_SIZE = 4;
+
+/** Recent line diffs not read again yet, the most recent last. A text of at
+ *  most {@link MAX_PATCH_FILE_BYTES} is held until it is read again or later
+ *  diffs push it out. */
+const diffMemo: { oldText: string | null; newText: string; hunks: PatchHunk[] }[] = [];
+
+/**
+ * {@link diffHunks} of a recent pair of texts again, without a second diff.
+ *
+ * The approval of an Edit or Write diffs the file against the predicted text,
+ * and the PostToolUse hook diffs the same texts once Claude wrote them. On a
+ * large file each diff can block the event loop for its whole time budget.
+ * Only a diff that completed is remembered: one that ran out of its budget
+ * says nothing about the texts, so a later call diffs them again. The hunks
+ * are shared, and no caller changes them.
+ */
+function memoizedDiffHunks(oldText: string | null, newText: string): PatchHunk[] | undefined {
+  const index = diffMemo.findIndex(
+    (entry) => entry.oldText === oldText && entry.newText === newText,
+  );
+  if (index >= 0) {
+    // The hook is normally the last reader of a pair, so the entry goes: the
+    // memo does not hold two large texts longer than needed.
+    const [entry] = diffMemo.splice(index, 1);
+    return entry.hunks;
+  }
+  const hunks = diffHunks(oldText, newText);
+  if (hunks) {
+    diffMemo.push({ oldText, newText, hunks });
+    if (diffMemo.length > DIFF_MEMO_SIZE) diffMemo.shift();
+  }
+  return hunks;
+}
+
+/** Forget the remembered line diffs. For tests. */
+export function clearDiffMemo(): void {
+  diffMemo.length = 0;
+}
+
 /**
  * Reads a file as patch input.
  *

src/mcp-command.ts

@@ -29,6 +29,10 @@ export type McpCommand =
  *  or `/mcp disable` with an optional server name or `all`. Any other `/mcp`
  *  argument gives null, and the prompt gets no replacement. */
 export function parseMcpCommand(text: string): McpCommand | null {
+  // The first word is `/mcp` only when the text starts with it after the
+  // leading whitespace. Checking that first keeps a long prompt from being
+  // split into words. `trimStart` removes the characters that `\s` matches.
+  if (!text.trimStart().startsWith("/mcp")) return null;
   const words = text.trim().split(/\s+/);
   if (words[0] !== "/mcp") return null;
   if (words.length === 1) return { action: "status" };

src/session-titles.ts

@@ -42,7 +42,22 @@ type TitleCapableQuery = Query & {
  *  unavailable or yields nothing. */
 type TitleFallback = { title: string; lastModified: number };
 
+/** How much of a long text {@link sanitizeTitle} reads first. Collapsed, it
+ *  is longer than a title unless it is mostly whitespace. */
+const TITLE_SCAN_LENGTH = 4096;
+
 export function sanitizeTitle(text: string): string {
+  if (text.length > TITLE_SCAN_LENGTH) {
+    // A stored summary can be a whole first prompt of megabytes. The title
+    // depends only on the start of the text: collapsing the whitespace of a
+    // prefix gives a prefix of the collapsed text. Two more characters than a
+    // title hold at least one that is not a space (spaces are collapsed), so
+    // the full text is longer than a title and gets the same cut.
+    const head = text.slice(0, TITLE_SCAN_LENGTH).replace(/\s+/g, " ").trimStart();
+    if (head.length >= MAX_TITLE_LENGTH + 2) {
+      return head.slice(0, MAX_TITLE_LENGTH - 1) + "…";
+    }
+  }
   // Replace newlines and collapse whitespace
   const sanitized = text
     .replace(/[\r\n]+/g, " ")

src/tests/diff-memo.test.ts

@@ -0,0 +1,111 @@
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import * as diff from "diff";
+import { clearDiffMemo, patchUpdateFromDiffToolResponse, previewPatchContent } from "../diff.js";
+
+vi.mock("diff", async (importOriginal) => {
+  const original = await importOriginal<typeof import("diff")>();
+  return { ...original, structuredPatch: vi.fn(original.structuredPatch) };
+});
+
+const structuredPatch = vi.mocked(diff.structuredPatch);
+const tempDirectories: string[] = [];
+
+beforeEach(() => {
+  clearDiffMemo();
+  structuredPatch.mockClear();
+});
+
+afterEach(async () => {
+  await Promise.all(
+    tempDirectories.splice(0).map((directory) => rm(directory, { recursive: true })),
+  );
+});
+
+async function temporaryFile(content: string): Promise<string> {
+  const directory = await mkdtemp(path.join(os.tmpdir(), "claude-acp-diff-memo-"));
+  tempDirectories.push(directory);
+  const filePath = path.join(directory, "file.txt");
+  await writeFile(filePath, content);
+  return filePath;
+}
+
+function patchText(content: unknown): string {
+  const block = Array.isArray(content) ? content[0] : content;
+  return (block._meta as any).jetbrains.air.diffPatch.text;
+}
+
+const before = Array.from({ length: 2000 }, (_, i) => `line ${i}\n`).join("");
+const after = before.replace("line 1000\n", "line one thousand\n");
+
+describe("line diff reuse between the approval and the PostToolUse hook", () => {
+  it("diffs an Edit once and gives the hook the same patch", async () => {
+    const filePath = await temporaryFile(before);
+    const preview = await previewPatchContent("Edit", {
+      file_path: filePath,
+      old_string: "line 1000\n",
+      new_string: "line one thousand\n",
+    });
+    expect(structuredPatch).toHaveBeenCalledTimes(1);
+
+    await writeFile(filePath, after);
+    const hook = await patchUpdateFromDiffToolResponse({
+      filePath,
+      originalFile: before,
+      oldString: "line 1000\n",
+    });
+
+    expect(structuredPatch).toHaveBeenCalledTimes(1);
+    expect(patchText(hook!.content)).toBe(patchText(preview));
+    expect(hook!.locations).toEqual([{ path: filePath, line: 998 }]);
+  });
+
+  it("diffs again when the written file differs from the preview", async () => {
+    const filePath = await temporaryFile(before);
+    await previewPatchContent("Edit", {
+      file_path: filePath,
+      old_string: "line 1000\n",
+      new_string: "line one thousand\n",
+    });
+    const written = before.replace("line 1000\n", "line 1000 changed later\n");
+    await writeFile(filePath, written);
+
+    const hook = await patchUpdateFromDiffToolResponse({ filePath, originalFile: before });
+
+    expect(structuredPatch).toHaveBeenCalledTimes(2);
+    expect(patchText(hook!.content)).toContain("+line 1000 changed later\n");
+  });
+
+  it("diffs again after a diff that ran out of its budget", async () => {
+    structuredPatch.mockImplementationOnce(() => undefined as never);
+    const filePath = await temporaryFile(before);
+    const input = { file_path: filePath, content: after };
+
+    const first = await previewPatchContent("Write", input);
+    const second = await previewPatchContent("Write", input);
+
+    expect(structuredPatch).toHaveBeenCalledTimes(2);
+    // Without a patch, the approval falls back to the standard diff.
+    expect(first).toEqual([{ type: "diff", path: filePath, oldText: before, newText: after }]);
+    // The second diff completes and gets the exact patch.
+    expect(patchText(second)).toContain("+line one thousand\n");
+  });
+
+  it("keeps only a few recent diffs, each until it is read again", async () => {
+    const filePath = await temporaryFile(before);
+    const write = (i: number) =>
+      previewPatchContent("Write", { file_path: filePath, content: `${before}tail ${i}\n` });
+    for (let i = 0; i < 6; i++) await write(i);
+    expect(structuredPatch).toHaveBeenCalledTimes(6);
+    // The most recent diffs are kept; the oldest were pushed out.
+    await write(5);
+    expect(structuredPatch).toHaveBeenCalledTimes(6);
+    await write(0);
+    expect(structuredPatch).toHaveBeenCalledTimes(7);
+    // A diff that was read again is gone.
+    await write(5);
+    expect(structuredPatch).toHaveBeenCalledTimes(8);
+  });
+});

src/tests/event-loop-yield.test.ts

@@ -0,0 +1,260 @@
+import { randomUUID } from "node:crypto";
+import { describe, expect, it, vi } from "vitest";
+import type { SessionNotification } from "@agentclientprotocol/sdk";
+import { ClaudeAcpAgent, type AcpClient } from "../acp-agent.js";
+import { Writable } from "node:stream";
+import { EventLoopYielder, nodeToWebWritable, Pushable } from "../utils.js";
+import {
+  mockSessionState,
+  successfulResultMessage,
+  userEcho,
+  wrapQuery,
+} from "./session-doubles.js";
+
+const sessionId = "test-session";
+
+/** Busy-waits, so a test can make one message cost real time. */
+function spin(ms: number): void {
+  const end = performance.now() + ms;
+  while (performance.now() < end) {
+    // spin
+  }
+}
+
+function textDelta(text: string) {
+  return {
+    type: "stream_event",
+    event: { type: "content_block_delta", index: 0, delta: { type: "text_delta", text } },
+    parent_tool_use_id: null,
+    uuid: randomUUID(),
+    session_id: sessionId,
+  };
+}
+
+/** A query whose whole answer is already buffered: every `next()` resolves at
+ *  once, as the SDK queue does when the CLI is ahead of the adapter. */
+function bufferedAnswer(chunks: string[]) {
+  return (input: Pushable<any>) =>
+    (async function* () {
+      const user = await input[Symbol.asyncIterator]().next();
+      yield userEcho(user.value);
+      for (const chunk of chunks) yield textDelta(chunk);
+      yield successfulResultMessage();
+      yield { type: "system", subtype: "session_state_changed", state: "idle" };
+    })();
+}
+
+function createAgent(
+  makeGenerator: (input: Pushable<any>) => AsyncGenerator<any>,
+  onUpdate: (notification: SessionNotification) => void,
+) {
+  const agent = new ClaudeAcpAgent(
+    {
+      sessionUpdate: async (notification: SessionNotification) => onUpdate(notification),
+    } as unknown as AcpClient,
+    { log: () => {}, error: () => {} },
+  );
+  const input = new Pushable<any>();
+  agent.sessions[sessionId] = mockSessionState({
+    query: wrapQuery(makeGenerator(input)),
+    input,
+  });
+  return agent;
+}
+
+function chunkTexts(updates: SessionNotification[]): string[] {
+  return updates
+    .filter((n) => n.update.sessionUpdate === "agent_message_chunk")
+    .map((n) => (n.update as { content: { text: string } }).content.text);
+}
+
+describe("EventLoopYielder", () => {
+  it("lets a loop run on within its budget", () => {
+    let time = 0;
+    const yielder = new EventLoopYielder(8, () => time);
+    expect(yielder.maybeYield()).toBeUndefined();
+    time = 7;
+    expect(yielder.maybeYield()).toBeUndefined();
+  });
+
+  it("yields to the event loop past its budget", async () => {
+    let time = 0;
+    const yielder = new EventLoopYielder(8, () => time);
+    expect(yielder.maybeYield()).toBeUndefined();
+    let timerRan = false;
+    setImmediate(() => (timerRan = true));
+    time = 8;
+    const pause = yielder.maybeYield();
+    expect(pause).toBeInstanceOf(Promise);
+    await pause;
+    expect(timerRan).toBe(true);
+  });
+
+  it("starts a new budget once the event loop has turned", async () => {
+    let time = 0;
+    const yielder = new EventLoopYielder(8, () => time);
+    yielder.maybeYield();
+    time = 20;
+    await yielder.maybeYield();
+    // The pause let the event loop turn, so the budget starts again here.
+    expect(yielder.maybeYield()).toBeUndefined();
+    time = 27;
+    expect(yielder.maybeYield()).toBeUndefined();
+    time = 28;
+    expect(yielder.maybeYield()).toBeInstanceOf(Promise);
+  });
+
+  it("does not yield for a loop that waits on I/O between items", async () => {
+    let time = 0;
+    const yielder = new EventLoopYielder(8, () => time);
+    for (let i = 0; i < 5; i++) {
+      expect(yielder.maybeYield()).toBeUndefined();
+      time += 20;
+      // A wait on I/O lets the event loop run its phases. A timer alone can
+      // resume within the same timers phase on a slow machine, before the
+      // check phase that ends the slice, so the wait ends in that phase.
+      await new Promise((resolve) => setTimeout(resolve, 1));
+      await new Promise((resolve) => setImmediate(resolve));
+    }
+  });
+});
+
+describe("consumer under a buffered backlog", () => {
+  it("lets timers run while it drains, and keeps every update in order", async () => {
+    const chunks = Array.from({ length: 30 }, (_, i) => `chunk-${i} `);
+    const updates: SessionNotification[] = [];
+    let chunksBeforeTimer: number | undefined;
+    let timerArmed = false;
+    const agent = createAgent(bufferedAnswer(chunks), (notification) => {
+      updates.push(notification);
+      if (notification.update.sessionUpdate !== "agent_message_chunk") return;
+      if (!timerArmed) {
+        timerArmed = true;
+        setTimeout(() => (chunksBeforeTimer = chunkTexts(updates).length), 0);
+      }
+      // Each chunk costs 2 ms: the backlog takes far longer than one budget.
+      spin(2);
+    });
+
+    const response = await agent.prompt({ sessionId, prompt: [{ type: "text", text: "go" }] });
+
+    expect(response.stopReason).toBe("end_turn");
+    expect(chunkTexts(updates)).toEqual(chunks);
+    expect(chunksBeforeTimer).toBeDefined();
+    expect(chunksBeforeTimer!).toBeLessThan(chunks.length);
+  });
+
+  it("runs a cancel that arrives mid-backlog before the backlog ends", async () => {
+    const chunks = Array.from({ length: 30 }, (_, i) => `chunk-${i} `);
+    const updates: SessionNotification[] = [];
+    let chunksAtCancel: number | undefined;
+    let cancelScheduled = false;
+    const agent = createAgent(bufferedAnswer(chunks), (notification) => {
+      updates.push(notification);
+      if (notification.update.sessionUpdate !== "agent_message_chunk") return;
+      if (!cancelScheduled) {
+        cancelScheduled = true;
+        // A `session/cancel` arrives as I/O, so it can only run in a new macrotask.
+        setImmediate(() => {
+          chunksAtCancel = chunkTexts(updates).length;
+          void agent.cancel({ sessionId });
+        });
+      }
+      spin(2);
+    });
+    const query = agent.sessions[sessionId].query as unknown as {
+      interrupt: ReturnType<typeof vi.fn>;
+    };
+
+    const response = await agent.prompt({ sessionId, prompt: [{ type: "text", text: "go" }] });
+
+    expect(chunksAtCancel).toBeDefined();
+    expect(chunksAtCancel!).toBeLessThan(chunks.length);
+    expect(query.interrupt).toHaveBeenCalled();
+    // The turn ends cancelled. The chunks that were already buffered still
+    // reach the client, in order, as they did before the cancel.
+    expect(response.stopReason).toBe("cancelled");
+    expect(chunkTexts(updates)).toEqual(chunks);
+  });
+});
+
+describe("consumer teardown under a buffered backlog", () => {
+  it("settles the turn when the session closes mid-backlog", async () => {
+    const chunks = Array.from({ length: 30 }, (_, i) => `chunk-${i} `);
+    const updates: SessionNotification[] = [];
+    let closeScheduled = false;
+    let chunksAtClose: number | undefined;
+    let closed: Promise<unknown> | undefined;
+    const agent = createAgent(bufferedAnswer(chunks), (notification) => {
+      updates.push(notification);
+      if (notification.update.sessionUpdate !== "agent_message_chunk") return;
+      if (!closeScheduled) {
+        closeScheduled = true;
+        setImmediate(() => {
+          chunksAtClose = chunkTexts(updates).length;
+          closed = agent.closeSession({ sessionId });
+        });
+      }
+      spin(2);
+    });
+
+    const response = await agent.prompt({ sessionId, prompt: [{ type: "text", text: "go" }] });
+
+    expect(chunksAtClose).toBeDefined();
+    expect(chunksAtClose!).toBeLessThan(chunks.length);
+    expect(response.stopReason).toBe("cancelled");
+    await closed;
+    expect(agent.sessions[sessionId]).toBeUndefined();
+    // Whatever reached the client is a prefix of the answer, in order, and
+    // the rest of the backlog is not forwarded after the close.
+    const received = chunkTexts(updates);
+    expect(received.length).toBeLessThan(chunks.length);
+    expect(received).toEqual(chunks.slice(0, received.length));
+    for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve));
+    expect(chunkTexts(updates)).toEqual(received);
+  });
+});
+
+describe("nodeToWebWritable", () => {
+  it("writes the bytes of each chunk, including a view into a larger buffer", async () => {
+    const received: Buffer[] = [];
+    const sink = new Writable({
+      write(chunk: Buffer, _encoding, callback) {
+        received.push(Buffer.from(chunk));
+        callback();
+      },
+    });
+    const writer = nodeToWebWritable(sink).getWriter();
+    const backing = new TextEncoder().encode("xxhello\nworld\nyy");
+    await writer.write(backing.subarray(2, 8));
+    await writer.write(new TextEncoder().encode("world\n"));
+    expect(Buffer.concat(received).toString()).toBe("hello\nworld\n");
+  });
+
+  it("gives an object-mode stream a Buffer copy", async () => {
+    const received: unknown[] = [];
+    const sink = new Writable({
+      objectMode: true,
+      write(chunk, _encoding, callback) {
+        received.push(chunk);
+        callback();
+      },
+    });
+    const bytes = new TextEncoder().encode("hi");
+    await nodeToWebWritable(sink).getWriter().write(bytes);
+    expect(Buffer.isBuffer(received[0])).toBe(true);
+    expect(received[0]).not.toBe(bytes);
+    expect((received[0] as Buffer).toString()).toBe("hi");
+  });
+
+  it("rejects the write that fails", async () => {
+    const sink = new Writable({
+      write(_chunk, _encoding, callback) {
+        callback(new Error("EPIPE"));
+      },
+    });
+    sink.on("error", () => {});
+    const writer = nodeToWebWritable(sink).getWriter();
+    await expect(writer.write(new Uint8Array([1]))).rejects.toThrow("EPIPE");
+  });
+});

src/tests/mcp-command.test.ts

@@ -223,6 +223,16 @@ describe("parseMcpCommand", () => {
       expect(parseMcpCommand(text)).toBeNull();
     }
   });
+
+  it("reads past any leading whitespace, as before the prefix check", () => {
+    expect(parseMcpCommand("\n\t\u00a0\ufeff /mcp enable  my\nserver ")).toEqual({
+      action: "enable",
+      all: false,
+      server: "my server",
+    });
+    expect(parseMcpCommand("\u2028/mcp")).toEqual({ action: "status" });
+    expect(parseMcpCommand(`hello ${"word ".repeat(100_000)}/mcp`)).toBeNull();
+  });
 });
 
 const LONG_ERROR =

src/tests/session-titles.test.ts

@@ -2,7 +2,7 @@ import { describe, it, expect, beforeEach, vi } from "vitest";
 import { readFile } from "node:fs/promises";
 import { createRequire } from "node:module";
 import { ClaudeAcpAgent, type AcpClient } from "../acp-agent.js";
-import { appendTitleContext } from "../session-titles.js";
+import { appendTitleContext, sanitizeTitle } from "../session-titles.js";
 import { Pushable } from "../utils.js";
 import { getSessionInfo } from "@anthropic-ai/claude-agent-sdk";
 import {
@@ -17,6 +17,36 @@ vi.mock("@anthropic-ai/claude-agent-sdk", async (importOriginal) => {
   return { ...actual, getSessionInfo: vi.fn() };
 });
 
+describe("sanitizeTitle", () => {
+  /** The sanitizer before it learned to read only the start of a long text. */
+  function fullScanTitle(text: string): string {
+    const sanitized = text
+      .replace(/[\r\n]+/g, " ")
+      .replace(/\s+/g, " ")
+      .trim();
+    return sanitized.length <= 256 ? sanitized : sanitized.slice(0, 255) + "…";
+  }
+
+  it("gives a long text the same title as a scan of the whole text", () => {
+    const texts = [
+      "lorem ipsum dolor sit amet\n".repeat(400_000),
+      " \n\t".repeat(3000) + "word ".repeat(2000),
+      // The cut of the scanned prefix falls inside a run of whitespace.
+      "x".repeat(300) + " ".repeat(5000) + "tail",
+      // Mostly whitespace: the prefix collapses below a title's length.
+      "a ".repeat(100) + " ".repeat(10_000) + "b".repeat(300),
+      " ".repeat(10_000) + "short",
+      "\u00a0\ufeff".repeat(2100) + "😀".repeat(2000),
+      "y".repeat(4095) + "😀".repeat(10),
+      "z".repeat(257) + " ".repeat(4096),
+      "z".repeat(256) + " ".repeat(4096),
+    ];
+    for (const text of texts) {
+      expect(sanitizeTitle(text)).toBe(fullScanTitle(text));
+    }
+  });
+});
+
 describe("SDK title generation contract", () => {
   // Smoke test for the one undeclared SDK method this feature rests on:
   // `generateSessionTitle` is on the runtime `Query` class but absent from

src/utils.ts

@@ -45,12 +45,20 @@ export class Pushable<T> implements AsyncIterable<T> {
   }
 }
 
-// Helper to convert Node.js streams to Web Streams
+// Helper to convert Node.js streams to Web Streams. A byte stream gets each
+// chunk without a copy, so the caller must not change a chunk until its write
+// resolves.
 export function nodeToWebWritable(nodeStream: Writable): WritableStream<Uint8Array> {
   return new WritableStream<Uint8Array>({
     write(chunk) {
       return new Promise<void>((resolve, reject) => {
-        nodeStream.write(Buffer.from(chunk), (err) => {
+        // A Uint8Array is written as it is: Node wraps it in a Buffer view
+        // without a copy. The ACP encoder hands each message a fresh array,
+        // so nothing changes the bytes while the write is pending. An
+        // object-mode stream would get the array itself, so it keeps
+        // getting a Buffer copy, as before.
+        const data = nodeStream.writableObjectMode ? Buffer.from(chunk) : chunk;
+        nodeStream.write(data, (err) => {
           if (err) {
             reject(err);
           } else {
@@ -84,6 +92,58 @@ export function unreachable(value: never, logger: Logger = console) {
   logger.error(`Unexpected case: ${valueAsString}`);
 }
 
+/**
+ * Yields to the event loop when the current macrotask has run for too long.
+ *
+ * An await of an already-resolved promise continues in a microtask, so a loop
+ * that drains a backlog of buffered messages never lets timers or I/O run:
+ * incoming requests, `session/cancel` among them, wait until the backlog is
+ * gone. A loop calls {@link maybeYield} once per item. Within the budget it
+ * returns undefined and the loop goes on at once. Past the budget it returns a
+ * promise that resolves in the check phase of the event loop, after pending
+ * I/O has run.
+ *
+ * The budget is measured from the first call since the event loop last
+ * reached its check phase, so a loop that already waits on real I/O never
+ * yields, and all loops that share one instance share one budget per
+ * event-loop iteration.
+ *
+ * A test that fakes `setImmediate` and drives such a loop for longer than the
+ * budget must advance the fake timers, or the loop waits for them.
+ */
+export class EventLoopYielder {
+  /** Whether a check-phase marker is pending. While it is, the event loop has
+   *  not reached its check phase since {@link sliceStart}. */
+  private markerPending = false;
+  private sliceStart = 0;
+
+  constructor(
+    private readonly budgetMs: number,
+    private readonly now: () => number = () => performance.now(),
+  ) {}
+
+  maybeYield(): Promise<void> | undefined {
+    const time = this.now();
+    if (!this.markerPending) {
+      this.markerPending = true;
+      this.sliceStart = time;
+      setImmediate(() => {
+        this.markerPending = false;
+      });
+      return undefined;
+    }
+    if (time - this.sliceStart < this.budgetMs) return undefined;
+    return new Promise<void>((resolve) => setImmediate(resolve));
+  }
+}
+
+/** The longest stretch, in milliseconds, that a message loop keeps the event
+ *  loop before it yields. */
+const MESSAGE_LOOP_BUDGET_MS = 8;
+
+/** The yielder shared by the loops that forward messages to the client. */
+export const messageLoopYielder = new EventLoopYielder(MESSAGE_LOOP_BUDGET_MS);
+
 export function sleep(time: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, time));
 }