diff --git a/extensions/signal/src/monitor.tool-result.pairs-uuid-only-senders-uuid-allowlist-entry.test.ts b/extensions/signal/src/monitor.tool-result.pairs-uuid-only-senders-uuid-allowlist-entry.test.ts index 6c8a742ad7ef..935981f231cd 100644 --- a/extensions/signal/src/monitor.tool-result.pairs-uuid-only-senders-uuid-allowlist-entry.test.ts +++ b/extensions/signal/src/monitor.tool-result.pairs-uuid-only-senders-uuid-allowlist-entry.test.ts @@ -169,6 +169,75 @@ describe("monitorSignalProvider tool results", () => { } }); + it("drains an inline inbound message accepted before the monitor stops", async () => { + const abortController = new AbortController(); + setSignalToolResultTestConfig({ + channels: { signal: { autoStart: false, dmPolicy: "open", allowFrom: ["*"] } }, + }); + replyMock.mockResolvedValue({ text: "accepted reply" }); + streamMock.mockImplementation(async ({ onEvent }) => { + onEvent({ + event: "receive", + data: JSON.stringify({ + envelope: { + sourceNumber: "+15550001111", + sourceName: "Ada", + timestamp: 1, + dataMessage: { message: "accepted message" }, + }, + }), + }); + abortController.abort(new Error("monitor stopped")); + }); + + await monitorSignalProvider({ + autoStart: false, + baseUrl: "http://127.0.0.1:8080", + abortSignal: abortController.signal, + }); + + expect(replyMock).toHaveBeenCalledTimes(1); + expect(sendMock).toHaveBeenCalledWith("+15550001111", "accepted reply", expect.anything()); + }); + + it("does not dispatch a buffered inbound message after the monitor stops", async () => { + vi.useFakeTimers(); + const abortController = new AbortController(); + setSignalToolResultTestConfig({ + messages: { inbound: { debounceMs: 10 } }, + channels: { signal: { autoStart: false, dmPolicy: "open", allowFrom: ["*"] } }, + }); + replyMock.mockResolvedValue({ text: "late reply" }); + streamMock.mockImplementation(async ({ onEvent }) => { + onEvent({ + event: "receive", + data: JSON.stringify({ + envelope: { + sourceNumber: "+15550001111", + sourceName: "Ada", + timestamp: 1, + dataMessage: { message: "wait for more" }, + }, + }), + }); + abortController.abort(new Error("monitor stopped")); + }); + + try { + await monitorSignalProvider({ + autoStart: false, + baseUrl: "http://127.0.0.1:8080", + abortSignal: abortController.signal, + }); + await vi.advanceTimersByTimeAsync(10); + + expect(replyMock).not.toHaveBeenCalled(); + expect(sendMock).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + it("sizes attachment RPC response caps from mediaMaxMb", async () => { const abortController = new AbortController(); const maxBytes = 2 * 1024 * 1024; diff --git a/extensions/signal/src/monitor/event-handler.ts b/extensions/signal/src/monitor/event-handler.ts index c98c389a3664..54fb93be75ec 100644 --- a/extensions/signal/src/monitor/event-handler.ts +++ b/extensions/signal/src/monitor/event-handler.ts @@ -236,6 +236,7 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { replyToSender?: string; replyToIsQuote?: boolean; }; + const activeEnqueueEntries = new WeakSet(); async function handleSignalInboundMessage(entry: SignalInboundEntry) { const fromLabel = formatInboundFromLabel({ @@ -756,18 +757,21 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { }); }, onFlush: async (entries) => { - if (deps.abortSignal?.aborted) { + // enqueue() awaits inline and overflow flushes, but not timer-backed work. + // Drain tracked inline work on shutdown; stop delayed work with no owner. + const hasActiveEnqueue = entries.some((entry) => activeEnqueueEntries.has(entry)); + if (!hasActiveEnqueue && deps.abortSignal?.aborted) { return; } try { await flushSignalInboundEntries(entries); } catch (err) { - if (deps.abortSignal?.aborted) { - return; - } if (!isSignalReplySessionInitConflictError(err)) { throw err; } + if (deps.abortSignal?.aborted) { + return; + } // Keep the current keyed debounce task reserved through backoff so a // newer same-conversation flush cannot overtake this failed batch. const retryTask = retrySignalInboundFlush(entries, err); @@ -1259,7 +1263,7 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { typeof nativeReplyTargetTimestamp === "number" ? String(nativeReplyTargetTimestamp) : undefined; - await debouncer.enqueue({ + const entry: SignalInboundEntry = { senderName, senderDisplay, senderRecipient, @@ -1283,6 +1287,12 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { replyToBody: visibleQuoteText || undefined, replyToSender: visibleQuoteSender, replyToIsQuote: visibleQuoteText ? true : undefined, - }); + }; + activeEnqueueEntries.add(entry); + try { + await debouncer.enqueue(entry); + } finally { + activeEnqueueEntries.delete(entry); + } }; }