From a555ad759815ad74a136e3287d17cddbbe871cf0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 5 Jul 2026 01:56:41 -0700 Subject: [PATCH] fix(reply): prevent foreground delivery deadlock (#100205) * fix(reply): prevent foreground delivery deadlock * fix(reply): prevent foreground delivery deadlock --- src/auto-reply/dispatch.freshness.test.ts | 113 ++++++++++++++++++ src/auto-reply/dispatch.ts | 31 +++++ src/auto-reply/reply/followup-runner.test.ts | 44 +++++++ src/auto-reply/reply/followup-runner.ts | 1 + src/auto-reply/reply/get-reply-run.ts | 1 + src/auto-reply/reply/get-reply.types.ts | 2 + .../reply/queue/drain.identity-guard.test.ts | 44 +++++++ src/auto-reply/reply/queue/drain.ts | 17 +++ src/auto-reply/reply/queue/types.ts | 2 + .../reply/reply-turn-admission.test.ts | 10 ++ src/auto-reply/reply/reply-turn-admission.ts | 19 ++- 11 files changed, 280 insertions(+), 4 deletions(-) diff --git a/src/auto-reply/dispatch.freshness.test.ts b/src/auto-reply/dispatch.freshness.test.ts index 935bd997909..7d64cb01bd3 100644 --- a/src/auto-reply/dispatch.freshness.test.ts +++ b/src/auto-reply/dispatch.freshness.test.ts @@ -190,6 +190,119 @@ describe("foreground reply freshness", () => { expect(deliveries).toEqual([{ kind: "final", text: "old rewritten final" }]); }); + it("does not fence an older final behind a newer inbound waiting for its delivery", async () => { + const deliveries: Delivery[] = []; + const olderStarted = createDeferred(); + const newerStarted = createDeferred(); + const releaseOlderFinal = createDeferred(); + const olderDelivered = createDeferred(); + + hoisted.dispatchReplyFromConfigMock.mockImplementation( + async (params: DispatchReplyFromConfigParams) => { + if (params.ctx.MessageSid === "old-message") { + olderStarted.resolve(); + await releaseOlderFinal.promise; + params.dispatcher.sendFinalReply({ text: "old final" }); + return queuedFinalResult(); + } + if (params.ctx.MessageSid === "new-message") { + newerStarted.resolve(); + // Same-session follow-up admission waits for the owning final delivery. + params.replyOptions?.onFollowupAdmissionWaitChange?.(true); + try { + await olderDelivered.promise; + } finally { + params.replyOptions?.onFollowupAdmissionWaitChange?.(false); + } + return { + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }; + } + throw new Error(`unexpected test message ${params.ctx.MessageSid ?? ""}`); + }, + ); + + const olderDispatch = dispatchWithDeliveries( + buildForegroundCtx({ MessageSid: "old-message" }), + deliveries, + { + deliver: async (payload, info) => { + deliveries.push({ kind: info.kind, text: payload.text }); + olderDelivered.resolve(); + }, + }, + ); + await olderStarted.promise; + + const newerDispatch = dispatchWithDeliveries( + buildForegroundCtx({ MessageSid: "new-message" }), + deliveries, + ); + await newerStarted.promise; + releaseOlderFinal.resolve(); + + await expect(olderDispatch).resolves.toEqual(queuedFinalResult()); + await expect(newerDispatch).resolves.toEqual({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(deliveries).toEqual([{ kind: "final", text: "old final" }]); + }); + + it("keeps an older final fenced while a newer independent turn resolves", async () => { + const deliveries: Delivery[] = []; + const olderBeforeDeliverStarted = createDeferred(); + const releaseOlderBeforeDeliver = createDeferred(); + const newerStarted = createDeferred(); + const releaseNewerFinal = createDeferred(); + + hoisted.dispatchReplyFromConfigMock.mockImplementation( + async (params: DispatchReplyFromConfigParams) => { + if (params.ctx.MessageSid === "old-message") { + params.dispatcher.sendFinalReply({ text: "old final" }); + return queuedFinalResult(); + } + if (params.ctx.MessageSid === "new-message") { + newerStarted.resolve(); + await releaseNewerFinal.promise; + params.dispatcher.sendFinalReply({ text: "new final" }); + return queuedFinalResult(); + } + throw new Error(`unexpected test message ${params.ctx.MessageSid ?? ""}`); + }, + ); + + const olderDispatch = dispatchWithDeliveries( + buildForegroundCtx({ MessageSid: "old-message" }), + deliveries, + { + beforeDeliver: () => { + olderBeforeDeliverStarted.resolve(); + return releaseOlderBeforeDeliver.promise; + }, + }, + ); + await olderBeforeDeliverStarted.promise; + + const newerDispatch = dispatchWithDeliveries( + buildForegroundCtx({ MessageSid: "new-message" }), + deliveries, + ); + await newerStarted.promise; + releaseOlderBeforeDeliver.resolve({ text: "old final" }); + await Promise.resolve(); + expect(deliveries).toEqual([]); + + releaseNewerFinal.resolve(); + await expect(newerDispatch).resolves.toEqual(queuedFinalResult()); + await expect(olderDispatch).resolves.toEqual({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(deliveries).toEqual([{ kind: "final", text: "new final" }]); + }); + it("keeps an older foreground final fenced while a newer visible delivery is unresolved", async () => { const deliveries: Delivery[] = []; const beforeDeliverStarted = createDeferred(); diff --git a/src/auto-reply/dispatch.ts b/src/auto-reply/dispatch.ts index 907ca9a969b..f07eb939afd 100644 --- a/src/auto-reply/dispatch.ts +++ b/src/auto-reply/dispatch.ts @@ -49,12 +49,14 @@ type ForegroundReplyFenceState = { visibleDeliveryGeneration: number; activeDispatches: number; activeGenerations: Map; + suspendedGenerations: Set; waiters: Set<() => void>; }; type ForegroundReplyFenceSnapshot = { key: string; generation: number; + state: ForegroundReplyFenceState; }; type ReplyPayloadRunState = { @@ -124,6 +126,7 @@ function beginForegroundReplyFence( visibleDeliveryGeneration: 0, activeDispatches: 0, activeGenerations: new Map(), + suspendedGenerations: new Set(), waiters: new Set<() => void>(), }; // Generation ordering lets newer foreground replies suppress stale visible deliveries. @@ -137,6 +140,7 @@ function beginForegroundReplyFence( return { key, generation: state.generation, + state, }; } @@ -148,6 +152,27 @@ function notifyForegroundReplyFenceWaiters(state: ForegroundReplyFenceState): vo } } +function setForegroundReplyFenceAdmissionWaiting( + snapshot: ForegroundReplyFenceSnapshot | undefined, + waiting: boolean, +): void { + if (!snapshot) { + return; + } + const state = foregroundReplyFenceByKey.get(snapshot.key); + if (state !== snapshot.state) { + return; + } + if (waiting) { + if (state.activeGenerations.delete(snapshot.generation)) { + state.suspendedGenerations.add(snapshot.generation); + } + } else if (state.suspendedGenerations.delete(snapshot.generation)) { + state.activeGenerations.set(snapshot.generation, 1); + } + notifyForegroundReplyFenceWaiters(state); +} + function hasNewerActiveForegroundReplyFenceGeneration( state: ForegroundReplyFenceState, generation: number, @@ -279,6 +304,7 @@ function endForegroundReplyFence(snapshot: ForegroundReplyFenceSnapshot): void { } else { state.activeGenerations.set(snapshot.generation, activeGenerationCount - 1); } + state.suspendedGenerations.delete(snapshot.generation); state.activeDispatches -= 1; notifyForegroundReplyFenceWaiters(state); if (state.activeDispatches <= 0) { @@ -635,6 +661,11 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: { replyOptions: { ...params.replyOptions, ...replyOptions, + onFollowupAdmissionWaitChange: (waiting) => { + // An admission wait depends on the older owner finishing delivery. + // Suspending only that generation breaks the cycle without weakening newer-turn fencing. + setForegroundReplyFenceAdmissionWaiting(foregroundReplyFence, waiting); + }, }, replyPayloadRunState, onSessionMetadataChanges: params.onSessionMetadataChanges, diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index 179e15ea217..df7465eccf4 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/src/auto-reply/reply/followup-runner.test.ts @@ -734,6 +734,50 @@ describe("createFollowupRunner reply-lane admission", () => { expect(call.sessionFile).toBe("/tmp/post-compact.jsonl"); }); + it("marks only the delivery-dependent follow-up admission wait", async () => { + const waitChanges: boolean[] = []; + const active = createReplyOperationForTest({ + sessionKey: "main", + sessionId: "active-session", + resetTriggered: false, + }); + let releaseBarrier = () => {}; + const barrier = new Promise((resolve) => { + releaseBarrier = resolve; + }); + runEmbeddedAgentMock.mockResolvedValueOnce({ payloads: [], meta: {} }); + const runner = createFollowupRunner({ + typing: createMockTypingController(), + typingMode: "instant", + sessionKey: "main", + defaultModel: "anthropic/claude", + }); + + const pending = runner( + createQueuedRun({ + onFollowupAdmissionWaitChange: (waiting) => waitChanges.push(waiting), + run: { + sessionId: "queued-session", + sessionKey: "main", + provider: "anthropic", + model: "claude", + }, + }), + ); + await Promise.resolve(); + expect(waitChanges).toEqual([]); + + active.completeWithAfterClearBarrier(barrier); + await vi.waitFor(() => { + expect(waitChanges).toEqual([true]); + }); + + releaseBarrier(); + await pending; + expect(waitChanges).toEqual([true, false]); + expect(runEmbeddedAgentMock).toHaveBeenCalledOnce(); + }); + it("uses an admission session hint while refreshing the queued session file", async () => { runEmbeddedAgentMock.mockResolvedValueOnce({ payloads: [], diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index fda57580894..19789b8cec0 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -619,6 +619,7 @@ export function createFollowupRunner(params: { resetTriggered: false, routeThreadId: queued.originatingThreadId, upstreamAbortSignal: resolveFollowupAbortSignal(queued), + onFollowupAdmissionWaitChange: effectiveQueued.onFollowupAdmissionWaitChange, }); if (admission.status === "skipped") { if (admission.reason === "active-run") { diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts index 390ff20d7db..7b5fdf017fa 100644 --- a/src/auto-reply/reply/get-reply-run.ts +++ b/src/auto-reply/reply/get-reply-run.ts @@ -1363,6 +1363,7 @@ export async function runPreparedReply( ...(queuedFollowupAbortSignal ? { abortSignal: queuedFollowupAbortSignal } : {}), deliveryCorrelations: opts?.queuedDeliveryCorrelations, queuedLifecycle: opts?.queuedFollowupLifecycle, + onFollowupAdmissionWaitChange: opts?.onFollowupAdmissionWaitChange, messageId: sessionCtx.MessageSidFull ?? sessionCtx.MessageSid, summaryLine: baseBodyTrimmedRaw, enqueuedAt: Date.now(), diff --git a/src/auto-reply/reply/get-reply.types.ts b/src/auto-reply/reply/get-reply.types.ts index 849d9a7211f..fcf17e49ecf 100644 --- a/src/auto-reply/reply/get-reply.types.ts +++ b/src/auto-reply/reply/get-reply.types.ts @@ -15,6 +15,8 @@ export type InternalReplySessionOptions = { requestedSessionId?: string; resumeRequestedSession?: boolean; sessionPromptSourceReplyDeliveryMode?: GetReplyOptions["sourceReplyDeliveryMode"]; + /** Marks queued follow-up admission waits on an older owner's delivery barrier. */ + onFollowupAdmissionWaitChange?: (waiting: boolean) => void; }; export type InternalGetReplyOptions = GetReplyOptions & diff --git a/src/auto-reply/reply/queue/drain.identity-guard.test.ts b/src/auto-reply/reply/queue/drain.identity-guard.test.ts index e962b1f4c45..d4a81e0aed0 100644 --- a/src/auto-reply/reply/queue/drain.identity-guard.test.ts +++ b/src/auto-reply/reply/queue/drain.identity-guard.test.ts @@ -117,4 +117,48 @@ describe("drain finally identity guard — late D1 must not orphan Q2", () => { expect(calls).toHaveLength(1); expect(calls[0]?.prompt).toBe("msg1"); }); + + it("keeps each inbound admission callback when it joins an active drain", async () => { + const key = `test-drain-callback-${Date.now()}-${Math.random()}`; + keysToCleanup.push(key); + const settings: QueueSettings = { mode: "followup", debounceMs: 0, cap: 50 }; + const gate = createDeferred(); + const firstEntered = createDeferred(); + const firstCallback = () => {}; + const secondCallback = () => {}; + const callbacks: Array = []; + const firstRunner = async (run: FollowupRun) => { + callbacks.push(run.onFollowupAdmissionWaitChange); + if (callbacks.length === 1) { + firstEntered.resolve(); + await gate.promise; + } + }; + const secondRunner = async () => { + throw new Error("active drain must retain its current runner"); + }; + + enqueueFollowupRun( + key, + { ...createRun({ prompt: "msg1" }), onFollowupAdmissionWaitChange: firstCallback }, + settings, + "message-id", + firstRunner, + ); + scheduleFollowupDrain(key, firstRunner); + await firstEntered.promise; + + enqueueFollowupRun( + key, + { ...createRun({ prompt: "msg2" }), onFollowupAdmissionWaitChange: secondCallback }, + settings, + "message-id", + secondRunner, + ); + scheduleFollowupDrain(key, secondRunner); + gate.resolve(); + + await expect.poll(() => callbacks.length).toBe(2); + expect(callbacks).toEqual([firstCallback, secondCallback]); + }); }); diff --git a/src/auto-reply/reply/queue/drain.ts b/src/auto-reply/reply/queue/drain.ts index 6751c87fd10..bbf3f2e7190 100644 --- a/src/auto-reply/reply/queue/drain.ts +++ b/src/auto-reply/reply/queue/drain.ts @@ -243,6 +243,7 @@ type FollowupRuntimeMetadata = Pick< | "queueAbortSignal" | "deliveryCorrelations" | "queuedLifecycle" + | "onFollowupAdmissionWaitChange" >; function hasCurrentTurnRuntimeMetadata(item: FollowupRun): boolean { @@ -478,6 +479,11 @@ function collectRuntimeMetadata( ): FollowupRuntimeMetadata { const currentTurnSource = items.find(hasCurrentTurnRuntimeMetadata); const deliveryCorrelations = items.flatMap((item) => item.deliveryCorrelations ?? []); + const admissionWaitCallbacks = new Set( + items.flatMap((item) => + item.onFollowupAdmissionWaitChange ? [item.onFollowupAdmissionWaitChange] : [], + ), + ); return { currentInboundEventKind: currentTurnSource?.currentInboundEventKind, currentInboundAudio: currentTurnSource?.currentInboundAudio, @@ -486,6 +492,14 @@ function collectRuntimeMetadata( queueAbortSignal: items.find((item) => item.queueAbortSignal)?.queueAbortSignal, deliveryCorrelations: deliveryCorrelations.length > 0 ? deliveryCorrelations : undefined, queuedLifecycle: items.length === 1 ? items[0]?.queuedLifecycle : undefined, + onFollowupAdmissionWaitChange: + admissionWaitCallbacks.size > 0 + ? (waiting) => { + for (const callback of admissionWaitCallbacks) { + callback(waiting); + } + } + : undefined, }; } @@ -797,6 +811,7 @@ export function createOverflowSummaryRetrySource(source: FollowupRun): FollowupR originatingChatType: source.originatingChatType, abortSignal: source.abortSignal, queuedLifecycle: source.queuedLifecycle, + onFollowupAdmissionWaitChange: source.onFollowupAdmissionWaitChange, ...(source.currentInboundEventKind === "room_event" ? { currentInboundEventKind: "room_event" } : {}), @@ -856,6 +871,8 @@ async function runSyntheticOverflowSummary(params: { run: params.source.run, enqueuedAt: Date.now(), abortSignal: params.abortSignal, + onFollowupAdmissionWaitChange: collectRuntimeMetadata(params.sources) + .onFollowupAdmissionWaitChange, ...(params.onAdmitted ? { queuedLifecycle: { diff --git a/src/auto-reply/reply/queue/types.ts b/src/auto-reply/reply/queue/types.ts index d7cc3ca8675..51186a64d55 100644 --- a/src/auto-reply/reply/queue/types.ts +++ b/src/auto-reply/reply/queue/types.ts @@ -66,6 +66,8 @@ export type FollowupRun = { queueAbortSignal?: AbortSignal; deliveryCorrelations?: QueuedReplyDeliveryCorrelation[]; queuedLifecycle?: QueuedReplyLifecycle; + /** Dispatch-scoped freshness owner for a queued delivery-barrier wait. */ + onFollowupAdmissionWaitChange?: (waiting: boolean) => void; /** Provider message ID, when available (for deduplication). */ messageId?: string; summaryLine?: string; diff --git a/src/auto-reply/reply/reply-turn-admission.test.ts b/src/auto-reply/reply/reply-turn-admission.test.ts index 9cd292ed0bb..62677e0556b 100644 --- a/src/auto-reply/reply/reply-turn-admission.test.ts +++ b/src/auto-reply/reply/reply-turn-admission.test.ts @@ -431,6 +431,7 @@ describe("reply turn admission", () => { }); it("waits for visible turns and reuses the active session id", async () => { + const waitChanges: boolean[] = []; const active = createReplyOperation({ sessionKey: "agent:main:telegram:topic:42", sessionId: "active-session", @@ -443,6 +444,7 @@ describe("reply turn admission", () => { sessionId: "new-session", kind: "visible", resetTriggered: false, + onFollowupAdmissionWaitChange: (waiting) => waitChanges.push(waiting), }); let settled = false; @@ -453,9 +455,11 @@ describe("reply turn admission", () => { setImmediate(resolve); }); expect(settled).toBe(false); + expect(waitChanges).toEqual([]); active.complete(); const result = await admitted; + expect(waitChanges).toEqual([]); expect(result.status).toBe("owned"); if (result.status === "owned") { @@ -533,6 +537,7 @@ describe("reply turn admission", () => { }); it("keeps an already-waiting follow-up behind the delivery barrier", async () => { + const waitChanges: boolean[] = []; const active = createReplyOperation({ sessionKey: "agent:main:discord:channel:42", sessionId: "active-session", @@ -547,6 +552,7 @@ describe("reply turn admission", () => { sessionId: "queued-session", kind: "queued_followup", resetTriggered: false, + onFollowupAdmissionWaitChange: (waiting) => waitChanges.push(waiting), }); let settled = false; void admitted.then(() => { @@ -558,9 +564,13 @@ describe("reply turn admission", () => { await Promise.resolve(); expect(settled).toBe(false); + await vi.waitFor(() => { + expect(waitChanges).toEqual([true]); + }); releaseBarrier(); const result = await admitted; + expect(waitChanges).toEqual([true, false]); expect(result.status).toBe("owned"); if (result.status === "owned") { result.operation.complete(); diff --git a/src/auto-reply/reply/reply-turn-admission.ts b/src/auto-reply/reply/reply-turn-admission.ts index c68307d848e..a8a0f052cea 100644 --- a/src/auto-reply/reply/reply-turn-admission.ts +++ b/src/auto-reply/reply/reply-turn-admission.ts @@ -69,12 +69,21 @@ export async function admitReplyTurn(params: { waitForActive?: boolean; retainLifecycleAdmissionOnActive?: boolean; onLifecycleInterrupt?: () => void; + onFollowupAdmissionWaitChange?: (waiting: boolean) => void; }): Promise { let sessionId = params.sessionId; let expectedSessionId = params.expectedSessionId; const waitTimeoutMs = params.waitTimeoutMs ?? (params.kind === "queued_followup" ? REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS : undefined); + const waitForFollowupAdmission = async (wait: () => Promise): Promise => { + params.onFollowupAdmissionWaitChange?.(true); + try { + return await wait(); + } finally { + params.onFollowupAdmissionWaitChange?.(false); + } + }; while (true) { if (isAbortSignalAborted(params.upstreamAbortSignal)) { return { status: "skipped", reason: "aborted" }; @@ -214,10 +223,12 @@ export async function admitReplyTurn(params: { if (params.kind === "heartbeat") { return { status: "skipped", reason: "active-run" }; } - const followupAdmission = await waitForReplyRunFollowupAdmission( - params.sessionKey, - waitTimeoutMs ?? REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, - { signal: params.upstreamAbortSignal }, + const followupAdmission = await waitForFollowupAdmission(() => + waitForReplyRunFollowupAdmission( + params.sessionKey, + waitTimeoutMs ?? REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, + { signal: params.upstreamAbortSignal }, + ), ); if (!followupAdmission.settled) { return {