diff --git a/extensions/telegram/src/bot-message-dispatch.test.ts b/extensions/telegram/src/bot-message-dispatch.test.ts index e61bef7d2269..b146619ed404 100644 --- a/extensions/telegram/src/bot-message-dispatch.test.ts +++ b/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((resolve) => { + adoptionStarted = resolve; + }); + let releaseAdoption: (() => void) | undefined; + const adoptionGate = new Promise((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; + onComplete?: () => void; + } + | undefined; + let deliverQueuedRoomEvent: + | DispatchReplyWithBufferedBlockDispatcherArgs["dispatcherOptions"]["deliver"] + | undefined; + let adoptionStarted: (() => void) | undefined; + const adoptionStartGate = new Promise((resolve) => { + adoptionStarted = resolve; + }); + let releaseAdoption: (() => void) | undefined; + const adoptionGate = new Promise((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?.(); }); diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index 4aa8f54fb8fb..de43d9fcd806 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/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; diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index 42bb9800c05a..a5ee3f9cebb7 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/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((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", diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index a087cc58087a..20c43d2bea8f 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/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 };