From 1eda44ee3627e2a0c149fbbfa972e01665f6ed96 Mon Sep 17 00:00:00 2001 From: NianJiu <3235467914@qq.com> Date: Fri, 17 Jul 2026 01:01:10 +0800 Subject: [PATCH] fix(acp): show durable notice after gateway disconnect (#108827) * fix(acp): record disconnect interruptions * test(acp): tighten disconnect settlement coverage Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * fix(acp): preserve replay order under backpressure Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * test(acp): await accepted prompt ledger record Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * fix(acp): isolate ledger mutation queues Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> --------- Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> Co-authored-by: Peter Steinberger --- src/acp/translator.event-ledger.test.ts | 9 + src/acp/translator.session-updates.test.ts | 106 +++++++++- src/acp/translator.session-updates.ts | 119 ++++++++---- src/acp/translator.stop-reason.test.ts | 215 +++++++++++++++++++-- src/acp/translator.ts | 68 +++++-- 5 files changed, 447 insertions(+), 70 deletions(-) diff --git a/src/acp/translator.event-ledger.test.ts b/src/acp/translator.event-ledger.test.ts index f8f0b74d952b..0692a1fcd951 100644 --- a/src/acp/translator.event-ledger.test.ts +++ b/src/acp/translator.event-ledger.test.ts @@ -115,6 +115,15 @@ describe("ACP translator event ledger replay", () => { const promptPromise = firstAgent.prompt(createPromptRequest(created.sessionId, "Question")); await waitForChatSend(firstRequestMock); + await vi.waitFor(async () => { + const replay = await eventLedger.readReplay({ + sessionId: created.sessionId, + sessionKey: firstSession.sessionKey, + }); + expect( + replay.events.some((event) => event.update.sessionUpdate === "user_message_chunk"), + ).toBe(true); + }); const runId = firstSessionStore.getSession(created.sessionId)?.activeRunId; if (!runId) { throw new Error("Expected active ACP run"); diff --git a/src/acp/translator.session-updates.test.ts b/src/acp/translator.session-updates.test.ts index e66bf8f76c75..cc1a18e39a52 100644 --- a/src/acp/translator.session-updates.test.ts +++ b/src/acp/translator.session-updates.test.ts @@ -34,7 +34,7 @@ const update: SessionUpdate = { availableCommands: [], }; -describe("AcpTranslatorSessionUpdates shutdown", () => { +describe("AcpTranslatorSessionUpdates", () => { it("blocks ledger reads and writes after shutdown starts", async () => { const ledger = createLedger(); const sessionUpdate = vi.fn(async () => {}); @@ -101,4 +101,108 @@ describe("AcpTranslatorSessionUpdates shutdown", () => { expect(ledger.recordUpdate).not.toHaveBeenCalled(); }); + + it("preserves ledger order without waiting for ACP delivery", async () => { + let releaseFirstDelivery!: () => void; + const firstDelivery = new Promise((resolve) => { + releaseFirstDelivery = resolve; + }); + let deliveryCount = 0; + const ledger = createLedger(); + const updates = createUpdates({ + ledger, + sessionUpdate: vi.fn(async () => { + deliveryCount += 1; + if (deliveryCount === 1) { + await firstDelivery; + } + }), + }); + const firstUpdate: SessionUpdate = { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "first" }, + }; + const interruptionUpdate: SessionUpdate = { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "interrupted" }, + }; + + const firstEmit = updates.emit({ + sessionId: "session-1", + sessionKey: "agent:main:session-1", + record: true, + update: firstUpdate, + }); + await vi.waitFor(() => { + expect(ledger.recordUpdate).toHaveBeenCalledTimes(1); + }); + + await updates.emit({ + sessionId: "session-1", + sessionKey: "agent:main:session-1", + record: true, + waitForDelivery: false, + update: interruptionUpdate, + }); + + expect(vi.mocked(ledger.recordUpdate).mock.calls.map(([params]) => params.update)).toEqual([ + firstUpdate, + interruptionUpdate, + ]); + await expect(Promise.race([firstEmit, Promise.resolve("pending")])).resolves.toBe("pending"); + releaseFirstDelivery(); + await firstEmit; + }); + + it("does not let a stalled ledger session block another session", async () => { + let releaseFirstWrite!: () => void; + const firstWrite = new Promise((resolve) => { + releaseFirstWrite = resolve; + }); + const ledger = createLedger(); + vi.mocked(ledger.recordUpdate).mockImplementation(async ({ sessionId }) => { + if (sessionId === "session-1") { + await firstWrite; + } + }); + const updates = createUpdates({ ledger }); + + const stalledEmit = updates.emit({ + sessionId: "session-1", + sessionKey: "agent:main:session-1", + record: true, + waitForDelivery: false, + update, + }); + await vi.waitFor(() => { + expect(ledger.recordUpdate).toHaveBeenCalledTimes(1); + }); + + const queuedSameSessionEmit = updates.emit({ + sessionId: "session-1", + sessionKey: "agent:main:session-1", + record: true, + waitForDelivery: false, + update, + }); + await updates.emit({ + sessionId: "session-2", + sessionKey: "agent:main:session-2", + record: true, + waitForDelivery: false, + update, + }); + + expect(vi.mocked(ledger.recordUpdate).mock.calls.map(([params]) => params.sessionId)).toEqual([ + "session-1", + "session-2", + ]); + releaseFirstWrite(); + await Promise.all([stalledEmit, queuedSameSessionEmit]); + expect(vi.mocked(ledger.recordUpdate).mock.calls.map(([params]) => params.sessionId)).toEqual([ + "session-1", + "session-2", + "session-1", + ]); + }); }); diff --git a/src/acp/translator.session-updates.ts b/src/acp/translator.session-updates.ts index ab39712551c5..8e09c2357c78 100644 --- a/src/acp/translator.session-updates.ts +++ b/src/acp/translator.session-updates.ts @@ -33,6 +33,9 @@ function resolveLedgerSessionId(session: { sessionId: string; ledgerSessionId?: /** Helper that keeps ACP client updates and replay ledger writes in sync. */ export class AcpTranslatorSessionUpdates { private stopped = false; + // Queue each ledger session at emission time so a detached disconnect notice + // cannot overtake its older update or block unrelated session settlement. + private ledgerMutationTails = new Map>(); constructor(private options: AcpTranslatorSessionUpdatesOptions) {} @@ -108,22 +111,24 @@ export class AcpTranslatorSessionUpdates { runId: string, prompt: PromptRequest["prompt"], ): Promise { - if (this.stopped) { - return; - } - try { - await this.options.eventLedger.recordUserPrompt({ - sessionId: resolveLedgerSessionId(session), - sessionKey: session.sessionKey, - runId, - prompt, - }); - } catch (err) { - this.options.log( - `event ledger prompt record failed for ${session.sessionId}: ${String(err)}`, - ); - await this.markLedgerIncomplete(session); - } + await this.enqueueLedgerMutation(resolveLedgerSessionId(session), async () => { + if (this.stopped) { + return; + } + try { + await this.options.eventLedger.recordUserPrompt({ + sessionId: resolveLedgerSessionId(session), + sessionKey: session.sessionKey, + runId, + prompt, + }); + } catch (err) { + this.options.log( + `event ledger prompt record failed for ${session.sessionId}: ${String(err)}`, + ); + await this.markLedgerIncomplete(session); + } + }); } async emit(params: { @@ -133,23 +138,33 @@ export class AcpTranslatorSessionUpdates { runId?: string; update: SessionUpdate; record?: boolean; + waitForDelivery?: boolean; }): Promise { if (this.stopped) { return; } - await this.options.connection.sessionUpdate({ + const delivery = this.options.connection.sessionUpdate({ sessionId: params.sessionId, update: params.update, }); - if (params.record && params.sessionKey) { - await this.recordLedgerUpdate({ - sessionId: params.sessionId, - sessionKey: params.sessionKey, - ...(params.ledgerSessionId ? { ledgerSessionId: params.ledgerSessionId } : {}), - ...(params.runId ? { runId: params.runId } : {}), - update: params.update, + const recording = + params.record && params.sessionKey + ? this.recordLedgerUpdate({ + sessionId: params.sessionId, + sessionKey: params.sessionKey, + ...(params.ledgerSessionId ? { ledgerSessionId: params.ledgerSessionId } : {}), + ...(params.runId ? { runId: params.runId } : {}), + update: params.update, + }) + : undefined; + if (params.waitForDelivery === false) { + void delivery.catch((err: unknown) => { + this.options.log(`session update delivery failed for ${params.sessionId}: ${String(err)}`); }); + } else { + await delivery; } + await recording; } async sendAvailableCommands( @@ -175,24 +190,44 @@ export class AcpTranslatorSessionUpdates { runId?: string; update: SessionUpdate; }): Promise { - if (this.stopped) { - return; - } - try { - await this.options.eventLedger.recordUpdate({ - sessionId: params.ledgerSessionId ?? params.sessionId, - sessionKey: params.sessionKey, - ...(params.runId ? { runId: params.runId } : {}), - update: params.update, - }); - } catch (err) { - this.options.log(`event ledger update record failed for ${params.sessionId}: ${String(err)}`); - await this.markLedgerIncomplete({ - sessionId: params.sessionId, - sessionKey: params.sessionKey, - ...(params.ledgerSessionId ? { ledgerSessionId: params.ledgerSessionId } : {}), - }); - } + await this.enqueueLedgerMutation(params.ledgerSessionId ?? params.sessionId, async () => { + if (this.stopped) { + return; + } + try { + await this.options.eventLedger.recordUpdate({ + sessionId: params.ledgerSessionId ?? params.sessionId, + sessionKey: params.sessionKey, + ...(params.runId ? { runId: params.runId } : {}), + update: params.update, + }); + } catch (err) { + this.options.log( + `event ledger update record failed for ${params.sessionId}: ${String(err)}`, + ); + await this.markLedgerIncomplete({ + sessionId: params.sessionId, + sessionKey: params.sessionKey, + ...(params.ledgerSessionId ? { ledgerSessionId: params.ledgerSessionId } : {}), + }); + } + }); + } + + private enqueueLedgerMutation( + ledgerSessionId: string, + mutation: () => Promise, + ): Promise { + const previous = this.ledgerMutationTails.get(ledgerSessionId) ?? Promise.resolve(); + const pending = previous.then(mutation, mutation); + const tail = pending.catch(() => {}); + this.ledgerMutationTails.set(ledgerSessionId, tail); + void tail.then(() => { + if (this.ledgerMutationTails.get(ledgerSessionId) === tail) { + this.ledgerMutationTails.delete(ledgerSessionId); + } + }); + return pending; } private async markLedgerIncomplete(session: AcpTranslatorSessionRef): Promise { diff --git a/src/acp/translator.stop-reason.test.ts b/src/acp/translator.stop-reason.test.ts index 1f845158ffee..e0c6d86171c7 100644 --- a/src/acp/translator.stop-reason.test.ts +++ b/src/acp/translator.stop-reason.test.ts @@ -3,6 +3,7 @@ import type { PromptRequest } from "@agentclientprotocol/sdk"; import { createInMemorySessionStore } from "@openclaw/acp-core/session"; import { describe, expect, it, vi } from "vitest"; import type { GatewayClient } from "../gateway/client.js"; +import { createInMemoryAcpEventLedger } from "./event-ledger.js"; import { AcpGatewayAgent } from "./translator.js"; import { createChatEvent, @@ -38,6 +39,64 @@ function requireFirstRequestIdempotencyKey(requestMock: { return idempotencyKey; } +async function createDisconnectNoticeHarness(params: { sendAccepted: boolean }) { + const sessionId = "session-1"; + const sessionKey = "agent:main:main"; + const sessionStore = createInMemorySessionStore(); + sessionStore.createSession({ sessionId, sessionKey, cwd: "/tmp" }); + const eventLedger = createInMemoryAcpEventLedger(); + await eventLedger.startSession({ + sessionId, + sessionKey, + cwd: "/tmp", + complete: true, + }); + const connection = createAcpConnection(); + const sessionUpdate = vi.fn( + async (_params: Parameters[0]) => {}, + ); + connection.sessionUpdate = sessionUpdate as typeof connection.sessionUpdate; + const request = vi.fn(async (method: string) => { + if (method === "chat.send") { + if (!params.sendAccepted) { + throw new Error("gateway closed (1006): connection lost"); + } + return {}; + } + if (method === "agent.wait") { + throw new Error("gateway closed (1006): connection lost"); + } + return {}; + }) as GatewayClient["request"]; + const agent = new AcpGatewayAgent(connection, createAcpGateway(request), { + eventLedger, + sessionStore, + }); + const promptPromise = promptAgent(agent, sessionId); + void promptPromise.catch(() => {}); + await vi.waitFor(() => { + expect(request).toHaveBeenCalledWith("chat.send", expect.objectContaining({ sessionKey }), { + timeoutMs: null, + }); + }); + if (params.sendAccepted) { + await vi.waitFor(async () => { + const replay = await eventLedger.readReplay({ sessionId, sessionKey }); + expect( + replay.events.some((event) => event.update.sessionUpdate === "user_message_chunk"), + ).toBe(true); + }); + } + return { + agent, + eventLedger, + promptPromise, + sessionId, + sessionKey, + sessionUpdate, + }; +} + describe("acp translator stop reason mapping", () => { it("error state resolves as end_turn, not refusal", async () => { const { agent, promptPromise, runId } = await createPendingPromptHarness(); @@ -169,32 +228,83 @@ describe("acp translator stop reason mapping", () => { it("rejects in-flight prompts when the gateway does not reconnect before the grace window", async () => { vi.useFakeTimers(); try { - const { agent, promptPromise } = await createPendingPromptHarness(); - void promptPromise.catch(() => {}); + const { agent, eventLedger, promptPromise, sessionId, sessionKey, sessionUpdate } = + await createDisconnectNoticeHarness({ sendAccepted: true }); agent.handleGatewayDisconnect("1006: connection lost"); await vi.advanceTimersByTimeAsync(5_000); await expect(promptPromise).rejects.toThrow("Gateway disconnected: 1006: connection lost"); + const expectedText = + "[OpenClaw interruption] The Gateway disconnected after accepting this message, so its final outcome is unknown. Check the session before retrying."; + expect(sessionUpdate).toHaveBeenCalledWith({ + sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: expectedText }, + }, + }); + const replay = await eventLedger.readReplay({ sessionId, sessionKey }); + const recordedNotices = replay.events.filter( + (event) => + event.update.sessionUpdate === "agent_message_chunk" && + event.update.content.type === "text" && + event.update.content.text === expectedText, + ); + expect(recordedNotices).toHaveLength(1); } finally { vi.useRealTimers(); } }); + it("makes the disconnect notice durable without waiting for ACP delivery", async () => { + vi.useFakeTimers(); + let releaseDelivery: (() => void) | undefined; + try { + const { agent, eventLedger, promptPromise, sessionUpdate } = + await createDisconnectNoticeHarness({ sendAccepted: true }); + const settleSpy = observeSettlement(promptPromise); + const originalRecordUpdate = eventLedger.recordUpdate; + let releaseRecord: (() => void) | undefined; + const recordBlocked = new Promise((resolve) => { + releaseRecord = resolve; + }); + let noticeRecordStarted = false; + eventLedger.recordUpdate = async (params) => { + if (params.update.sessionUpdate === "agent_message_chunk") { + noticeRecordStarted = true; + await recordBlocked; + } + await originalRecordUpdate(params); + }; + const deliveryBlocked = new Promise((resolve) => { + releaseDelivery = resolve; + }); + sessionUpdate.mockImplementation(async () => await deliveryBlocked); + + agent.handleGatewayDisconnect("1006: connection lost"); + await vi.advanceTimersByTimeAsync(5_000); + await vi.waitFor(() => { + expect(noticeRecordStarted).toBe(true); + expect(sessionUpdate).toHaveBeenCalledTimes(1); + }); + expect(settleSpy).not.toHaveBeenCalled(); + + releaseRecord?.(); + await expect(promptPromise).rejects.toThrow("Gateway disconnected: 1006: connection lost"); + } finally { + releaseDelivery?.(); + vi.useRealTimers(); + } + }); + it("keeps pre-ack send disconnects inside the reconnect grace window", async () => { vi.useFakeTimers(); try { - const request = vi.fn(async (method: string) => { - if (method === "chat.send") { - throw new Error("gateway closed (1006): connection lost"); - } - return {}; - }) as GatewayClient["request"]; - const { agent, sessionId } = createSessionAgentHarness(request); - const promptPromise = promptAgent(agent, sessionId); + const { agent, eventLedger, promptPromise, sessionId, sessionKey, sessionUpdate } = + await createDisconnectNoticeHarness({ sendAccepted: false }); const settleSpy = observeSettlement(promptPromise); - await Promise.resolve(); expect(settleSpy).not.toHaveBeenCalled(); agent.handleGatewayDisconnect("1006: connection lost"); @@ -203,6 +313,24 @@ describe("acp translator stop reason mapping", () => { await vi.advanceTimersByTimeAsync(1); await expect(promptPromise).rejects.toThrow("Gateway disconnected: 1006: connection lost"); + const expectedText = + "[OpenClaw interruption] The Gateway disconnected before OpenClaw could confirm whether this message was accepted, so its final outcome is unknown. Check the session before retrying."; + expect(sessionUpdate).toHaveBeenCalledWith({ + sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: expectedText }, + }, + }); + const replay = await eventLedger.readReplay({ sessionId, sessionKey }); + expect( + replay.events.filter( + (event) => + event.update.sessionUpdate === "agent_message_chunk" && + event.update.content.type === "text" && + event.update.content.text === expectedText, + ), + ).toHaveLength(1); } finally { vi.useRealTimers(); } @@ -620,6 +748,71 @@ describe("acp translator stop reason mapping", () => { await expect(promptPromise).resolves.toEqual({ stopReason: "end_turn" }); }); + it("keeps a replacement prompt while a stale send failure races disconnect rejection", async () => { + vi.useFakeTimers(); + try { + let rejectFirstSend: ((error: Error) => void) | undefined; + const firstSend = new Promise((_, reject) => { + rejectFirstSend = reject; + }); + const runIds: string[] = []; + const request = vi.fn( + (method: string, params?: Record): Promise => { + if (method === "chat.send") { + const runId = params?.idempotencyKey; + if (typeof runId === "string") { + runIds.push(runId); + } + return runIds.length === 1 ? firstSend : Promise.resolve({}); + } + if (method === "agent.wait") { + return Promise.reject(new Error("gateway closed (1006): connection lost")); + } + return Promise.resolve({}); + }, + ) as GatewayClient["request"]; + const sessionStore = createInMemorySessionStore(); + const sessionId = "session-1"; + const sessionKey = "agent:main:main"; + sessionStore.createSession({ sessionId, sessionKey, cwd: "/tmp" }); + const agent = new AcpGatewayAgent(createAcpConnection(), createAcpGateway(request), { + sessionStore, + }); + + const firstPrompt = promptAgent(agent, sessionId, "first"); + void firstPrompt.catch(() => {}); + await vi.waitFor(() => { + expect(runIds).toHaveLength(1); + }); + agent.handleGatewayDisconnect("1006: connection lost"); + await vi.advanceTimersByTimeAsync(5_000); + await expect(firstPrompt).rejects.toThrow("Gateway disconnected: 1006: connection lost"); + + const secondPrompt = promptAgent(agent, sessionId, "second"); + await vi.waitFor(() => { + expect(runIds).toHaveLength(2); + }); + + rejectFirstSend?.(new Error("gateway closed (1006): connection lost")); + await Promise.resolve(); + await expect(Promise.race([secondPrompt, Promise.resolve("pending")])).resolves.toBe( + "pending", + ); + + await agent.handleGatewayEvent( + createChatEvent({ + runId: runIds[1], + sessionKey, + seq: 1, + state: "final", + }), + ); + await expect(secondPrompt).resolves.toEqual({ stopReason: "end_turn" }); + } finally { + vi.useRealTimers(); + } + }); + it("does not let a stale disconnect deadline reject a newer prompt on the same session", async () => { vi.useFakeTimers(); try { diff --git a/src/acp/translator.ts b/src/acp/translator.ts index df624299e71c..07f89996952d 100644 --- a/src/acp/translator.ts +++ b/src/acp/translator.ts @@ -721,7 +721,7 @@ export class AcpGatewayAgent implements Agent { if (status === "error") { const current = pending(); if (current) { - this.rejectPendingPrompt( + await this.rejectPendingPrompt( current, new Error("Chat failed before the run started; try again."), ); @@ -775,21 +775,21 @@ export class AcpGatewayAgent implements Agent { } }; - void sendWithProvenanceFallback().catch((err: unknown) => { + void sendWithProvenanceFallback().catch(async (err: unknown) => { const promptKey = this.pendingPromptKey(params.sessionId, runId); - if ( - isGatewayCloseError(err) && - (this.getPendingPrompt(params.sessionId, runId) || this.settlingPromptKeys.has(promptKey)) - ) { + if (this.settlingPromptKeys.has(promptKey)) { return; } - this.clearApprovalRelaysForPrompt(params.sessionId, runId, { denyActive: true }); - this.pendingPrompts.delete(params.sessionId); - this.sessionStore.clearActiveRun(params.sessionId); - if (this.pendingPrompts.size === 0) { - this.clearDisconnectTimer(); + if (isGatewayCloseError(err) && this.getPendingPrompt(params.sessionId, runId)) { + return; } - reject(err instanceof Error ? err : new Error(String(err))); + const error = err instanceof Error ? err : new Error(String(err)); + const current = this.getPendingPrompt(params.sessionId, runId); + if (current) { + await this.rejectPendingPrompt(current, error); + return; + } + reject(error); }); }); } @@ -1299,11 +1299,19 @@ export class AcpGatewayAgent implements Agent { this.disconnectTimer.unref?.(); } - private rejectPendingPrompt(pending: PendingPrompt, error: Error): void { + private async rejectPendingPrompt( + pending: PendingPrompt, + error: Error, + options: { recordDisconnectNotice?: boolean } = {}, + ): Promise { const currentPending = this.getPendingPrompt(pending.sessionId, pending.idempotencyKey); if (currentPending !== pending) { return; } + + const promptKey = this.pendingPromptKey(pending.sessionId, pending.idempotencyKey); + // Claim before emitting so late Gateway events cannot settle this prompt twice. + this.settlingPromptKeys.add(promptKey); this.clearApprovalRelaysForPrompt(pending.sessionId, pending.idempotencyKey, { denyActive: true, }); @@ -1312,7 +1320,33 @@ export class AcpGatewayAgent implements Agent { if (this.pendingPrompts.size === 0) { this.clearDisconnectTimer(); } - pending.reject(error); + + try { + if (options.recordDisconnectNotice) { + const text = pending.sendAccepted + ? "[OpenClaw interruption] The Gateway disconnected after accepting this message, so its final outcome is unknown. Check the session before retrying." + : "[OpenClaw interruption] The Gateway disconnected before OpenClaw could confirm whether this message was accepted, so its final outcome is unknown. Check the session before retrying."; + // Make replay durable before rejecting, but do not let ACP client backpressure + // extend the disconnect deadline indefinitely. + await this.sessionUpdates.emit({ + sessionId: pending.sessionId, + sessionKey: pending.sessionKey, + ...(pending.ledgerSessionId ? { ledgerSessionId: pending.ledgerSessionId } : {}), + runId: pending.idempotencyKey, + record: true, + waitForDelivery: false, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text }, + }, + }); + } + } catch (noticeError) { + this.log(`disconnect notice failed for ${pending.idempotencyKey}: ${String(noticeError)}`); + } finally { + pending.reject(error); + this.settlingPromptKeys.delete(promptKey); + } } private clearPendingDisconnectState( @@ -1394,9 +1428,10 @@ export class AcpGatewayAgent implements Agent { this.log(`agent.wait reconcile failed for ${pending.idempotencyKey}: ${String(err)}`); if (deadlineExpired) { if (this.shouldRejectPendingAtDisconnectDeadline(pending, disconnectContext)) { - this.rejectPendingPrompt( + await this.rejectPendingPrompt( pending, new Error(`Gateway disconnected: ${disconnectContext.reason}`), + { recordDisconnectNotice: true }, ); return false; } @@ -1424,9 +1459,10 @@ export class AcpGatewayAgent implements Agent { if (!currentDisconnectContext) { return false; } - this.rejectPendingPrompt( + await this.rejectPendingPrompt( currentPending, new Error(`Gateway disconnected: ${currentDisconnectContext.reason}`), + { recordDisconnectNotice: true }, ); return false; }