diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index 403054dc3..558e9c442 100644 --- a/src/session/approval-resume.test.ts +++ b/src/session/approval-resume.test.ts @@ -595,3 +595,111 @@ describe("approval identity ordering", () => { expect(ctx.deliver).not.toHaveBeenCalled(); }); }); + +describe("approval resume retry re-await", () => { + function retryHarness(outcome: { allow: boolean; message?: string }) { + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + const resolveSuspended = mock(async () => outcome); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + return { resume, delivered, deliver, resolveSuspended }; + } + + function onlyDecision(delivered: InboundMessage[]): InboundMessage { + const message = delivered[0]; + if (message === undefined) throw new Error("expected a delivered decision"); + return message; + } + + test("retry after a delivered acceptance reuses it exactly once", async () => { + const { resume, delivered, resolveSuspended } = retryHarness({ + allow: true, + }); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + expect(decisionBody(onlyDecision(delivered)).outcome).toBe("approved"); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); + + test("late duplicate acceptance after a rejected decision is a no-op", async () => { + const { resume, delivered, resolveSuspended } = retryHarness({ + allow: false, + message: "not today", + }); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(delivered).toHaveLength(1); + expect(decisionBody(onlyDecision(delivered)).outcome).toBe("rejected"); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); + + test("retry without acceptance still gates", async () => { + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + deliver.mockImplementationOnce(() => { + throw new Error("agent is done"); + }); + const resolveSuspended = mock(async () => ({ allow: true })); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + await expect( + resume.handle(suspension("corr-A", "echo alpha")), + ).rejects.toThrow("agent is done"); + expect(delivered).toHaveLength(0); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(2); + expect(delivered).toHaveLength(1); + }); + + test("concurrent duplicate handles share one gate and one deliver", async () => { + const gate = Promise.withResolvers<{ allow: boolean }>(); + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + const resolveSuspended = mock(() => gate.promise); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + const first = resume.handle(suspension("corr-A", "echo alpha")); + const second = resume.handle(suspension("corr-A", "echo alpha")); + for (let i = 0; i < 10; i++) await Promise.resolve(); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + gate.resolve({ allow: true }); + expect(await first).toBe(true); + expect(await second).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); +}); diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index bb6db8773..ab3c0a300 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -142,100 +142,131 @@ export function createApprovalResume(args: { ) => string | undefined | Promise; gate: PermissionGate; }): ApprovalResume { - return { - handle: async (result) => { - if (result.type !== "suspended") return false; - const generationCurrent = args.captureGeneration?.() ?? (() => true); - const parkedAgent = args.getAgent(); - if (parkedAgent === undefined) - throw new Error("approval resume: no live agent"); - const { correlationId, approvalSnapshot } = result; - let cancelled = false; - let canReject = false; - const stillCurrent = (): boolean => !cancelled && generationCurrent(); - const cancelParked = (): void => { - if (cancelled) return; - cancelled = true; - if (canReject) { - parkedAgent.deliver( - decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), - ); - } - }; - const dropParked = (): void => { - args.onDropped?.(APPROVAL_DROPPED_NOTICE); - cancelParked(); - }; - args.registerParkedCancel?.(cancelParked); - try { - // The resolver captures the paired store synchronously before its first await. - const parkedCallId = await args.resolveParkedCallId(correlationId); - if (!stillCurrent()) { - dropParked(); - return true; - } - if (parkedCallId === undefined) return true; - const initialHistory = await parkedAgent.history(); - if (!stillCurrent()) { - dropParked(); - return true; - } - if (timeoutResult(initialHistory, parkedCallId)) return true; - canReject = true; + // Retry re-await wiring: correlation ids whose decision was handed to the + // reactor reuse that acceptance. A retry after an observed acceptance + // returns without opening the gate or delivering again, so the parked call + // resumes exactly once and a late duplicate acceptance is a no-op. Ids are + // recorded only when the decision is actually handed over — a failed send + // (deliver threw, so nothing reached the reactor) retries as before. + const handedOver = new Set(); + // Settlements currently gating-and-delivering, keyed by correlation id. A + // concurrent duplicate handle shares the one in-flight outcome instead of + // opening a second gate, so no waiter is lost and none double-resumes. + const inflight = new Map>(); - const deliverDecision = async ( - message: InboundMessage, - ): Promise => { - if (!stillCurrent()) return; - if (args.deliver !== undefined) { - await args.deliver(message, stillCurrent); - } else { - parkedAgent.deliver(message); - } - }; - const request = - approvalSnapshot === undefined - ? null - : requestFromApprovalSnapshot(approvalSnapshot, correlationId); - if (request === null) { - args.registerParkedCancel?.(undefined); - await deliverDecision( - decisionMessage( - correlationId, - "rejected", - "approval surface unavailable", - ), - ); - return true; - } - const outcome = await args.gate.resolveSuspended(request, stillCurrent); - if (!stillCurrent()) { - dropParked(); - return true; + const settleSuspended = async ( + result: Extract, + ): Promise => { + const generationCurrent = args.captureGeneration?.() ?? (() => true); + const parkedAgent = args.getAgent(); + if (parkedAgent === undefined) + throw new Error("approval resume: no live agent"); + const { correlationId, approvalSnapshot } = result; + let cancelled = false; + let canReject = false; + const stillCurrent = (): boolean => !cancelled && generationCurrent(); + const cancelParked = (): void => { + if (cancelled) return; + cancelled = true; + if (canReject) { + parkedAgent.deliver( + decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), + ); + } + }; + const dropParked = (): void => { + args.onDropped?.(APPROVAL_DROPPED_NOTICE); + cancelParked(); + }; + args.registerParkedCancel?.(cancelParked); + try { + // The resolver captures the paired store synchronously before its first await. + const parkedCallId = await args.resolveParkedCallId(correlationId); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (parkedCallId === undefined) return true; + const initialHistory = await parkedAgent.history(); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (timeoutResult(initialHistory, parkedCallId)) return true; + canReject = true; + + const deliverDecision = async ( + message: InboundMessage, + ): Promise => { + if (!stillCurrent()) return false; + if (args.deliver !== undefined) { + await args.deliver(message, stillCurrent); + } else { + parkedAgent.deliver(message); } + handedOver.add(correlationId); + return true; + }; + const request = + approvalSnapshot === undefined + ? null + : requestFromApprovalSnapshot(approvalSnapshot, correlationId); + if (request === null) { args.registerParkedCancel?.(undefined); - const history = await parkedAgent.history(); - const timedOut = timeoutResult(history, parkedCallId); - if (timedOut) canReject = false; - if (!stillCurrent()) { - dropParked(); - return true; - } - if (timedOut) { - logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${parkedCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; - return true; - } await deliverDecision( decisionMessage( correlationId, - outcome?.allow === true ? "approved" : "rejected", - outcome?.allow === true ? undefined : outcome?.message, + "rejected", + "approval surface unavailable", ), ); return true; - } finally { - args.registerParkedCancel?.(undefined); } + const outcome = await args.gate.resolveSuspended(request, stillCurrent); + if (!stillCurrent()) { + dropParked(); + return true; + } + args.registerParkedCancel?.(undefined); + const history = await parkedAgent.history(); + const timedOut = timeoutResult(history, parkedCallId); + if (timedOut) canReject = false; + if (!stillCurrent()) { + dropParked(); + return true; + } + if (timedOut) { + logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${parkedCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; + return true; + } + await deliverDecision( + decisionMessage( + correlationId, + outcome?.allow === true ? "approved" : "rejected", + outcome?.allow === true ? undefined : outcome?.message, + ), + ); + return true; + } finally { + args.registerParkedCancel?.(undefined); + } + }; + + return { + handle: (result) => { + if (result.type !== "suspended") return Promise.resolve(false); + const { correlationId } = result; + if (handedOver.has(correlationId)) return Promise.resolve(true); + const ongoing = inflight.get(correlationId); + if (ongoing !== undefined) return ongoing; + const task = settleSuspended(result); + inflight.set(correlationId, task); + const forget = (): void => { + if (inflight.get(correlationId) === task) + inflight.delete(correlationId); + }; + task.then(forget, forget); + return task; }, }; }