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

@@ -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<void>();
+    const newerStarted = createDeferred<void>();
+    const releaseOlderFinal = createDeferred<void>();
+    const olderDelivered = createDeferred<void>();
+
+    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 ?? "<missing>"}`);
+      },
+    );
+
+    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<void>();
+    const releaseOlderBeforeDeliver = createDeferred<ReplyPayload | null>();
+    const newerStarted = createDeferred<void>();
+    const releaseNewerFinal = createDeferred<void>();
+
+    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 ?? "<missing>"}`);
+      },
+    );
+
+    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<void>();

src/auto-reply/dispatch.ts

@@ -49,12 +49,14 @@ type ForegroundReplyFenceState = {
   visibleDeliveryGeneration: number;
   activeDispatches: number;
   activeGenerations: Map<number, number>;
+  suspendedGenerations: Set<number>;
   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<number, number>(),
+    suspendedGenerations: new Set<number>(),
     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,

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<void>((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: [],

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") {

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(),

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 &

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<void>();
+    const firstEntered = createDeferred<void>();
+    const firstCallback = () => {};
+    const secondCallback = () => {};
+    const callbacks: Array<FollowupRun["onFollowupAdmissionWaitChange"]> = [];
+    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]);
+  });
 });

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: {

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;

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();

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<ReplyTurnAdmission> {
   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 <T>(wait: () => Promise<T>): Promise<T> => {
+    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 {