refactor: consolidate remaining bounded concurrency (#106265)

extensions/memory-core/package.json

@@ -7,6 +7,7 @@
   "dependencies": {
     "chokidar": "5.0.0",
     "json5": "2.2.3",
+    "p-limit": "7.3.0",
     "typebox": "1.3.3"
   },
   "devDependencies": {

extensions/memory-core/src/dreaming-narrative.ts

@@ -16,6 +16,7 @@ import { resolveStateDir } from "openclaw/plugin-sdk/memory-core-host-runtime-co
 import { getRuntimeConfig } from "openclaw/plugin-sdk/runtime-config-snapshot";
 import { cleanupSessionLifecycleArtifacts } from "openclaw/plugin-sdk/session-store-runtime";
 import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
+import pLimit from "p-limit";
 import { readDreamsFile, resolveDreamsPath, updateDreamsFile } from "./dreaming-dreams-file.js";
 
 // ── Types ──────────────────────────────────────────────────────────────
@@ -997,44 +998,20 @@ export async function generateAndAppendDreamNarrative(params: {
 // write-lock while it runs and burns a model slot, which caused lock
 // contention (>30 s) and cascading narrative timeouts (#73198).
 //
-// `runDetachedDreamNarrative` wraps `generateAndAppendDreamNarrative` with a
-// FIFO queue capped at `DETACHED_NARRATIVE_CONCURRENCY` so the total in-flight
-// detached narratives across phases/workspaces stays bounded.
+// `runDetachedDreamNarrative` caps total in-flight detached narratives across
+// phases/workspaces so cron sweeps cannot exhaust model and session-lock slots.
 const DETACHED_NARRATIVE_CONCURRENCY = 3;
-
-let activeDetachedNarratives = 0;
-const detachedNarrativeQueue: Array<() => void> = [];
-
-function releaseDetachedNarrativeSlot(): void {
-  activeDetachedNarratives -= 1;
-  detachedNarrativeQueue.shift()?.();
-}
-
-async function acquireDetachedNarrativeSlot(): Promise<void> {
-  if (activeDetachedNarratives >= DETACHED_NARRATIVE_CONCURRENCY) {
-    await new Promise<void>((resolve) => {
-      detachedNarrativeQueue.push(resolve);
-    });
-  }
-  activeDetachedNarratives += 1;
-}
+const detachedNarrativeLimit = pLimit(DETACHED_NARRATIVE_CONCURRENCY);
 
 export function runDetachedDreamNarrative(
   params: Parameters<typeof generateAndAppendDreamNarrative>[0],
 ): void {
   queueMicrotask(() => {
-    void (async () => {
-      await acquireDetachedNarrativeSlot();
-      try {
-        await generateAndAppendDreamNarrative(params);
-      } catch {
-        // Detached narratives intentionally swallow errors — callers (cron
-        // sweeps) cannot recover, and surfacing here would only cause noisy
-        // unhandled rejections. Logging happens inside
-        // generateAndAppendDreamNarrative.
-      } finally {
-        releaseDetachedNarrativeSlot();
-      }
-    })();
+    void detachedNarrativeLimit(() => generateAndAppendDreamNarrative(params)).catch(() => {
+      // Detached narratives intentionally swallow errors — callers (cron
+      // sweeps) cannot recover, and surfacing here would only cause noisy
+      // unhandled rejections. Logging happens inside
+      // generateAndAppendDreamNarrative.
+    });
   });
 }

extensions/memory-core/src/short-term-promotion.ts

@@ -18,6 +18,7 @@ import {
   uniqueStrings,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
 import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
+import pLimit from "p-limit";
 import {
   deriveConceptTags,
   MAX_CONCEPT_TAGS,
@@ -56,6 +57,8 @@ const PROMOTED_SNIPPET_CHARS_PER_TOKEN_ESTIMATE = 4;
 const MAX_QUERY_HASHES = 32;
 const MAX_RECALL_DAYS = 16;
 const SHORT_TERM_RECALL_MAX_ENTRIES = 512;
+// One recall batch can inspect every retained entry; cap filesystem pressure.
+const SHORT_TERM_SOURCE_FILE_CHECK_CONCURRENCY = 32;
 const SHORT_TERM_RECALL_MAX_SNIPPET_CHARS = 800;
 export const SHORT_TERM_STORE_RELATIVE_PATH = path.join(
   "memory",
@@ -1333,12 +1336,13 @@ export async function filterLiveShortTermRecallEntries(params: {
     return [];
   }
   const sourceFileChecks = new Map<string, Promise<boolean>>();
+  const sourceFileLimit = pLimit(SHORT_TERM_SOURCE_FILE_CHECK_CONCURRENCY);
   const checkSourceFile = (sourcePath: string): Promise<boolean> => {
     const existing = sourceFileChecks.get(sourcePath);
     if (existing) {
       return existing;
     }
-    const check = shortTermRecallSourceIsFile(sourcePath);
+    const check = sourceFileLimit(() => shortTermRecallSourceIsFile(sourcePath));
     sourceFileChecks.set(sourcePath, check);
     return check;
   };

extensions/memory-wiki/package.json

@@ -6,6 +6,7 @@
   "type": "module",
   "dependencies": {
     "mdast-util-from-markdown": "2.0.3",
+    "p-map": "7.0.5",
     "typebox": "1.3.3",
     "yaml": "2.9.0",
     "zod": "4.4.3"

extensions/memory-wiki/src/import-runs-state.ts

@@ -6,6 +6,9 @@ import type {
   OpenKeyedStoreOptions,
   PluginStateKeyedStore,
 } from "openclaw/plugin-sdk/plugin-state-runtime";
+import pMap from "p-map";
+
+const LEGACY_IMPORT_RUN_READ_CONCURRENCY = 16;
 
 type ChatGptImportRunEntry = {
   path: string;
@@ -471,13 +474,13 @@ export async function readLegacyMemoryWikiImportRunRecords(
       }
       throw error;
     });
-  const records = await Promise.all(
-    entries
-      .filter((entry) => entry.isFile() && entry.name.endsWith(".json"))
-      .map(async (entry) => {
-        const raw = await fs.readFile(path.join(importRunsDir, entry.name), "utf8");
-        return normalizeMemoryWikiImportRunRecord(JSON.parse(raw) as unknown);
-      }),
+  const records = await pMap(
+    entries.filter((entry) => entry.isFile() && entry.name.endsWith(".json")),
+    async (entry) => {
+      const raw = await fs.readFile(path.join(importRunsDir, entry.name), "utf8");
+      return normalizeMemoryWikiImportRunRecord(JSON.parse(raw) as unknown);
+    },
+    { concurrency: LEGACY_IMPORT_RUN_READ_CONCURRENCY, stopOnError: true },
   );
   return records.filter((entry): entry is ChatGptImportRunRecord => entry !== null);
 }

extensions/memory-wiki/src/query.ts

@@ -18,6 +18,7 @@ import {
   normalizeLowercaseStringOrEmpty,
   uniqueStrings,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import type { OpenClawConfig } from "../api.js";
 import { assessClaimFreshness, isClaimContestedStatus } from "./claim-health.js";
 import type { ResolvedMemoryWikiConfig, WikiSearchBackend, WikiSearchCorpus } from "./config.js";
@@ -33,6 +34,7 @@ import { initializeMemoryWikiVault } from "./vault.js";
 const QUERY_DIRS = ["entities", "concepts", "sources", "syntheses", "reports"] as const;
 const AGENT_DIGEST_PATH = ".openclaw-wiki/cache/agent-digest.json";
 const CLAIMS_DIGEST_PATH = ".openclaw-wiki/cache/claims.jsonl";
+const QUERY_PAGE_READ_CONCURRENCY = 16;
 const RELATED_BLOCK_PATTERN =
   /<!-- openclaw:wiki:related:start -->[\s\S]*?<!-- openclaw:wiki:related:end -->/g;
 const MARKDOWN_FRONTMATTER_PATTERN = /^\s*---\r?\n[\s\S]*?\r?\n---\r?\n?/;
@@ -271,13 +273,15 @@ async function readQueryableWikiPagesByPaths(
   rootDir: string,
   files: string[],
 ): Promise<QueryableWikiPage[]> {
-  const pages = await Promise.all(
-    files.map(async (relativePath) => {
+  const pages = await pMap(
+    files,
+    async (relativePath) => {
       const absolutePath = path.join(rootDir, relativePath);
       const raw = await fs.readFile(absolutePath, "utf8");
       const summary = toWikiPageSummary({ absolutePath, relativePath, raw });
       return summary ? { ...summary, raw } : null;
-    }),
+    },
+    { concurrency: QUERY_PAGE_READ_CONCURRENCY, stopOnError: true },
   );
   return pages.flatMap((page) => (page ? [page] : []));
 }

extensions/memory-wiki/src/unsafe-local.ts

@@ -3,6 +3,7 @@ import { createHash } from "node:crypto";
 import fs from "node:fs/promises";
 import path from "node:path";
 import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
+import pMap from "p-map";
 import type { BridgeMemoryWikiResult } from "./bridge.js";
 import type { ResolvedMemoryWikiConfig } from "./config.js";
 import { appendMemoryWikiLog } from "./log.js";
@@ -30,6 +31,7 @@ type UnsafeLocalArtifact = {
 };
 
 const DIRECTORY_TEXT_EXTENSIONS = new Set([".json", ".jsonl", ".md", ".txt", ".yaml", ".yml"]);
+const UNSAFE_LOCAL_SYNC_CONCURRENCY = 16;
 
 function detectFenceLanguage(filePath: string): string {
   const ext = normalizeLowercaseStringOrEmpty(path.extname(filePath));
@@ -222,8 +224,9 @@ export async function syncMemoryWikiUnsafeLocalSources(
     incomingCount: artifacts.length,
   });
   const activeKeys = new Set<string>();
-  const results = await Promise.all(
-    artifacts.map(async (artifact) => {
+  const results = await pMap(
+    artifacts,
+    async (artifact) => {
       const stats = await fs.stat(artifact.absolutePath);
       activeKeys.add(artifact.syncKey);
       return await writeUnsafeLocalSourcePage({
@@ -233,7 +236,8 @@ export async function syncMemoryWikiUnsafeLocalSources(
         sourceSize: stats.size,
         state,
       });
-    }),
+    },
+    { concurrency: UNSAFE_LOCAL_SYNC_CONCURRENCY, stopOnError: true },
   );
 
   const removedCount = await pruneImportedSourceEntries({

pnpm-lock.yaml

@@ -1093,6 +1093,9 @@ importers:
       json5:
         specifier: 2.2.3
         version: 2.2.3
+      p-limit:
+        specifier: 7.3.0
+        version: 7.3.0
       typebox:
         specifier: 1.3.3
         version: 1.3.3
@@ -1128,6 +1131,9 @@ importers:
       mdast-util-from-markdown:
         specifier: 2.0.3
         version: 2.0.3
+      p-map:
+        specifier: 7.0.5
+        version: 7.0.5
       typebox:
         specifier: 1.3.3
         version: 1.3.3

scripts/run-additional-boundary-checks.mjs

@@ -3,6 +3,7 @@
 // timeout handling, and grouped CI output.
 import { spawn } from "node:child_process";
 import { performance } from "node:perf_hooks";
+import pMap from "p-map";
 import prettyMilliseconds from "pretty-ms";
 
 const DEFAULT_CHECK_TIMEOUT_MS = 10 * 60 * 1000;
@@ -499,43 +500,23 @@ export async function runChecks(
     outputMaxBytes = DEFAULT_OUTPUT_MAX_BYTES,
   } = {},
 ) {
-  const results = Array.from({ length: checks.length });
   const activeChildren = new Set();
   const removeActiveChildCleanup = installActiveChildCleanup(activeChildren);
-  let nextIndex = 0;
-  let active = 0;
+  let results;
 
   try {
-    await new Promise((resolve) => {
-      const launch = () => {
-        if (nextIndex >= checks.length && active === 0) {
-          resolve();
-          return;
-        }
-
-        while (active < concurrency && nextIndex < checks.length) {
-          const index = nextIndex;
-          const check = checks[nextIndex++];
-          active += 1;
-          void runSingleCheck(check, {
-            activeChildren,
-            checkTimeoutMs,
-            cwd,
-            env,
-            outputMaxBytes,
-          })
-            .then((result) => {
-              results[index] = result;
-            })
-            .finally(() => {
-              active -= 1;
-              launch();
-            });
-        }
-      };
-
-      launch();
-    });
+    results = await pMap(
+      checks,
+      (check) =>
+        runSingleCheck(check, {
+          activeChildren,
+          checkTimeoutMs,
+          cwd,
+          env,
+          outputMaxBytes,
+        }),
+      { concurrency, stopOnError: true },
+    );
   } finally {
     removeActiveChildCleanup();
   }

scripts/run-oxlint-shards.mjs

@@ -3,6 +3,7 @@ import { spawn, spawnSync } from "node:child_process";
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
+import pMap from "p-map";
 import {
   acquireLocalHeavyCheckLockSync,
   resolveLocalHeavyCheckEnv,
@@ -272,21 +273,13 @@ export async function main(extraArgs = process.argv.slice(2), runtimeEnv = proce
         platform: process.platform,
         splitCore: shardArgs.splitCore,
       });
-      const results =
-        shardConcurrency <= 1
-          ? await runShardsSerial({
-              entries: selectedShards,
-              env,
-              extraArgs: shardArgs.oxlintArgs,
-              runner,
-            })
-          : await runShardsParallel({
-              concurrency: Math.min(shardConcurrency, selectedShards.length),
-              entries: selectedShards,
-              env,
-              extraArgs: shardArgs.oxlintArgs,
-              runner,
-            });
+      const results = await runShards({
+        concurrency: Math.max(1, Math.min(shardConcurrency, selectedShards.length)),
+        entries: selectedShards,
+        env,
+        extraArgs: shardArgs.oxlintArgs,
+        runner,
+      });
       process.exitCode = results.find((status) => status !== 0) ?? 0;
     }
   } finally {
@@ -385,38 +378,17 @@ export function resolveOxlintShardConcurrency({
   );
 }
 
-async function runShardsSerial({ entries, env, extraArgs, runner }) {
-  const results = [];
-  for (const shard of entries) {
-    results.push(await runShard({ env, extraArgs, runner, shard }));
-    if (isParentTerminationRequested()) {
-      break;
-    }
-  }
-  return results;
-}
-
-async function runShardsParallel({ concurrency, entries, env, extraArgs, runner }) {
-  const results = [];
-  results.length = entries.length;
-  let nextIndex = 0;
-
-  const workers = Array.from({ length: concurrency }, async () => {
-    for (;;) {
+async function runShards({ concurrency, entries, env, extraArgs, runner }) {
+  const results = await pMap(
+    entries,
+    async (shard) => {
       if (isParentTerminationRequested()) {
-        return;
+        return undefined;
       }
-      const currentIndex = nextIndex;
-      nextIndex += 1;
-      const shard = entries[currentIndex];
-      if (!shard) {
-        return;
-      }
-      results[currentIndex] = await runShard({ env, extraArgs, runner, shard });
-    }
-  });
-
-  await Promise.all(workers);
+      return await runShard({ env, extraArgs, runner, shard });
+    },
+    { concurrency, stopOnError: true },
+  );
   return results.filter((status) => status !== undefined);
 }
 

scripts/test-extension-batch.mjs

@@ -2,6 +2,7 @@
 
 // Runs grouped Vitest plans for one or more bundled plugins.
 import path from "node:path";
+import pMap from "p-map";
 import {
   listTrackedTestFilesForRoots,
   resolveExtensionBatchPlan,
@@ -196,14 +197,11 @@ export async function runExtensionBatchPlan(batchPlan, params = {}) {
     console.log(`[test-extension-batch] Running up to ${parallelism} config groups in parallel`);
   }
 
-  let nextGroupIndex = 0;
   let exitCode = 0;
-  async function worker() {
-    while (exitCode === 0) {
-      const groupIndex = nextGroupIndex;
-      nextGroupIndex += 1;
-      const group = orderedGroups[groupIndex];
-      if (!group) {
+  await pMap(
+    orderedGroups,
+    async (group, groupIndex) => {
+      if (exitCode !== 0) {
         return;
       }
       const groupExitCode = await runPlanGroup(group, {
@@ -215,14 +213,12 @@ export async function runExtensionBatchPlan(batchPlan, params = {}) {
         useDedicatedCache,
         vitestArgs,
       });
-      if (groupExitCode !== 0) {
+      if (groupExitCode !== 0 && exitCode === 0) {
         exitCode = groupExitCode;
-        return;
       }
-    }
-  }
-
-  await Promise.all(Array.from({ length: parallelism }, () => worker()));
+    },
+    { concurrency: parallelism, stopOnError: true },
+  );
   return exitCode;
 }
 

scripts/test-group-report.mjs

@@ -4,6 +4,7 @@ import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
+import pMap from "p-map";
 import { parsePositiveInt } from "./lib/numeric-options.mjs";
 import {
   buildGroupedTestComparison,
@@ -980,19 +981,15 @@ export async function runReportPlans(params) {
   const concurrency = resolveRunPlanConcurrency(params.args, params.runPlans.length);
   const runSpecs = resolveReportRunSpecs(params.args, params.runPlans, { concurrency });
   const runVitest = params.runVitestJsonReport ?? runVitestJsonReport;
-  const results = [];
-  results.length = runSpecs.length;
-  const runs = [];
-  runs.length = runSpecs.length;
-  let nextIndex = 0;
   let failed = false;
   let exitCode = 0;
 
-  async function worker() {
-    while (nextIndex < runSpecs.length && exitCode === 0) {
-      const index = nextIndex;
-      nextIndex += 1;
-      const plan = runSpecs[index];
+  const results = await pMap(
+    runSpecs,
+    async (plan) => {
+      if (exitCode !== 0) {
+        return null;
+      }
       const slug = sanitizePathSegment(plan.label);
       const run = await runVitest({
         config: plan.config,
@@ -1006,7 +1003,6 @@ export async function runReportPlans(params) {
         killGraceMs: params.args.killGraceMs,
         vitestArgs: plan.vitestArgs,
       });
-      runs[index] = run;
       printRunLine(run);
       let includeEntry = true;
       if (run.status !== 0) {
@@ -1035,24 +1031,19 @@ export async function runReportPlans(params) {
         }
       }
       const entry = includeEntry ? { config: plan.label, reportPath: run.reportPath, run } : null;
-      results[index] = entry;
       if (entry) {
         printSlowTestsForRun(entry, params.args.maxTestMs);
       }
-    }
-  }
-
-  await Promise.all(
-    Array.from({ length: concurrency }, async () => {
-      await worker();
-    }),
+      return { entry, run };
+    },
+    { concurrency, stopOnError: true },
   );
 
   return {
     failed,
     exitCode,
-    runEntries: results.filter(Boolean),
-    runs: runs.filter(Boolean),
+    runEntries: results.flatMap((result) => (result?.entry ? [result.entry] : [])),
+    runs: results.flatMap((result) => (result ? [result.run] : [])),
   };
 }
 

scripts/test-projects.mjs

@@ -2,6 +2,7 @@
 // full local suite.
 import fs from "node:fs";
 import { performance } from "node:perf_hooks";
+import pMap from "p-map";
 import { formatMs } from "./lib/check-timing-summary.mjs";
 import { acquireLocalHeavyCheckLockSync } from "./lib/local-heavy-check-runtime.mjs";
 import {
@@ -207,21 +208,21 @@ function printNoChangedTestTargets(args, cwd, baseEnv) {
 }
 
 async function runVitestSpecsParallel(specs, concurrency) {
-  let nextIndex = 0;
   let exitCode = 0;
+  let stopScheduling = false;
   const failures = [];
   const timings = [];
 
-  const runWorker = async () => {
-    for (;;) {
-      const index = nextIndex;
-      nextIndex += 1;
-      const spec = specs[index];
-      if (!spec) {
+  await pMap(
+    specs,
+    async (spec, index) => {
+      if (stopScheduling) {
         return;
       }
       const result = await runLoggedVitestSpec(spec);
       if (!result) {
+        // A forwarded termination signal must not admit replacement shards during shutdown.
+        stopScheduling = true;
         return;
       }
       if (result.code !== 0) {
@@ -238,10 +239,9 @@ async function runVitestSpecsParallel(specs, concurrency) {
       if (result.timing) {
         timings.push(result.timing);
       }
-    }
-  };
-
-  await Promise.all(Array.from({ length: concurrency }, () => runWorker()));
+    },
+    { concurrency, stopOnError: true },
+  );
   return { exitCode, failures, timings };
 }
 

src/gateway/server-startup-post-attach.ts

@@ -2,6 +2,7 @@
 // Schedules warmups, sentinels, update checks, memory backend, and plugin services.
 import { monitorEventLoopDelay, performance } from "node:perf_hooks";
 import { setTimeout as sleep } from "node:timers/promises";
+import pMap from "p-map";
 import type { CliDeps } from "../cli/deps.types.js";
 import { resolveStateDir } from "../config/paths.js";
 import type { GatewayTailscaleMode } from "../config/types.gateway.js";
@@ -448,7 +449,6 @@ async function cleanupStaleSessionLocks(params: {
       Math.floor(params.concurrency ?? SESSION_LOCK_CLEANUP_CONCURRENCY),
     ),
   );
-  let nextIndex = 0;
   let markRestartAbortedMainSessionsFromLocks =
     params.markRestartAbortedMainSessionsFromLocks ?? null;
   const getMarker = async () => {
@@ -456,11 +456,10 @@ async function cleanupStaleSessionLocks(params: {
       .markRestartAbortedMainSessionsFromLocks;
     return markRestartAbortedMainSessionsFromLocks;
   };
-  const worker = async () => {
-    while (!params.isStopped()) {
-      const sessionsDir = params.sessionDirs[nextIndex];
-      nextIndex += 1;
-      if (!sessionsDir) {
+  await pMap(
+    params.sessionDirs,
+    async (sessionsDir) => {
+      if (params.isStopped()) {
         return;
       }
       const result = await params.cleanStaleLockFiles({
@@ -470,16 +469,16 @@ async function cleanupStaleSessionLocks(params: {
         log: { warn: (message) => params.log.warn(message) },
       });
       if (result.cleaned.length === 0) {
-        continue;
+        return;
       }
       const markRestartAbortedMainSessionsFromLocksLocal = await getMarker();
       await markRestartAbortedMainSessionsFromLocksLocal({
         sessionsDir,
         cleanedLocks: result.cleaned,
       });
-    }
-  };
-  await Promise.all(Array.from({ length: concurrency }, () => worker()));
+    },
+    { concurrency, stopOnError: true },
+  );
 }
 
 function scheduleTranscriptsAutoStartSidecar(params: {

src/infra/package-dist-inventory.ts

@@ -2,6 +2,7 @@
 import fs from "node:fs/promises";
 import path from "node:path";
 import { sortUniqueStrings } from "@openclaw/normalization-core/string-normalization";
+import pLimit, { type LimitFunction } from "p-limit";
 import { isLocalBuildMetadataDistPath } from "../../scripts/lib/local-build-metadata-paths.mjs";
 import { readJsonIfExists, writeJson } from "./json-files.js";
 
@@ -91,38 +92,6 @@ type PackageDistInventoryRules = {
   externalizedExtensionIds: ExternalizedBundledExtensionIds;
   exclusions: PackageDistExclusionRules;
 };
-type PackageDistInventoryScanContext = {
-  activeFsOps: number;
-  fsConcurrency: number;
-  waiters: Array<() => void>;
-};
-
-function createPackageDistInventoryScanContext(): PackageDistInventoryScanContext {
-  return {
-    activeFsOps: 0,
-    fsConcurrency: PACKAGE_DIST_INVENTORY_SCAN_CONCURRENCY,
-    waiters: [],
-  };
-}
-
-async function withPackageDistInventoryFsSlot<T>(
-  context: PackageDistInventoryScanContext,
-  task: () => Promise<T>,
-): Promise<T> {
-  while (context.activeFsOps >= context.fsConcurrency) {
-    await new Promise<void>((resolve) => {
-      context.waiters.push(resolve);
-    });
-  }
-  context.activeFsOps += 1;
-  try {
-    return await task();
-  } finally {
-    context.activeFsOps -= 1;
-    context.waiters.shift()?.();
-  }
-}
-
 function normalizeRelativePath(value: string): string {
   return value.replace(/\\/g, "/");
 }
@@ -349,22 +318,20 @@ async function collectRelativeFiles(
   rootDir: string,
   baseDir: string,
   rules: PackageDistInventoryRules,
-  context: PackageDistInventoryScanContext,
+  fsLimit: LimitFunction,
 ): Promise<string[]> {
   const rootRelativePath = normalizeRelativePath(path.relative(baseDir, rootDir));
   if (rootRelativePath && isOmittedDistSubtree(rootRelativePath, rules)) {
     return [];
   }
   try {
-    const rootStats = await withPackageDistInventoryFsSlot(context, () => fs.lstat(rootDir));
+    const rootStats = await fsLimit(() => fs.lstat(rootDir));
     if (!rootStats.isDirectory() || rootStats.isSymbolicLink()) {
       throw new Error(
         `Unsafe package dist path: ${normalizeRelativePath(path.relative(baseDir, rootDir))}`,
       );
     }
-    const entries = await withPackageDistInventoryFsSlot(context, () =>
-      fs.readdir(rootDir, { withFileTypes: true }),
-    );
+    const entries = await fsLimit(() => fs.readdir(rootDir, { withFileTypes: true }));
     const files = await Promise.all(
       entries.map(async (entry) => {
         const entryPath = path.join(rootDir, entry.name);
@@ -373,7 +340,7 @@ async function collectRelativeFiles(
           throw new Error(`Unsafe package dist path: ${relativePath}`);
         }
         if (entry.isDirectory()) {
-          return await collectRelativeFiles(entryPath, baseDir, rules, context);
+          return await collectRelativeFiles(entryPath, baseDir, rules, fsLimit);
         }
         if (entry.isFile()) {
           return isPackagedDistPath(relativePath, rules) ? [relativePath] : [];
@@ -393,13 +360,8 @@ async function collectRelativeFiles(
 /** Collects package dist files that should be present after install/update publication. */
 export async function collectPackageDistInventory(packageRoot: string): Promise<string[]> {
   const rules = await collectPackageDistInventoryRulesForRoot(packageRoot);
-  const scanContext = createPackageDistInventoryScanContext();
-  return await collectRelativeFiles(
-    path.join(packageRoot, "dist"),
-    packageRoot,
-    rules,
-    scanContext,
-  );
+  const fsLimit = pLimit(PACKAGE_DIST_INVENTORY_SCAN_CONCURRENCY);
+  return await collectRelativeFiles(path.join(packageRoot, "dist"), packageRoot, rules, fsLimit);
 }
 
 /** Lists legacy plugin dependency staging directories that must not ship in package dist. */

test/scripts/run-oxlint.test.ts

@@ -94,14 +94,6 @@ describe("run-oxlint", () => {
     expect(childSkipIndex).toBeGreaterThan(lockIndex);
   });
 
-  it("keeps a serial oxlint shard path available", () => {
-    const shardedLintRunner = readFileSync("scripts/run-oxlint-shards.mjs", "utf8");
-
-    expect(shardedLintRunner).toContain("OPENCLAW_OXLINT_SHARDS_SERIAL");
-    expect(shardedLintRunner).toContain('platform === "win32"');
-    expect(shardedLintRunner).toContain("runShardsSerial");
-  });
-
   it("serializes broad oxlint shards on constrained local hosts", () => {
     expect(
       shouldRunOxlintShardsSerial({

test/scripts/test-extension.test.ts

@@ -39,6 +39,29 @@ type RunGroupParams = {
   env: Record<string, string | undefined>;
   targets: string[];
 };
+
+function createConcurrentExtensionBatchPlan() {
+  const groups = [
+    ["light", 10, "one", 1],
+    ["heavy", 30, "two", 3],
+    ["middle", 20, "three", 2],
+  ] as const;
+  return {
+    extensionCount: groups.length,
+    extensionIds: groups.map(([, , extensionId]) => extensionId),
+    estimatedCost: 60,
+    hasTests: true,
+    planGroups: groups.map(([config, estimatedCost, extensionId, testFileCount]) => ({
+      config,
+      estimatedCost,
+      extensionIds: [extensionId],
+      roots: [`extensions/${extensionId}`],
+      testFileCount,
+    })),
+    testFileCount: 6,
+  };
+}
+
 function runScriptResult(args: string[], cwd = process.cwd()) {
   return spawnSync(process.execPath, [scriptPath, ...args], {
     cwd,
@@ -561,48 +584,17 @@ describe("scripts/test-extension.mjs", () => {
         resolvers.push(() => resolve(0));
       });
     });
-    const runPromise = runExtensionBatchPlan(
-      {
-        extensionCount: 3,
-        extensionIds: ["one", "two", "three"],
-        estimatedCost: 60,
-        hasTests: true,
-        planGroups: [
-          {
-            config: "light",
-            estimatedCost: 10,
-            extensionIds: ["one"],
-            roots: ["extensions/one"],
-            testFileCount: 1,
-          },
-          {
-            config: "heavy",
-            estimatedCost: 30,
-            extensionIds: ["two"],
-            roots: ["extensions/two"],
-            testFileCount: 3,
-          },
-          {
-            config: "middle",
-            estimatedCost: 20,
-            extensionIds: ["three"],
-            roots: ["extensions/three"],
-            testFileCount: 2,
-          },
-        ],
-        testFileCount: 6,
-      },
-      {
-        env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" },
-        runGroup: runGroup as NonNullable<
-          NonNullable<Parameters<typeof runExtensionBatchPlan>[1]>["runGroup"]
-        >,
-        vitestArgs: ["--reporter=dot"],
-      },
-    );
+    const runPromise = runExtensionBatchPlan(createConcurrentExtensionBatchPlan(), {
+      env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" },
+      runGroup: runGroup as NonNullable<
+        NonNullable<Parameters<typeof runExtensionBatchPlan>[1]>["runGroup"]
+      >,
+      vitestArgs: ["--reporter=dot"],
+    });
 
-    await Promise.resolve();
-    expect(started).toEqual(["heavy", "middle"]);
+    await vi.waitFor(() => {
+      expect(started).toEqual(["heavy", "middle"]);
+    });
     resolvers.shift()?.();
     await new Promise<void>((resolve) => {
       setImmediate(resolve);
@@ -631,6 +623,40 @@ describe("scripts/test-extension.mjs", () => {
     });
   });
 
+  it("stops admitting extension batch groups after a parallel failure", async () => {
+    const started: string[] = [];
+    let resolveHeavy: ((code: number) => void) | undefined;
+    let resolveMiddle: ((code: number) => void) | undefined;
+    const runGroup = vi.fn((params: RunGroupParams) => {
+      started.push(params.config);
+      return new Promise<number>((resolve) => {
+        if (params.config === "heavy") {
+          resolveHeavy = resolve;
+        } else if (params.config === "middle") {
+          resolveMiddle = resolve;
+        }
+      });
+    });
+    const runPromise = runExtensionBatchPlan(createConcurrentExtensionBatchPlan(), {
+      env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" },
+      runGroup: runGroup as NonNullable<
+        NonNullable<Parameters<typeof runExtensionBatchPlan>[1]>["runGroup"]
+      >,
+    });
+
+    await vi.waitFor(() => {
+      expect(started).toEqual(["heavy", "middle"]);
+    });
+    resolveHeavy?.(7);
+    await new Promise<void>((resolve) => {
+      setImmediate(resolve);
+    });
+    expect(started).toEqual(["heavy", "middle"]);
+    resolveMiddle?.(0);
+    await expect(runPromise).resolves.toBe(7);
+    expect(runGroup).toHaveBeenCalledTimes(2);
+  });
+
   it("keeps extension batch parallelism bounded by group count", () => {
     expect(resolveExtensionBatchParallelism(3, { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" })).toBe(2);
     expect(resolveExtensionBatchParallelism(1, { OPENCLAW_EXTENSION_BATCH_PARALLEL: "4" })).toBe(1);

test/scripts/test-group-report.test.ts

@@ -400,6 +400,67 @@ describe("scripts/test-group-report aggregation", () => {
     }
   });
 
+  it("stops admitting report plans after a parallel failure", async () => {
+    const tempDir = makeTempDir();
+    const labels = ["first", "second", "third"];
+    const started: string[] = [];
+    const resolvers = new Map<string, (status: number) => void>();
+    try {
+      const runPromise = runReportPlans({
+        args: parseTestGroupReportArgs([
+          ...labels.flatMap((label) => ["--config", `${label}.config.ts`]),
+          "--concurrency",
+          "2",
+          "--no-rss",
+        ]),
+        logDir: path.join(tempDir, "logs"),
+        reportDir: path.join(tempDir, "reports"),
+        runPlans: labels.map((label) => ({
+          config: `${label}.config.ts`,
+          forwardedArgs: [],
+          label,
+        })),
+        runVitestJsonReport: async (params: {
+          config: string;
+          label: string;
+          logPath: string;
+          reportPath: string;
+        }) => {
+          started.push(params.label);
+          const status = await new Promise<number>((resolve) => {
+            resolvers.set(params.label, resolve);
+          });
+          return {
+            config: params.config,
+            elapsedMs: 10,
+            label: params.label,
+            logPath: params.logPath,
+            maxRssBytes: null,
+            reportPath: params.reportPath,
+            status,
+          };
+        },
+      });
+
+      await vi.waitFor(() => {
+        expect(started).toStrictEqual(["first", "second"]);
+      });
+      resolvers.get("first")?.(1);
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+      expect(started).toStrictEqual(["first", "second"]);
+      resolvers.get("second")?.(0);
+
+      const result = await runPromise;
+      expect(result.exitCode).toBe(1);
+      expect(result.failed).toBe(true);
+      expect(result.runs.map((run) => run.label)).toStrictEqual(["first", "second"]);
+    } finally {
+      fs.rmSync(tempDir, { recursive: true, force: true });
+    }
+  });
+
   it("prints slow tests as soon as each config report completes", async () => {
     const tempDir = makeTempDir();
     const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});