refactor: consolidate bounded concurrency on p-limit and p-map (#106002)

* refactor: adopt p-limit and p-map for bounded concurrency

* chore: normalize lockfile ordering

* fix: preserve memory host concurrency export

* refactor: keep media apply within size budget

* fix: drain memory concurrency before failure

* fix: satisfy memory concurrency guards

* chore: scope memory host dead-code checks

* chore: lower concurrency LOC baselines

* fix: preserve concurrency API surfaces

* chore: reconcile concurrency LOC baseline

* test: keep memory concurrency gate portable

* style: simplify concurrency drain

* ci: refresh deadcode export baseline

config/knip.config.ts

@@ -240,6 +240,10 @@ const config = {
       entry: ["src/*.ts!"],
       project: ["src/**/*.ts!"],
     },
+    "packages/memory-host-sdk": {
+      entry: ["src/*.ts!", "src/host/embeddings-worker-child.ts!"],
+      project: ["src/**/*.ts!"],
+    },
     "packages/speech-core": {
       entry: ["api.ts!", "runtime-api.ts!", "speaker.ts!", "voice-models.ts!"],
       project: ["**/*.ts!"],

extensions/discord/npm-shrinkwrap.json

@@ -11,6 +11,7 @@
         "@discordjs/voice": "0.19.2",
         "discord-api-types": "0.38.49",
         "libopus-wasm": "0.2.0",
+        "p-map": "7.0.5",
         "typebox": "1.3.3",
         "undici": "8.5.0",
         "ws": "8.21.0"
@@ -360,6 +361,18 @@
         "node": ">=20"
       }
     },
+    "node_modules/p-map": {
+      "version": "7.0.5",
+      "resolved": "https://registry.npmjs.org/p-map/-/p-map-7.0.5.tgz",
+      "integrity": "sha512-e8vJF4XdVkzqqSHguEMz41mQO1wKwxKm5ENrUJQUu9kLDCtn83cxbyHZcszr4QC5zEA7WffRRC4gsTecC7J9oA==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=18"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/prism-media": {
       "version": "1.3.5",
       "resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.3.5.tgz",

extensions/discord/package.json

@@ -11,6 +11,7 @@
     "@discordjs/voice": "0.19.2",
     "discord-api-types": "0.38.49",
     "libopus-wasm": "0.2.0",
+    "p-map": "7.0.5",
     "typebox": "1.3.3",
     "undici": "8.5.0",
     "ws": "8.21.0"

extensions/discord/src/monitor/thread-bindings.lifecycle.test.ts

@@ -1793,8 +1793,9 @@ describe("thread binding lifecycle", () => {
       },
     });
 
-    await Promise.resolve();
-    await Promise.resolve();
+    await vi.waitFor(() => {
+      expect(probeCallCount).toBe(2);
+    });
     const observedParallelStart = secondProbeStartedBeforeFirstResolved;
 
     resolveFirstProbe?.({ status: "healthy" });

extensions/discord/src/monitor/thread-bindings.lifecycle.ts

@@ -1,12 +1,12 @@
 import { readAcpSessionEntry, type AcpSessionStoreEntry } from "openclaw/plugin-sdk/acp-runtime";
 import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 // Discord plugin module implements thread bindings.lifecycle behavior.
-import { expectDefined } from "openclaw/plugin-sdk/expect-runtime";
 import {
   normalizeOptionalLowercaseString,
   normalizeOptionalString,
   uniqueStrings,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import { parseDiscordTarget } from "../targets.js";
 import { resolveChannelIdForBinding } from "./thread-bindings.discord-api.js";
 import { getThreadBindingManager } from "./thread-bindings.manager.js";
@@ -53,37 +53,6 @@ type AcpThreadBindingHealthProbe = (params: {
 // Cap startup fan-out so large binding sets do not create unbounded ACP probe spikes.
 const ACP_STARTUP_HEALTH_PROBE_CONCURRENCY_LIMIT = 8;
 
-async function mapWithConcurrency<TItem, TResult>(params: {
-  items: TItem[];
-  limit: number;
-  worker: (item: TItem, index: number) => Promise<TResult>;
-}): Promise<TResult[]> {
-  if (params.items.length === 0) {
-    return [];
-  }
-  const limit = Math.max(1, Math.floor(params.limit));
-  const resultsByIndex = new Map<number, TResult>();
-  let nextIndex = 0;
-
-  const runWorker = async () => {
-    for (;;) {
-      const index = nextIndex;
-      nextIndex += 1;
-      if (index >= params.items.length) {
-        return;
-      }
-      const item = expectDefined(params.items[index], "bounded worker item index");
-      resultsByIndex.set(index, await params.worker(item, index));
-    }
-  };
-
-  const workers = Array.from({ length: Math.min(limit, params.items.length) }, () => runWorker());
-  await Promise.all(workers);
-  return params.items.map((_item, index) =>
-    expectDefined(resultsByIndex.get(index), "completed bounded worker result"),
-  );
-}
-
 export function listThreadBindingsForAccount(accountId?: string): ThreadBindingRecord[] {
   const manager = getThreadBindingManager(accountId);
   if (!manager) {
@@ -297,10 +266,9 @@ export async function reconcileAcpThreadBindingsOnStartup(params: {
   }
 
   if (params.healthProbe && probeTargets.length > 0) {
-    const probeResults = await mapWithConcurrency({
-      items: probeTargets,
-      limit: ACP_STARTUP_HEALTH_PROBE_CONCURRENCY_LIMIT,
-      worker: async ({ binding, sessionKey, session }) => {
+    const probeResults = await pMap(
+      probeTargets,
+      async ({ binding, sessionKey, session }) => {
         try {
           const result = await params.healthProbe?.({
             cfg: params.cfg,
@@ -321,7 +289,11 @@ export async function reconcileAcpThreadBindingsOnStartup(params: {
           };
         }
       },
-    });
+      {
+        concurrency: ACP_STARTUP_HEALTH_PROBE_CONCURRENCY_LIMIT,
+        stopOnError: true,
+      },
+    );
 
     for (const probeResult of probeResults) {
       if (probeResult.status === "stale") {

extensions/openshell/npm-shrinkwrap.json

@@ -8,9 +8,37 @@
       "name": "@openclaw/openshell-sandbox",
       "version": "2026.7.2",
       "dependencies": {
+        "p-limit": "7.3.0",
         "zod": "4.4.3"
       }
     },
+    "node_modules/p-limit": {
+      "version": "7.3.0",
+      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.3.0.tgz",
+      "integrity": "sha512-7cIXg/Z0M5WZRblrsOla88S4wAK+zOQQWeBYfV3qJuJXMr+LnbYjaadrFaS0JILfEDPVqHyKnZ1Z/1d6J9VVUw==",
+      "license": "MIT",
+      "dependencies": {
+        "yocto-queue": "^1.2.1"
+      },
+      "engines": {
+        "node": ">=20"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
+    "node_modules/yocto-queue": {
+      "version": "1.2.2",
+      "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz",
+      "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=12.20"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/zod": {
       "version": "4.4.3",
       "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz",

extensions/openshell/package.json

@@ -8,6 +8,7 @@
   },
   "type": "module",
   "dependencies": {
+    "p-limit": "7.3.0",
     "zod": "4.4.3"
   },
   "devDependencies": {

extensions/openshell/src/mirror.ts

@@ -3,6 +3,7 @@ import fs from "node:fs/promises";
 import path from "node:path";
 import { movePathWithCopyFallback } from "openclaw/plugin-sdk/security-runtime";
 import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pLimit from "p-limit";
 
 export const DEFAULT_OPEN_SHELL_MIRROR_EXCLUDE_DIRS = ["hooks", "git-hooks", ".git"] as const;
 const COPY_TREE_FS_CONCURRENCY = 16;
@@ -12,31 +13,7 @@ function createExcludeMatcher(excludeDirs?: readonly string[]) {
   return (name: string) => excluded.has(normalizeLowercaseStringOrEmpty(name));
 }
 
-function createConcurrencyLimiter(limit: number) {
-  let active = 0;
-  const queue: Array<() => void> = [];
-
-  const release = () => {
-    active -= 1;
-    queue.shift()?.();
-  };
-
-  return async <T>(task: () => Promise<T>): Promise<T> => {
-    if (active >= limit) {
-      await new Promise<void>((resolve) => {
-        queue.push(resolve);
-      });
-    }
-    active += 1;
-    try {
-      return await task();
-    } finally {
-      release();
-    }
-  };
-}
-
-const runLimitedFs = createConcurrencyLimiter(COPY_TREE_FS_CONCURRENCY);
+const runLimitedFs = pLimit(COPY_TREE_FS_CONCURRENCY);
 
 async function lstatIfExists(targetPath: string) {
   return await runLimitedFs(async () => await fs.lstat(targetPath)).catch(() => null);

extensions/qa-lab/package.json

@@ -7,6 +7,8 @@
   "dependencies": {
     "@copilotkit/aimock": "1.35.0",
     "@modelcontextprotocol/sdk": "1.29.0",
+    "p-limit": "7.3.0",
+    "p-map": "7.0.5",
     "playwright-core": "1.61.1",
     "pretty-ms": "9.3.0",
     "semver": "7.8.5",

extensions/qa-lab/src/character-eval.ts

@@ -3,6 +3,7 @@ import fs from "node:fs/promises";
 import path from "node:path";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import { normalizeStringEntries, uniqueStrings } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import prettyMilliseconds from "pretty-ms";
 import { createQaArtifactRunId } from "./artifact-run-id.js";
 import { isQaFastModeModelRef, type QaProviderMode } from "./model-selection.js";
@@ -190,28 +191,6 @@ function normalizeConcurrency(value: number | undefined, fallback = 1) {
   return Math.max(1, Math.floor(value));
 }
 
-async function mapWithConcurrency<T, U>(
-  items: readonly T[],
-  concurrency: number,
-  mapper: (item: T, index: number) => Promise<U>,
-) {
-  const results = Array.from<U>({ length: items.length });
-  const pendingItems = items.entries();
-  const workerCount = Math.min(normalizeConcurrency(concurrency), items.length);
-  const workers = Array.from({ length: workerCount }, async () => {
-    while (true) {
-      const next = pendingItems.next();
-      if (next.done) {
-        return;
-      }
-      const [index, item] = next.value;
-      results[index] = await mapper(item, index);
-    }
-  });
-  await Promise.all(workers);
-  return results;
-}
-
 function extractTranscript(result: QaSuiteResult) {
   let longestDetail: string | undefined;
   for (const scenario of result.scenarios) {
@@ -534,77 +513,83 @@ export async function runQaCharacterEval(params: QaCharacterEvalParams) {
     `start scenario=${scenarioId} candidates=${models.length} candidateConcurrency=${candidateConcurrency} output=${outputDir}`,
   );
   const candidatesStartedAt = Date.now();
-  const runs = await mapWithConcurrency(models, candidateConcurrency, async (model, index) => {
-    const thinkingDefault = resolveCandidateThinkingDefault({
-      model,
-      candidateThinkingDefault: params.candidateThinkingDefault,
-      candidateThinkingByModel: params.candidateThinkingByModel,
-      candidateModelOptions: params.candidateModelOptions,
-    });
-    const fastMode = resolveCandidateFastMode({
-      model,
-      candidateFastMode: params.candidateFastMode,
-      candidateModelOptions: params.candidateModelOptions,
-    });
-    const modelOutputDir = path.join(runsDir, sanitizePathPart(model));
-    const runStartedAt = Date.now();
-    logCharacterEvalProgress(
-      params.progress,
-      `candidate start ${formatEvalIndex(index, models.length)} model=${model} thinking=${thinkingDefault} fast=${fastMode ? "on" : "off"}`,
-    );
-    try {
-      const result = await runSuite({
-        repoRoot,
-        outputDir: modelOutputDir,
-        providerMode: "live-frontier",
-        primaryModel: model,
-        alternateModel: model,
-        fastMode,
-        thinkingDefault,
-        scenarioIds: [scenarioId],
+  const runs = await pMap(
+    models,
+    async (model, index) => {
+      const thinkingDefault = resolveCandidateThinkingDefault({
+        model,
+        candidateThinkingDefault: params.candidateThinkingDefault,
+        candidateThinkingByModel: params.candidateThinkingByModel,
+        candidateModelOptions: params.candidateModelOptions,
       });
-      const transcript = extractTranscript(result);
-      const transcriptFailure = detectTranscriptFailure(transcript);
-      const failedScenarioCount = await readQaSuiteFailedScenarioCountFromFile(result.summaryPath);
-      const status = failedScenarioCount > 0 || transcriptFailure ? "fail" : "pass";
-      const run = {
+      const fastMode = resolveCandidateFastMode({
         model,
-        status,
-        durationMs: Date.now() - runStartedAt,
-        outputDir: modelOutputDir,
-        thinkingDefault,
-        fastMode,
-        reportPath: result.reportPath,
-        summaryPath: result.summaryPath,
-        transcript,
-        stats: collectTranscriptStats(transcript),
-        ...(transcriptFailure ? { error: transcriptFailure } : {}),
-      } satisfies QaCharacterEvalRun;
+        candidateFastMode: params.candidateFastMode,
+        candidateModelOptions: params.candidateModelOptions,
+      });
+      const modelOutputDir = path.join(runsDir, sanitizePathPart(model));
+      const runStartedAt = Date.now();
       logCharacterEvalProgress(
         params.progress,
-        `candidate done ${formatEvalIndex(index, models.length)} model=${model} ${summarizeRunStats(run)}`,
+        `candidate start ${formatEvalIndex(index, models.length)} model=${model} thinking=${thinkingDefault} fast=${fastMode ? "on" : "off"}`,
       );
-      return run;
-    } catch (error) {
-      const transcript = "";
-      const run = {
-        model,
-        status: "fail",
-        durationMs: Date.now() - runStartedAt,
-        outputDir: modelOutputDir,
-        thinkingDefault,
-        fastMode,
-        transcript,
-        stats: collectTranscriptStats(transcript),
-        error: formatErrorMessage(error),
-      } satisfies QaCharacterEvalRun;
-      logCharacterEvalProgress(
-        params.progress,
-        `candidate done ${formatEvalIndex(index, models.length)} model=${model} ${summarizeRunStats(run)}`,
-      );
-      return run;
-    }
-  });
+      try {
+        const result = await runSuite({
+          repoRoot,
+          outputDir: modelOutputDir,
+          providerMode: "live-frontier",
+          primaryModel: model,
+          alternateModel: model,
+          fastMode,
+          thinkingDefault,
+          scenarioIds: [scenarioId],
+        });
+        const transcript = extractTranscript(result);
+        const transcriptFailure = detectTranscriptFailure(transcript);
+        const failedScenarioCount = await readQaSuiteFailedScenarioCountFromFile(
+          result.summaryPath,
+        );
+        const status = failedScenarioCount > 0 || transcriptFailure ? "fail" : "pass";
+        const run = {
+          model,
+          status,
+          durationMs: Date.now() - runStartedAt,
+          outputDir: modelOutputDir,
+          thinkingDefault,
+          fastMode,
+          reportPath: result.reportPath,
+          summaryPath: result.summaryPath,
+          transcript,
+          stats: collectTranscriptStats(transcript),
+          ...(transcriptFailure ? { error: transcriptFailure } : {}),
+        } satisfies QaCharacterEvalRun;
+        logCharacterEvalProgress(
+          params.progress,
+          `candidate done ${formatEvalIndex(index, models.length)} model=${model} ${summarizeRunStats(run)}`,
+        );
+        return run;
+      } catch (error) {
+        const transcript = "";
+        const run = {
+          model,
+          status: "fail",
+          durationMs: Date.now() - runStartedAt,
+          outputDir: modelOutputDir,
+          thinkingDefault,
+          fastMode,
+          transcript,
+          stats: collectTranscriptStats(transcript),
+          error: formatErrorMessage(error),
+        } satisfies QaCharacterEvalRun;
+        logCharacterEvalProgress(
+          params.progress,
+          `candidate done ${formatEvalIndex(index, models.length)} model=${model} ${summarizeRunStats(run)}`,
+        );
+        return run;
+      }
+    },
+    { concurrency: candidateConcurrency, stopOnError: true },
+  );
   const failedCandidateCount = runs.filter((run) => run.status === "fail").length;
   logCharacterEvalProgress(
     params.progress,
@@ -629,9 +614,8 @@ export async function runQaCharacterEval(params: QaCharacterEvalParams) {
     `judges start judges=${judgeModels.length} judgeConcurrency=${judgeConcurrency} timeout=${formatDuration(judgeTimeoutMs)} labels=${params.judgeBlindModels === true ? "blind" : "visible"}`,
   );
   const judgesStartedAt = Date.now();
-  const judgments = await mapWithConcurrency(
+  const judgments = await pMap(
     judgeModels,
-    judgeConcurrency,
     async (judgeModel, index) => {
       const judgeOptions = resolveJudgeOptions({
         model: judgeModel,
@@ -685,6 +669,7 @@ export async function runQaCharacterEval(params: QaCharacterEvalParams) {
       );
       return judgment;
     },
+    { concurrency: judgeConcurrency, stopOnError: true },
   );
   const failedJudgeCount = judgments.filter((judgment) => judgment.rankings.length === 0).length;
   logCharacterEvalProgress(

extensions/qa-lab/src/evidence-gallery.ts

@@ -4,6 +4,7 @@ import os from "node:os";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
+import pLimit from "p-limit";
 import type {
   QaEvidenceArtifactView,
   QaEvidenceGalleryEntryView,
@@ -853,25 +854,6 @@ async function buildProducerContext(params: {
   };
 }
 
-function createConcurrencyLimit(limit: number) {
-  let active = 0;
-  const queue: Array<() => void> = [];
-  return async function runLimited<T>(task: () => Promise<T>): Promise<T> {
-    if (active >= limit) {
-      await new Promise<void>((resolve) => {
-        queue.push(resolve);
-      });
-    }
-    active += 1;
-    try {
-      return await task();
-    } finally {
-      active -= 1;
-      queue.shift()?.();
-    }
-  };
-}
-
 export async function buildQaEvidenceGalleryModel(params: {
   evidencePath: string;
   repoRoot: string;
@@ -900,7 +882,7 @@ export async function buildQaEvidenceGalleryModel(params: {
     repoRoot,
     summaryEntries: summary.entries,
   });
-  const limitArtifactView = createConcurrencyLimit(ARTIFACT_VIEW_CONCURRENCY);
+  const limitArtifactView = pLimit(ARTIFACT_VIEW_CONCURRENCY);
   const entries = await Promise.all(
     summary.entries.map(async (entry, entryIndex): Promise<QaEvidenceGalleryEntryView> => {
       counts[entry.result.status] += 1;

extensions/qa-lab/src/suite-planning.ts

@@ -2,6 +2,7 @@
 import path from "node:path";
 import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime";
 import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import { createQaArtifactRunId } from "./artifact-run-id.js";
 import { ensureRepoBoundDirectory, resolveRepoRelativeOutputDir } from "./cli-paths.js";
 import type { QaCliBackendAuthMode } from "./gateway-child.js";
@@ -411,10 +412,7 @@ async function mapQaSuiteWithConcurrency<T, U>(
     sleepImpl?: (ms: number) => Promise<unknown>;
   },
 ) {
-  const results = Array.from<U>({ length: items.length });
-  let nextIndex = 0;
   let nextStartGate = Promise.resolve();
-  const workerCount = Math.min(Math.max(1, Math.floor(concurrency)), items.length);
   const startStaggerMs = Math.max(0, Math.floor(opts?.startStaggerMs ?? 0));
   const sleepImpl =
     opts?.sleepImpl ??
@@ -444,20 +442,17 @@ async function mapQaSuiteWithConcurrency<T, U>(
       }
     })();
   }
-  const workers = Array.from({ length: workerCount }, async () => {
-    while (nextIndex < items.length) {
-      const index = nextIndex;
-      nextIndex += 1;
-      await waitForStartSlot(nextIndex < items.length);
-      const item = items[index];
-      if (item === undefined) {
-        return;
-      }
-      results[index] = await mapper(item, index);
-    }
-  });
-  await Promise.all(workers);
-  return results;
+  return await pMap(
+    items,
+    async (item, index) => {
+      await waitForStartSlot(index < items.length - 1);
+      return await mapper(item, index);
+    },
+    {
+      concurrency: Math.max(1, Math.floor(concurrency)),
+      stopOnError: true,
+    },
+  );
 }
 
 async function resolveQaSuiteOutputDir(repoRoot: string, outputDir?: string) {

extensions/qqbot/npm-shrinkwrap.json

@@ -10,6 +10,7 @@
       "dependencies": {
         "@tencent-connect/qqbot-connector": "1.1.0",
         "mpg123-decoder": "1.0.3",
+        "p-map": "7.0.5",
         "pretty-ms": "9.3.0",
         "silk-wasm": "3.7.1",
         "ws": "8.21.0",
@@ -65,6 +66,18 @@
         "url": "https://github.com/sponsors/eshaz"
       }
     },
+    "node_modules/p-map": {
+      "version": "7.0.5",
+      "resolved": "https://registry.npmjs.org/p-map/-/p-map-7.0.5.tgz",
+      "integrity": "sha512-e8vJF4XdVkzqqSHguEMz41mQO1wKwxKm5ENrUJQUu9kLDCtn83cxbyHZcszr4QC5zEA7WffRRC4gsTecC7J9oA==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=18"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/parse-ms": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/parse-ms/-/parse-ms-4.0.0.tgz",

extensions/qqbot/package.json

@@ -11,6 +11,7 @@
   "dependencies": {
     "@tencent-connect/qqbot-connector": "1.1.0",
     "mpg123-decoder": "1.0.3",
+    "p-map": "7.0.5",
     "pretty-ms": "9.3.0",
     "silk-wasm": "3.7.1",
     "ws": "8.21.0",

extensions/qqbot/src/engine/api/media-chunked.ts

@@ -40,6 +40,7 @@ import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http";
 import { sleep } from "openclaw/plugin-sdk/runtime-env";
 import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime";
 import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
+import pMap from "p-map";
 import type { MediaSource, OpenedLocalFile } from "../messaging/media-source.js";
 import { openLocalFile } from "../messaging/media-source.js";
 import {
@@ -286,10 +287,10 @@ export class ChunkedMediaApi {
         });
       };
 
-      await runWithConcurrency(
-        parts.map((part) => () => uploadPart(part)),
-        maxConcurrent,
-      );
+      await pMap(parts, uploadPart, {
+        concurrency: maxConcurrent,
+        stopOnError: true,
+      });
 
       this.logger?.info?.(`${prefix} all parts uploaded, completing...`);
 
@@ -613,23 +614,3 @@ async function putToPresignedUrl(
 
   throw lastError ?? new Error(`Part ${partIndex}/${totalParts} upload failed`);
 }
-
-// ============ Concurrency ============
-
-/**
- * Batch-mode concurrency limiter. Deliberately simple: dispatch N tasks at
- * a time and wait for the whole batch to settle before the next batch.
- *
- * A pool / queue implementation would recover some throughput when tasks
- * have heavy variance, but part uploads are size-uniform (last part can be
- * short) so the extra complexity is not worth it.
- */
-async function runWithConcurrency(
-  tasks: Array<() => Promise<void>>,
-  maxConcurrent: number,
-): Promise<void> {
-  for (let i = 0; i < tasks.length; i += maxConcurrent) {
-    const batch = tasks.slice(i, i + maxConcurrent);
-    await Promise.all(batch.map((task) => task()));
-  }
-}

extensions/slack/npm-shrinkwrap.json

@@ -11,6 +11,7 @@
         "@slack/bolt": "4.7.3",
         "@slack/types": "2.21.1",
         "@slack/web-api": "7.18.0",
+        "p-map": "7.0.5",
         "typebox": "1.3.3",
         "ws": "8.21.0",
         "zod": "4.4.3"
@@ -982,6 +983,18 @@
         "node": ">=4"
       }
     },
+    "node_modules/p-map": {
+      "version": "7.0.5",
+      "resolved": "https://registry.npmjs.org/p-map/-/p-map-7.0.5.tgz",
+      "integrity": "sha512-e8vJF4XdVkzqqSHguEMz41mQO1wKwxKm5ENrUJQUu9kLDCtn83cxbyHZcszr4QC5zEA7WffRRC4gsTecC7J9oA==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=18"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/p-queue": {
       "version": "6.6.2",
       "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz",

extensions/slack/package.json

@@ -11,6 +11,7 @@
     "@slack/bolt": "4.7.3",
     "@slack/types": "2.21.1",
     "@slack/web-api": "7.18.0",
+    "p-map": "7.0.5",
     "typebox": "1.3.3",
     "ws": "8.21.0",
     "zod": "4.4.3"

extensions/slack/src/monitor/media.ts

@@ -8,6 +8,7 @@ import {
   normalizeLowercaseStringOrEmpty,
   normalizeOptionalLowercaseString,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import { formatSlackFileReference } from "../file-reference.js";
 import type { SlackAttachment, SlackFile } from "../types.js";
 export { MAX_SLACK_MEDIA_FILES, type SlackMediaResult } from "./media-types.js";
@@ -300,33 +301,6 @@ function resolveForwardedAttachmentImageUrl(attachment: SlackAttachment): string
   }
 }
 
-async function mapLimit<T, R>(
-  items: T[],
-  limit: number,
-  fn: (item: T) => Promise<R>,
-): Promise<R[]> {
-  if (items.length === 0) {
-    return [];
-  }
-  const results: R[] = [];
-  results.length = items.length;
-  const pendingItems = items.entries();
-  const workerCount = Math.max(1, Math.min(limit, items.length));
-  await Promise.all(
-    Array.from({ length: workerCount }, async () => {
-      while (true) {
-        const next = pendingItems.next();
-        if (next.done) {
-          return;
-        }
-        const [idx, item] = next.value;
-        results[idx] = await fn(item);
-      }
-    }),
-  );
-  return results;
-}
-
 /**
  * Downloads all files attached to a Slack message and returns them as an array.
  * Returns `null` when no files could be downloaded.
@@ -345,9 +319,8 @@ export async function resolveSlackMedia(params: {
   const limitedFiles =
     files.length > MAX_SLACK_MEDIA_FILES ? files.slice(0, MAX_SLACK_MEDIA_FILES) : files;
 
-  const resolved = await mapLimit<SlackFile, SlackMediaResult | null>(
+  const resolved = await pMap(
     limitedFiles,
-    MAX_SLACK_MEDIA_CONCURRENCY,
     async (file) => {
       // Audio preflight keys the original event file object so admission can
       // reuse that exact download without turning this into a persistent cache.
@@ -387,6 +360,7 @@ export async function resolveSlackMedia(params: {
         abortSignal: params.abortSignal,
       }).catch(() => null);
     },
+    { concurrency: MAX_SLACK_MEDIA_CONCURRENCY, stopOnError: true },
   );
 
   const results = resolved.filter((entry): entry is SlackMediaResult => Boolean(entry));

npm-shrinkwrap.json

@@ -50,6 +50,8 @@
         "ms": "2.1.3",
         "node-edge-tts": "1.2.10",
         "openai": "6.45.0",
+        "p-limit": "7.3.0",
+        "p-map": "7.0.5",
         "partial-json": "0.1.7",
         "playwright-core": "1.61.1",
         "pretty-ms": "9.3.0",
@@ -2426,15 +2428,15 @@
       }
     },
     "node_modules/p-limit": {
-      "version": "2.3.0",
-      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-2.3.0.tgz",
-      "integrity": "sha512-//88mFWSJx8lxCzwdAABTJL2MyWB12+eIY7MDL2SqLmAkeKU9qxRvWuSyTjm3FUmpBEMuFfckAIqEaVGUDxb6w==",
+      "version": "7.3.0",
+      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.3.0.tgz",
+      "integrity": "sha512-7cIXg/Z0M5WZRblrsOla88S4wAK+zOQQWeBYfV3qJuJXMr+LnbYjaadrFaS0JILfEDPVqHyKnZ1Z/1d6J9VVUw==",
       "license": "MIT",
       "dependencies": {
-        "p-try": "^2.0.0"
+        "yocto-queue": "^1.2.1"
       },
       "engines": {
-        "node": ">=6"
+        "node": ">=20"
       },
       "funding": {
         "url": "https://github.com/sponsors/sindresorhus"
@@ -2452,6 +2454,33 @@
         "node": ">=8"
       }
     },
+    "node_modules/p-locate/node_modules/p-limit": {
+      "version": "2.3.0",
+      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-2.3.0.tgz",
+      "integrity": "sha512-//88mFWSJx8lxCzwdAABTJL2MyWB12+eIY7MDL2SqLmAkeKU9qxRvWuSyTjm3FUmpBEMuFfckAIqEaVGUDxb6w==",
+      "license": "MIT",
+      "dependencies": {
+        "p-try": "^2.0.0"
+      },
+      "engines": {
+        "node": ">=6"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
+    "node_modules/p-map": {
+      "version": "7.0.5",
+      "resolved": "https://registry.npmjs.org/p-map/-/p-map-7.0.5.tgz",
+      "integrity": "sha512-e8vJF4XdVkzqqSHguEMz41mQO1wKwxKm5ENrUJQUu9kLDCtn83cxbyHZcszr4QC5zEA7WffRRC4gsTecC7J9oA==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=18"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/p-retry": {
       "version": "4.6.2",
       "resolved": "https://registry.npmjs.org/p-retry/-/p-retry-4.6.2.tgz",
@@ -3624,6 +3653,18 @@
         "node": ">=12"
       }
     },
+    "node_modules/yocto-queue": {
+      "version": "1.2.2",
+      "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz",
+      "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=12.20"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/zod": {
       "version": "4.4.3",
       "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz",

package.json

@@ -2045,6 +2045,8 @@
     "node-edge-tts": "1.2.10",
     "openai": "6.45.0",
     "partial-json": "0.1.7",
+    "p-limit": "7.3.0",
+    "p-map": "7.0.5",
     "playwright-core": "1.61.1",
     "pretty-ms": "9.3.0",
     "proper-lockfile": "4.1.2",

packages/memory-host-sdk/package.json

@@ -5,7 +5,8 @@
   "type": "module",
   "dependencies": {
     "@openclaw/normalization-core": "workspace:*",
-    "@openclaw/retry": "workspace:*"
+    "@openclaw/retry": "workspace:*",
+    "p-map": "7.0.5"
   },
   "exports": {
     "./runtime-core": "./src/runtime-core.ts",

packages/memory-host-sdk/src/host/backend-config.ts

@@ -63,14 +63,14 @@ export type ResolvedMemoryBackendConfig = {
   qmd?: ResolvedQmdConfig;
 };
 
-export type ResolvedQmdCollection = {
+/** @public */ export type ResolvedQmdCollection = {
   name: string;
   path: string;
   pattern: string;
   kind: "memory" | "custom" | "sessions";
 };
 
-export type ResolvedQmdUpdateConfig = {
+/** @public */ export type ResolvedQmdUpdateConfig = {
   intervalMs: number;
   debounceMs: number;
   onBoot: boolean;
@@ -83,14 +83,14 @@ export type ResolvedQmdUpdateConfig = {
   embedTimeoutMs: number;
 };
 
-export type ResolvedQmdLimitsConfig = {
+/** @public */ export type ResolvedQmdLimitsConfig = {
   maxResults: number;
   maxSnippetChars: number;
   maxInjectedChars: number;
   timeoutMs: number;
 };
 
-export type ResolvedQmdSessionConfig = {
+/** @public */ export type ResolvedQmdSessionConfig = {
   enabled: boolean;
   exportDir?: string;
   retentionDays?: number;

packages/memory-host-sdk/src/host/concurrency.ts

@@ -0,0 +1,32 @@
+// Memory Host SDK concurrency helpers preserve stop-and-drain semantics.
+import pMap from "p-map";
+
+/** Run tasks with bounded concurrency, stopping admission and draining active work on failure. */
+export async function runWithConcurrency<T>(
+  tasks: Array<() => Promise<T>>,
+  limit: number,
+): Promise<T[]> {
+  const inFlight = new Set<Promise<T>>();
+  try {
+    return await pMap(
+      tasks,
+      (task) => {
+        const run = Promise.resolve().then(task);
+        inFlight.add(run);
+        void run.then(
+          () => inFlight.delete(run),
+          () => inFlight.delete(run),
+        );
+        return run;
+      },
+      {
+        concurrency: Math.max(1, Math.floor(limit)),
+        stopOnError: true,
+      },
+    );
+  } catch (error) {
+    // p-map stops dequeuing on error, but active memory writes must drain before callers recover.
+    await Promise.allSettled(inFlight);
+    throw error;
+  }
+}

packages/memory-host-sdk/src/host/embeddings.types.ts

@@ -19,11 +19,11 @@ export type EmbeddingProviderCallOptions = {
   signal?: AbortSignal;
 };
 
-export type EmbeddingProviderId = string;
-export type EmbeddingProviderRequest = string;
-export type EmbeddingProviderFallback = string;
+/** @public */ export type EmbeddingProviderId = string;
+/** @public */ export type EmbeddingProviderRequest = string;
+/** @public */ export type EmbeddingProviderFallback = string;
 
-export type GeminiTaskType =
+/** @public */ export type GeminiTaskType =
   | "RETRIEVAL_QUERY"
   | "RETRIEVAL_DOCUMENT"
   | "SEMANTIC_SIMILARITY"

packages/memory-host-sdk/src/host/internal.test.ts

@@ -14,6 +14,7 @@ import {
   listMemoryFiles,
   normalizeExtraMemoryPaths,
   remapChunkLines,
+  runWithConcurrency,
 } from "./internal.js";
 import { normalizeMemoryMultimodalSettings, type MemoryMultimodalSettings } from "./multimodal.js";
 
@@ -78,6 +79,46 @@ const multimodal: MemoryMultimodalSettings = normalizeMemoryMultimodalSettings({
 describe("memory host SDK package internals", () => {
   const getTmpDir = setupTempDirLifecycle("memory-package-");
 
+  it("drains in-flight work before propagating a concurrency failure", async () => {
+    const failure = new Error("embedding failed");
+    let releaseTask!: () => void;
+    const release = new Promise<void>((resolve) => {
+      releaseTask = resolve;
+    });
+    const started: number[] = [];
+    const completed: number[] = [];
+    const run = runWithConcurrency(
+      [
+        async () => {
+          started.push(0);
+          await release;
+          completed.push(0);
+          return 0;
+        },
+        async () => {
+          started.push(1);
+          throw failure;
+        },
+        async () => {
+          started.push(2);
+          return 2;
+        },
+      ],
+      2,
+    );
+
+    await vi.waitFor(() => expect(started).toEqual([0, 1]));
+    const onSettled = vi.fn();
+    void run.then(onSettled, onSettled);
+    await Promise.resolve();
+    expect(onSettled).not.toHaveBeenCalled();
+
+    releaseTask();
+    await expect(run).rejects.toBe(failure);
+    expect(completed).toEqual([0]);
+    expect(started).toEqual([0, 1]);
+  });
+
   it("propagates directory creation failures", () => {
     const mkdirError = new Error("disk full");
     const targetDir = path.join(getTmpDir(), "blocked");

packages/memory-host-sdk/src/host/internal.ts

@@ -4,6 +4,7 @@ import fsSync from "node:fs";
 import fs from "node:fs/promises";
 import { homedir } from "node:os";
 import path from "node:path";
+import { runWithConcurrency as runWithConcurrencyImpl } from "./concurrency.js";
 import { CANONICAL_ROOT_MEMORY_FILENAME } from "./config-utils.js";
 import { estimateStructuredEmbeddingInputBytes } from "./embedding-input-limits.js";
 import { buildTextEmbeddingInput, type EmbeddingInput } from "./embedding-inputs.js";
@@ -24,7 +25,6 @@ import {
   CHARS_PER_TOKEN_ESTIMATE,
   detectMime,
   estimateStringChars,
-  runTasksWithConcurrency,
   truncateUtf16Safe,
 } from "./openclaw-runtime-io.js";
 import {
@@ -539,17 +539,6 @@ export function cosineSimilarity(a: number[], b: number[]): number {
   return dot / (Math.sqrt(normA) * Math.sqrt(normB));
 }
 
-export async function runWithConcurrency<T>(
-  tasks: Array<() => Promise<T>>,
-  limit: number,
-): Promise<T[]> {
-  const { results, firstError, hasError } = await runTasksWithConcurrency({
-    tasks,
-    limit,
-    errorMode: "stop",
-  });
-  if (hasError) {
-    throw firstError;
-  }
-  return results;
+export function runWithConcurrency<T>(tasks: Array<() => Promise<T>>, limit: number): Promise<T[]> {
+  return runWithConcurrencyImpl(tasks, limit);
 }

packages/memory-host-sdk/src/host/types.ts

@@ -51,7 +51,7 @@ export type MemorySyncParams = {
   progress?: (update: MemorySyncProgressUpdate) => void;
 };
 
-/** Runtime backend/mode diagnostics for memory search. */
+/** @public Runtime backend/mode diagnostics for memory search. */
 export type MemorySearchRuntimeQmdCollectionValidationDebug = {
   cacheState?: "hit" | "miss" | "write" | "bypass-force" | "error";
   elapsedMs: number;
@@ -60,20 +60,20 @@ export type MemorySearchRuntimeQmdCollectionValidationDebug = {
   showCalls?: number;
 };
 
-export type MemorySearchRuntimeQmdMultiCollectionProbeDebug = {
+/** @public */ export type MemorySearchRuntimeQmdMultiCollectionProbeDebug = {
   cacheState?: "hit" | "miss" | "write" | "error";
   elapsedMs: number;
   supported: boolean;
 };
 
-export type MemorySearchRuntimeQmdSearchPlanDebug = {
+/** @public */ export type MemorySearchRuntimeQmdSearchPlanDebug = {
   command?: "query" | "search" | "vsearch";
   collectionCount?: number;
   groupCount?: number;
   sources?: MemorySource[];
 };
 
-export type MemorySearchRuntimeQmdDebug = {
+/** @public */ export type MemorySearchRuntimeQmdDebug = {
   collectionValidation?: MemorySearchRuntimeQmdCollectionValidationDebug;
   multiCollectionProbe?: MemorySearchRuntimeQmdMultiCollectionProbeDebug;
   searchPlan?: MemorySearchRuntimeQmdSearchPlanDebug;

pnpm-lock.yaml

@@ -166,6 +166,12 @@ importers:
       openai:
         specifier: 6.45.0
         version: 6.45.0(@aws-sdk/credential-provider-node@3.972.61)(@smithy/hash-node@4.4.5)(@smithy/signature-v4@5.6.1)(ws@8.21.0)(zod@4.4.3)
+      p-limit:
+        specifier: 7.3.0
+        version: 7.3.0
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       partial-json:
         specifier: 0.1.7
         version: 0.1.7
@@ -711,6 +717,9 @@ importers:
       libopus-wasm:
         specifier: 0.2.0
         version: 0.2.0
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       typebox:
         specifier: 1.3.3
         version: 1.3.3
@@ -1345,6 +1354,9 @@ importers:
 
   extensions/openshell:
     dependencies:
+      p-limit:
+        specifier: 7.3.0
+        version: 7.3.0
       zod:
         specifier: 4.4.3
         version: 4.4.3
@@ -1411,6 +1423,12 @@ importers:
       '@modelcontextprotocol/sdk':
         specifier: 1.29.0
         version: 1.29.0(zod@4.4.3)
+      p-limit:
+        specifier: 7.3.0
+        version: 7.3.0
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       playwright-core:
         specifier: 1.61.1
         version: 1.61.1
@@ -1479,6 +1497,9 @@ importers:
       mpg123-decoder:
         specifier: 1.0.3
         version: 1.0.3
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       pretty-ms:
         specifier: 9.3.0
         version: 9.3.0
@@ -1566,6 +1587,9 @@ importers:
       '@slack/web-api':
         specifier: 7.18.0
         version: 7.18.0
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       typebox:
         specifier: 1.3.3
         version: 1.3.3
@@ -2019,6 +2043,9 @@ importers:
       '@openclaw/retry':
         specifier: workspace:*
         version: link:../retry
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
 
   packages/model-catalog-core: {}
 
@@ -6751,6 +6778,10 @@ packages:
     resolution: {integrity: sha512-//88mFWSJx8lxCzwdAABTJL2MyWB12+eIY7MDL2SqLmAkeKU9qxRvWuSyTjm3FUmpBEMuFfckAIqEaVGUDxb6w==}
     engines: {node: '>=6'}
 
+  p-limit@7.3.0:
+    resolution: {integrity: sha512-7cIXg/Z0M5WZRblrsOla88S4wAK+zOQQWeBYfV3qJuJXMr+LnbYjaadrFaS0JILfEDPVqHyKnZ1Z/1d6J9VVUw==}
+    engines: {node: '>=20'}
+
   p-locate@4.1.0:
     resolution: {integrity: sha512-R79ZZ/0wAxKGu3oYMlz8jy/kbhsNrS7SKZ7PxEHBgJ5+F2mtFW2fK2cOtBh1cHYkQsbzFV7I+EoRKe6Yt0oK7A==}
     engines: {node: '>=8'}
@@ -7982,6 +8013,10 @@ packages:
     resolution: {integrity: sha512-GZtjxm/J/4TSxuL3FNYjCmLktBTnIw/rVmKSIyKeYAZpmJB2ig9VauCC5xsa82GNKVKDAqpOn3KVzNt0zmrU0g==}
     engines: {node: '>=12'}
 
+  yocto-queue@1.2.2:
+    resolution: {integrity: sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==}
+    engines: {node: '>=12.20'}
+
   yoctocolors@2.1.2:
     resolution: {integrity: sha512-CzhO+pFNo8ajLM2d2IW/R93ipy99LWjtwblvC1RsoSUMZgyLbYFr221TnSNT7GjGdYui6P459mw9JH/g/zW2ug==}
     engines: {node: '>=18'}
@@ -13211,6 +13246,10 @@ snapshots:
     dependencies:
       p-try: 2.2.0
 
+  p-limit@7.3.0:
+    dependencies:
+      yocto-queue: 1.2.2
+
   p-locate@4.1.0:
     dependencies:
       p-limit: 2.3.0
@@ -14475,6 +14514,8 @@ snapshots:
       y18n: 5.0.8
       yargs-parser: 21.1.1
 
+  yocto-queue@1.2.2: {}
+
   yoctocolors@2.1.2:
     optional: true
 

scripts/check-architecture-smells.mjs

@@ -4,6 +4,7 @@
 import { promises as fs } from "node:fs";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
+import pMap from "p-map";
 import { BUNDLED_PLUGIN_PATH_PREFIX } from "./lib/bundled-plugin-paths.mjs";
 import {
   collectModuleReferencesFromSource,
@@ -11,7 +12,6 @@ import {
   resolveRepoSpecifier,
   writeLine,
 } from "./lib/guard-inventory-utils.mjs";
-import { mapWithConcurrency } from "./lib/source-file-scan-cache.mjs";
 import {
   collectTypeScriptFilesFromRoots,
   resolveSourceRoots,
@@ -175,13 +175,17 @@ export async function collectArchitectureSmells() {
       const files = (await collectTypeScriptFilesFromRoots(scanRoots)).toSorted((left, right) =>
         normalizeRepoPath(repoRoot, left).localeCompare(normalizeRepoPath(repoRoot, right)),
       );
-      const entriesByFile = await mapWithConcurrency(files, undefined, async (filePath) => {
-        const source = await fs.readFile(filePath, "utf8");
-        const entries = scanPluginSdkExtensionFacadeSmells(source, filePath);
-        entries.push(...scanRuntimeTypeImplementationSmells(source, filePath));
-        entries.push(...scanRuntimeServiceLocatorSmells(source, filePath));
-        return entries;
-      });
+      const entriesByFile = await pMap(
+        files,
+        async (filePath) => {
+          const source = await fs.readFile(filePath, "utf8");
+          const entries = scanPluginSdkExtensionFacadeSmells(source, filePath);
+          entries.push(...scanRuntimeTypeImplementationSmells(source, filePath));
+          entries.push(...scanRuntimeServiceLocatorSmells(source, filePath));
+          return entries;
+        },
+        { concurrency: 32, stopOnError: true },
+      );
       return entriesByFile.flat().toSorted(compareEntries);
     })();
     try {

scripts/check-temp-path-guardrails.ts

@@ -2,6 +2,7 @@
 import { execFileSync } from "node:child_process";
 import fs from "node:fs/promises";
 import path from "node:path";
+import pMap from "p-map";
 
 type QuoteChar = "'" | '"' | "`";
 
@@ -225,42 +226,22 @@ async function readRuntimeSourceFiles(
   repoRoot: string,
   absolutePaths: string[],
 ): Promise<RuntimeSourceGuardrailFile[]> {
-  const output: Array<RuntimeSourceGuardrailFile | undefined> = Array.from({
-    length: absolutePaths.length,
-  });
-  let nextIndex = 0;
-
-  const worker = async () => {
-    for (;;) {
-      const index = nextIndex;
-      nextIndex += 1;
-      if (index >= absolutePaths.length) {
-        return;
-      }
-      const absolutePath = absolutePaths[index];
-      if (!absolutePath) {
-        continue;
-      }
-      let source: string;
+  const output = await pMap(
+    absolutePaths,
+    async (absolutePath): Promise<RuntimeSourceGuardrailFile | null> => {
       try {
-        source = await fs.readFile(absolutePath, "utf8");
+        return {
+          relativePath: path.relative(repoRoot, absolutePath),
+          source: await fs.readFile(absolutePath, "utf8"),
+        };
       } catch {
         // File tracked by git but deleted on disk (e.g. pending deletion).
-        continue;
+        return null;
       }
-      output[index] = {
-        relativePath: path.relative(repoRoot, absolutePath),
-        source,
-      };
-    }
-  };
-
-  const workers = Array.from(
-    { length: Math.min(FILE_READ_CONCURRENCY, Math.max(1, absolutePaths.length)) },
-    () => worker(),
+    },
+    { concurrency: FILE_READ_CONCURRENCY, stopOnError: false },
   );
-  await Promise.all(workers);
-  return output.filter((entry): entry is RuntimeSourceGuardrailFile => entry !== undefined);
+  return output.filter((entry): entry is RuntimeSourceGuardrailFile => entry !== null);
 }
 
 async function main() {

scripts/deadcode-exports.baseline.mjs

@@ -2931,12 +2931,9 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [
   "src/infra/sqlite-snapshot.ts: PublishVerifiedSqliteFileOptions",
   "src/infra/sqlite-snapshot.ts: SqliteFileContent",
   "src/infra/sqlite-snapshot.ts: VerifiedSqliteSnapshot",
-  "src/infra/sqlite-wal.ts: configureSqliteWalMaintenance",
   "src/infra/sqlite-wal.ts: DEFAULT_SQLITE_WAL_AUTOCHECKPOINT_PAGES",
   "src/infra/sqlite-wal.ts: DEFAULT_SQLITE_WAL_CHECKPOINT_INTERVAL_MS",
   "src/infra/sqlite-wal.ts: enableIncrementalAutoVacuumForFreshDatabase",
-  "src/infra/sqlite-wal.ts: SqliteConnectionPragmaOptions",
-  "src/infra/sqlite-wal.ts: SqliteWalMaintenanceOptions",
   "src/infra/stale-lock-file.ts: LockFileOwnerPayload",
   "src/infra/state-migrations.debug-proxy.ts: LegacyDebugProxyCaptureDetection",
   "src/infra/state-migrations.fs.ts: isLegacyWhatsAppAuthFile",
@@ -3011,8 +3008,6 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [
   "src/media-understanding/runner.entries.ts: formatMissingProviderHint",
   "src/media/input-files.ts: fetchWithGuard",
   "src/media/parse.ts: SplitMediaFromOutputOptions",
-  "src/memory/root-memory-files.ts: resolveCanonicalRootMemoryFile",
-  "src/memory/root-memory-files.ts: shouldSkipRootMemoryAuxiliaryPath",
   "src/music-generation/capabilities.ts: resolveMusicGenerationMode",
   "src/music-generation/runtime-types.ts: ListRuntimeMusicGenerationProvidersParams",
   "src/music-generation/runtime-types.ts: RuntimeMusicGenerationProvider",

scripts/generate-npm-shrinkwrap.d.mts

@@ -48,11 +48,6 @@ export function resolvePackageDirs(args: string[]): {
   jobs: number;
   packageDirs: unknown[];
 };
-export function runBoundedTasks<Item, Result>(
-  items: Item[],
-  jobs: number,
-  runTask: (item: Item) => Promise<Result>,
-): Promise<Result[]>;
 export function resolveShrinkwrapJobs(
   rawValue: unknown,
   env?: NodeJS.ProcessEnv,

scripts/generate-npm-shrinkwrap.mjs

@@ -7,6 +7,7 @@ import { tmpdir } from "node:os";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { isMainThread, parentPort, Worker, workerData } from "node:worker_threads";
+import pMap from "p-map";
 import { parse as parseYaml } from "yaml";
 import { listChangedPathsFromGit, listStagedChangedPaths } from "./changed-lanes.mjs";
 import { resolveNpmRunner } from "./npm-runner.mjs";
@@ -1347,21 +1348,6 @@ function updateOrCheckPackage(packageDir, check, changedPaths = []) {
   return `${label}: npm-shrinkwrap.json is current.`;
 }
 
-/** @internal Directly tested script implementation detail. */
-export async function runBoundedTasks(items, jobs, runTask) {
-  const results = Array.from({ length: items.length });
-  let nextIndex = 0;
-  const workers = Array.from({ length: Math.min(jobs, items.length) }, async () => {
-    while (nextIndex < items.length) {
-      const index = nextIndex;
-      nextIndex += 1;
-      results[index] = await runTask(items[index], index);
-    }
-  });
-  await Promise.all(workers);
-  return results;
-}
-
 /** @internal Directly tested script implementation detail. */
 export function resolveShrinkwrapJobs(
   rawValue,
@@ -1407,19 +1393,23 @@ async function runPackageWorker(packageDir, check, changedPaths) {
 }
 
 async function updateOrCheckPackages({ check, changedPaths, jobs, packageDirs }) {
-  const outcomes = await runBoundedTasks(packageDirs, jobs, async (packageDir) => {
-    try {
-      const output =
-        jobs === 1
-          ? updateOrCheckPackage(packageDir, check, changedPaths)
-          : await runPackageWorker(packageDir, check, changedPaths);
-      return { output };
-    } catch (error) {
-      return {
-        error: error instanceof Error ? error.message : String(error),
-      };
-    }
-  });
+  const outcomes = await pMap(
+    packageDirs,
+    async (packageDir) => {
+      try {
+        const output =
+          jobs === 1
+            ? updateOrCheckPackage(packageDir, check, changedPaths)
+            : await runPackageWorker(packageDir, check, changedPaths);
+        return { output };
+      } catch (error) {
+        return {
+          error: error instanceof Error ? error.message : String(error),
+        };
+      }
+    },
+    { concurrency: jobs, stopOnError: false },
+  );
 
   const errors = [];
   for (const outcome of outcomes) {

scripts/lib/extension-import-boundary-checker.mjs

@@ -1,5 +1,6 @@
 // Creates reusable import-boundary guards for bundled extension source trees.
 import { promises as fs } from "node:fs";
+import pMap from "p-map";
 import { BUNDLED_PLUGIN_PATH_PREFIX } from "./bundled-plugin-paths.mjs";
 import {
   collectModuleReferencesFromSource,
@@ -9,7 +10,6 @@ import {
   resolveRepoSpecifier,
   writeLine,
 } from "./guard-inventory-utils.mjs";
-import { mapWithConcurrency } from "./source-file-scan-cache.mjs";
 import {
   collectTypeScriptFilesFromRoots,
   resolveRepoRoot,
@@ -101,21 +101,25 @@ export function createExtensionImportBoundaryChecker(params) {
       .toSorted((left, right) =>
         normalizeRepoPath(repoRoot, left).localeCompare(normalizeRepoPath(repoRoot, right)),
       );
-    const entriesByFile = await mapWithConcurrency(files, undefined, async (filePath) => {
-      const source = await readBoundedSourceFile(filePath, maxSourceBytes);
-      if (
-        params.skipSourcesWithoutBundledPluginPrefix &&
-        !source.includes(BUNDLED_PLUGIN_PATH_PREFIX)
-      ) {
-        return [];
-      }
-      return scanImportBoundaryViolations(
-        source,
-        filePath,
-        params.boundaryLabel,
-        params.allowResolvedPath,
-      );
-    });
+    const entriesByFile = await pMap(
+      files,
+      async (filePath) => {
+        const source = await readBoundedSourceFile(filePath, maxSourceBytes);
+        if (
+          params.skipSourcesWithoutBundledPluginPrefix &&
+          !source.includes(BUNDLED_PLUGIN_PATH_PREFIX)
+        ) {
+          return [];
+        }
+        return scanImportBoundaryViolations(
+          source,
+          filePath,
+          params.boundaryLabel,
+          params.allowResolvedPath,
+        );
+      },
+      { concurrency: 32, stopOnError: true },
+    );
     const inventory = entriesByFile.flat();
     return inventory.toSorted(compareEntries);
   });

scripts/lib/source-file-scan-cache.d.mts

@@ -1,9 +1,3 @@
-export function mapWithConcurrency<Item, Result>(
-  items: Item[],
-  concurrency: number,
-  mapper: (item: Item, index: number) => Promise<Result>,
-): Promise<Result[]>;
-
 export function collectSourceFileContents(params: {
   repoRoot: string;
   scanRoots: string[];

scripts/lib/source-file-scan-cache.mjs

@@ -1,6 +1,7 @@
 // Caches source file discovery and bounded-concurrency reads for guard scripts.
 import { promises as fs } from "node:fs";
 import path from "node:path";
+import pMap from "p-map";
 
 const DEFAULT_SOURCE_FILE_READ_CONCURRENCY = 32;
 const DEFAULT_SOURCE_FILE_MAX_BYTES = 2 * 1024 * 1024;
@@ -73,29 +74,6 @@ async function readBoundedSourceFile(params, filePath, readFile, statFile, maxFi
   };
 }
 
-/**
- * Maps items with bounded worker concurrency while preserving input order.
- */
-export async function mapWithConcurrency(items, concurrency, mapper) {
-  const out = Array.from({ length: items.length });
-  const workerCount = Math.min(normalizeConcurrency(concurrency), items.length);
-  let nextIndex = 0;
-
-  async function worker() {
-    for (;;) {
-      const index = nextIndex;
-      nextIndex += 1;
-      if (index >= items.length) {
-        return;
-      }
-      out[index] = await mapper(items[index], index);
-    }
-  }
-
-  await Promise.all(Array.from({ length: workerCount }, () => worker()));
-  return out;
-}
-
 /**
  * Collects sorted source files and cached contents for configured scan roots.
  */
@@ -133,8 +111,13 @@ export async function collectSourceFileContents(params) {
     const readFile = params.readFile ?? fs.readFile;
     const statFile = params.statFile ?? fs.stat;
     const maxFileBytes = normalizeMaxFileBytes(params.maxFileBytes);
-    return await mapWithConcurrency(files, params.maxConcurrentReads, async (filePath) =>
-      readBoundedSourceFile(params, filePath, readFile, statFile, maxFileBytes),
+    return await pMap(
+      files,
+      async (filePath) => readBoundedSourceFile(params, filePath, readFile, statFile, maxFileBytes),
+      {
+        concurrency: normalizeConcurrency(params.maxConcurrentReads),
+        stopOnError: true,
+      },
     );
   })();
 

scripts/native-app-i18n.ts

@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
 import { mkdir, readdir, readFile, writeFile } from "node:fs/promises";
 import path from "node:path";
 import { fileURLToPath, pathToFileURL } from "node:url";
+import pMap from "p-map";
 import { expectDefined } from "../packages/normalization-core/src/expect.js";
 import { translateNativeEntries } from "./control-ui-i18n.ts";
 
@@ -1185,31 +1186,6 @@ async function readNativeI18nInventory(): Promise<{
   return { entries: inventory.entries as NativeI18nEntry[], raw };
 }
 
-async function mapWithConcurrency<T, R>(
-  values: readonly T[],
-  limit: number,
-  run: (value: T) => Promise<R>,
-): Promise<R[]> {
-  const results = Array<R>(values.length);
-  let nextIndex = 0;
-  const workerCount = Math.min(limit, values.length);
-  await Promise.all(
-    Array.from({ length: workerCount }, async () => {
-      for (;;) {
-        const index = nextIndex;
-        nextIndex += 1;
-        if (index >= values.length) {
-          return;
-        }
-        results[index] = await run(
-          expectDefined(values[index], `native i18n concurrency input at index ${index}`),
-        );
-      }
-    }),
-  );
-  return results;
-}
-
 export async function collectNativeI18nEntries(
   previousEntries?: readonly NativeI18nEntry[],
 ): Promise<NativeI18nEntry[]> {
@@ -1225,14 +1201,17 @@ export async function collectNativeI18nEntries(
       surface,
     })),
   );
-  const sources = await mapWithConcurrency(
+  const sources = await pMap(
     filesByRoot.flatMap(({ files, surface }) => files.map((filePath) => ({ filePath, surface }))),
-    NATIVE_SOURCE_READ_CONCURRENCY,
     async ({ filePath, surface }) => ({
       repoPath: path.relative(ROOT, filePath).split(path.sep).join("/"),
       source: await readFile(filePath, "utf8"),
       surface,
     }),
+    {
+      concurrency: NATIVE_SOURCE_READ_CONCURRENCY,
+      stopOnError: true,
+    },
   );
   const typedSources: Array<{
     repoPath: string;

scripts/profile-extension-memory.mjs

@@ -8,6 +8,7 @@ import { existsSync, mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync
 import os from "node:os";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
+import pMap from "p-map";
 import { ensureExtensionMemoryBuild } from "./ensure-extension-memory-build.mjs";
 import { stripLeadingPackageManagerSeparator } from "./lib/arg-utils.mjs";
 import { formatErrorMessage } from "./lib/error-format.mjs";
@@ -495,15 +496,9 @@ async function main() {
           timeoutMs: options.combinedTimeoutMs,
         });
 
-    const pending = [...selectedEntries];
-    const results = [];
-
-    async function worker() {
-      while (pending.length > 0) {
-        const next = pending.shift();
-        if (next === undefined) {
-          return;
-        }
+    const results = await pMap(
+      selectedEntries,
+      async (next) => {
         const result = await runCase({
           repoRoot,
           env,
@@ -512,7 +507,7 @@ async function main() {
           body: buildImportBody([next.file], "IMPORTED"),
           timeoutMs: options.timeoutMs,
         });
-        results.push({
+        const entry = {
           dir: next.dir,
           file: next.file,
           status: result.timedOut ? "timeout" : result.code === 0 ? "ok" : "fail",
@@ -522,16 +517,14 @@ async function main() {
               ? result.maxRssMb - baseline.maxRssMb
               : null,
           stderrPreview: summarizeStderr(result.stderr),
-        });
+        };
 
         const status = result.timedOut ? "timeout" : result.code === 0 ? "ok" : "fail";
         const rss = result.maxRssMb === null ? "n/a" : `${result.maxRssMb.toFixed(1)} MB`;
         console.log(`[extension-memory] ${next.dir}: ${status} ${rss}`);
-      }
-    }
-
-    await Promise.all(
-      Array.from({ length: Math.min(options.concurrency, selectedEntries.length) }, () => worker()),
+        return entry;
+      },
+      { concurrency: options.concurrency, stopOnError: true },
     );
 
     results.sort((a, b) => a.dir.localeCompare(b.dir));

scripts/update-clawtributors.ts

@@ -2,6 +2,7 @@
 import { execFileSync, execSync } from "node:child_process";
 import { readFileSync, writeFileSync } from "node:fs";
 import { resolve } from "node:path";
+import pMap from "p-map";
 import { expectDefined } from "../packages/normalization-core/src/expect.js";
 import type { ApiContributor, Entry, MapConfig, User } from "./update-clawtributors.types.js";
 
@@ -625,41 +626,24 @@ async function filterVisibleEntries(
   entriesResult: Entry[],
   hiddenLogins: ReadonlySet<string>,
 ): Promise<Entry[]> {
-  const results = await mapConcurrent(entriesResult, 8, async (entry) => {
-    const login = entry.login ?? entry.key;
-    if (!login) {
-      return entry;
-    }
-    const normalized = normalizeLogin(login)?.toLowerCase();
-    if (normalized && hiddenLogins.has(normalized)) {
-      return null;
-    }
-    return (await isDefaultGitHubAvatar(login)) ? null : entry;
-  });
+  const results = await pMap(
+    entriesResult,
+    async (entry) => {
+      const login = entry.login ?? entry.key;
+      if (!login) {
+        return entry;
+      }
+      const normalized = normalizeLogin(login)?.toLowerCase();
+      if (normalized && hiddenLogins.has(normalized)) {
+        return null;
+      }
+      return (await isDefaultGitHubAvatar(login)) ? null : entry;
+    },
+    { concurrency: 8, stopOnError: true },
+  );
   return results.filter((entry): entry is Entry => entry !== null);
 }
 
-async function mapConcurrent<T, R>(
-  items: T[],
-  limit: number,
-  mapper: (item: T, index: number) => Promise<R>,
-): Promise<R[]> {
-  const results: R[] = [];
-  results.length = items.length;
-  let nextIndex = 0;
-  const workers = Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, async () => {
-    while (nextIndex < items.length) {
-      const index = nextIndex++;
-      results[index] = await mapper(
-        expectDefined(items[index], `clawtributor concurrency input at index ${index}`),
-        index,
-      );
-    }
-  });
-  await Promise.all(workers);
-  return results;
-}
-
 function readImageDimensions(buffer: Buffer): { width: number; height: number } | null {
   if (isPng(buffer)) {
     return readPngDimensions(buffer);

scripts/write-cli-startup-metadata.ts

@@ -4,7 +4,7 @@ import { createHash } from "node:crypto";
 import { existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from "node:fs";
 import path from "node:path";
 import { fileURLToPath, pathToFileURL } from "node:url";
-import { expectDefined } from "../packages/normalization-core/src/expect.js";
+import pMap from "p-map";
 import type { RootHelpRenderOptions } from "../src/cli/program/root-help.js";
 import type { OpenClawConfig } from "../src/config/config.js";
 import { resolveWindowsTaskkillPath } from "./lib/windows-taskkill.mjs";
@@ -348,32 +348,6 @@ function createIsolatedRootHelpRenderContext(
   return { config, env };
 }
 
-async function mapWithConcurrency<T, R>(
-  values: readonly T[],
-  limit: number,
-  run: (value: T) => Promise<R>,
-): Promise<R[]> {
-  const results: R[] = [];
-  results.length = values.length;
-  let nextIndex = 0;
-  const workerCount = Math.min(Math.max(1, limit), values.length);
-  await Promise.all(
-    Array.from({ length: workerCount }, async () => {
-      for (;;) {
-        const index = nextIndex;
-        nextIndex += 1;
-        if (index >= values.length) {
-          return;
-        }
-        results[index] = await run(
-          expectDefined(values[index], `CLI metadata concurrency input at index ${index}`),
-        );
-      }
-    }),
-  );
-  return results;
-}
-
 async function spawnText(
   args: string[],
   options: {
@@ -757,10 +731,13 @@ async function renderSourceCommandHelpTextRecord(
   commands: readonly SourceCommandHelpCommand[],
   renderContext: RootHelpRenderContext = createIsolatedRootHelpRenderContext(),
 ): Promise<SourceCommandHelpText> {
-  const helpTexts = await mapWithConcurrency(
+  const helpTexts = await pMap(
     commands,
-    COMMAND_HELP_RENDER_CONCURRENCY,
     async (commandName) => await renderSourceCommandHelpText(commandName, renderContext),
+    {
+      concurrency: COMMAND_HELP_RENDER_CONCURRENCY,
+      stopOnError: true,
+    },
   );
   return Object.fromEntries(
     commands.map((commandName, index) => [commandName, helpTexts[index]]),
@@ -941,7 +918,6 @@ function hasAllPrecomputedSubcommandHelpText(value: unknown): boolean {
 }
 
 export const testing = {
-  mapWithConcurrency,
   signalCliStartupMetadataProcessTree,
   spawnText,
 };

src/agents/model-scan.ts

@@ -13,6 +13,7 @@ import {
   normalizeStringEntries,
   uniqueStrings,
 } from "@openclaw/normalization-core/string-normalization";
+import pMap from "p-map";
 import { Type } from "typebox";
 import { formatErrorMessage } from "../infra/errors.js";
 /**
@@ -411,43 +412,6 @@ function buildOpenRouterScanResult(params: {
   };
 }
 
-async function mapWithConcurrency<T, R>(
-  items: T[],
-  concurrency: number,
-  fn: (item: T, index: number) => Promise<R>,
-  opts?: { onProgress?: (completed: number, total: number) => void },
-): Promise<R[]> {
-  const limit = Math.max(1, Math.floor(concurrency));
-  const results: R[] = Array.from({ length: items.length }, () => undefined as R);
-  let nextIndex = 0;
-  let completed = 0;
-
-  const worker = async () => {
-    while (true) {
-      const current = nextIndex;
-      nextIndex += 1;
-      if (current >= items.length) {
-        return;
-      }
-      const item = items.at(current);
-      if (item === undefined) {
-        return;
-      }
-      results[current] = await fn(item, current);
-      completed += 1;
-      opts?.onProgress?.(completed, items.length);
-    }
-  };
-
-  if (items.length === 0) {
-    opts?.onProgress?.(0, 0);
-    return results;
-  }
-
-  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, () => worker()));
-  return results;
-}
-
 export async function scanOpenRouterModels(
   options: OpenRouterScanOptions = {},
 ): Promise<ModelScanResult[]> {
@@ -514,49 +478,46 @@ export async function scanOpenRouterModels(
     total: filtered.length,
   });
 
-  return mapWithConcurrency(
+  let completed = 0;
+  return pMap(
     filtered,
-    concurrency,
     async (entry) => {
       const isFree = isFreeOpenRouterModel(entry);
+      let result: ModelScanResult;
       if (!probe) {
-        return buildOpenRouterScanResult({
+        result = buildOpenRouterScanResult({
           entry,
           isFree,
           tool: { ok: false, latencyMs: null, skipped: true },
           image: { ok: false, latencyMs: null, skipped: true },
         });
+      } else {
+        const model: OpenAIModel = {
+          ...baseModel,
+          id: entry.id,
+          name: entry.name || entry.id,
+          contextWindow: entry.contextLength ?? baseModel.contextWindow,
+          maxTokens: entry.maxCompletionTokens ?? baseModel.maxTokens,
+          input: parseModality(entry.modality),
+          reasoning: baseModel.reasoning,
+        };
+
+        const toolResult = await probeTool(model, apiKey, timeoutMs);
+        const imageResult = model.input?.includes("image")
+          ? await probeImage(ensureImageInput(model), apiKey, timeoutMs)
+          : { ok: false, latencyMs: null, skipped: true };
+
+        result = buildOpenRouterScanResult({
+          entry,
+          isFree,
+          tool: toolResult,
+          image: imageResult,
+        });
       }
-
-      const model: OpenAIModel = {
-        ...baseModel,
-        id: entry.id,
-        name: entry.name || entry.id,
-        contextWindow: entry.contextLength ?? baseModel.contextWindow,
-        maxTokens: entry.maxCompletionTokens ?? baseModel.maxTokens,
-        input: parseModality(entry.modality),
-        reasoning: baseModel.reasoning,
-      };
-
-      const toolResult = await probeTool(model, apiKey, timeoutMs);
-      const imageResult = model.input?.includes("image")
-        ? await probeImage(ensureImageInput(model), apiKey, timeoutMs)
-        : { ok: false, latencyMs: null, skipped: true };
-
-      return buildOpenRouterScanResult({
-        entry,
-        isFree,
-        tool: toolResult,
-        image: imageResult,
-      });
-    },
-    {
-      onProgress: (completed, total) =>
-        options.onProgress?.({
-          phase: "probe",
-          completed,
-          total,
-        }),
+      completed += 1;
+      options.onProgress?.({ phase: "probe", completed, total: filtered.length });
+      return result;
     },
+    { concurrency, stopOnError: true },
   );
 }

src/agents/sessions/session-manager.ts

@@ -19,6 +19,7 @@ import {
 import { readdir, readFile, stat } from "node:fs/promises";
 import { join, resolve } from "node:path";
 import { isProxy } from "node:util/types";
+import pMap from "p-map";
 import {
   appendTranscriptEventSync,
   appendTranscriptMessageSync,
@@ -1449,46 +1450,6 @@ export type SessionListProgress = (loaded: number, total: number) => void;
 
 const MAX_CONCURRENT_SESSION_INFO_LOADS = 10;
 
-async function buildSessionInfosWithConcurrency(
-  files: string[],
-  onLoaded: () => void,
-): Promise<(SessionInfo | null)[]> {
-  const results: (SessionInfo | null)[] = Array.from({ length: files.length }, () => null);
-  const inFlight = new Set<Promise<void>>();
-  let nextIndex = 0;
-
-  const startNext = (): void => {
-    const index = nextIndex++;
-    const file = files[index];
-    if (!file) {
-      return;
-    }
-    const task: Promise<void> = buildSessionInfo(file)
-      .then((info) => {
-        results[index] = info;
-      })
-      .catch(() => {
-        results[index] = null;
-      })
-      .finally(() => {
-        inFlight.delete(task);
-        onLoaded();
-      });
-    inFlight.add(task);
-  };
-
-  while (nextIndex < files.length || inFlight.size > 0) {
-    while (nextIndex < files.length && inFlight.size < MAX_CONCURRENT_SESSION_INFO_LOADS) {
-      startNext();
-    }
-    if (inFlight.size > 0) {
-      await Promise.race(inFlight);
-    }
-  }
-
-  return results;
-}
-
 async function listSessionsFromDir(
   dir: string,
   onProgress?: SessionListProgress,
@@ -1507,10 +1468,20 @@ async function listSessionsFromDir(
     const total = progressTotal ?? files.length;
 
     let loaded = 0;
-    const results = await buildSessionInfosWithConcurrency(files, () => {
-      loaded++;
-      onProgress?.(progressOffset + loaded, total);
-    });
+    const results = await pMap(
+      files,
+      async (file) => {
+        try {
+          return await buildSessionInfo(file);
+        } catch {
+          return null;
+        } finally {
+          loaded++;
+          onProgress?.(progressOffset + loaded, total);
+        }
+      },
+      { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false },
+    );
     for (const info of results) {
       if (info && sessionInfoMatchesCwd(info, cwd)) {
         sessions.push(info);
@@ -3322,10 +3293,20 @@ export class SessionManager {
       const sessions: SessionInfo[] = [];
       const allFiles = dirFiles.flat();
 
-      const results = await buildSessionInfosWithConcurrency(allFiles, () => {
-        loaded++;
-        onProgress?.(loaded, totalFiles);
-      });
+      const results = await pMap(
+        allFiles,
+        async (file) => {
+          try {
+            return await buildSessionInfo(file);
+          } catch {
+            return null;
+          } finally {
+            loaded++;
+            onProgress?.(loaded, totalFiles);
+          }
+        },
+        { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false },
+      );
 
       for (const info of results) {
         if (info) {

src/agents/tools/sessions-list-tool.ts

@@ -9,6 +9,7 @@ import {
   normalizeOptionalLowercaseString,
   readStringValue,
 } from "@openclaw/normalization-core/string-coerce";
+import pMap from "p-map";
 import { Type } from "typebox";
 import { getRuntimeConfig } from "../../config/config.js";
 import {
@@ -419,19 +420,9 @@ export function createSessionsListTool(opts?: {
       }
 
       if (titleTargets.length > 0) {
-        const maxConcurrent = Math.min(4, titleTargets.length);
-        let index = 0;
-        const worker = async () => {
-          while (true) {
-            const next = index;
-            index += 1;
-            if (next >= titleTargets.length) {
-              return;
-            }
-            const target = titleTargets.at(next);
-            if (!target) {
-              return;
-            }
+        await pMap(
+          titleTargets,
+          async (target) => {
             const fields = await readSessionTitleFieldsFromTranscriptAsync({
               agentId: target.agentId,
               sessionEntry: target.sessionEntry,
@@ -448,25 +439,15 @@ export function createSessionsListTool(opts?: {
             if (includeLastMessage && fields.lastMessagePreview) {
               target.row.lastMessagePreview = fields.lastMessagePreview;
             }
-          }
-        };
-        await Promise.all(Array.from({ length: maxConcurrent }, () => worker()));
+          },
+          { concurrency: 4, stopOnError: true },
+        );
       }
 
       if (messageLimit > 0 && historyTargets.length > 0) {
-        const maxConcurrent = Math.min(4, historyTargets.length);
-        let index = 0;
-        const worker = async () => {
-          while (true) {
-            const next = index;
-            index += 1;
-            if (next >= historyTargets.length) {
-              return;
-            }
-            const target = historyTargets.at(next);
-            if (!target) {
-              return;
-            }
+        await pMap(
+          historyTargets,
+          async (target) => {
             const history = await gatewayCall<{ messages: Array<unknown> }>({
               method: "chat.history",
               params: { sessionKey: target.resolvedKey, limit: messageLimit },
@@ -475,9 +456,9 @@ export function createSessionsListTool(opts?: {
             const filtered = stripToolMessages(rawMessages);
             target.row.messages =
               filtered.length > messageLimit ? filtered.slice(-messageLimit) : filtered;
-          }
-        };
-        await Promise.all(Array.from({ length: maxConcurrent }, () => worker()));
+          },
+          { concurrency: 4, stopOnError: true },
+        );
       }
 
       const visibilityMetadata =

src/commands/models/list.probe.ts

@@ -3,6 +3,7 @@ import crypto from "node:crypto";
 import fs from "node:fs/promises";
 import { expectDefined } from "@openclaw/normalization-core";
 import { normalizeUniqueStringEntries } from "@openclaw/normalization-core/string-normalization";
+import pMap from "p-map";
 import {
   resolveAgentDir,
   resolveAgentWorkspaceDir,
@@ -592,17 +593,9 @@ async function runTargetsWithConcurrency(params: {
   await fs.mkdir(workspaceDir, { recursive: true });
 
   let completed = 0;
-  const results: Array<AuthProbeResult | undefined> = Array.from({ length: targets.length });
-  let cursor = 0;
-
-  const worker = async () => {
-    while (true) {
-      const index = cursor;
-      cursor += 1;
-      if (index >= targets.length) {
-        return;
-      }
-      const target = expectDefined(targets[index], "targets entry at index");
+  return await pMap(
+    targets,
+    async (target) => {
       onProgress?.({
         completed,
         total: targets.length,
@@ -618,15 +611,12 @@ async function runTargetsWithConcurrency(params: {
         timeoutMs,
         maxTokens,
       });
-      results[index] = result;
       completed += 1;
       onProgress?.({ completed, total: targets.length });
-    }
-  };
-
-  await Promise.all(Array.from({ length: concurrency }, () => worker()));
-
-  return results.filter((entry): entry is AuthProbeResult => Boolean(entry));
+      return result;
+    },
+    { concurrency, stopOnError: true },
+  );
 }
 
 /** Runs all auth probes with bounded concurrency and returns a summary. */

src/media-understanding/apply-capability.ts

@@ -0,0 +1,16 @@
+// Keeps one provider failure from blocking the remaining media capabilities.
+import { logVerbose, shouldLogVerbose } from "../globals.js";
+import { runCapability } from "./runner.js";
+
+export async function runMediaCapability(
+  params: Parameters<typeof runCapability>[0],
+): Promise<Awaited<ReturnType<typeof runCapability>> | undefined> {
+  try {
+    return await runCapability(params);
+  } catch (err) {
+    if (shouldLogVerbose()) {
+      logVerbose(`Media understanding task failed: ${String(err)}`);
+    }
+    return undefined;
+  }
+}

src/media-understanding/apply.ts

@@ -6,6 +6,7 @@ import {
   normalizeLowercaseStringOrEmpty,
   normalizeOptionalString,
 } from "@openclaw/normalization-core/string-coerce";
+import pMap from "p-map";
 import type { ActiveMediaModel } from "../../packages/media-understanding-common/src/active-model.js";
 import {
   extractMediaUserText,
@@ -19,8 +20,8 @@ import { logVerbose, shouldLogVerbose } from "../globals.js";
 import { renderFileContextBlock } from "../media/file-context.js";
 import { extractFileContentFromSource, normalizeMimeType } from "../media/input-files.js";
 import { wrapExternalContent } from "../security/external-content.js";
+import { runMediaCapability } from "./apply-capability.js";
 import { resolveAttachmentKind } from "./attachments.js";
-import { runWithConcurrency } from "./concurrency.js";
 import { DEFAULT_ECHO_TRANSCRIPT_FORMAT, sendTranscriptEcho } from "./echo-transcript.js";
 import type { ExtractedFileImage } from "./extracted-file-images.js";
 import {
@@ -33,7 +34,6 @@ import {
   createMediaAttachmentCache,
   normalizeMediaAttachments,
   resolveMediaAttachmentLocalRoots,
-  runCapability,
 } from "./runner.js";
 import type {
   MediaUnderstandingCapability,
@@ -560,24 +560,24 @@ export async function applyMediaUnderstanding(params: {
   });
 
   try {
-    const tasks = CAPABILITY_ORDER.map((capability) => async () => {
-      const config = cfg.tools?.media?.[capability];
-      return await runCapability({
-        capability,
-        cfg,
-        ctx,
-        attachments: cache,
-        media: attachments,
-        agentId: params.agentId,
-        agentDir: params.agentDir,
-        workspaceDir: params.workspaceDir,
-        providerRegistry,
-        config,
-        activeModel: params.activeModel,
-      });
-    });
-
-    const results = await runWithConcurrency(tasks, resolveConcurrency(cfg));
+    const results = await pMap(
+      CAPABILITY_ORDER,
+      async (capability) =>
+        await runMediaCapability({
+          capability,
+          cfg,
+          ctx,
+          attachments: cache,
+          media: attachments,
+          agentId: params.agentId,
+          agentDir: params.agentDir,
+          workspaceDir: params.workspaceDir,
+          providerRegistry,
+          config: cfg.tools?.media?.[capability],
+          activeModel: params.activeModel,
+        }),
+      { concurrency: resolveConcurrency(cfg), stopOnError: false },
+    );
     const outputs: MediaUnderstandingOutput[] = [];
     const decisions: MediaUnderstandingDecision[] = [];
     for (const entry of results) {

src/media-understanding/concurrency.ts

@@ -1,22 +0,0 @@
-// Concurrency wrapper for media-understanding tasks that keeps successful
-// outputs while verbose-logging per-provider failures.
-import { logVerbose, shouldLogVerbose } from "../globals.js";
-import { runTasksWithConcurrency } from "../utils/run-with-concurrency.js";
-
-/** Runs media tasks under a fixed concurrency limit while preserving successful results. */
-export async function runWithConcurrency<T>(
-  tasks: Array<() => Promise<T>>,
-  limit: number,
-): Promise<T[]> {
-  const { results } = await runTasksWithConcurrency({
-    tasks,
-    limit,
-    // Media understanding tries every eligible entry; verbose mode keeps per-entry failures visible.
-    onTaskError(err) {
-      if (shouldLogVerbose()) {
-        logVerbose(`Media understanding task failed: ${String(err)}`);
-      }
-    },
-  });
-  return results;
-}

src/utils/run-with-concurrency.test.ts

@@ -4,10 +4,6 @@ import { runTasksWithConcurrency } from "./run-with-concurrency.js";
 
 describe("runTasksWithConcurrency", () => {
   it("preserves task order with bounded worker count", async () => {
-    const flushMicrotasks = async () => {
-      await Promise.resolve();
-      await Promise.resolve();
-    };
     let running = 0;
     let peak = 0;
     const resolvers: Array<(() => void) | undefined> = [];
@@ -22,7 +18,10 @@ describe("runTasksWithConcurrency", () => {
     });
 
     const resultPromise = runTasksWithConcurrency({ tasks, limit: 2 });
-    const takeResolver = (index: number): (() => void) => {
+    const takeResolver = async (index: number): Promise<() => void> => {
+      await vi.waitFor(() => {
+        expect(resolvers[index]).toBeTypeOf("function");
+      });
       const resolver = resolvers[index];
       if (!resolver) {
         throw new Error(`expected task ${index} to be running`);
@@ -30,17 +29,14 @@ describe("runTasksWithConcurrency", () => {
       return resolver;
     };
 
-    await flushMicrotasks();
-    const resolveFirst = takeResolver(0);
-    const resolveSecond = takeResolver(1);
+    const resolveFirst = await takeResolver(0);
+    const resolveSecond = await takeResolver(1);
 
     resolveSecond();
-    await flushMicrotasks();
-    const resolveThird = takeResolver(2);
+    const resolveThird = await takeResolver(2);
 
     resolveFirst();
-    await flushMicrotasks();
-    const resolveFourth = takeResolver(3);
+    const resolveFourth = await takeResolver(3);
 
     resolveThird();
     resolveFourth();

src/utils/run-with-concurrency.ts

@@ -1,4 +1,6 @@
-import { expectDefined } from "@openclaw/normalization-core"; /** Controls whether the worker pool keeps scheduling after a task failure. */
+import pLimit from "p-limit";
+
+/** Controls whether the worker pool keeps scheduling after a task failure. */
 export type ConcurrencyErrorMode = "continue" | "stop";
 
 /** Options for running a fixed list of promise factories through a bounded worker pool. */
@@ -33,39 +35,31 @@ export async function runTasksWithConcurrency<T>(
     return { results: [], firstError: undefined, hasError: false };
   }
 
-  const resolvedLimit = Math.max(1, Math.min(limit, tasks.length));
+  const resolvedLimit = Number.isFinite(limit)
+    ? Math.max(1, Math.min(Math.floor(limit), tasks.length))
+    : tasks.length;
   const results: T[] = Array.from({ length: tasks.length });
-  let next = 0;
   let firstError: unknown = undefined;
   let hasError = false;
+  const limiter = pLimit(resolvedLimit);
 
-  const workers = Array.from({ length: resolvedLimit }, async () => {
-    while (true) {
+  const runs = tasks.map((task, index) =>
+    limiter(async () => {
       if (errorMode === "stop" && hasError) {
         return;
       }
-      // Synchronous cursor adoption is the whole scheduling lock: each worker
-      // claims one stable index before awaiting task work.
-      const index = next;
-      next += 1;
-      if (index >= tasks.length) {
-        return;
-      }
       try {
-        results[index] = await expectDefined(tasks[index], "tasks entry at index")();
+        results[index] = await task();
       } catch (error) {
         if (!hasError) {
           firstError = error;
           hasError = true;
         }
         onTaskError?.(error, index);
-        if (errorMode === "stop") {
-          return;
-        }
       }
-    }
-  });
+    }),
+  );
 
-  await Promise.allSettled(workers);
+  await Promise.allSettled(runs);
   return { results, firstError, hasError };
 }

test/scripts/generate-npm-shrinkwrap.test.ts

@@ -21,7 +21,6 @@ import {
   resolvePackageDirs,
   resolveShrinkwrapJobs,
   restoreCurrentPnpmLockedPackages,
-  runBoundedTasks,
   shouldUseLegacyPeerDepsForShrinkwrap,
   shrinkwrapPackageDirsForChangedPaths,
 } from "../../scripts/generate-npm-shrinkwrap.mjs";
@@ -106,23 +105,6 @@ describe("generate-npm-shrinkwrap", () => {
     );
   });
 
-  it("bounds shrinkwrap package concurrency while preserving result order", async () => {
-    let active = 0;
-    let maxActive = 0;
-    const results = await runBoundedTasks(["slow", "fast", "last"], 2, async (value) => {
-      active += 1;
-      maxActive = Math.max(maxActive, active);
-      await new Promise<void>((resolve) => {
-        setTimeout(resolve, value === "slow" ? 30 : 5);
-      });
-      active -= 1;
-      return value;
-    });
-
-    expect(maxActive).toBe(2);
-    expect(results).toEqual(["slow", "fast", "last"]);
-  });
-
   it("validates shrinkwrap worker counts from flags and environment", () => {
     expect(resolveShrinkwrapJobs("3", {})).toBe(3);
     expect(resolveShrinkwrapJobs(undefined, { OPENCLAW_NPM_SHRINKWRAP_JOBS: "2" })).toBe(2);

test/scripts/write-cli-startup-metadata.test.ts

@@ -114,24 +114,6 @@ async function waitForChildClose(
 describe("write-cli-startup-metadata", () => {
   const { createTempDir } = createScriptTestHarness();
 
-  it("caps concurrent metadata render workers while preserving result order", async () => {
-    let active = 0;
-    let peakActive = 0;
-
-    const result = await __testing.mapWithConcurrency([1, 2, 3, 4, 5], 2, async (value) => {
-      active += 1;
-      peakActive = Math.max(peakActive, active);
-      await new Promise((resolve) => {
-        setTimeout(resolve, 1);
-      });
-      active -= 1;
-      return `rendered-${value}`;
-    });
-
-    expect(result).toEqual(["rendered-1", "rendered-2", "rendered-3", "rendered-4", "rendered-5"]);
-    expect(peakActive).toBe(2);
-  });
-
   it("fails command help rendering when captured output exceeds the byte limit", async () => {
     await expect(
       __testing.spawnText(["--eval", "process.stdout.write('x'.repeat(2048))"], {