diff --git a/extensions/codex/src/app-server/session-binding.ts b/extensions/codex/src/app-server/session-binding.ts
index 33b0cb29ea47..46fa7cfe4cb5 100644
--- a/extensions/codex/src/app-server/session-binding.ts
+++ b/extensions/codex/src/app-server/session-binding.ts
@@ -12,6 +12,7 @@ import {
   type AuthProfileStore,
 } from "openclaw/plugin-sdk/agent-runtime";
 import { type FileLockOptions, withFileLock } from "openclaw/plugin-sdk/file-lock";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import {
   CODEX_PLUGINS_MARKETPLACE_NAME,
   normalizeCodexServiceTier,
@@ -40,7 +41,7 @@ const CODEX_APP_SERVER_BINDING_LOCK_OPTIONS: FileLockOptions = {
   },
   stale: CODEX_APP_SERVER_BINDING_GUARDED_REQUEST_TIMEOUT_MS * 2,
 };
-const bindingMutationQueues = new Map<string, Promise<void>>();
+const bindingMutationQueue = new KeyedAsyncQueue();
 const bindingMutationContext = new AsyncLocalStorage<Set<string>>();
 
 type ProviderAuthAliasLookupParams = Parameters<typeof resolveProviderIdForAuth>[1];
@@ -118,30 +119,13 @@ export async function withCodexAppServerBindingLock<T>(
   // The SDK file lock is process-reentrant, so pair it with a local queue.
   // Nested writes from the same guarded mutation can proceed, but unrelated
   // same-process tasks cannot slip between compare/clear/start.
-  const previous = bindingMutationQueues.get(bindingPath) ?? Promise.resolve();
-  let releaseCurrent!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseCurrent = resolve;
-  });
-  const queued = previous.then(
-    () => current,
-    () => current,
-  );
-  bindingMutationQueues.set(bindingPath, queued);
-  await previous.catch(() => undefined);
-
   const nestedOwnedBindings = new Set(ownedBindings);
   nestedOwnedBindings.add(bindingPath);
-  try {
-    return await bindingMutationContext.run(nestedOwnedBindings, () =>
+  return await bindingMutationQueue.enqueue(bindingPath, () =>
+    bindingMutationContext.run(nestedOwnedBindings, () =>
       withFileLock(bindingPath, CODEX_APP_SERVER_BINDING_LOCK_OPTIONS, run),
-    );
-  } finally {
-    releaseCurrent();
-    if (bindingMutationQueues.get(bindingPath) === queued) {
-      bindingMutationQueues.delete(bindingPath);
-    }
-  }
+    ),
+  );
 }
 
 /** Reads and normalizes a Codex app-server binding sidecar, returning undefined on stale data. */
diff --git a/extensions/codex/src/conversation-binding.ts b/extensions/codex/src/conversation-binding.ts
index 5c3af84e821c..c48a9c90e6a7 100644
--- a/extensions/codex/src/conversation-binding.ts
+++ b/extensions/codex/src/conversation-binding.ts
@@ -5,6 +5,7 @@ import {
 } from "openclaw/plugin-sdk/agent-harness-runtime";
 import { resolveSessionAgentIds } from "openclaw/plugin-sdk/agent-runtime";
 import { loadExecApprovals } from "openclaw/plugin-sdk/exec-approvals-runtime";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import type {
   PluginConversationBindingResolvedEvent,
   PluginHookInboundClaimContext,
@@ -120,7 +121,7 @@ type CodexConversationConfig = Parameters<
 type ResolvedCodexConversationConfig = NonNullable<CodexConversationConfig>;
 
 type CodexConversationGlobalState = {
-  queues: Map<string, Promise<void>>;
+  queue: KeyedAsyncQueue;
 };
 
 async function resolveConversationAppServerRuntime(params: {
@@ -169,7 +170,7 @@ function getGlobalState(): CodexConversationGlobalState {
   const globalState = globalThis as typeof globalThis & {
     [CODEX_CONVERSATION_GLOBAL_STATE]?: CodexConversationGlobalState;
   };
-  globalState[CODEX_CONVERSATION_GLOBAL_STATE] ??= { queues: new Map() };
+  globalState[CODEX_CONVERSATION_GLOBAL_STATE] ??= { queue: new KeyedAsyncQueue() };
   return globalState[CODEX_CONVERSATION_GLOBAL_STATE];
 }
 
@@ -974,22 +975,7 @@ function isCodexThreadNotFoundError(error: unknown): boolean {
 }
 
 function enqueueBoundTurn<T>(key: string, run: () => Promise<T>): Promise<T> {
-  const state = getGlobalState();
-  const previous = state.queues.get(key) ?? Promise.resolve();
-  const next = previous.then(run, run);
-  const queued = next.then(
-    () => undefined,
-    () => undefined,
-  );
-  state.queues.set(key, queued);
-  void next
-    .finally(() => {
-      if (state.queues.get(key) === queued) {
-        state.queues.delete(key);
-      }
-    })
-    .catch(() => undefined);
-  return next;
+  return getGlobalState().queue.enqueue(key, run);
 }
 
 function resolveThreadRequestModelProvider(params: {
diff --git a/extensions/discord/src/internal/command-deploy.ts b/extensions/discord/src/internal/command-deploy.ts
index 4f81f543b94b..9989d783b82a 100644
--- a/extensions/discord/src/internal/command-deploy.ts
+++ b/extensions/discord/src/internal/command-deploy.ts
@@ -2,6 +2,7 @@
 import { createHash } from "node:crypto";
 import path from "node:path";
 import { ApplicationCommandType, type APIApplicationCommand } from "discord-api-types/v10";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { privateFileStore } from "openclaw/plugin-sdk/security-runtime";
 import {
   createApplicationCommand,
@@ -33,25 +34,10 @@ type SerializedCommand = ReturnType<BaseCommand["serialize"]>;
  * file lock. Discord deployers only run inside the gateway process, so an
  * in-process mutex is sufficient for the documented concurrency surface.
  */
-const cachePersistLocks = new Map<string, Promise<void>>();
+const cachePersistLocks = new KeyedAsyncQueue();
 
 async function withCachePersistLock<T>(storePath: string, fn: () => Promise<T>): Promise<T> {
-  const previous = cachePersistLocks.get(storePath) ?? Promise.resolve();
-  let release: () => void = () => {};
-  const next = new Promise<void>((resolve) => {
-    release = resolve;
-  });
-  const chained = previous.then(() => next);
-  cachePersistLocks.set(storePath, chained);
-  try {
-    await previous;
-    return await fn();
-  } finally {
-    release();
-    if (cachePersistLocks.get(storePath) === chained) {
-      cachePersistLocks.delete(storePath);
-    }
-  }
+  return await cachePersistLocks.enqueue(storePath, fn);
 }
 
 export class DiscordCommandDeployer {
diff --git a/extensions/imessage/src/monitor/catchup.ts b/extensions/imessage/src/monitor/catchup.ts
index 56e459274b5c..0821b4bbc046 100644
--- a/extensions/imessage/src/monitor/catchup.ts
+++ b/extensions/imessage/src/monitor/catchup.ts
@@ -1,5 +1,6 @@
 // Imessage plugin module implements catchup behavior.
 import { createHash } from "node:crypto";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
 import { getIMessageRuntime } from "../runtime.js";
 
@@ -29,7 +30,7 @@ const MAX_FAILURE_RETRY_MAP_JSON_BYTES = 48_000;
 const textEncoder = new TextEncoder();
 export const IMESSAGE_CATCHUP_CURSOR_NAMESPACE = "imessage.catchup-cursors";
 export const IMESSAGE_CATCHUP_CURSOR_MAX_ENTRIES = 256;
-const cursorWriteQueues = new Map<string, Promise<unknown>>();
+const cursorWriteQueue = new KeyedAsyncQueue();
 
 export type IMessageCatchupConfig = {
   enabled?: boolean;
@@ -119,17 +120,7 @@ function updateCatchupCursorStore(
 
 function enqueueCursorWrite<T>(accountId: string, fn: () => Promise<T>): Promise<T> {
   const key = resolveIMessageCatchupCursorKey(accountId);
-  const prev = cursorWriteQueues.get(key) ?? Promise.resolve();
-  const next = prev.then(fn, fn);
-  cursorWriteQueues.set(key, next);
-  next
-    .finally(() => {
-      if (cursorWriteQueues.get(key) === next) {
-        cursorWriteQueues.delete(key);
-      }
-    })
-    .catch(() => {});
-  return next;
+  return cursorWriteQueue.enqueue(key, fn);
 }
 
 function sanitizeFailureRetriesInput(raw: unknown): Record<string, number> {
diff --git a/extensions/matrix/src/matrix/monitor/handler.ts b/extensions/matrix/src/matrix/monitor/handler.ts
index 76d73983a617..49243ed79a8a 100644
--- a/extensions/matrix/src/matrix/monitor/handler.ts
+++ b/extensions/matrix/src/matrix/monitor/handler.ts
@@ -27,6 +27,7 @@ import {
   resolveChannelContextVisibilityMode,
 } from "openclaw/plugin-sdk/context-visibility-runtime";
 import { isDangerousNameMatchingEnabled } from "openclaw/plugin-sdk/dangerous-name-runtime";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
 import {
   isFutureDateTimestampMs,
@@ -529,7 +530,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
     logVerboseMessage,
   });
   const roomHistoryTracker = createRoomHistoryTracker();
-  const roomIngressTails = new Map<string, Promise<void>>();
+  const roomIngressQueue = new KeyedAsyncQueue();
   const sharedDmContextNoticeRooms = new Set<string>();
 
   const readStoreAllowFrom = async (): Promise<string[]> => {
@@ -576,22 +577,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
   };
 
   const runRoomIngress = async <T>(roomId: string, task: () => Promise<T>): Promise<T> => {
-    const previous = roomIngressTails.get(roomId) ?? Promise.resolve();
-    let releaseCurrent!: () => void;
-    const current = new Promise<void>((resolve) => {
-      releaseCurrent = resolve;
-    });
-    const chain = previous.catch(() => {}).then(() => current);
-    roomIngressTails.set(roomId, chain);
-    await previous.catch(() => {});
-    try {
-      return await task();
-    } finally {
-      releaseCurrent();
-      if (roomIngressTails.get(roomId) === chain) {
-        roomIngressTails.delete(roomId);
-      }
-    }
+    return await roomIngressQueue.enqueue(roomId, task);
   };
 
   return async (roomId: string, event: MatrixRawEvent) => {
diff --git a/extensions/memory-core/src/short-term-promotion.ts b/extensions/memory-core/src/short-term-promotion.ts
index 565d21a61f8f..0d76a6c3b153 100644
--- a/extensions/memory-core/src/short-term-promotion.ts
+++ b/extensions/memory-core/src/short-term-promotion.ts
@@ -2,6 +2,7 @@
 import { createHash } from "node:crypto";
 import fs from "node:fs/promises";
 import path from "node:path";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import type { MemorySearchResult } from "openclaw/plugin-sdk/memory-core-host-runtime-files";
 import {
   DEFAULT_MEMORY_DEEP_DREAMING_MAX_PROMOTED_SNIPPET_TOKENS,
@@ -88,7 +89,7 @@ const GENERIC_DAY_HEADING_RE =
   /^(?:(?:mon|monday|tue|tues|tuesday|wed|wednesday|thu|thur|thurs|thursday|fri|friday|sat|saturday|sun|sunday)(?:,\s+)?)?(?:(?:jan|january|feb|february|mar|march|apr|april|may|jun|june|jul|july|aug|august|sep|sept|september|oct|october|nov|november|dec|december)\s+\d{1,2}(?:st|nd|rd|th)?(?:,\s*\d{4})?|\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?|\d{4}[/-]\d{2}[/-]\d{2})$/i;
 const PROMOTION_LIST_MARKER_RE = /^(?:\d+\.\s+|[-*+]\s+)/;
 const MANAGED_DREAMING_HEADINGS = new Set(["light sleep", "rem sleep"]);
-const inProcessShortTermLocks = new Map<string, Promise<void>>();
+const inProcessShortTermLocks = new KeyedAsyncQueue();
 
 type PromotionWeights = {
   frequency: number;
@@ -849,23 +850,7 @@ function isProcessLikelyAlive(pid: number): boolean {
 }
 
 async function withInProcessShortTermLock<T>(lockPath: string, task: () => Promise<T>): Promise<T> {
-  const previous = inProcessShortTermLocks.get(lockPath) ?? Promise.resolve();
-  let releaseCurrent!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseCurrent = resolve;
-  });
-  const queued = previous.catch(() => undefined).then(() => current);
-  inProcessShortTermLocks.set(lockPath, queued);
-
-  await previous.catch(() => undefined);
-  try {
-    return await task();
-  } finally {
-    releaseCurrent();
-    if (inProcessShortTermLocks.get(lockPath) === queued) {
-      inProcessShortTermLocks.delete(lockPath);
-    }
-  }
+  return await inProcessShortTermLocks.enqueue(lockPath, task);
 }
 
 async function withShortTermLock<T>(workspaceDir: string, task: () => Promise<T>): Promise<T> {
diff --git a/extensions/msteams/src/sqlite-state.ts b/extensions/msteams/src/sqlite-state.ts
index 7e9086d5a8c9..889d33e069b6 100644
--- a/extensions/msteams/src/sqlite-state.ts
+++ b/extensions/msteams/src/sqlite-state.ts
@@ -1,5 +1,6 @@
 // Msteams plugin module implements sqlite state behavior.
 import path from "node:path";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { getMSTeamsRuntime } from "./runtime.js";
 import { withFileLock } from "./store-fs.js";
 
@@ -55,28 +56,10 @@ export function resolveMSTeamsSqliteStateDir(
   );
 }
 
-const sqliteMutationLocks = new Map<string, Promise<unknown>>();
+const sqliteMutationLocks = new KeyedAsyncQueue();
 
 async function withProcessMutationLock<T>(lockPath: string, fn: () => Promise<T>): Promise<T> {
-  const previous = sqliteMutationLocks.get(lockPath) ?? Promise.resolve();
-  let release: () => void = () => {};
-  const next = new Promise<void>((resolve) => {
-    release = resolve;
-  });
-  const chained = previous.then(
-    () => next,
-    () => next,
-  );
-  sqliteMutationLocks.set(lockPath, chained);
-  await previous.catch(() => undefined);
-  try {
-    return await fn();
-  } finally {
-    release();
-    if (sqliteMutationLocks.get(lockPath) === chained) {
-      sqliteMutationLocks.delete(lockPath);
-    }
-  }
+  return await sqliteMutationLocks.enqueue(lockPath, fn);
 }
 
 export async function withMSTeamsSqliteMutationLock<T>(
diff --git a/extensions/nostr/src/nostr-profile-http.ts b/extensions/nostr/src/nostr-profile-http.ts
index 9cc030e0cc81..f991f966217c 100644
--- a/extensions/nostr/src/nostr-profile-http.ts
+++ b/extensions/nostr/src/nostr-profile-http.ts
@@ -8,6 +8,7 @@
  */
 
 import type { IncomingMessage, ServerResponse } from "node:http";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import {
   normalizeLowercaseStringOrEmpty,
   normalizeOptionalLowercaseString,
@@ -77,31 +78,10 @@ function checkRateLimit(accountId: string): boolean {
 // Mutex for Concurrent Publish Prevention
 // ============================================================================
 
-const publishLocks = new Map<string, Promise<void>>();
+const publishLocks = new KeyedAsyncQueue();
 
 async function withPublishLock<T>(accountId: string, fn: () => Promise<T>): Promise<T> {
-  // Atomic mutex using promise chaining - prevents TOCTOU race condition
-  const prev = publishLocks.get(accountId) ?? Promise.resolve();
-  let resolve: () => void;
-  const next = new Promise<void>((r) => {
-    resolve = r;
-  });
-  // Atomically replace the lock before awaiting - any concurrent request
-  // will now wait on our `next` promise
-  publishLocks.set(accountId, next);
-
-  // Wait for previous operation to complete
-  await prev.catch(() => {});
-
-  try {
-    return await fn();
-  } finally {
-    resolve!();
-    // Clean up if we're the last in chain
-    if (publishLocks.get(accountId) === next) {
-      publishLocks.delete(accountId);
-    }
-  }
+  return await publishLocks.enqueue(accountId, fn);
 }
 
 // ============================================================================
diff --git a/extensions/slack/src/send.ts b/extensions/slack/src/send.ts
index 857e41d308fe..e0bf43a83c7d 100644
--- a/extensions/slack/src/send.ts
+++ b/extensions/slack/src/send.ts
@@ -9,6 +9,7 @@ import {
 } from "openclaw/plugin-sdk/channel-outbound";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { withTrustedEnvProxyGuardedFetchMode } from "openclaw/plugin-sdk/fetch-runtime";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime";
 import { requireRuntimeConfig } from "openclaw/plugin-sdk/plugin-config-runtime";
 import {
@@ -47,7 +48,7 @@ const SLACK_DNS_RETRY_CODES = new Set(["EAI_AGAIN", "ENOTFOUND", "UND_ERR_DNS_RE
 const SLACK_DNS_RETRY_ATTEMPTS = 2;
 const SLACK_DNS_RETRY_BASE_DELAY_MS = 250;
 const slackDmChannelCache = new Map<string, string>();
-const slackSendQueues = new Map<string, Promise<void>>();
+const slackSendQueue = new KeyedAsyncQueue();
 
 type SlackRecipient =
   | {
@@ -488,22 +489,7 @@ function createSlackSendQueueKey(params: {
 }
 
 async function runQueuedSlackSend<T>(key: string, task: () => Promise<T>): Promise<T> {
-  const previous = slackSendQueues.get(key) ?? Promise.resolve();
-  let releaseCurrent!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseCurrent = resolve;
-  });
-  const queuedCurrent = previous.catch(() => undefined).then(() => current);
-  slackSendQueues.set(key, queuedCurrent);
-  await previous.catch(() => undefined);
-  try {
-    return await task();
-  } finally {
-    releaseCurrent();
-    if (slackSendQueues.get(key) === queuedCurrent) {
-      slackSendQueues.delete(key);
-    }
-  }
+  return await slackSendQueue.enqueue(key, task);
 }
 
 function createSlackDmCacheKey(params: {
@@ -602,10 +588,6 @@ export function clearSlackDmChannelCache(): void {
   slackDmChannelCache.clear();
 }
 
-export function clearSlackSendQueuesForTest(): void {
-  slackSendQueues.clear();
-}
-
 export function clearSlackDefaultSendIdentitiesForTest(): void {
   slackDefaultSendIdentities.clear();
 }
diff --git a/extensions/slack/src/send.upload.test.ts b/extensions/slack/src/send.upload.test.ts
index bfc5731527e4..d0ef73f875e2 100644
--- a/extensions/slack/src/send.upload.test.ts
+++ b/extensions/slack/src/send.upload.test.ts
@@ -54,8 +54,7 @@ vi.mock("./runtime-api.js", async () => {
   };
 });
 
-const { sendMessageSlack, clearSlackDmChannelCache, clearSlackSendQueuesForTest } =
-  await import("./send.js");
+const { sendMessageSlack, clearSlackDmChannelCache } = await import("./send.js");
 const SLACK_TEST_CFG = { channels: { slack: { botToken: "xoxb-test" } } };
 
 type UploadTestClient = WebClient & {
@@ -167,7 +166,6 @@ describe("sendMessageSlack file upload with user IDs", () => {
     fetchWithSsrFGuard.mockClear();
     loadOutboundMediaFromUrlMock.mockClear();
     clearSlackDmChannelCache();
-    clearSlackSendQueuesForTest();
     clearSlackThreadParticipationCache();
   });
 
diff --git a/extensions/telegram/src/bot-handlers.runtime.ts b/extensions/telegram/src/bot-handlers.runtime.ts
index 09df6ac9f75d..ca4bf88aae39 100644
--- a/extensions/telegram/src/bot-handlers.runtime.ts
+++ b/extensions/telegram/src/bot-handlers.runtime.ts
@@ -31,6 +31,7 @@ import {
   resolvePluginConversationBindingApproval,
 } from "openclaw/plugin-sdk/conversation-runtime";
 import { isApprovalNotFoundError } from "openclaw/plugin-sdk/error-runtime";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { applyModelOverrideToSessionEntry } from "openclaw/plugin-sdk/model-session-runtime";
 import { formatModelsAvailableHeader } from "openclaw/plugin-sdk/models-provider-runtime";
 import { parseStrictPositiveInteger } from "openclaw/plugin-sdk/number-runtime";
@@ -261,7 +262,7 @@ export const registerTelegramHandlers = ({
   type PromptContextMessageSelection = ReadonlyMap<string, "include" | "exclude">;
 
   const mediaGroupBuffer = new Map<string, BufferedMediaGroupEntry>();
-  const mediaGroupProcessingByKey = new Map<string, Promise<void>>();
+  const mediaGroupProcessingQueue = new KeyedAsyncQueue();
   const messageCache = createTelegramMessageCache({
     scope: resolveTelegramMessageCacheScope(telegramDeps.resolveStorePath(cfg.session?.store)),
   });
@@ -282,20 +283,16 @@ export const registerTelegramHandlers = ({
     timer: ReturnType<typeof setTimeout>;
   };
   const textFragmentBuffer = new Map<string, TextFragmentEntry>();
-  const textFragmentProcessingByKey = new Map<string, Promise<void>>();
+  const textFragmentProcessingQueue = new KeyedAsyncQueue();
 
   const queueBufferedProcessing = async (
-    processingByKey: Map<string, Promise<void>>,
+    queue: KeyedAsyncQueue,
     key: string,
     task: () => Promise<void>,
   ) => {
-    const previous = processingByKey.get(key) ?? Promise.resolve();
-    const current = previous.then(task).catch(() => undefined);
-    processingByKey.set(key, current);
-    await current;
-    if (processingByKey.get(key) === current) {
-      processingByKey.delete(key);
-    }
+    await queue.enqueue(key, async () => {
+      await task().catch(() => undefined);
+    });
   };
 
   const debounceMs = resolveInboundDebounceMs({ cfg, channel: "telegram" });
@@ -1177,7 +1174,7 @@ export const registerTelegramHandlers = ({
   };
 
   const queueTextFragmentFlush = async (entry: TextFragmentEntry) => {
-    await queueBufferedProcessing(textFragmentProcessingByKey, entry.key, async () => {
+    await queueBufferedProcessing(textFragmentProcessingQueue, entry.key, async () => {
       await flushTextFragments(entry);
     });
   };
@@ -2252,7 +2249,7 @@ export const registerTelegramHandlers = ({
         );
         existing.timer = setTimeout(() => {
           mediaGroupBuffer.delete(mediaGroupKey);
-          void queueBufferedProcessing(mediaGroupProcessingByKey, mediaGroupKey, async () => {
+          void queueBufferedProcessing(mediaGroupProcessingQueue, mediaGroupKey, async () => {
             await processMediaGroup(existing);
           });
         }, mediaGroupTimeoutMs);
@@ -2280,7 +2277,7 @@ export const registerTelegramHandlers = ({
           ),
           timer: setTimeout(() => {
             mediaGroupBuffer.delete(mediaGroupKey);
-            void queueBufferedProcessing(mediaGroupProcessingByKey, mediaGroupKey, async () => {
+            void queueBufferedProcessing(mediaGroupProcessingQueue, mediaGroupKey, async () => {
               await processMediaGroup(entry);
             });
           }, mediaGroupTimeoutMs),
diff --git a/extensions/telegram/src/thread-bindings.ts b/extensions/telegram/src/thread-bindings.ts
index d4b1a0d8e6f5..7ee8773a10d4 100644
--- a/extensions/telegram/src/thread-bindings.ts
+++ b/extensions/telegram/src/thread-bindings.ts
@@ -16,6 +16,7 @@ import {
   type SessionBindingRecord,
 } from "openclaw/plugin-sdk/conversation-runtime";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
+import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
 import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
 import { normalizeAccountId, isAcpSessionKey } from "openclaw/plugin-sdk/routing";
 import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
@@ -435,21 +436,14 @@ function enqueuePersistBindings(params: {
   if (!params.persist) {
     return Promise.resolve();
   }
-  const previous =
-    getThreadBindingsState().persistQueueByAccountId.get(params.accountId) ?? Promise.resolve();
-  const next = previous
-    .catch(() => undefined)
-    .then(async () => {
+  const state = getThreadBindingsState();
+  return enqueueKeyedTask({
+    tails: state.persistQueueByAccountId,
+    key: params.accountId,
+    task: async () => {
       await persistBindingsToStore(params);
-    });
-  getThreadBindingsState().persistQueueByAccountId.set(params.accountId, next);
-  const cleanup = () => {
-    if (getThreadBindingsState().persistQueueByAccountId.get(params.accountId) === next) {
-      getThreadBindingsState().persistQueueByAccountId.delete(params.accountId);
-    }
-  };
-  next.then(cleanup, cleanup);
-  return next;
+    },
+  });
 }
 
 function persistBindingsSafely(params: {
diff --git a/extensions/whatsapp/src/creds-persistence.ts b/extensions/whatsapp/src/creds-persistence.ts
index 668de15a0ce9..c4f35cddff3a 100644
--- a/extensions/whatsapp/src/creds-persistence.ts
+++ b/extensions/whatsapp/src/creds-persistence.ts
@@ -1,4 +1,5 @@
 // Whatsapp plugin module implements creds persistence behavior.
+import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
 import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
 import { replaceFileAtomic } from "openclaw/plugin-sdk/security-runtime";
 import { assertWebCredsPathRegularFileOrMissing, resolveWebCredsPath } from "./creds-files.js";
@@ -48,18 +49,17 @@ export function enqueueCredsSave(
   saveCreds: () => Promise<void> | void,
   onError: (error: unknown) => void,
 ): void {
-  const previous = credsSaveQueues.get(authDir) ?? Promise.resolve();
-  const next = previous
-    .then(() => saveCreds())
-    .catch((error: unknown) => {
-      onError(error);
-    })
-    .finally(() => {
-      if (credsSaveQueues.get(authDir) === next) {
-        credsSaveQueues.delete(authDir);
+  void enqueueKeyedTask({
+    tails: credsSaveQueues,
+    key: authDir,
+    task: async () => {
+      try {
+        await saveCreds();
+      } catch (error) {
+        onError(error);
       }
-    });
-  credsSaveQueues.set(authDir, next);
+    },
+  });
 }
 
 export function waitForCredsSaveQueue(authDir?: string): Promise<void> {
diff --git a/src/acp/control-plane/manager.test.ts b/src/acp/control-plane/manager.test.ts
index fb8fde97d164..e30d2807ce4e 100644
--- a/src/acp/control-plane/manager.test.ts
+++ b/src/acp/control-plane/manager.test.ts
@@ -1442,49 +1442,6 @@ describe("AcpSessionManager", () => {
     expect(snapshot.errorsByCode.ACP_TURN_FAILED).toBe(1);
   });
 
-  it("cleans actor-tail bookkeeping after session turns complete", async () => {
-    const runtimeState = createRuntime();
-    hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
-      id: "acpx",
-      runtime: runtimeState.runtime,
-    });
-    hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => {
-      const sessionKey = (paramsUnknown as { sessionKey?: string }).sessionKey ?? "";
-      return {
-        sessionKey,
-        storeSessionKey: sessionKey,
-        acp: {
-          ...readySessionMeta(),
-          runtimeSessionName: `runtime:${sessionKey}`,
-        },
-      };
-    });
-    runtimeState.runTurn.mockImplementation(async function* () {
-      yield { type: "done" as const };
-    });
-
-    const manager = new AcpSessionManager();
-    await manager.runTurn({
-      cfg: baseCfg,
-      sessionKey: "agent:codex:acp:session-a",
-      text: "first",
-      mode: "prompt",
-      requestId: "r1",
-    });
-    await manager.runTurn({
-      cfg: baseCfg,
-      sessionKey: "agent:codex:acp:session-b",
-      text: "second",
-      mode: "prompt",
-      requestId: "r2",
-    });
-
-    const internals = manager as unknown as {
-      actorQueue: { getTailMapForTesting(): Map<string, Promise<void>> };
-    };
-    expect(internals.actorQueue.getTailMapForTesting().size).toBe(0);
-  });
-
   it("surfaces backend failures raised after a done event", async () => {
     const runtimeState = createRuntime();
     hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
diff --git a/src/acp/control-plane/session-actor-queue.ts b/src/acp/control-plane/session-actor-queue.ts
index 1dc495e42b94..bfc5552c22a5 100644
--- a/src/acp/control-plane/session-actor-queue.ts
+++ b/src/acp/control-plane/session-actor-queue.ts
@@ -6,10 +6,6 @@ export class SessionActorQueue {
   private readonly queue = new KeyedAsyncQueue();
   private readonly pendingBySession = new Map<string, number>();
 
-  getTailMapForTesting(): Map<string, Promise<void>> {
-    return this.queue.getTailMapForTesting();
-  }
-
   getTotalPendingCount(): number {
     let total = 0;
     for (const count of this.pendingBySession.values()) {
diff --git a/src/agents/auth-profiles/oauth-manager.ts b/src/agents/auth-profiles/oauth-manager.ts
index f49b8245b320..865ce08d592a 100644
--- a/src/agents/auth-profiles/oauth-manager.ts
+++ b/src/agents/auth-profiles/oauth-manager.ts
@@ -1,3 +1,4 @@
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 /**
  * OAuth credential manager.
  * Resolves usable access tokens, refreshes expired credentials under global
@@ -357,7 +358,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
     return null;
   }
 
-  const refreshQueues = new Map<string, Promise<unknown>>();
+  let refreshQueue = new KeyedAsyncQueue();
 
   function refreshQueueKey(provider: string, profileId: string): string {
     return `${provider}\u0000${profileId}`;
@@ -653,21 +654,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
     attemptedCredentials?: OAuthCredential[];
   }): Promise<ResolvedOAuthAccess | null> {
     const key = refreshQueueKey(params.provider, params.profileId);
-    const prev = refreshQueues.get(key) ?? Promise.resolve();
-    let release!: () => void;
-    const gate = new Promise<void>((resolve) => {
-      release = resolve;
-    });
-    refreshQueues.set(key, gate);
-    try {
-      await prev;
-      return await doRefreshOAuthTokenWithLock(params);
-    } finally {
-      release();
-      if (refreshQueues.get(key) === gate) {
-        refreshQueues.delete(key);
-      }
-    }
+    return await refreshQueue.enqueue(key, () => doRefreshOAuthTokenWithLock(params));
   }
 
   async function resolveOAuthAccess(params: {
@@ -818,7 +805,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
   }
 
   function resetRefreshQueuesForTest(): void {
-    refreshQueues.clear();
+    refreshQueue = new KeyedAsyncQueue();
   }
 
   return {
diff --git a/src/agents/auth-profiles/oauth-refresh-queue.test.ts b/src/agents/auth-profiles/oauth-refresh-queue.test.ts
index ddd02ede6b59..d23ba770db62 100644
--- a/src/agents/auth-profiles/oauth-refresh-queue.test.ts
+++ b/src/agents/auth-profiles/oauth-refresh-queue.test.ts
@@ -115,21 +115,10 @@ describe("OAuth refresh in-process queue", () => {
     });
   });
 
-  it("resetOAuthRefreshQueuesForTest drains pending gates", () => {
-    // We can't observe the internal map, but we can assert that calling the
-    // reset is idempotent and safe from any state.
-    expect(resetOAuthRefreshQueuesForTest()).toBeUndefined();
-    expect(resetOAuthRefreshQueuesForTest()).toBeUndefined();
-  });
-
   it("serializes a 10-caller burst so later arrivals never pass an earlier caller", async () => {
     // Burst-arrival stress: 10 same-PID callers all fire concurrently.
     // The queue must chain them so each refresh completes fully before the
     // next one begins — i.e. no overlap between running refresh calls.
-    // This pins the invariant that the map-overwrite pattern in the queue
-    // wrapper does not let later arrivals skip ahead (see review P2: the
-    // `refreshQueues.set(key, gate)` overwrites only the *map head*, while
-    // FIFO ordering is enforced via the `await prev` chain).
     const profileId = "openai:default";
     const provider = "openai";
     saveAuthProfileStore(createExpiredOauthStore({ profileId, provider }), agentDir);
diff --git a/src/agents/models-config-state.ts b/src/agents/models-config-state.ts
index 54c9c67ec77a..5a0faa97cfac 100644
--- a/src/agents/models-config-state.ts
+++ b/src/agents/models-config-state.ts
@@ -1,5 +1,7 @@
 // Process-wide models.json coordination state. Dynamic imports can load this
 // module multiple times, so Symbol.for keeps write locks and ready-cache shared.
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
+
 const MODELS_JSON_STATE_KEY = Symbol.for("openclaw.modelsJsonState");
 
 export type ModelsJsonReadyResult = {
@@ -13,7 +15,7 @@ export type ModelsJsonReadyState = {
 };
 
 type ModelsJsonState = {
-  writeLocks: Map<string, Promise<void>>;
+  writeQueue: KeyedAsyncQueue;
   readyCache: Map<string, Promise<ModelsJsonReadyState>>;
 };
 
@@ -23,7 +25,7 @@ export const MODELS_JSON_STATE = (() => {
   };
   if (!globalState[MODELS_JSON_STATE_KEY]) {
     globalState[MODELS_JSON_STATE_KEY] = {
-      writeLocks: new Map<string, Promise<void>>(),
+      writeQueue: new KeyedAsyncQueue(),
       readyCache: new Map<string, Promise<ModelsJsonReadyState>>(),
     };
   }
@@ -32,6 +34,6 @@ export const MODELS_JSON_STATE = (() => {
 
 /** Clear models.json write/ready caches for tests. */
 export function resetModelsJsonReadyCacheForTest(): void {
-  MODELS_JSON_STATE.writeLocks.clear();
+  MODELS_JSON_STATE.writeQueue = new KeyedAsyncQueue();
   MODELS_JSON_STATE.readyCache.clear();
 }
diff --git a/src/agents/models-config.ts b/src/agents/models-config.ts
index 001a1b68a20d..e690ed1e6b21 100644
--- a/src/agents/models-config.ts
+++ b/src/agents/models-config.ts
@@ -340,22 +340,7 @@ export async function buildModelsJsonSourceFingerprint(
 }
 
 async function withModelsJsonWriteLock<T>(targetPath: string, run: () => Promise<T>): Promise<T> {
-  const prior = MODELS_JSON_STATE.writeLocks.get(targetPath) ?? Promise.resolve();
-  let release: () => void = () => {};
-  const gate = new Promise<void>((resolve) => {
-    release = resolve;
-  });
-  const pending = prior.then(() => gate);
-  MODELS_JSON_STATE.writeLocks.set(targetPath, pending);
-  try {
-    await prior;
-    return await run();
-  } finally {
-    release();
-    if (MODELS_JSON_STATE.writeLocks.get(targetPath) === pending) {
-      MODELS_JSON_STATE.writeLocks.delete(targetPath);
-    }
-  }
+  return await MODELS_JSON_STATE.writeQueue.enqueue(targetPath, run);
 }
 
 /** Ensures models.json and plugin catalog sidecars are current for an agent. */
diff --git a/src/agents/sessions/tools/file-mutation-queue.ts b/src/agents/sessions/tools/file-mutation-queue.ts
index 69e4153bb0a3..afe2554af1df 100644
--- a/src/agents/sessions/tools/file-mutation-queue.ts
+++ b/src/agents/sessions/tools/file-mutation-queue.ts
@@ -5,8 +5,9 @@
  */
 import { realpathSync } from "node:fs";
 import { resolve } from "node:path";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 
-const fileMutationQueues = new Map<string, Promise<void>>();
+const fileMutationQueue = new KeyedAsyncQueue();
 
 function getMutationQueueKey(filePath: string): string {
   const resolvedPath = resolve(filePath);
@@ -23,22 +24,5 @@ function getMutationQueueKey(filePath: string): string {
  */
 export async function withFileMutationQueue<T>(filePath: string, fn: () => Promise<T>): Promise<T> {
   const key = getMutationQueueKey(filePath);
-  const currentQueue = fileMutationQueues.get(key) ?? Promise.resolve();
-
-  let releaseNext!: () => void;
-  const nextQueue = new Promise<void>((resolveQueue) => {
-    releaseNext = resolveQueue;
-  });
-  const chainedQueue = currentQueue.then(() => nextQueue);
-  fileMutationQueues.set(key, chainedQueue);
-
-  await currentQueue;
-  try {
-    return await fn();
-  } finally {
-    releaseNext();
-    if (fileMutationQueues.get(key) === chainedQueue) {
-      fileMutationQueues.delete(key);
-    }
-  }
+  return await fileMutationQueue.enqueue(key, fn);
 }
diff --git a/src/config/mutate.ts b/src/config/mutate.ts
index ce9387ad8911..bc29c73123b0 100644
--- a/src/config/mutate.ts
+++ b/src/config/mutate.ts
@@ -3,6 +3,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import fs from "node:fs/promises";
 import path from "node:path";
 import { isDeepStrictEqual } from "node:util";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { formatErrorMessage } from "../infra/errors.js";
 import { withFileLock } from "../infra/file-lock.js";
 import { root as createFsRoot, type Root as FsSafeRoot } from "../infra/fs-safe.js";
@@ -68,7 +69,7 @@ const CONFIG_MUTATION_LOCK_OPTIONS = {
 
 const DEFAULT_CONFIG_MUTATION_RETRY_ATTEMPTS = 5;
 const activeConfigMutationLocks = new AsyncLocalStorage<Set<string>>();
-const configMutationQueueTails = new Map<string, Promise<void>>();
+const configMutationQueue = new KeyedAsyncQueue();
 
 export { ConfigMutationConflictError } from "./mutation-conflict.js";
 
@@ -187,28 +188,14 @@ async function withConfigMutationLock<T>(
   assertConfigWriteAllowedInCurrentMode({ configPath });
   await fs.mkdir(path.dirname(configPath), { recursive: true, mode: 0o700 });
 
-  const previousTail = configMutationQueueTails.get(configPath) ?? Promise.resolve();
-  let releaseQueueSlot!: () => void;
-  const currentRun = new Promise<void>((resolve) => {
-    releaseQueueSlot = resolve;
-  });
-  const currentTail = previousTail.catch(() => undefined).then(() => currentRun);
-  configMutationQueueTails.set(configPath, currentTail);
-
-  await previousTail.catch(() => undefined);
-  try {
-    const nextActiveLocks = new Set(activeLocks ?? []);
-    nextActiveLocks.add(configPath);
-    return await activeConfigMutationLocks.run(
+  const nextActiveLocks = new Set(activeLocks ?? []);
+  nextActiveLocks.add(configPath);
+  return await configMutationQueue.enqueue(configPath, () =>
+    activeConfigMutationLocks.run(
       nextActiveLocks,
       async () => await withFileLock(configPath, CONFIG_MUTATION_LOCK_OPTIONS, fn),
-    );
-  } finally {
-    releaseQueueSlot();
-    if (configMutationQueueTails.get(configPath) === currentTail) {
-      configMutationQueueTails.delete(configPath);
-    }
-  }
+    ),
+  );
 }
 
 function markActiveConfigMutationPath(configPath: string): void {
diff --git a/src/config/sessions/transcript-append.ts b/src/config/sessions/transcript-append.ts
index 12f96acedbe9..f8b71fac1869 100644
--- a/src/config/sessions/transcript-append.ts
+++ b/src/config/sessions/transcript-append.ts
@@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
 import fs from "node:fs/promises";
 import path from "node:path";
 import { resolveTimestampMsToIsoString } from "@openclaw/normalization-core/number-coercion";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import type { AgentMessage } from "../../agents/runtime/index.js";
 import {
   acquireSessionWriteLock,
@@ -34,7 +35,7 @@ import { CURRENT_SESSION_VERSION } from "./version.js";
 
 const SESSION_MANAGER_APPEND_MAX_BYTES = 8 * 1024 * 1024;
 
-const transcriptAppendQueues = new Map<string, Promise<void>>();
+const transcriptAppendQueue = new KeyedAsyncQueue();
 
 type TranscriptLeafInfo = {
   leafId?: string;
@@ -397,24 +398,9 @@ export async function withSessionTranscriptAppendQueue<T>(
   fn: () => Promise<T>,
 ): Promise<T> {
   const queueKey = await resolveTranscriptAppendQueueKey(transcriptPath);
-  const previous = transcriptAppendQueues.get(queueKey) ?? Promise.resolve();
-  let releaseCurrent!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseCurrent = resolve;
-  });
-  const tail = previous.catch(() => undefined).then(() => current);
   // Per-file queue is in-process only; the external session write lock still owns cross-process
   // ordering.
-  transcriptAppendQueues.set(queueKey, tail);
-  await previous.catch(() => undefined);
-  try {
-    return await fn();
-  } finally {
-    releaseCurrent();
-    if (transcriptAppendQueues.get(queueKey) === tail) {
-      transcriptAppendQueues.delete(queueKey);
-    }
-  }
+  return await transcriptAppendQueue.enqueue(queueKey, fn);
 }
 
 export type AppendSessionTranscriptMessageParams<TMessage = unknown> = {
diff --git a/src/cron/run-log.ts b/src/cron/run-log.ts
index 40d5dddc7905..536cd15b8eb8 100644
--- a/src/cron/run-log.ts
+++ b/src/cron/run-log.ts
@@ -5,6 +5,7 @@ import {
   normalizeStringifiedOptionalString,
 } from "@openclaw/normalization-core/string-coerce";
 import { uniqueValues } from "@openclaw/normalization-core/string-normalization";
+import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
 import { parseByteSize } from "../cli/parse-bytes.js";
 import type { CronConfig } from "../config/types.cron.js";
 import {
@@ -150,11 +151,11 @@ export async function appendCronRunLog(params: {
       : { ...params.entry, jobId: normalizedJobId };
   const storeKey = cronStoreKey(params.storePath);
   const writeKey = cronRunLogWriteKey(params.storePath, entry.jobId);
-  const prev = writesByTarget.get(writeKey) ?? Promise.resolve();
   // Keep writes for the same store/job ordered so prune-by-count cannot race a later insert.
-  const next = prev
-    .catch(() => undefined)
-    .then(async () => {
+  await enqueueKeyedTask({
+    tails: writesByTarget,
+    key: writeKey,
+    task: async () => {
       runOpenClawStateWriteTransaction(({ db }) => {
         insertCronRunLogEntry(db, storeKey, entry);
         if (params.opts?.keepLines !== false) {
@@ -166,15 +167,8 @@ export async function appendCronRunLog(params: {
           );
         }
       });
-    });
-  writesByTarget.set(writeKey, next);
-  try {
-    await next;
-  } finally {
-    if (writesByTarget.get(writeKey) === next) {
-      writesByTarget.delete(writeKey);
-    }
-  }
+    },
+  });
 }
 
 /** Reads recent run-log entries synchronously for startup/task reconciliation paths. */
diff --git a/src/gateway/rate-limit-attempt-serialization.ts b/src/gateway/rate-limit-attempt-serialization.ts
index e51484c2a077..317f9f172640 100644
--- a/src/gateway/rate-limit-attempt-serialization.ts
+++ b/src/gateway/rate-limit-attempt-serialization.ts
@@ -1,8 +1,9 @@
 // Gateway auth rate-limit serialization.
 // Serializes limiter attempts per IP/scope so concurrent failures count correctly.
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { AUTH_RATE_LIMIT_SCOPE_DEFAULT, normalizeRateLimitClientIp } from "./auth-rate-limit.js";
 
-const pendingAttempts = new Map<string, Promise<void>>();
+const pendingAttempts = new KeyedAsyncQueue();
 
 function normalizeScope(scope: string | undefined): string {
   return (scope ?? AUTH_RATE_LIMIT_SCOPE_DEFAULT).trim() || AUTH_RATE_LIMIT_SCOPE_DEFAULT;
@@ -17,24 +18,7 @@ export async function withSerializedKeyedAttempt<T>(params: {
   key: string;
   run: () => Promise<T>;
 }): Promise<T> {
-  const key = params.key;
-  const previous = pendingAttempts.get(key) ?? Promise.resolve();
-  let releaseCurrent!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseCurrent = resolve;
-  });
-  const tail = previous.catch(() => {}).then(() => current);
-  pendingAttempts.set(key, tail);
-
-  await previous.catch(() => {});
-  try {
-    return await params.run();
-  } finally {
-    releaseCurrent();
-    if (pendingAttempts.get(key) === tail) {
-      pendingAttempts.delete(key);
-    }
-  }
+  return await pendingAttempts.enqueue(params.key, params.run);
 }
 
 /** Runs one rate-limit attempt after prior attempts for the same IP/scope finish. */
diff --git a/src/gateway/server-methods/send.ts b/src/gateway/server-methods/send.ts
index 7e3a9ae2ddfe..af130b5d67b6 100644
--- a/src/gateway/server-methods/send.ts
+++ b/src/gateway/server-methods/send.ts
@@ -5,6 +5,7 @@ import {
   normalizeOptionalString,
   readStringValue,
 } from "@openclaw/normalization-core/string-coerce";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import {
   ErrorCodes,
   errorShape,
@@ -410,7 +411,7 @@ async function mirrorDeliveredSourceReplyToTranscriptBestEffort(params: {
   }
 }
 
-const sourceReplyTranscriptMirrorQueues = new Map<string, Promise<void>>();
+const sourceReplyTranscriptMirrorQueue = new KeyedAsyncQueue();
 
 function resolveSourceReplyTranscriptMirrorQueueKey(
   mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0],
@@ -424,22 +425,11 @@ function scheduleDeliveredSourceReplyTranscriptMirror(params: {
   mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0];
 }): Promise<void> {
   const queueKey = resolveSourceReplyTranscriptMirrorQueueKey(params.mirror);
-  const previous = sourceReplyTranscriptMirrorQueues.get(queueKey);
   // Queue per session so current-conversation source replies are visible before
   // a following turn can read the transcript.
-  const queued = (async () => {
-    await previous?.catch(() => undefined);
-    await mirrorDeliveredSourceReplyToTranscriptBestEffort(params);
-  })();
-  sourceReplyTranscriptMirrorQueues.set(queueKey, queued);
-  void queued
-    .finally(() => {
-      if (sourceReplyTranscriptMirrorQueues.get(queueKey) === queued) {
-        sourceReplyTranscriptMirrorQueues.delete(queueKey);
-      }
-    })
-    .catch(() => undefined);
-  return queued;
+  return sourceReplyTranscriptMirrorQueue.enqueue(queueKey, () =>
+    mirrorDeliveredSourceReplyToTranscriptBestEffort(params),
+  );
 }
 
 export const sendHandlers: GatewayRequestHandlers = {
diff --git a/src/plugin-sdk/keyed-async-queue.test.ts b/src/plugin-sdk/keyed-async-queue.test.ts
index ac11e2aa733f..489b4d90ee63 100644
--- a/src/plugin-sdk/keyed-async-queue.test.ts
+++ b/src/plugin-sdk/keyed-async-queue.test.ts
@@ -116,17 +116,29 @@ describe("enqueueKeyedTask", () => {
 });
 
 describe("KeyedAsyncQueue", () => {
-  it("exposes tail map for observability", async () => {
+  it("serializes tasks while preserving their return values", async () => {
     const queue = new KeyedAsyncQueue();
     const gate = createDeferred();
-    const run = queue.enqueue("actor", async () => {
+    const order: string[] = [];
+    const first = queue.enqueue("actor", async () => {
+      order.push("first:start");
       await gate.promise;
       return 1;
     });
-    expect(queue.getTailMapForTesting().has("actor")).toBe(true);
-    gate.resolve();
-    await run;
+    const second = queue.enqueue("actor", async () => {
+      order.push("second:start");
+      return 2;
+    });
     await Promise.resolve();
-    expect(queue.getTailMapForTesting().has("actor")).toBe(false);
+    await Promise.resolve();
+    expect(order).toEqual(["first:start"]);
+    gate.resolve();
+    await expect(Promise.all([first, second])).resolves.toEqual([1, 2]);
+    expect(order).toEqual(["first:start", "second:start"]);
+  });
+
+  it("retains the deprecated tail-map accessor for Plugin SDK compatibility", () => {
+    const queue = new KeyedAsyncQueue();
+    expect(queue.getTailMapForTesting()).toBeInstanceOf(Map);
   });
 });
diff --git a/src/plugin-sdk/keyed-async-queue.ts b/src/plugin-sdk/keyed-async-queue.ts
index 63e8da9f0da2..0c4aca8ef10c 100644
--- a/src/plugin-sdk/keyed-async-queue.ts
+++ b/src/plugin-sdk/keyed-async-queue.ts
@@ -38,6 +38,10 @@ export function enqueueKeyedTask<T>(params: {
 export class KeyedAsyncQueue {
   private readonly tails = new Map<string, Promise<void>>();
 
+  /**
+   * @deprecated Retained for shipped Plugin SDK compatibility. New callers must
+   * not depend on queue storage; remove in a declared Plugin SDK breaking window.
+   */
   getTailMapForTesting(): Map<string, Promise<void>> {
     return this.tails;
   }
diff --git a/src/skills/loading/serialize.ts b/src/skills/loading/serialize.ts
index 39608e30c149..7b88a0df2ef3 100644
--- a/src/skills/loading/serialize.ts
+++ b/src/skills/loading/serialize.ts
@@ -1,16 +1,9 @@
 // Skill serialization helpers compact skill metadata and coordinate sync queue updates.
-const SKILLS_SYNC_QUEUE = new Map<string, Promise<unknown>>();
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
+
+const skillsSyncQueue = new KeyedAsyncQueue();
 
 /** Serializes async work by key so repeated skill loads do not race on shared files. */
 export async function serializeByKey<T>(key: string, task: () => Promise<T>) {
-  const prev = SKILLS_SYNC_QUEUE.get(key) ?? Promise.resolve();
-  const next = prev.then(task, task);
-  SKILLS_SYNC_QUEUE.set(key, next);
-  try {
-    return await next;
-  } finally {
-    if (SKILLS_SYNC_QUEUE.get(key) === next) {
-      SKILLS_SYNC_QUEUE.delete(key);
-    }
-  }
+  return await skillsSyncQueue.enqueue(key, task);
 }
diff --git a/src/skills/workshop/store.ts b/src/skills/workshop/store.ts
index 7c6fe07ff787..c10343d9f56f 100644
--- a/src/skills/workshop/store.ts
+++ b/src/skills/workshop/store.ts
@@ -2,6 +2,7 @@ import crypto from "node:crypto";
 import fs from "node:fs/promises";
 import path from "node:path";
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { resolveStateDir } from "../../config/paths.js";
 import { sha256Hex } from "../../infra/crypto-digest.js";
 import { type FileLockOptions, withFileLock } from "../../infra/file-lock.js";
@@ -49,7 +50,7 @@ const SKILL_WORKSHOP_LOCK_OPTIONS: FileLockOptions = {
   },
   stale: 60_000,
 };
-const skillWorkshopProcessLocks = new Map<string, Promise<void>>();
+const skillWorkshopProcessLocks = new KeyedAsyncQueue();
 
 type SkillWorkshopStoreOptions = {
   env?: NodeJS.ProcessEnv;
@@ -363,24 +364,10 @@ async function withSkillProposalManifestLock<T>(
 
 async function withSkillWorkshopLock<T>(lockFile: string, fn: () => Promise<T>): Promise<T> {
   const lockKey = path.resolve(lockFile);
-  const previous = skillWorkshopProcessLocks.get(lockKey) ?? Promise.resolve();
-  let releaseQueued!: () => void;
-  const current = new Promise<void>((resolve) => {
-    releaseQueued = resolve;
-  });
-  const previousDone = previous.catch(() => undefined);
-  const queued = previousDone.then(() => current);
-  skillWorkshopProcessLocks.set(lockKey, queued);
-  await previousDone;
-  await fs.mkdir(path.dirname(lockFile), { recursive: true });
-  try {
+  return await skillWorkshopProcessLocks.enqueue(lockKey, async () => {
+    await fs.mkdir(path.dirname(lockFile), { recursive: true });
     return await withFileLock(lockFile, SKILL_WORKSHOP_LOCK_OPTIONS, fn);
-  } finally {
-    releaseQueued();
-    if (skillWorkshopProcessLocks.get(lockKey) === queued) {
-      skillWorkshopProcessLocks.delete(lockKey);
-    }
-  }
+  });
 }
 
 export async function readProposalSupportFiles(
diff --git a/src/trajectory/runtime.ts b/src/trajectory/runtime.ts
index 4ce5e6613f2a..26aa70c96552 100644
--- a/src/trajectory/runtime.ts
+++ b/src/trajectory/runtime.ts
@@ -1,6 +1,7 @@
 // Trajectory runtime records runtime events into trajectory log files.
 import fs from "node:fs";
 import path from "node:path";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 import { sanitizeDiagnosticPayload } from "../agents/payload-redaction.js";
 import type {
   QueuedFileWriter,
@@ -46,7 +47,7 @@ type TrajectoryRuntimeRecorder = {
 };
 
 const writers = new Map<string, TrajectoryRuntimeWriter>();
-const windowFlushes = new Map<string, Promise<void>>();
+const windowFlushes = new KeyedAsyncQueue();
 const MAX_TRAJECTORY_WRITERS = 100;
 const TRAJECTORY_RUNTIME_DATA_STRING_MAX_CHARS = 32_768;
 const TRAJECTORY_RUNTIME_DATA_ARRAY_MAX_ITEMS = 64;
@@ -389,19 +390,9 @@ async function queueTrajectoryWindowFlush(params: {
   maxFileBytes: number;
   appendedLines: string[];
 }): Promise<void> {
-  const previous = windowFlushes.get(params.filePath) ?? Promise.resolve();
-  const current = previous
-    .catch(() => undefined)
-    .then(async () => {
-      await replaceTrajectoryWindow(params);
-    })
-    .finally(() => {
-      if (windowFlushes.get(params.filePath) === current) {
-        windowFlushes.delete(params.filePath);
-      }
-    });
-  windowFlushes.set(params.filePath, current);
-  await current;
+  await windowFlushes.enqueue(params.filePath, async () => {
+    await replaceTrajectoryWindow(params);
+  });
 }
 
 function createTrajectoryWindowWriter(