fix(signal): drain accepted messages on shutdown (#103967)

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;

extensions/signal/src/monitor/event-handler.ts

@@ -236,6 +236,7 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
     replyToSender?: string;
     replyToIsQuote?: boolean;
   };
+  const activeEnqueueEntries = new WeakSet<SignalInboundEntry>();
 
   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);
+    }
   };
 }