fix(signal): keep active-run controls responsive (#107422)

* fix(signal): keep active-run controls responsive

Co-authored-by: arduano <leonid.shchurov@gmail.com>

* test(signal): satisfy control-lane lint

* fix(signal): serialize queue mutations

* fix(signal): separate control ingress lane

* fix(auto-reply): cancel queued inbound work

* fix(signal): unblock approval responses

* fix(signal): cancel group pending lanes on abort

* refactor(signal): keep ingress lane setup scoped

* docs(changelog): note Signal control ingress fix

---------

Co-authored-by: arduano <leonid.shchurov@gmail.com>

CHANGELOG.md

@@ -38,6 +38,7 @@ Docs: https://docs.openclaw.ai
 ### Fixes
 
 - **Tlon custom S3 uploads:** pass storage endpoints through the AWS SDK's native parser so custom S3-compatible uploads no longer fail before presigning.
+- **Signal active-run controls:** keep authorized stop, status, approval, and queue-read controls responsive during active turns while preserving ordinary and stateful turns in canonical session admission, and cancel every pending group sender lane on stop. (#107422) Thanks @arduano.
 - **Agent auth storage locks:** surface normal release failures while avoiding redundant release attempts after `proper-lockfile` reports a compromised lock.
 - **Paired-node session catalogs:** authorize bundled Anthropic and Codex catalog requests to invoke their read-only node commands from Control UI read flows, restoring remote Claude/Codex rows and terminal resume availability. Fixes #107406.
 - **Sandbox recreate confirmation:** treat Clack cancellation as a decline so Ctrl-C cannot proceed with container removal.

extensions/signal/src/monitor/event-handler.control-lane.test.ts

@@ -0,0 +1,329 @@
+// Signal tests cover ordered control delivery around active inbound work.
+import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
+import type { MsgContext } from "openclaw/plugin-sdk/reply-runtime";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const {
+  dispatchInboundMessageMock,
+  recordInboundSessionMock,
+  sendReadReceiptMock,
+  sendTypingMock,
+} = vi.hoisted(() => ({
+  dispatchInboundMessageMock: vi.fn(),
+  recordInboundSessionMock: vi.fn(),
+  sendReadReceiptMock: vi.fn(),
+  sendTypingMock: vi.fn(),
+}));
+
+vi.mock("../send.js", () => ({
+  sendMessageSignal: vi.fn(),
+  sendTypingSignal: sendTypingMock,
+  sendReadReceiptSignal: sendReadReceiptMock,
+}));
+
+vi.mock("openclaw/plugin-sdk/reply-runtime", async () => {
+  const actual = await vi.importActual<typeof import("openclaw/plugin-sdk/reply-runtime")>(
+    "openclaw/plugin-sdk/reply-runtime",
+  );
+  return {
+    ...actual,
+    dispatchInboundMessage: dispatchInboundMessageMock,
+    dispatchInboundMessageWithDispatcher: dispatchInboundMessageMock,
+    dispatchInboundMessageWithBufferedDispatcher: dispatchInboundMessageMock,
+  };
+});
+
+vi.mock("openclaw/plugin-sdk/conversation-runtime", async () => {
+  const actual = await vi.importActual<typeof import("openclaw/plugin-sdk/conversation-runtime")>(
+    "openclaw/plugin-sdk/conversation-runtime",
+  );
+  return {
+    ...actual,
+    recordInboundSession: recordInboundSessionMock,
+    readChannelAllowFromStore: vi.fn().mockResolvedValue([]),
+    upsertChannelPairingRequest: vi.fn(),
+  };
+});
+
+const [
+  { createBaseSignalEventHandlerDeps, createSignalReceiveEvent },
+  { createSignalEventHandler },
+  {
+    createSignalPendingInboundRegistry,
+    resolveSignalControlLaneKey,
+    resolveSignalInboundDebounceKey,
+  },
+] = await Promise.all([
+  import("./event-handler.test-harness.js"),
+  import("./event-handler.js"),
+  import("./event-handler.control-lane.js"),
+]);
+
+type DispatchParams = { ctx: MsgContext };
+
+const dispatchResult = {
+  queuedFinal: false,
+  counts: { tool: 0, block: 0, final: 1 },
+};
+
+function createHandler(debounceMs: number) {
+  const dmPolicy = "allowlist";
+  const allowFrom = ["+15550001111"];
+  return createSignalEventHandler(
+    createBaseSignalEventHandlerDeps({
+      cfg: {
+        messages: { inbound: { debounceMs } },
+        channels: { signal: { dmPolicy, allowFrom } },
+      } as OpenClawConfig,
+      dmPolicy,
+      allowFrom,
+      historyLimit: 0,
+    }),
+  );
+}
+
+function signalText(message: string, timestamp: number) {
+  return createSignalReceiveEvent({
+    timestamp,
+    dataMessage: { message, attachments: [] },
+  });
+}
+
+function signalGroupText(message: string, timestamp: number, sourceNumber: string) {
+  return createSignalReceiveEvent({
+    sourceNumber,
+    sourceName: sourceNumber,
+    timestamp,
+    dataMessage: {
+      message,
+      attachments: [],
+      groupInfo: { groupId: "group-1", groupName: "Test Group" },
+    },
+  });
+}
+
+function dispatchedCommandBody(index: number): string | undefined {
+  const call = dispatchInboundMessageMock.mock.calls[index];
+  if (!call) {
+    throw new Error(`missing dispatch call ${index}`);
+  }
+  return (call[0] as DispatchParams).ctx.CommandBody;
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, ms);
+  });
+}
+
+describe("Signal active-run control lane", () => {
+  beforeEach(() => {
+    vi.useRealTimers();
+    dispatchInboundMessageMock.mockReset().mockResolvedValue(dispatchResult);
+    recordInboundSessionMock.mockReset().mockResolvedValue(undefined);
+    sendReadReceiptMock.mockReset().mockResolvedValue(true);
+    sendTypingMock.mockReset().mockResolvedValue(true);
+  });
+
+  it.each([
+    "stop",
+    "/approve abc12345 allow-once",
+    "/status",
+    "/queue",
+    "/QUEUE",
+    "/steer keep going",
+  ])("dispatches active-run-safe control %s while normal work is active", async (controlText) => {
+    let releaseActive!: () => void;
+    const activeGate = new Promise<void>((resolve) => {
+      releaseActive = resolve;
+    });
+    dispatchInboundMessageMock.mockImplementationOnce(async () => {
+      await activeGate;
+      return dispatchResult;
+    });
+    const handler = createHandler(5);
+
+    await handler(signalText("start a long task", 1));
+    await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1));
+
+    const controlHandled = handler(signalText(controlText, 2));
+    await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2));
+    expect(dispatchedCommandBody(1)).toBe(controlText);
+
+    releaseActive();
+    await controlHandled;
+  });
+
+  it("serializes repeated aborts on the control lane", async () => {
+    let releaseFirstAbort!: () => void;
+    const firstAbortGate = new Promise<void>((resolve) => {
+      releaseFirstAbort = resolve;
+    });
+    dispatchInboundMessageMock.mockImplementationOnce(async () => {
+      await firstAbortGate;
+      return dispatchResult;
+    });
+    const handler = createHandler(0);
+
+    const first = handler(signalText("stop", 1));
+    await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1));
+    const second = handler(signalText("halt", 2));
+    await delay(20);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+
+    releaseFirstAbort();
+    await Promise.all([first, second]);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2);
+    expect(dispatchedCommandBody(1)).toBe("halt");
+  });
+
+  it.each(["one more detail", "/reset"])(
+    "leaves zero-debounce turn %s to core session admission",
+    async (followupText) => {
+      let releaseActive!: () => void;
+      const activeGate = new Promise<void>((resolve) => {
+        releaseActive = resolve;
+      });
+      dispatchInboundMessageMock.mockImplementationOnce(async () => {
+        await activeGate;
+        return dispatchResult;
+      });
+      const handler = createHandler(0);
+
+      const active = handler(signalText("start a long task", 1));
+      await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1));
+      const followup = handler(signalText(followupText, 2));
+      await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2));
+      expect(dispatchedCommandBody(1)).toBe(followupText);
+
+      releaseActive();
+      await Promise.all([active, followup]);
+    },
+  );
+
+  it("does not promote or cancel an unauthorized abort", () => {
+    const entry = {
+      senderName: "Alice",
+      senderDisplay: "+15550001111",
+      senderRecipient: "+15550001111",
+      senderPeerId: "+15550001111",
+      isGroup: false,
+      bodyText: "stop",
+      commandBody: "stop",
+      commandAuthorized: false,
+    };
+    const cancelKey = vi.fn(() => true);
+    const pendingInboundRegistry = createSignalPendingInboundRegistry("default");
+
+    expect(resolveSignalInboundDebounceKey("default", entry)).toBe(
+      "signal:default:+15550001111:+15550001111",
+    );
+    expect(resolveSignalControlLaneKey("default", entry)).toBeNull();
+    pendingInboundRegistry.track(entry);
+    pendingInboundRegistry.cancelPendingOnAbort(entry, cancelKey);
+    expect(cancelKey).not.toHaveBeenCalled();
+  });
+
+  it("shares one group control lane without merging normal sender batches", () => {
+    const entry = {
+      senderName: "Alice",
+      senderDisplay: "+15550001111",
+      senderRecipient: "+15550001111",
+      senderPeerId: "+15550001111",
+      groupId: "group-1",
+      isGroup: true,
+      bodyText: "stop",
+      commandBody: "stop",
+      commandAuthorized: true,
+    };
+    const otherSender = { ...entry, senderPeerId: "+15550002222" };
+
+    expect(resolveSignalControlLaneKey("default", entry)).toBe(
+      resolveSignalControlLaneKey("default", otherSender),
+    );
+    expect(resolveSignalInboundDebounceKey("default", entry)).not.toBe(
+      resolveSignalInboundDebounceKey("default", otherSender),
+    );
+  });
+
+  it.each([
+    "/reset",
+    "/queue status",
+    "/queue collect",
+    "/queue interrupt",
+    "/queue reset",
+    "/queue debounce:2s",
+    "/queue cap:5",
+    "/queue drop:summarize",
+  ])("keeps stateful command %s behind active conversation work", async (commandText) => {
+    let releaseActive!: () => void;
+    const activeGate = new Promise<void>((resolve) => {
+      releaseActive = resolve;
+    });
+    dispatchInboundMessageMock.mockImplementationOnce(async () => {
+      await activeGate;
+      return dispatchResult;
+    });
+    const handler = createHandler(5);
+
+    const active = handler(signalText("start a long task", 1));
+    await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1));
+    const statefulCommand = handler(signalText(commandText, 2));
+    await delay(20);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+
+    releaseActive();
+    await Promise.all([active, statefulCommand]);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2);
+    expect(dispatchedCommandBody(1)).toBe(commandText);
+  });
+
+  it("cancels ordinary text still waiting in the debounce window", async () => {
+    const handler = createHandler(50);
+
+    await handler(signalText("queued work", 1));
+    await handler(signalText("stop", 2));
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+    expect(dispatchedCommandBody(0)).toBe("stop");
+
+    await delay(75);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("cancels pending normal work from every sender in a group conversation", async () => {
+    const handler = createHandler(50);
+
+    await handler(signalGroupText("queued work", 1, "+15550001111"));
+    await handler(signalGroupText("stop", 2, "+15550002222"));
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+    expect(dispatchedCommandBody(0)).toBe("stop");
+
+    await delay(75);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("cancels ordinary text released from debounce but still waiting on active work", async () => {
+    let releaseActive!: () => void;
+    const activeGate = new Promise<void>((resolve) => {
+      releaseActive = resolve;
+    });
+    dispatchInboundMessageMock.mockImplementationOnce(async () => {
+      await activeGate;
+      return dispatchResult;
+    });
+    const handler = createHandler(5);
+
+    await handler(signalText("start a long task", 1));
+    await vi.waitFor(() => expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(1));
+    await handler(signalText("queued followup", 2));
+    await delay(20);
+
+    await handler(signalText("stop", 3));
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2);
+    expect(dispatchedCommandBody(1)).toBe("stop");
+
+    releaseActive();
+    await delay(20);
+    expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(2);
+  });
+});

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

@@ -0,0 +1,173 @@
+// Signal plugin helpers isolate active-run control scheduling from the inbound handler.
+import {
+  listChatCommands,
+  maybeResolveTextAlias,
+  normalizeCommandBody,
+} from "openclaw/plugin-sdk/command-auth-native";
+import { isAbortRequestText } from "openclaw/plugin-sdk/command-primitives-runtime";
+
+export type SignalInboundEntry = {
+  senderName: string;
+  senderDisplay: string;
+  senderRecipient: string;
+  senderPeerId: string;
+  groupId?: string;
+  groupName?: string;
+  isGroup: boolean;
+  bodyText: string;
+  nativeReplyBody?: string;
+  commandBody: string;
+  timestamp?: number;
+  messageId?: string;
+  replyToId?: string;
+  isBatched?: boolean;
+  mediaPath?: string;
+  mediaType?: string;
+  mediaPaths?: string[];
+  mediaTypes?: string[];
+  commandAuthorized: boolean;
+  canDetectMention?: boolean;
+  requireMention?: boolean;
+  wasMentioned?: boolean;
+  replyToBody?: string;
+  replyToSender?: string;
+  replyToIsQuote?: boolean;
+};
+
+type TrackedSignalInboundLane = {
+  conversationKey: string;
+  inboundKey: string;
+};
+
+const SIGNAL_ACTIVE_RUN_CONTROL_COMMAND_KEYS = new Set([
+  "approve",
+  "commands",
+  "context",
+  "help",
+  "status",
+  "steer",
+  "tasks",
+  "tools",
+  "whoami",
+]);
+
+function resolveSignalConversationId(entry: SignalInboundEntry): string | null {
+  const conversationId = entry.isGroup ? entry.groupId : entry.senderPeerId;
+  return conversationId?.trim() || null;
+}
+
+export function resolveSignalInboundDebounceKey(
+  accountId: string,
+  entry: SignalInboundEntry,
+): string | null {
+  const conversationId = resolveSignalConversationId(entry);
+  if (!conversationId || !entry.senderPeerId) {
+    return null;
+  }
+  return `signal:${accountId}:${conversationId}:${entry.senderPeerId}`;
+}
+
+function resolveSignalInboundConversationKey(
+  accountId: string,
+  entry: SignalInboundEntry,
+): string | null {
+  const conversationId = resolveSignalConversationId(entry);
+  return conversationId ? `signal:${accountId}:${conversationId}` : null;
+}
+
+function isSignalActiveRunControlText(text: string): boolean {
+  if (isAbortRequestText(text)) {
+    return true;
+  }
+  const normalizedBody = normalizeCommandBody(text.trim());
+  const alias = maybeResolveTextAlias(normalizedBody);
+  if (!alias) {
+    return false;
+  }
+  const command = listChatCommands().find((entry) =>
+    entry.textAliases.some((candidate) => candidate.trim().toLowerCase() === alias),
+  );
+  if (command?.key === "queue") {
+    // Bare `/queue` only reads current settings. Every argument form can mutate them.
+    return normalizedBody.slice(alias.length).trim() === "";
+  }
+  return command ? SIGNAL_ACTIVE_RUN_CONTROL_COMMAND_KEYS.has(command.key) : false;
+}
+
+export function resolveSignalControlLaneKey(
+  accountId: string,
+  entry: SignalInboundEntry,
+): string | null {
+  if (!entry.commandAuthorized || !isSignalActiveRunControlText(entry.commandBody)) {
+    return null;
+  }
+  const conversationId = resolveSignalConversationId(entry);
+  return conversationId ? `signal:${accountId}:${conversationId}:control` : null;
+}
+
+export function createSignalPendingInboundRegistry(accountId: string) {
+  const trackedEntries = new WeakMap<SignalInboundEntry, TrackedSignalInboundLane>();
+  const countsByConversation = new Map<string, Map<string, number>>();
+
+  const track = (entry: SignalInboundEntry) => {
+    if (trackedEntries.has(entry)) {
+      return;
+    }
+    const conversationKey = resolveSignalInboundConversationKey(accountId, entry);
+    const inboundKey = resolveSignalInboundDebounceKey(accountId, entry);
+    if (!conversationKey || !inboundKey) {
+      return;
+    }
+    const counts = countsByConversation.get(conversationKey) ?? new Map<string, number>();
+    counts.set(inboundKey, (counts.get(inboundKey) ?? 0) + 1);
+    countsByConversation.set(conversationKey, counts);
+    trackedEntries.set(entry, { conversationKey, inboundKey });
+  };
+
+  const complete = (entries: SignalInboundEntry[]) => {
+    for (const entry of entries) {
+      const tracked = trackedEntries.get(entry);
+      if (!tracked) {
+        continue;
+      }
+      trackedEntries.delete(entry);
+      const counts = countsByConversation.get(tracked.conversationKey);
+      const nextCount = (counts?.get(tracked.inboundKey) ?? 0) - 1;
+      if (nextCount > 0) {
+        counts?.set(tracked.inboundKey, nextCount);
+        continue;
+      }
+      counts?.delete(tracked.inboundKey);
+      if (counts?.size === 0) {
+        countsByConversation.delete(tracked.conversationKey);
+      }
+    }
+  };
+
+  const cancelPendingOnAbort = (entry: SignalInboundEntry, cancelKey: (key: string) => boolean) => {
+    if (!entry.commandAuthorized || !isAbortRequestText(entry.commandBody)) {
+      return;
+    }
+    const conversationKey = resolveSignalInboundConversationKey(accountId, entry);
+    if (!conversationKey) {
+      return;
+    }
+    // Group members have distinct normal debounce keys, but stop applies to the shared session.
+    // Cancel every still-tracked sender lane before core interrupts the active run.
+    for (const inboundKey of countsByConversation.get(conversationKey)?.keys() ?? []) {
+      cancelKey(inboundKey);
+    }
+  };
+
+  const completeAfter =
+    (flush: (entries: SignalInboundEntry[]) => Promise<void>) =>
+    async (entries: SignalInboundEntry[]) => {
+      try {
+        await flush(entries);
+      } finally {
+        complete(entries);
+      }
+    };
+
+  return { track, complete, completeAfter, cancelPendingOnAbort };
+}

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

@@ -32,7 +32,7 @@ import {
   resolveChannelGroupPolicy,
   resolveChannelGroupRequireMention,
 } from "openclaw/plugin-sdk/channel-policy";
-import { hasControlCommand } from "openclaw/plugin-sdk/command-auth-native";
+import { isControlCommandMessage } from "openclaw/plugin-sdk/command-detection";
 import { recordInboundSession } from "openclaw/plugin-sdk/conversation-runtime";
 import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import {
@@ -84,6 +84,12 @@ import {
 } from "../send-reactions.js";
 import { sendMessageSignal, sendReadReceiptSignal, sendTypingSignal } from "../send.js";
 import { handleSignalDirectMessageAccess, resolveSignalAccessState } from "./access-policy.js";
+import {
+  createSignalPendingInboundRegistry,
+  resolveSignalControlLaneKey,
+  resolveSignalInboundDebounceKey,
+  type SignalInboundEntry,
+} from "./event-handler.control-lane.js";
 import type {
   SignalEnvelope,
   SignalEventHandlerDeps,
@@ -208,33 +214,6 @@ async function finalizeSignalStatusReaction(params: {
 }
 
 export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
-  type SignalInboundEntry = {
-    senderName: string;
-    senderDisplay: string;
-    senderRecipient: string;
-    senderPeerId: string;
-    groupId?: string;
-    groupName?: string;
-    isGroup: boolean;
-    bodyText: string;
-    nativeReplyBody?: string;
-    commandBody: string;
-    timestamp?: number;
-    messageId?: string;
-    replyToId?: string;
-    isBatched?: boolean;
-    mediaPath?: string;
-    mediaType?: string;
-    mediaPaths?: string[];
-    mediaTypes?: string[];
-    commandAuthorized: boolean;
-    canDetectMention?: boolean;
-    requireMention?: boolean;
-    wasMentioned?: boolean;
-    replyToBody?: string;
-    replyToSender?: string;
-    replyToIsQuote?: boolean;
-  };
   const activeEnqueueEntries = new WeakSet<SignalInboundEntry>();
 
   async function handleSignalInboundMessage(entry: SignalInboundEntry) {
@@ -741,16 +720,41 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
     throw lastError;
   }
 
+  const flushDebouncedSignalInboundEntries = async (entries: SignalInboundEntry[]) => {
+    // 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 (!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);
+      deps.runTrackedTask?.(() => retryTask.catch(() => undefined));
+      await retryTask;
+    }
+  };
+  const reportSignalInboundFlushError = (err: unknown) => {
+    deps.runtime.error?.(`signal debounce flush failed: ${String(err)}`);
+  };
+  const pendingInboundRegistry = createSignalPendingInboundRegistry(deps.accountId);
+  const flushNormalSignalInboundEntries = pendingInboundRegistry.completeAfter(
+    flushDebouncedSignalInboundEntries,
+  );
+
   const { debouncer } = createChannelInboundDebouncer<SignalInboundEntry>({
     cfg: deps.cfg,
     channel: "signal",
-    buildKey: (entry) => {
-      const conversationId = entry.isGroup ? (entry.groupId ?? "unknown") : entry.senderPeerId;
-      if (!conversationId || !entry.senderPeerId) {
-        return null;
-      }
-      return `signal:${deps.accountId}:${conversationId}:${entry.senderPeerId}`;
-    },
+    buildKey: (entry) => resolveSignalInboundDebounceKey(deps.accountId, entry),
     shouldDebounce: (entry) => {
       return shouldDebounceTextInbound({
         text: entry.commandBody,
@@ -758,32 +762,19 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
         hasMedia: Boolean(entry.mediaPath || entry.mediaType || entry.mediaPaths?.length),
       });
     },
-    onFlush: async (entries) => {
-      // 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 (!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);
-        deps.runTrackedTask?.(() => retryTask.catch(() => undefined));
-        await retryTask;
-      }
-    },
-    onError: (err) => {
-      deps.runtime.error?.(`signal debounce flush failed: ${String(err)}`);
-    },
+    onFlush: flushNormalSignalInboundEntries,
+    onError: reportSignalInboundFlushError,
+    onCancel: pendingInboundRegistry.complete,
+  });
+  const { debouncer: controlDebouncer } = createChannelInboundDebouncer<SignalInboundEntry>({
+    cfg: deps.cfg,
+    channel: "signal",
+    // Controls bypass normal batching but retain FIFO ordering with each other.
+    serializeImmediate: true,
+    buildKey: (entry) => resolveSignalControlLaneKey(deps.accountId, entry),
+    shouldDebounce: () => false,
+    onFlush: flushDebouncedSignalInboundEntries,
+    onError: reportSignalInboundFlushError,
   });
 
   async function handleReactionOnlyInbound(params: {
@@ -940,7 +931,7 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
     const messageText = normalizedMessage.trim();
     const groupId = dataMessage?.groupInfo?.groupId ?? reaction?.groupInfo?.groupId ?? undefined;
     const isGroup = Boolean(groupId);
-    const hasControlCommandInMessage = hasControlCommand(messageText, deps.cfg);
+    const hasControlCommandInMessage = isControlCommandMessage(messageText, deps.cfg);
 
     const senderDisplay = formatSignalSenderDisplay(sender);
     const { senderAccess, commandAccess } = await resolveSignalAccessState({
@@ -1310,9 +1301,18 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
       replyToSender: visibleQuoteSender,
       replyToIsQuote: visibleQuoteText ? true : undefined,
     };
+    pendingInboundRegistry.cancelPendingOnAbort(entry, debouncer.cancelKey);
+    // Normal and stateful turns stay on the existing ingress path so core session admission owns
+    // queueing and lifecycle mutations; only the narrow safe set uses channel-level serialization.
+    const inboundLane = resolveSignalControlLaneKey(deps.accountId, entry)
+      ? controlDebouncer
+      : debouncer;
+    if (inboundLane === debouncer) {
+      pendingInboundRegistry.track(entry);
+    }
     activeEnqueueEntries.add(entry);
     try {
-      await debouncer.enqueue(entry);
+      await inboundLane.enqueue(entry);
     } finally {
       activeEnqueueEntries.delete(entry);
     }

src/auto-reply/inbound-debounce.ts

@@ -63,6 +63,7 @@ export type InboundDebounceCreateParams<T> = {
 export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>) {
   const buffers = new Map<string, DebounceBuffer<T>>();
   const keyChains = new Map<string, Promise<void>>();
+  const keyGenerations = new Map<string, number>();
   const defaultDebounceMs = resolveNonNegativeIntegerOption(params.debounceMs, 0);
   const maxTrackedKeys = Math.max(1, Math.trunc(params.maxTrackedKeys ?? DEFAULT_MAX_TRACKED_KEYS));
 
@@ -84,6 +85,25 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
     }
   };
 
+  const cancelItems = (items: T[]) => {
+    try {
+      params.onCancel?.(items);
+    } catch {
+      // Cancellation observers release caller-owned resources; debounce state
+      // must still drain even if an observer fails.
+    }
+  };
+
+  const resolveKeyGeneration = (key: string) => keyGenerations.get(key) ?? 0;
+
+  const runQueuedFlush = async (key: string, generation: number, items: T[]) => {
+    if (resolveKeyGeneration(key) !== generation) {
+      cancelItems(items);
+      return;
+    }
+    await runFlush(items);
+  };
+
   const enqueueKeyTask = (key: string, task: () => Promise<void>) => {
     const previous = keyChains.get(key) ?? Promise.resolve();
     const next = previous.catch(() => undefined).then(task);
@@ -92,6 +112,9 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
     const cleanup = () => {
       if (keyChains.get(key) === settled) {
         keyChains.delete(key);
+        if (!buffers.has(key)) {
+          keyGenerations.delete(key);
+        }
       }
     };
     settled.then(cleanup, cleanup);
@@ -108,6 +131,9 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
       resolveSettled();
       if (keyChains.get(key) === settled) {
         keyChains.delete(key);
+        if (!buffers.has(key)) {
+          keyGenerations.delete(key);
+        }
       }
     };
     let next: Promise<void>;
@@ -174,9 +200,15 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
 
   const cancelKey = (key: string): boolean => {
     const buffer = buffers.get(key);
-    if (!buffer) {
+    if (!buffer && !keyChains.has(key)) {
       return false;
     }
+    // Invalidate released tasks still waiting behind an active same-key flush.
+    // The active task has already crossed this check and remains caller-owned.
+    keyGenerations.set(key, resolveKeyGeneration(key) + 1);
+    if (!buffer) {
+      return true;
+    }
     if (buffers.get(key) === buffer) {
       buffers.delete(key);
     }
@@ -186,12 +218,7 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
     }
     const canceledItems = buffer.items;
     buffer.items = [];
-    try {
-      params.onCancel?.(canceledItems);
-    } catch {
-      // Cancellation observers release caller-owned resources; debounce state
-      // must still drain even if an observer fails.
-    }
+    cancelItems(canceledItems);
     releaseBuffer(buffer);
     return true;
   };
@@ -223,8 +250,9 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
         if (buffers.has(key)) {
           // Reserve the keyed immediate slot before forcing the pending buffer
           // to flush so fire-and-forget callers cannot be overtaken.
+          const generation = resolveKeyGeneration(key);
           const reservedTask = enqueueReservedKeyTask(key, async () => {
-            await runFlush([item]);
+            await runQueuedFlush(key, generation, [item]);
           });
           try {
             await flushKey(key);
@@ -235,8 +263,9 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
           return;
         }
         if (keyChains.has(key)) {
+          const generation = resolveKeyGeneration(key);
           await enqueueKeyTask(key, async () => {
-            await runFlush([item]);
+            await runQueuedFlush(key, generation, [item]);
           });
           return;
         }
@@ -263,16 +292,22 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
     if (!canTrackKey(key)) {
       // When the debounce map is saturated, fall back to immediate keyed work
       // instead of buffering, but still preserve same-key ordering.
+      const generation = resolveKeyGeneration(key);
       await enqueueKeyTask(key, async () => {
-        await runFlush([item]);
+        await runQueuedFlush(key, generation, [item]);
       });
       return;
     }
+    const generation = resolveKeyGeneration(key);
     const reservedTask = enqueueReservedKeyTask(key, async () => {
       if (buffer.items.length === 0) {
         return;
       }
-      await runFlush(buffer.items);
+      const items = buffer.items;
+      if (resolveKeyGeneration(key) !== generation) {
+        buffer.items = [];
+      }
+      await runQueuedFlush(key, generation, items);
     });
     const buffer: DebounceBuffer<T> = {
       items: [item],

src/auto-reply/inbound.test.ts

@@ -534,6 +534,45 @@ describe("createInboundDebouncer", () => {
     vi.useRealTimers();
   });
 
+  it("cancels a released flush still waiting behind active same-key work", async () => {
+    const calls: Array<string[]> = [];
+    const canceled: Array<string[]> = [];
+    let releaseFirst!: () => void;
+    const firstGate = new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    const debouncer = createInboundDebouncer<{ key: string; id: string }>({
+      debounceMs: 50,
+      buildKey: (item) => item.key,
+      onFlush: async (items) => {
+        const ids = items.map((entry) => entry.id);
+        calls.push(ids);
+        if (ids[0] === "1") {
+          await firstGate;
+        }
+      },
+      onCancel: (items) => {
+        canceled.push(items.map((entry) => entry.id));
+      },
+    });
+
+    await debouncer.enqueue({ key: "a", id: "1" });
+    const firstFlush = debouncer.flushKey("a");
+    await vi.waitFor(() => expect(calls).toEqual([["1"]]));
+
+    await debouncer.enqueue({ key: "a", id: "2" });
+    const secondFlush = debouncer.flushKey("a");
+    expect(debouncer.cancelKey("a")).toBe(true);
+
+    await debouncer.enqueue({ key: "a", id: "3" });
+    const thirdFlush = debouncer.flushKey("a");
+    releaseFirst();
+    await Promise.all([firstFlush, secondFlush, thirdFlush]);
+
+    expect(canceled).toEqual([["2"]]);
+    expect(calls).toEqual([["1"], ["3"]]);
+  });
+
   it("flushes buffered items before non-debounced item", async () => {
     vi.useFakeTimers();
     const calls: Array<string[]> = [];