Confirm ACP interaction replies reach the provider pipe before settlement

Bind response receipts to the active stream, request and prompt; reject cancellation, disconnect, write failure and mismatched responses before acknowledging durable delivery.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/paperclip-runner/src/cli/acpx-response-delivery.test.ts

@@ -0,0 +1,29 @@
+import { describe, expect, it, vi } from "vitest";
+import { deliverAcpxResponse, requireAcpxResponseDelivery } from "./acpx-response-delivery.js";
+
+describe("ACP interaction response delivery", () => {
+  it("does not acknowledge callback settlement before the provider pipe write", async () => {
+    const written = Promise.withResolvers<void>();
+    const settle = vi.fn();
+    let acknowledged = false;
+    const pending = deliverAcpxResponse(requireAcpxResponseDelivery({ responseDelivery: written.promise }), settle)
+      .then(value => { acknowledged = true; return value; });
+    expect(settle).toHaveBeenCalledOnce();
+    await Promise.resolve();
+    expect(acknowledged).toBe(false);
+    written.resolve();
+    await expect(pending).resolves.toEqual({ resolved: true });
+  });
+
+  it.each(["cancelled", "disconnected", "write failed"])("does not report delivery when the response is %s", async reason => {
+    const written = Promise.withResolvers<void>();
+    const pending = deliverAcpxResponse(written.promise, () => {});
+    const failed = expect(pending).rejects.toThrow(reason);
+    written.reject(new Error(reason));
+    await failed;
+  });
+
+  it("requires a transport-owned receipt before presenting an interaction", () => {
+    expect(() => requireAcpxResponseDelivery({})).toThrow("no bound provider response");
+  });
+});

packages/paperclip-runner/src/cli/acpx-response-delivery.ts

@@ -0,0 +1,17 @@
+/** A local callback result is not evidence that the provider received its reply. */
+export function requireAcpxResponseDelivery(context: { responseDelivery?: Promise<void> }): Promise<void> {
+  if (!context.responseDelivery || typeof context.responseDelivery.then !== "function") {
+    throw new Error("ACP interaction has no bound provider response delivery receipt");
+  }
+  return context.responseDelivery;
+}
+
+/** Settle the callback first: the SDK can only write after that callback returns. */
+export async function deliverAcpxResponse(
+  responseDelivery: Promise<void>,
+  settle: () => void,
+): Promise<{ resolved: true }> {
+  settle();
+  await responseDelivery;
+  return { resolved: true };
+}

packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts

@@ -643,7 +643,7 @@ function loadWaitForTool(input: {
     "utf8",
   );
   const start = source.indexOf("async function waitForTool");
-  const end = source.indexOf("\nasync function waitForInput", start);
+  const end = source.indexOf("\nasync function waitForPermission", start);
   if (start < 0 || end < 0) throw new Error("waitForTool source not found");
   const functionSource = source
     .slice(start, end)

packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts

@@ -1,6 +1,7 @@
 #!/usr/bin/env node
 import { createHash } from "node:crypto";
 import { createInterface } from "node:readline";
+import { deliverAcpxResponse, requireAcpxResponseDelivery } from "./acpx-response-delivery.js";
 
 import type {
   AcpElicitationContext,
@@ -123,6 +124,7 @@ interface PendingTool {
 
 interface PendingInput {
   turnId: string;
+  responseDelivery: Promise<void>;
   questionSet: PaperclipQuestionSet;
   prepareResolution(resolution: HarnessRuntimeRequestResolution): () => void;
   cancel(): void;
@@ -131,6 +133,7 @@ interface PendingInput {
 
 interface PendingPermission {
   turnId: string;
+  responseDelivery: Promise<void>;
   normalized: NormalizedAcpxPermission;
   settle(response: AcpPermissionDecision): void;
   cleanup(): void;
@@ -372,7 +375,7 @@ async function dispatch(
         onElicitation: (providerRequest, context) =>
           waitForInput(currentTurnId, providerRequest, context),
         onPermissionRequest: (providerRequest, context) =>
-          waitForPermission(currentTurnId, providerRequest, context.signal),
+          waitForPermission(currentTurnId, providerRequest, context),
       });
     } catch (error) {
       turnId = null;
@@ -419,8 +422,7 @@ async function dispatch(
     const decision = pending.normalized.resolve(resolution);
     if (!permissions.delete(requestId)) throw new Error("permission request lost its settlement race");
     pending.cleanup();
-    pending.settle(decision);
-    return { resolved: true };
+    return await deliverAcpxResponse(pending.responseDelivery, () => pending.settle(decision));
   }
   if (request.command === "input.resolve") {
     const requestId = boundedIdentity(request.params.requestId, "requestId");
@@ -442,8 +444,7 @@ async function dispatch(
     if (!inputs.delete(requestId))
       throw new Error("input request lost its settlement race");
     pending.cleanup();
-    deliver();
-    return { resolved: true };
+    return await deliverAcpxResponse(pending.responseDelivery, deliver);
   }
   if (request.command === "tool.resolve") {
     const callId = boundedIdentity(request.params.callId, "callId");
@@ -731,12 +732,14 @@ async function waitForTool(call: RunnerToolCall): Promise<unknown> {
 async function waitForPermission(
   activeTurnId: string,
   request: AcpPermissionRequest,
-  signal: AbortSignal,
+  context: { signal: AbortSignal; responseDelivery?: Promise<void> },
 ): Promise<AcpPermissionDecision> {
+  const { signal } = context;
   if (turnId !== activeTurnId || signal.aborted || permissions.size >= MAX_PENDING_INPUTS) {
     return { outcome: "cancel" };
   }
   const normalized = normalizeAcpxPermission(request, openParams?.agent === "pi" ? { allowAlwaysScope: "session" } : {});
+  const responseDelivery = requireAcpxResponseDelivery(context);
   const requestId = stableRequestId(activeTurnId, ++requestSequence, normalized.toolCallId);
   return await new Promise((settle) => {
     const abort = () => {
@@ -745,7 +748,7 @@ async function waitForPermission(
       settle({ outcome: "cancel" });
     };
     permissions.set(requestId, {
-      turnId: activeTurnId, normalized, settle,
+      turnId: activeTurnId, normalized, settle, responseDelivery,
       cleanup: () => signal.removeEventListener("abort", abort),
     });
     signal.addEventListener("abort", abort, { once: true });
@@ -807,9 +810,10 @@ async function waitForInput(
 async function waitForExtensionInput(
   activeTurnId: string,
   input: AcpxExtensionInput,
-  context: { requestId: string | number | null; signal: AbortSignal },
+  context: { requestId: string | number | null; signal: AbortSignal; responseDelivery?: Promise<void> },
 ): Promise<Record<string, unknown>> {
   if (turnId !== activeTurnId || context.signal.aborted || inputs.size >= MAX_PENDING_INPUTS) return input.cancel();
+  const responseDelivery = requireAcpxResponseDelivery(context);
   const requestId = stableRequestId(activeTurnId, ++requestSequence, context.requestId);
   return await new Promise((settle) => {
     const abort = () => {
@@ -819,7 +823,7 @@ async function waitForExtensionInput(
       settle(input.cancel());
     };
     inputs.set(requestId, {
-      turnId: activeTurnId, questionSet: input.questionSet,
+      turnId: activeTurnId, questionSet: input.questionSet, responseDelivery,
       prepareResolution: resolution => {
         const response = input.resolve(resolution);
         return () => settle(response);

packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts

@@ -44,8 +44,10 @@ describe("Codex ACPX runtime adapter", () => {
       await expect(created.onPermissionRequest!(request as never, { signal })).resolves.toEqual({ outcome: "reject_once" });
     }
     expect(handler).not.toHaveBeenCalled();
-    await expect(created.onPermissionRequest!({ sessionId: "backend-1", raw: { sessionId: "backend-1" }, inferredKind: "edit" } as never, { signal })).resolves.toEqual({ outcome: "allow_once" });
+    const responseDelivery = Promise.resolve();
+    await expect(created.onPermissionRequest!({ sessionId: "backend-1", raw: { sessionId: "backend-1" }, inferredKind: "edit" } as never, { signal, responseDelivery })).resolves.toEqual({ outcome: "allow_once" });
     expect(handler).toHaveBeenCalledOnce();
+    expect(handler).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ responseDelivery }));
     options.permissionMode = "approve-all";
     await expect(created.onPermissionRequest!({ sessionId: "forged", raw: {}, inferredKind: "edit" } as never, { signal })).resolves.toEqual({ outcome: "reject_once" });
     pending.settle(); await turn.result; await port.close({ reason: "session checks complete" });
@@ -71,8 +73,9 @@ describe("Codex ACPX runtime adapter", () => {
     const signal = new AbortController().signal;
     await expect(created.onExtensionRequest!("cursor/ask_question", { sessionId: "backend-1" }, { requestId: 0, signal })).resolves.toEqual({ requestId: 0 });
     const sessionless = { toolCallId: "native-cursor-plan", plan: "Review the complete plan" };
-    await expect(created.onExtensionRequest!("cursor/create_plan", sessionless, { requestId: "plan", signal })).resolves.toEqual({ requestId: "plan" });
-    expect(request).toHaveBeenLastCalledWith("cursor/create_plan", { ...sessionless, sessionId: "backend-1" }, expect.objectContaining({ requestId: "plan" }));
+    const responseDelivery = Promise.resolve();
+    await expect(created.onExtensionRequest!("cursor/create_plan", sessionless, { requestId: "plan", signal, responseDelivery })).resolves.toEqual({ requestId: "plan" });
+    expect(request).toHaveBeenLastCalledWith("cursor/create_plan", { ...sessionless, sessionId: "backend-1" }, expect.objectContaining({ requestId: "plan", responseDelivery }));
     expect(sessionless).not.toHaveProperty("sessionId");
     await expect(created.onExtensionRequest!("pi/steer", {}, { requestId: 1, signal })).rejects.toThrow("admitted active turn");
     await expect(created.onExtensionRequest!("cursor/ask_question", { sessionId: "other" }, { requestId: 1, signal })).rejects.toThrow("admitted active turn");

packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts

@@ -324,7 +324,7 @@ export async function openQualifiedAcpxRuntime(
       // Cursor's native extensions omit sessionId. Only after admission may
       // that omission inherit this exact prompt's ACP wire session identity.
       const boundParams = params.sessionId === undefined ? { ...params, sessionId: active.sessionId } : params;
-      const response = await abortableExtensionResult(active.onRequest(method, boundParams, { requestId: context.requestId, signal }), signal);
+      const response = await abortableExtensionResult(active.onRequest(method, boundParams, { requestId: context.requestId, signal, responseDelivery: context.responseDelivery }), signal);
       signal.throwIfAborted();
       if (!ownsExtensionTurn(active, params)) throw new Error("ACPX extension request turn expired");
       return response;
@@ -377,6 +377,7 @@ export async function openQualifiedAcpxRuntime(
           // callback must never acquire the next turn's approval authority.
           const decision = await handler(request, {
             signal: AbortSignal.any([active.signal, context.signal]),
+            responseDelivery: context.responseDelivery,
           });
           if (permissionBoundary.active !== active || active.signal.aborted || context.signal.aborted) {
             return { outcome: "cancel" };

packages/paperclip-runner/src/drivers/acpx/profile-extensions.ts

@@ -77,7 +77,7 @@ export function bindAcpxExtensionTurn(input: {
   adapter: AcpxProfileExtensionAdapter | null;
   active(): boolean;
   sessionId: string;
-  waitForInput(input: AcpxExtensionInput, context: { requestId: string | number; signal: AbortSignal }): Promise<Record<string, unknown>>;
+  waitForInput(input: AcpxExtensionInput, context: { requestId: string | number; signal: AbortSignal; responseDelivery?: Promise<void> }): Promise<Record<string, unknown>>;
   emit(event: CanonicalProviderEvent): void;
 }) {
   let queue = Promise.resolve();
@@ -110,7 +110,7 @@ export function bindAcpxExtensionTurn(input: {
     if (!input.adapter) throw new Error("ACP extension adapter is unavailable");
   };
   return {
-    async onExtensionRequest(method: string, params: Record<string, unknown>, context: { requestId: string | number; signal: AbortSignal }) {
+    async onExtensionRequest(method: string, params: Record<string, unknown>, context: { requestId: string | number; signal: AbortSignal; responseDelivery?: Promise<void> }) {
       assertSession(params);
       if (context.signal.aborted) throw new Error("ACP extension was cancelled");
       const result = await enqueue(() => input.adapter!.request(method, params));

packages/paperclip-runner/test/acpx-response-delivery-package-contract.test.mjs

@@ -0,0 +1,100 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { mkdtemp, rm } from "node:fs/promises";
+import { createRequire } from "node:module";
+import { tmpdir } from "node:os";
+import { dirname, join } from "node:path";
+import { pathToFileURL } from "node:url";
+import test from "node:test";
+
+const require = createRequire(import.meta.url);
+const acpxRoot = process.env.PAPERCLIP_TEST_ACPX_PACKAGE_ROOT ?? dirname(require.resolve("acpx/package.json"));
+const { k: AcpClient } = await import(pathToFileURL(join(acpxRoot, "dist/live-checkpoint-BSIrfgVo.js")));
+const deferred = () => Promise.withResolvers();
+const nextTick = () => new Promise(resolve => setImmediate(resolve));
+
+function tapped(write) {
+  const client = new AcpClient({ agentCommand: "fixture", cwd: process.cwd(), permissionMode: "approve-all" });
+  client.activePrompt = { sessionId: "session", elicitationController: new AbortController() };
+  const stream = client.createTappedStream({ readable: new ReadableStream({ start() {} }), writable: new WritableStream({ write }) });
+  return { client, stream, controller: client.activePrompt.elicitationController };
+}
+
+test("receipt stays pending until the exact JSON-RPC result reaches the underlying writer", async () => {
+  const gate = deferred(); let receipt; let delivered = false;
+  const { stream } = tapped(() => gate.promise);
+  const result = await stream.withResponseDelivery(0, new AbortController().signal, async promise => {
+    receipt = promise; void receipt.then(() => { delivered = true; }); return { selected: "yes" };
+  });
+  const writer = stream.writable.getWriter();
+  const written = writer.write({ jsonrpc: "2.0", id: 0, result });
+  await nextTick(); assert.equal(delivered, false);
+  gate.resolve(); await written; await receipt; assert.equal(delivered, true);
+  stream.abortResponseDeliveries(); writer.releaseLock();
+});
+
+for (const failure of ["write", "cancel", "disconnect", "changed", "error"]) {
+  test(`receipt rejects ${failure} after callback resolution instead of falsely acknowledging delivery`, async () => {
+    const gate = deferred(); let receipt;
+    const { stream, controller } = tapped(() => gate.promise);
+    const result = await stream.withResponseDelivery("request", new AbortController().signal, async promise => { receipt = promise; return { selected: "yes" }; });
+    const rejected = assert.rejects(receipt);
+    const writer = stream.writable.getWriter();
+    const wire = failure === "error" ? { error: { code: -32000, message: "cancelled" } }
+      : { result: failure === "changed" ? { selected: "no" } : result };
+    const written = writer.write({ jsonrpc: "2.0", id: "request", ...wire });
+    if (failure === "write") { const writeFailed = assert.rejects(written); gate.reject(new Error("fixture write failure")); await writeFailed; }
+    else {
+      if (failure === "cancel") controller.abort();
+      if (failure === "disconnect") stream.abortResponseDeliveries();
+      gate.resolve(); await written;
+    }
+    await rejected; stream.abortResponseDeliveries(); writer.releaseLock();
+  });
+}
+
+test("duplicate unresolved IDs poison receipts; an old response cannot acknowledge a new request", async () => {
+  let receipt; const pending = deferred();
+  const { stream } = tapped(() => {});
+  const first = stream.withResponseDelivery(0, new AbortController().signal, async promise => { receipt = promise; await pending.promise; return {}; });
+  const rejected = assert.rejects(receipt);
+  await assert.rejects(stream.withResponseDelivery(0, new AbortController().signal, async () => ({})), /duplicated/);
+  await rejected; const expired = assert.rejects(first, /outlived/); pending.resolve(); await expired;
+});
+
+const fixture = String.raw`
+const readline = require('node:readline');
+const send = value => process.stdout.write(JSON.stringify({jsonrpc:'2.0',...value})+'\n');
+let promptId;
+readline.createInterface({input:process.stdin}).on('line', line => {
+ const message=JSON.parse(line);
+ if(message.method==='initialize') send({id:message.id,result:{protocolVersion:1,agentCapabilities:{},authMethods:[]}});
+ else if(message.method==='session/new') send({id:message.id,result:{sessionId:'session'}});
+ else if(message.method==='session/prompt') {
+  promptId=message.id; const mode=message.params.prompt[0].text;
+  if(mode==='permission') send({id:0,method:'session/request_permission',params:{sessionId:'session',toolCall:{toolCallId:'tool',title:'Edit',kind:'edit',status:'pending'},options:[{optionId:'yes',name:'Allow',kind:'allow_once'},{optionId:'no',name:'Deny',kind:'reject_once'}]}});
+  else if(mode==='elicitation') send({id:0,method:'elicitation/create',params:{sessionId:'session',mode:'form',message:'Choose',requestedSchema:{type:'object',properties:{name:{type:'string'}}}}});
+  else send({id:0,method:'fixture/question',params:{sessionId:'session'}});
+ } else if(message.id===0 && !message.method) {
+  send({method:'session/update',params:{sessionId:'session',update:{sessionUpdate:'agent_message_chunk',content:{type:'text',text:JSON.stringify(message)}}}});
+  send({id:promptId,result:{stopReason:'end_turn'}});
+ }
+});
+`;
+for (const mode of ["extension", "permission", "elicitation"]) {
+  test(`real ${mode} request supplies a stream-owned receipt without callback deadlock`, { timeout: 10000 }, async () => {
+    const cwd = await mkdtemp(join(tmpdir(), "acpx-response-delivery-"));
+    let delivery; let seen = 0;
+    const remember = context => { seen++; assert.ok(context.responseDelivery instanceof Promise); delivery = context.responseDelivery; };
+    const client = new AcpClient({ agentCommand: "fixture", cwd, authPolicy: "skip", permissionMode: "approve-all", fs: false, terminal: false, elicitationModes: ["form"], extensionMethods: ["fixture/question"],
+      spawnAgent: () => spawn(process.execPath, ["-e", fixture], { cwd, stdio: ["pipe", "pipe", "pipe"] }),
+      onExtensionRequest: async (_method, _params, context) => { remember(context); return { status: "accepted" }; },
+      onPermissionRequest: async (_params, context) => { remember(context); return { outcome: "allow_once" }; },
+    });
+    try {
+      await client.start(); await client.createSession();
+      const result = await client.prompt("session", mode, undefined, async (_params, context) => { remember(context); return { action: "accept", content: { name: "Paperclip" } }; });
+      assert.equal(result.stopReason, "end_turn"); assert.equal(seen, 1); await delivery;
+    } finally { await client.close(); await rm(cwd, { recursive: true, force: true }); }
+  });
+}

patches/acpx@0.13.1.patch

@@ -158,7 +158,15 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  			if (!this.options.verbose) return;
  			process.stderr.write(chunk);
  		});
-@@ -4253,7 +4304,12 @@
+@@ -4226,6 +4277,7 @@
+ 		const output = Readable.toWeb(child.stdout);
+ 		const stream = this.createTappedStream(createNdJsonMessageStream(this.options.agentCommand, input, output));
+ 		const connection = this.createConnection(stream, launch);
++		connection.signal.addEventListener("abort", () => stream.abortResponseDeliveries(), { once: true });
+ 		connection.signal.addEventListener("abort", () => {
+ 			this.recordAgentExit("connection_close", child.exitCode ?? null, child.signalCode ?? null);
+ 		}, { once: true });
+@@ -4253,7 +4305,12 @@
  			geminiAcp: isGeminiAcpCommand(spawnCommand, args),
  			copilotAcp: isCopilotAcpCommand(spawnCommand, args),
  			claudeAcp: isClaudeAcpCommand(spawnCommand, args),
@@ -172,7 +180,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  		};
  	}
  	logAgentLaunch(plan) {
-@@ -4279,10 +4335,17 @@
+@@ -4279,10 +4336,17 @@
  	}
  	async spawnAgentProcess(plan) {
  		const spawnCommand = buildAgentSpawnCommand(plan.spawnCommand, plan.args, process.platform, plan.spawnOptions.env);
@@ -192,19 +200,33 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  		try {
  			await waitForSpawn$1(spawnedChild);
  		} catch (error) {
-@@ -4312,6 +4375,11 @@
+@@ -4293,10 +4357,10 @@
+ 	createConnection(stream, launch) {
+ 		const app = client({ name: "acpx" }).onNotification(methods.client.session.update, async ({ params }) => {
+ 			await this.handleSessionUpdate(params);
+-		}).onNotification(methods.client.elicitation.complete, async () => {}).onRequest(methods.client.session.requestPermission, async ({ params }) => {
+-			return await this.handlePermissionRequest(params);
++		}).onNotification(methods.client.elicitation.complete, async () => {}).onRequest(methods.client.session.requestPermission, async ({ params, requestId, signal }) => {
++			return await stream.withResponseDelivery(requestId, signal, (responseDelivery) => this.handlePermissionRequest(params, responseDelivery));
+ 		}).onRequest(methods.client.elicitation.create, async ({ params, requestId, signal }) => {
+-			return await this.handleElicitationRequest(params, requestId, signal);
++			return await stream.withResponseDelivery(requestId, signal, (responseDelivery) => this.handleElicitationRequest(params, requestId, signal, responseDelivery));
+ 		}).onRequest(methods.client.fs.readTextFile, async ({ params }) => {
+ 			return await this.handleReadTextFile(params);
+ 		}).onRequest(methods.client.fs.writeTextFile, async ({ params }) => {
+@@ -4312,6 +4376,11 @@
  		}).onRequest(methods.client.terminal.release, async ({ params }) => {
  			return await this.handleReleaseTerminal(params);
  		});
 +		for (const method of this.options.extensionMethods) {
 +			if (this.options.onExtensionRequest) app.onRequest(method, boundedExtensionRecord, async ({ params, requestId, signal }) => {
-+				return await this.handleExtensionRequest(method, params, requestId, signal);
++				return await stream.withResponseDelivery(requestId, signal, (responseDelivery) => this.handleExtensionRequest(method, params, requestId, signal, responseDelivery));
 +			});
 +		}
  		if (launch.devinAcp) app.onRequest("_cognition.ai/request_diagnostics", (params) => {
  			return params && typeof params === "object" && !Array.isArray(params) ? params : {};
  		}, async () => ({}));
-@@ -4333,12 +4401,12 @@
+@@ -4333,12 +4402,12 @@
  	async initializeProtocolConnection(connection, launch) {
  		const initializePromise = connection.initialize({
  			protocolVersion: PROTOCOL_VERSION,
@@ -219,7 +241,44 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  			clientInfo: resolveClientInfo(launch.devinAcp)
  		});
  		const initialized = launch.geminiAcp ? await withTimeout(initializePromise, resolveGeminiAcpStartupTimeoutMs()) : await initializePromise;
-@@ -4367,8 +4435,9 @@
+@@ -4358,6 +4427,36 @@
+ 		throw normalizedError;
+ 	}
+ 	createTappedStream(base) {
++		const deliveries = new Map();
++		const scope = new AbortController();
++		const abortResponseDeliveries = () => scope.abort(new Error("ACP response connection closed before delivery"));
++		const withResponseDelivery = async (id, requestSignal, handler) => {
++			if (!(id === null || typeof id === "string" || typeof id === "number" && Number.isFinite(id))) throw new Error("ACP response request id is invalid");
++			if (deliveries.has(id) || deliveries.size >= 128) {
++				abortResponseDeliveries();
++				throw new Error("ACP response request id is duplicated or pending deliveries exceeded their bound");
++			}
++			const active = this.activePrompt;
++			const signal = AbortSignal.any([scope.signal, requestSignal, ...(active ? [active.elicitationController.signal] : [])]);
++			let settle, reject;
++			const responseDelivery = new Promise((resolve, fail) => { settle = resolve; reject = fail; });
++			// The host may decline the request without awaiting its receipt.
++			void responseDelivery.catch(() => {});
++			const abort = () => reject(new Error("ACP response delivery was cancelled or disconnected"));
++			const receipt = { settle, reject, expected: void 0, cleanup: () => signal.removeEventListener("abort", abort) };
++			deliveries.set(id, receipt);
++			signal.addEventListener("abort", abort, { once: true });
++			if (signal.aborted) abort();
++			try {
++				const result = await handler(responseDelivery);
++				if (signal.aborted || this.activePrompt !== active) throw new Error("ACP response outlived its admitted prompt");
++				receipt.expected = JSON.stringify(result);
++				return result;
++			} catch (error) {
++				reject(new Error("ACP request failed before response delivery"));
++				throw error;
++			}
++		};
+ 		const onAcpMessage = () => this.eventHandlers.onAcpMessage;
+ 		const onAcpOutputMessage = () => this.eventHandlers.onAcpOutputMessage;
+ 		const elicitationRequestIds = /* @__PURE__ */ new Set();
+@@ -4367,8 +4466,9 @@
  		const shouldSuppressInboundReplaySessionUpdate = (message) => {
  			return this.suppressReplaySessionUpdateMessages && isSessionUpdateNotification(message);
  		};
@@ -230,7 +289,15 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  			if (requestId !== void 0) {
  				elicitationRequestIds.add(requestId);
  				return;
-@@ -4387,7 +4456,10 @@
+@@ -4379,6 +4479,7 @@
+ 			}
+ 		};
+ 		return {
++			withResponseDelivery, abortResponseDeliveries,
+ 			readable: new ReadableStream({ async start(controller) {
+ 				const reader = base.readable.getReader();
+ 				try {
+@@ -4387,10 +4488,16 @@
  						if (done) break;
  						if (!value) continue;
  						observeInbound(value);
@@ -242,11 +309,51 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  						controller.enqueue(value);
  					}
  				} finally {
-@@ -4926,6 +4998,40 @@
++					abortResponseDeliveries();
++					for (const receipt of deliveries.values()) receipt.cleanup();
++					deliveries.clear();
+ 					reader.releaseLock();
+ 					controller.close();
+ 				}
+@@ -4403,10 +4510,20 @@
+ 					onAcpOutputMessage()?.("outbound", message);
+ 					onAcpMessage()?.("outbound", message);
+ 				}
++				const receipt = id === void 0 ? void 0 : deliveries.get(id);
+ 				const writer = base.writable.getWriter();
+ 				try {
+ 					await writer.write(message);
++					if (receipt) {
++						if (!("result" in message) || receipt.expected === void 0 || JSON.stringify(message.result) !== receipt.expected) receipt.reject(new Error("ACP wrote an error or changed response instead of the selected resolution"));
++						else receipt.settle();
++					}
++				} catch (error) {
++					receipt?.reject(new Error("ACP response write failed"));
++					abortResponseDeliveries();
++					throw error;
+ 				} finally {
++					if (receipt) { receipt.cleanup(); deliveries.delete(id); }
+ 					writer.releaseLock();
+ 				}
+ 			} })
+@@ -4915,9 +5032,9 @@
+ 		await connection.authenticate({ methodId: selected.methodId });
+ 		this.log(`authenticated with method ${selected.methodId} (${selected.source})`);
+ 	}
+-	async handlePermissionRequest(params) {
++	async handlePermissionRequest(params, responseDelivery) {
+ 		if (this.cancellingSessionIds.has(params.sessionId)) return cancelledPermissionResponse();
+-		const hostResponse = await this.tryHandlePermissionRequestWithHost(params);
++		const hostResponse = await this.tryHandlePermissionRequestWithHost(params, responseDelivery);
+ 		if (hostResponse) return hostResponse;
+ 		const { response, recorded } = await this.resolvePermissionRequestFromMode(params);
+ 		if (!recorded) {
+@@ -4926,14 +5043,49 @@
  		}
  		return response;
  	}
-+	async handleExtensionRequest(method, params, requestId, requestSignal) {
+-	async handleElicitationRequest(request, requestId, requestSignal) {
++	async handleExtensionRequest(method, params, requestId, requestSignal, responseDelivery) {
 +		const active = this.activePrompt;
 +		const connection = this.connection;
 +		if (!this.options.extensionMethods.includes(method) || !this.options.onExtensionRequest) throw new Error("ACP extension method is not enabled");
@@ -260,7 +367,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
 +			signal.addEventListener("abort", onAbort, { once: true });
 +		});
 +		try {
-+			const result = await Promise.race([Promise.resolve().then(() => this.options.onExtensionRequest(method, boundedExtensionRecord(params), { requestId, signal })), aborted]);
++			const result = await Promise.race([Promise.resolve().then(() => this.options.onExtensionRequest(method, boundedExtensionRecord(params), { requestId, signal, responseDelivery })), aborted]);
 +			signal.throwIfAborted();
 +			if (this.activePrompt !== active || this.connection !== connection) throw new Error("ACP extension request outlived its prompt");
 +			return boundedExtensionRecord(result);
@@ -280,10 +387,38 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
 +			this.log("ACP extension notification was rejected");
 +		}
 +	}
- 	async handleElicitationRequest(request, requestId, requestSignal) {
++	async handleElicitationRequest(request, requestId, requestSignal, responseDelivery) {
  		const resolved = this.resolveElicitationOwner(request);
  		if ("response" in resolved) return resolved.response;
-@@ -5028,6 +5134,12 @@
+ 		const { active, handler } = resolved.owner;
+ 		const signal = AbortSignal.any([requestSignal, active.elicitationController.signal]);
+ 		const handlerAttempt = Promise.resolve().then(async () => await handler(request, {
+ 			requestId,
+-			signal
++			signal,
++			responseDelivery
+ 		})).then((response) => ({
+ 			kind: "response",
+ 			response
+@@ -4970,7 +5122,7 @@
+ 	isElicitationSessionCancelling(sessionId) {
+ 		return this.closing || this.cancellingSessionIds.has(sessionId);
+ 	}
+-	async tryHandlePermissionRequestWithHost(params) {
++	async tryHandlePermissionRequestWithHost(params, responseDelivery) {
+ 		if (!this.options.onPermissionRequest) return;
+ 		const signal = this.cancellationSignalForSession(params.sessionId);
+ 		try {
+@@ -4978,7 +5130,7 @@
+ 				sessionId: params.sessionId,
+ 				raw: params,
+ 				inferredKind: inferToolKind(params)
+-			}, { signal });
++			}, { signal, responseDelivery });
+ 			return this.hostPermissionDecisionResponse(params, signal, decision);
+ 		} catch (error) {
+ 			return this.hostPermissionErrorResponse(params, signal, error);
+@@ -5028,6 +5180,12 @@
  	attachAgentLifecycleObservers(child) {
  		child.once("exit", (exitCode, signal) => {
  			this.recordAgentExit("process_exit", exitCode, signal);
@@ -296,7 +431,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  		});
  		child.once("close", (exitCode, signal) => {
  			this.recordAgentExit("process_close", exitCode, signal);
-@@ -5771,10 +5883,20 @@
+@@ -5771,10 +5929,20 @@
  	trimConversationForRuntime(conversation);
  	return acpx;
  }
@@ -319,7 +454,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  	updateConversationTimestamp(conversation, timestamp);
  	trimConversationForRuntime(conversation);
  	return true;
-@@ -6435,6 +6557,27 @@
+@@ -6435,6 +6603,27 @@
  //#region src/runtime/engine/prompt-turn.ts
  const SESSION_REPLY_IDLE_MS = 1e3;
  const SESSION_REPLY_DRAIN_TIMEOUT_MS = 5e3;
@@ -347,7 +482,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  async function runPromptTurn(params) {
  	try {
  		const promptPromise = params.client.prompt(params.sessionId, params.prompt, params.onPromptRequestStarted, params.onElicitation);
-@@ -6444,6 +6587,19 @@
+@@ -6444,6 +6633,19 @@
  			idleMs: SESSION_REPLY_IDLE_MS,
  			timeoutMs: SESSION_REPLY_DRAIN_TIMEOUT_MS
  		}).catch(() => {});
@@ -367,7 +502,7 @@ diff --git a/dist/live-checkpoint-BSIrfgVo.js b/dist/live-checkpoint-BSIrfgVo.js
  		recordPromptResponseUsage(params.conversation, response.usage, params.promptMessageId);
  		return {
  			stopReason: response.stopReason,
-@@ -6518,4 +6674,4 @@
+@@ -6518,4 +6720,4 @@
  //#endregion
  export { writeSessionRecord as $, PERMISSION_POLICY_ACTIONS as $t, REQUESTED_MODEL_UNSUPPORTED_ERROR_CODE as A, TimeoutError as At, getAcpxVersion as B, formatErrorMessage as Bt, mergeSessionOptions as C, PromptInputValidationError as Ct, applyLifecycleSnapshotToRecord as D, promptToDisplayText as Dt, applyConversation as E, parsePromptSource as Et, modelStateFromConfigOptions as F, normalizeAgentName$1 as Ft, findSession as G, toAcpErrorPayload as Gt, DEFAULT_HISTORY_LIMIT as H, normalizeOutputError as Ht, normalizeAgentCommandInput as I, resolveAgentArgv as It, listSessions as J, NON_INTERACTIVE_PERMISSION_POLICIES as Jt, findSessionByDirectoryWalk as K, AUTH_POLICIES as Kt, renderArgvIdentity as L, resolveAgentCommand as Lt, RequestedModelUnsupportedError as M, withTimeout as Mt, assertRequestedModelSupported as N, DEFAULT_AGENT_NAME as Nt, reconcileAgentSessionId as O, textPrompt as Ot, isRequestedModelUnsupportedError as P, listBuiltInAgents as Pt, resolveSessionRecord as Q, PERMISSION_MODES as Qt, runTimedExecFile as R, resolveCanonicalAgentName as Rt, advertisedModelState as S, parsePromptStopReason as St, sessionOptionsFromRecord as T, mergePromptSourceWithText as Tt, absolutePath as U, extractAcpError as Ut, permissionModeSatisfies as V, isRetryablePromptError as Vt, findGitRepositoryRoot as W, isAcpResourceNotFoundError as Wt, normalizeName as X, OUTPUT_ERROR_ORIGINS as Xt, listSessionsForAgent as Y, OUTPUT_ERROR_CODES as Yt, pruneSessions as Z, OUTPUT_FORMATS as Zt, createSessionConversation as _, sessionEventLockPath as _t, applyRequestedModelIfAdvertised as a, measurePerf as at, recordSessionUpdate as b, isAcpJsonRpcMessage as bt, setCurrentModelId as c, setPerfGauge as ct, setDesiredModelId as d, serializeSessionRecordForDisk as dt, SESSION_RECORD_SCHEMA as en, createAtomicWriteTempPath as et, syncAdvertisedModelState as f, normalizeRuntimeSessionId as ft, cloneSessionConversation as g, sessionEventActivePath as gt, cloneSessionAcpxState as h, sessionBaseDir$1 as ht, connectAndLoadSession as i, QueueProtocolError as in, incrementPerfCounter as it, REQUESTED_MODEL_UNSUPPORTED_REASONS as j, withInterrupt as jt, AcpClient as k, InterruptedError as kt, setDesiredConfigOption as l, startPerfTimer as lt, applyConfigOptionsToState as m, defaultSessionEventLog as mt, runPromptTurn as n, AgentSpawnError as nn, formatPerfMetric as nt, currentModelIdFromSetModelResponse as o, recordPerfDuration as ot, applyConfigOptionsToRecord as p, DEFAULT_EVENT_SEGMENT_MAX_BYTES as pt, isoNow$2 as q, EXIT_CODES as qt, withConnectedSession as r, QueueConnectionError as rn, getPerfMetricsSnapshot as rt, clearDesiredConfigOption as s, resetPerfMetrics as st, LiveSessionCheckpoint as t, AcpxOperationalError as tn, assertPersistedKeyPolicy as tt, setDesiredModeId as u, parseSessionRecord as ut, recordClientOperation as v, sessionEventSegmentPath as vt, persistSessionOptions as w, isPromptInput as wt, trimConversationForRuntime as x, parseJsonRpcErrorMessage as xt, recordPromptSubmission as y, extractSessionUpdateNotification as yt, splitCommandLine as z, exitCodeForOutputErrorCode as zt };
  
@@ -418,9 +553,12 @@ diff --git a/dist/runtime.d.ts b/dist/runtime.d.ts
  } | {
    type: "tool_call";
    text: string;
-@@ -313,6 +324,45 @@
+@@ -312,7 +323,49 @@
+   elicitationModes?: readonly AcpElicitationMode[];
    onPermissionRequest?: (req: AcpPermissionRequest, ctx: {
      signal: AbortSignal;
++    /** Resolves after this exact response is written to the current provider pipe. */
++    responseDelivery?: Promise<void>;
    }) => Promise<AcpPermissionDecision | undefined>;
 +  /** Ephemeral initialize capabilities; mandatory base capabilities take precedence. */
 +  clientCapabilities?: Record<string, unknown>;
@@ -430,6 +568,7 @@ diff --git a/dist/runtime.d.ts b/dist/runtime.d.ts
 +  onExtensionRequest?: (method: string, params: Record<string, unknown>, context: {
 +    requestId: string | number;
 +    signal: AbortSignal;
++    responseDelivery?: Promise<void>;
 +  }) => Promise<Record<string, unknown>>;
 +  onExtensionNotification?: (method: string, params: Record<string, unknown>) => void;
 +  /** Ephemeral allowlisted environment evaluated immediately before child spawn. */
@@ -464,7 +603,7 @@ diff --git a/dist/runtime.d.ts b/dist/runtime.d.ts
  };
  type AcpFileSessionStoreOptions = {
    stateDir: string;
-@@ -369,6 +419,7 @@
+@@ -369,6 +422,7 @@
    private createAndSaveRuntimeRecord;
    private retainInitializedSessionOwner;
    startTurn(input: {
@@ -472,7 +611,7 @@ diff --git a/dist/runtime.d.ts b/dist/runtime.d.ts
      handle: AcpRuntimeHandle;
      text: string;
      attachments?: AcpRuntimeTurnAttachment[];
-@@ -414,6 +465,7 @@
+@@ -414,6 +468,7 @@
    private cleanupRuntimeTurn;
    private finalizeRuntimeTurnRecord;
    runTurn(input: {
@@ -700,7 +839,16 @@ diff --git a/dist/session-options-DwRDODlr.d.ts b/dist/session-options-DwRDODlr.
  //#region src/prompt-content.d.ts
  type PromptInput = ContentBlock[];
  //#endregion
-@@ -116,6 +117,34 @@
+@@ -27,6 +28,8 @@
+   requestId: JsonRpcId;
+   /** Aborts with the request itself or its owning prompt turn/session. */
+   signal: AbortSignal;
++  /** Await only after returning the selected response from this callback. */
++  responseDelivery?: Promise<void>;
+ };
+ type AcpElicitationResponseMeta = {
+   _meta?: Record<string, unknown> | null;
+@@ -116,6 +119,35 @@
      };
      env?: Record<string, string>;
    };
@@ -712,6 +860,7 @@ diff --git a/dist/session-options-DwRDODlr.d.ts b/dist/session-options-DwRDODlr.
 +  onExtensionRequest?: (method: string, params: Record<string, unknown>, context: {
 +    requestId: string | number;
 +    signal: AbortSignal;
++    responseDelivery?: Promise<void>;
 +  }) => Promise<Record<string, unknown>>;
 +  onExtensionNotification?: (method: string, params: Record<string, unknown>) => void;
 +  /** Ephemeral child environment factory; its return value is never persisted. */
@@ -735,7 +884,16 @@ diff --git a/dist/session-options-DwRDODlr.d.ts b/dist/session-options-DwRDODlr.
    onAcpMessage?: (direction: AcpMessageDirection, message: AcpJsonRpcMessage) => void;
    onAcpOutputMessage?: (direction: AcpMessageDirection, message: AcpJsonRpcMessage) => void;
    onSessionUpdate?: (notification: SessionNotification) => void;
-@@ -263,12 +292,15 @@
+@@ -123,6 +155,8 @@
+   onPermissionEscalation?: (event: PermissionEscalationEvent) => void;
+   onPermissionRequest?: (req: AcpPermissionRequest, ctx: {
+     signal: AbortSignal;
++    /** Resolves after this exact response is written to the current provider pipe. */
++    responseDelivery?: Promise<void>;
+   }) => Promise<AcpPermissionDecision | undefined>;
+ };
+ declare const SESSION_RECORD_SCHEMA: "acpx.session.v1";
+@@ -263,12 +297,15 @@
    lastAgentDisconnectReason?: string;
    protocolVersion?: number;
    agentCapabilities?: AgentCapabilities;
@@ -752,7 +910,7 @@ diff --git a/dist/session-options-DwRDODlr.d.ts b/dist/session-options-DwRDODlr.
    acpx?: SessionAcpxState;
    importedFrom?: SessionImportedFrom;
  };
-@@ -295,4 +327,4 @@
+@@ -295,4 +332,4 @@
  };
  //#endregion
  export { SessionRecord as _, AcpElicitationHandler as a, AcpElicitationResponse as c, AuthPolicy as d, McpServer$1 as f, PermissionStats as g, PermissionPolicy as h, AcpElicitationContext as i, AcpPermissionDecision as l, PermissionMode as m, SystemPromptOption as n, AcpElicitationMode as o, NonInteractivePermissionPolicy as p, AcpClientOptions as r, AcpElicitationRequest as s, SessionAgentOptions as t, AcpPermissionRequest as u, PromptInput as v };