fix(telegram): suppress replies superseded during adoption (#103965)

* fix(telegram): preserve supersession through async adoption

* fix(auto-reply): stop superseded queued turns after adoption

* test(telegram): use dispatcher delivery return type

(cherry picked from commit 3f402f2c48)

extensions/telegram/src/bot-message-dispatch.test.ts

@@ -5903,6 +5903,61 @@ describe("dispatchTelegramMessage draft streaming", () => {
     await firstPromise;
   });
 
+  it("keeps supersession latched when it arrives during adoption", async () => {
+    const sessionKey = "agent:main:telegram:direct:adoption-race";
+    let adoptionStarted: (() => void) | undefined;
+    const adoptionStartGate = new Promise<void>((resolve) => {
+      adoptionStarted = resolve;
+    });
+    let releaseAdoption: (() => void) | undefined;
+    const adoptionGate = new Promise<void>((resolve) => {
+      releaseAdoption = resolve;
+    });
+    dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
+      async ({ dispatcherOptions, replyOptions }) => {
+        await replyOptions?.onTurnAdopted?.();
+        await dispatcherOptions.deliver({ text: "stale final" }, { kind: "final" });
+        return { queuedFinal: true };
+      },
+    );
+    deliverReplies.mockResolvedValue({ delivered: true });
+
+    const dispatchPromise = dispatchWithContext({
+      context: createContext({
+        ctxPayload: {
+          SessionKey: sessionKey,
+          ChatType: "direct",
+          MessageSid: "101",
+          RawBody: "long turn",
+          BodyForAgent: "long turn",
+          CommandBody: "long turn",
+          CommandAuthorized: true,
+        } as unknown as TelegramMessageContext["ctxPayload"],
+        msg: {
+          chat: { id: 123, type: "private" },
+          message_id: 101,
+          text: "long turn",
+        } as unknown as TelegramMessageContext["msg"],
+        chatId: 123,
+        isGroup: false,
+        threadSpec: { id: undefined, scope: "none" },
+      }),
+      streamMode: "off",
+      onTurnAdopted: async () => {
+        adoptionStarted?.();
+        await adoptionGate;
+      },
+    });
+    await adoptionStartGate;
+
+    const { supersedeTelegramReplyFence } = await import("./telegram-reply-fence.js");
+    expect(supersedeTelegramReplyFence(sessionKey)).toBe(true);
+    releaseAdoption?.();
+    await dispatchPromise;
+
+    expect(deliverReplies).not.toHaveBeenCalled();
+  });
+
   it("lets authorized /stop kill an adopted run without the released fence controller", async () => {
     const historyKey = "telegram:group:-100123";
     const groupHistories = new Map([[historyKey, []]]);
@@ -6221,11 +6276,29 @@ describe("dispatchTelegramMessage draft streaming", () => {
     const historyKey = "telegram:group:-100123";
     const groupHistories = new Map([[historyKey, []]]);
     let roomEventAbortSignal: AbortSignal | undefined;
-    let queuedLifecycle: { onEnqueued?: () => void; onComplete?: () => void } | undefined;
+    let queuedLifecycle:
+      | {
+          onEnqueued?: () => void;
+          onAdmitted?: () => Promise<void> | void;
+          onComplete?: () => void;
+        }
+      | undefined;
+    let deliverQueuedRoomEvent:
+      | DispatchReplyWithBufferedBlockDispatcherArgs["dispatcherOptions"]["deliver"]
+      | undefined;
+    let adoptionStarted: (() => void) | undefined;
+    const adoptionStartGate = new Promise<void>((resolve) => {
+      adoptionStarted = resolve;
+    });
+    let releaseAdoption: (() => void) | undefined;
+    const adoptionGate = new Promise<void>((resolve) => {
+      releaseAdoption = resolve;
+    });
     dispatchReplyWithBufferedBlockDispatcher
-      .mockImplementationOnce(async ({ replyOptions }) => {
+      .mockImplementationOnce(async ({ dispatcherOptions, replyOptions }) => {
         roomEventAbortSignal = replyOptions?.abortSignal;
         queuedLifecycle = replyOptions?.queuedFollowupLifecycle;
+        deliverQueuedRoomEvent = dispatcherOptions.deliver;
         queuedLifecycle?.onEnqueued?.();
         return {
           queuedFinal: false,
@@ -6272,15 +6345,26 @@ describe("dispatchTelegramMessage draft streaming", () => {
     await dispatchWithContext({
       context: createGroupContext("room_event", 99, "ambient chatter"),
       streamMode: "off",
+      onTurnAdopted: async () => {
+        adoptionStarted?.();
+        await adoptionGate;
+      },
     });
     expect(roomEventAbortSignal?.aborted).toBe(false);
 
+    const admissionPromise = queuedLifecycle?.onAdmitted?.();
+    await adoptionStartGate;
+
     await dispatchWithContext({
       context: createGroupContext("user_request", 100, "@bot answer now"),
       streamMode: "off",
     });
 
     expect(roomEventAbortSignal?.aborted).toBe(true);
+    releaseAdoption?.();
+    await admissionPromise;
+    await deliverQueuedRoomEvent?.({ text: "stale ambient answer" }, { kind: "final" });
+    expect(deliverReplies).toHaveBeenCalledTimes(1);
     queuedLifecycle?.onComplete?.();
   });
 

extensions/telegram/src/bot-message-dispatch.ts

@@ -887,12 +887,15 @@ export const dispatchTelegramMessage = async ({
   let replyAbortControllerQueued = false;
   let queuedTurnAdopted = false;
   let dispatchWasSuperseded;
+  // Queued source dispatches release their generation before admission but retain this controller.
+  // Its aborted bit preserves supersession across the later async adoption handoff.
   const isDispatchSuperseded = () =>
-    replyFenceGeneration !== undefined &&
-    isTelegramReplyFenceSuperseded({
-      key: activeReplyFenceKey,
-      generation: replyFenceGeneration,
-    });
+    replyAbortController.signal.aborted ||
+    (replyFenceGeneration !== undefined &&
+      isTelegramReplyFenceSuperseded({
+        key: activeReplyFenceKey,
+        generation: replyFenceGeneration,
+      }));
   const releaseReplyFence = () => {
     if (replyFenceGeneration === undefined) {
       return;

src/auto-reply/reply/followup-runner.test.ts

@@ -732,6 +732,49 @@ describe("createFollowupRunner reply-lane admission", () => {
     expect(events).toEqual(["admission-started", "admitted", "run", "complete"]);
   });
 
+  it("stops an aborted queued followup after asynchronous owner admission", async () => {
+    const events: string[] = [];
+    const abortController = new AbortController();
+    let releaseAdmission!: () => void;
+    const admissionBarrier = new Promise<void>((resolve) => {
+      releaseAdmission = resolve;
+    });
+    const onBlockReply = vi.fn(async () => {});
+    const runner = createFollowupRunner({
+      typing: createMockTypingController(),
+      typingMode: "instant",
+      sessionKey: "main",
+      defaultModel: "anthropic/claude",
+      opts: { onBlockReply },
+    });
+
+    const pending = runner(
+      createQueuedRun({
+        abortSignal: abortController.signal,
+        queuedLifecycle: {
+          onAdmitted: async () => {
+            events.push("admission-started");
+            await admissionBarrier;
+            events.push("admitted");
+          },
+          onComplete: () => events.push("complete"),
+        },
+        run: { provider: "anthropic", model: "claude" },
+      }),
+    );
+
+    await vi.waitFor(() => expect(events).toEqual(["admission-started"]));
+    abortController.abort();
+    releaseAdmission();
+    await pending;
+
+    expect(events).toEqual(["admission-started", "admitted", "complete"]);
+    expect(runPreflightCompactionIfNeededMock).not.toHaveBeenCalled();
+    expect(runEmbeddedAgentMock).not.toHaveBeenCalled();
+    expect(runCliAgentMock).not.toHaveBeenCalled();
+    expect(onBlockReply).not.toHaveBeenCalled();
+  });
+
   it("passes prepared media user turns to embedded runtime dispatch", async () => {
     const preparedUserTurnMessage = {
       role: "user",

src/auto-reply/reply/followup-runner.ts

@@ -684,6 +684,11 @@ export function createFollowupRunner(params: {
       // Multi-source collected turns become atomic at reply-lane admission.
       // Their queue owner uses this boundary to retire source cancellation ids.
       await admitFollowupRunLifecycle(effectiveQueued);
+      // Admission can await transport-owned durability. Supersession during that handoff is
+      // sticky; stop before preflight can emit notices or start provider work for the stale turn.
+      if (isFollowupRunAborted(effectiveQueued)) {
+        return;
+      }
       if (replyOperation.sessionId !== run.sessionId) {
         run = { ...run, sessionId: replyOperation.sessionId };
         effectiveQueued = { ...effectiveQueued, run };