From 113675717436a9aeaa905867adb19ca75c71c1bc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 13 Jul 2026 01:53:28 -0700 Subject: [PATCH] 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 | 4 + extensions/discord/npm-shrinkwrap.json | 13 ++ extensions/discord/package.json | 1 + .../monitor/thread-bindings.lifecycle.test.ts | 5 +- .../src/monitor/thread-bindings.lifecycle.ts | 46 +---- extensions/openshell/npm-shrinkwrap.json | 28 +++ extensions/openshell/package.json | 1 + extensions/openshell/src/mirror.ts | 27 +-- extensions/qa-lab/package.json | 2 + extensions/qa-lab/src/character-eval.ts | 165 ++++++++---------- extensions/qa-lab/src/evidence-gallery.ts | 22 +-- extensions/qa-lab/src/suite-planning.ts | 29 ++- extensions/qqbot/npm-shrinkwrap.json | 13 ++ extensions/qqbot/package.json | 1 + .../qqbot/src/engine/api/media-chunked.ts | 29 +-- extensions/slack/npm-shrinkwrap.json | 13 ++ extensions/slack/package.json | 1 + extensions/slack/src/monitor/media.ts | 32 +--- npm-shrinkwrap.json | 51 +++++- package.json | 2 + packages/memory-host-sdk/package.json | 3 +- .../src/host/backend-config.ts | 8 +- .../memory-host-sdk/src/host/concurrency.ts | 32 ++++ .../src/host/embeddings.types.ts | 8 +- .../memory-host-sdk/src/host/internal.test.ts | 41 +++++ packages/memory-host-sdk/src/host/internal.ts | 17 +- packages/memory-host-sdk/src/host/types.ts | 8 +- pnpm-lock.yaml | 41 +++++ scripts/check-architecture-smells.mjs | 20 ++- scripts/check-temp-path-guardrails.ts | 43 ++--- scripts/deadcode-exports.baseline.mjs | 5 - scripts/generate-npm-shrinkwrap.d.mts | 5 - scripts/generate-npm-shrinkwrap.mjs | 46 ++--- .../lib/extension-import-boundary-checker.mjs | 36 ++-- scripts/lib/source-file-scan-cache.d.mts | 6 - scripts/lib/source-file-scan-cache.mjs | 33 +--- scripts/native-app-i18n.ts | 33 +--- scripts/profile-extension-memory.mjs | 25 +-- scripts/update-clawtributors.ts | 48 ++--- scripts/write-cli-startup-metadata.ts | 36 +--- src/agents/model-scan.ts | 101 ++++------- src/agents/sessions/session-manager.ts | 77 +++----- src/agents/tools/sessions-list-tool.ts | 45 ++--- src/commands/models/list.probe.ts | 26 +-- src/media-understanding/apply-capability.ts | 16 ++ src/media-understanding/apply.ts | 40 ++--- src/media-understanding/concurrency.ts | 22 --- src/utils/run-with-concurrency.test.ts | 20 +-- src/utils/run-with-concurrency.ts | 32 ++-- test/scripts/generate-npm-shrinkwrap.test.ts | 18 -- .../write-cli-startup-metadata.test.ts | 18 -- 51 files changed, 612 insertions(+), 782 deletions(-) create mode 100644 packages/memory-host-sdk/src/host/concurrency.ts create mode 100644 src/media-understanding/apply-capability.ts delete mode 100644 src/media-understanding/concurrency.ts diff --git a/config/knip.config.ts b/config/knip.config.ts index 23380993aeb..536f0c7d719 100644 --- a/config/knip.config.ts +++ b/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!"], diff --git a/extensions/discord/npm-shrinkwrap.json b/extensions/discord/npm-shrinkwrap.json index 7c1596702ac..62373104c50 100644 --- a/extensions/discord/npm-shrinkwrap.json +++ b/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", diff --git a/extensions/discord/package.json b/extensions/discord/package.json index 467ac4c543a..ef5a970ba2a 100644 --- a/extensions/discord/package.json +++ b/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" diff --git a/extensions/discord/src/monitor/thread-bindings.lifecycle.test.ts b/extensions/discord/src/monitor/thread-bindings.lifecycle.test.ts index 54da04ece92..94d7beaea79 100644 --- a/extensions/discord/src/monitor/thread-bindings.lifecycle.test.ts +++ b/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" }); diff --git a/extensions/discord/src/monitor/thread-bindings.lifecycle.ts b/extensions/discord/src/monitor/thread-bindings.lifecycle.ts index 46bfba60985..f5e87c0b467 100644 --- a/extensions/discord/src/monitor/thread-bindings.lifecycle.ts +++ b/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(params: { - items: TItem[]; - limit: number; - worker: (item: TItem, index: number) => Promise; -}): Promise { - if (params.items.length === 0) { - return []; - } - const limit = Math.max(1, Math.floor(params.limit)); - const resultsByIndex = new Map(); - 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") { diff --git a/extensions/openshell/npm-shrinkwrap.json b/extensions/openshell/npm-shrinkwrap.json index 8ab9cd7b0aa..ec5d4493517 100644 --- a/extensions/openshell/npm-shrinkwrap.json +++ b/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", diff --git a/extensions/openshell/package.json b/extensions/openshell/package.json index 4ec786d9d23..6dbc449a7c3 100644 --- a/extensions/openshell/package.json +++ b/extensions/openshell/package.json @@ -8,6 +8,7 @@ }, "type": "module", "dependencies": { + "p-limit": "7.3.0", "zod": "4.4.3" }, "devDependencies": { diff --git a/extensions/openshell/src/mirror.ts b/extensions/openshell/src/mirror.ts index c811b28693c..caf547f4a31 100644 --- a/extensions/openshell/src/mirror.ts +++ b/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 (task: () => Promise): Promise => { - if (active >= limit) { - await new Promise((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); diff --git a/extensions/qa-lab/package.json b/extensions/qa-lab/package.json index 008d4a061de..d5546363886 100644 --- a/extensions/qa-lab/package.json +++ b/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", diff --git a/extensions/qa-lab/src/character-eval.ts b/extensions/qa-lab/src/character-eval.ts index cbd8dbdfc89..69728bdfc87 100644 --- a/extensions/qa-lab/src/character-eval.ts +++ b/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( - items: readonly T[], - concurrency: number, - mapper: (item: T, index: number) => Promise, -) { - const results = Array.from({ 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( diff --git a/extensions/qa-lab/src/evidence-gallery.ts b/extensions/qa-lab/src/evidence-gallery.ts index 487a271e121..7fe9e28ee8c 100644 --- a/extensions/qa-lab/src/evidence-gallery.ts +++ b/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(task: () => Promise): Promise { - if (active >= limit) { - await new Promise((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 => { counts[entry.result.status] += 1; diff --git a/extensions/qa-lab/src/suite-planning.ts b/extensions/qa-lab/src/suite-planning.ts index 1fb2828fae7..85bee822ecb 100644 --- a/extensions/qa-lab/src/suite-planning.ts +++ b/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( sleepImpl?: (ms: number) => Promise; }, ) { - const results = Array.from({ 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( } })(); } - 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) { diff --git a/extensions/qqbot/npm-shrinkwrap.json b/extensions/qqbot/npm-shrinkwrap.json index 634d40e5c29..e3c4d7fb890 100644 --- a/extensions/qqbot/npm-shrinkwrap.json +++ b/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", diff --git a/extensions/qqbot/package.json b/extensions/qqbot/package.json index 3b50ac46046..333730c01a5 100644 --- a/extensions/qqbot/package.json +++ b/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", diff --git a/extensions/qqbot/src/engine/api/media-chunked.ts b/extensions/qqbot/src/engine/api/media-chunked.ts index 29036623a5d..0794a89524a 100644 --- a/extensions/qqbot/src/engine/api/media-chunked.ts +++ b/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>, - maxConcurrent: number, -): Promise { - for (let i = 0; i < tasks.length; i += maxConcurrent) { - const batch = tasks.slice(i, i + maxConcurrent); - await Promise.all(batch.map((task) => task())); - } -} diff --git a/extensions/slack/npm-shrinkwrap.json b/extensions/slack/npm-shrinkwrap.json index f21962dfbe5..953ce290e67 100644 --- a/extensions/slack/npm-shrinkwrap.json +++ b/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", diff --git a/extensions/slack/package.json b/extensions/slack/package.json index 8a64c7317b4..ea30dcd71ac 100644 --- a/extensions/slack/package.json +++ b/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" diff --git a/extensions/slack/src/monitor/media.ts b/extensions/slack/src/monitor/media.ts index 1ae9fb0d09c..1c6e59df28c 100644 --- a/extensions/slack/src/monitor/media.ts +++ b/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( - items: T[], - limit: number, - fn: (item: T) => Promise, -): Promise { - 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( + 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)); diff --git a/npm-shrinkwrap.json b/npm-shrinkwrap.json index bfc1eb55140..f424ccb0e91 100644 --- a/npm-shrinkwrap.json +++ b/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", diff --git a/package.json b/package.json index e3674628986..7212f848966 100644 --- a/package.json +++ b/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", diff --git a/packages/memory-host-sdk/package.json b/packages/memory-host-sdk/package.json index 5435a532627..7a75838808e 100644 --- a/packages/memory-host-sdk/package.json +++ b/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", diff --git a/packages/memory-host-sdk/src/host/backend-config.ts b/packages/memory-host-sdk/src/host/backend-config.ts index ef0c3dfd12d..a5fa7e83cf5 100644 --- a/packages/memory-host-sdk/src/host/backend-config.ts +++ b/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; diff --git a/packages/memory-host-sdk/src/host/concurrency.ts b/packages/memory-host-sdk/src/host/concurrency.ts new file mode 100644 index 00000000000..fecefa27aab --- /dev/null +++ b/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( + tasks: Array<() => Promise>, + limit: number, +): Promise { + const inFlight = new Set>(); + 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; + } +} diff --git a/packages/memory-host-sdk/src/host/embeddings.types.ts b/packages/memory-host-sdk/src/host/embeddings.types.ts index 4c7824c0393..5c3fd42af12 100644 --- a/packages/memory-host-sdk/src/host/embeddings.types.ts +++ b/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" diff --git a/packages/memory-host-sdk/src/host/internal.test.ts b/packages/memory-host-sdk/src/host/internal.test.ts index 05120b1d3df..185cfa2245e 100644 --- a/packages/memory-host-sdk/src/host/internal.test.ts +++ b/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((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"); diff --git a/packages/memory-host-sdk/src/host/internal.ts b/packages/memory-host-sdk/src/host/internal.ts index fb35d925027..bd2e23cceb6 100644 --- a/packages/memory-host-sdk/src/host/internal.ts +++ b/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( - tasks: Array<() => Promise>, - limit: number, -): Promise { - const { results, firstError, hasError } = await runTasksWithConcurrency({ - tasks, - limit, - errorMode: "stop", - }); - if (hasError) { - throw firstError; - } - return results; +export function runWithConcurrency(tasks: Array<() => Promise>, limit: number): Promise { + return runWithConcurrencyImpl(tasks, limit); } diff --git a/packages/memory-host-sdk/src/host/types.ts b/packages/memory-host-sdk/src/host/types.ts index 837ffe59462..37e32ddf9e6 100644 --- a/packages/memory-host-sdk/src/host/types.ts +++ b/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; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c805a0dae92..1365d574f8d 100644 --- a/pnpm-lock.yaml +++ b/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 diff --git a/scripts/check-architecture-smells.mjs b/scripts/check-architecture-smells.mjs index ada61bb5048..062ea0c3284 100644 --- a/scripts/check-architecture-smells.mjs +++ b/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 { diff --git a/scripts/check-temp-path-guardrails.ts b/scripts/check-temp-path-guardrails.ts index 1d4f98e73a7..97f5d3b561f 100644 --- a/scripts/check-temp-path-guardrails.ts +++ b/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 { - const output: Array = 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 => { 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() { diff --git a/scripts/deadcode-exports.baseline.mjs b/scripts/deadcode-exports.baseline.mjs index 35366106415..011f27cf75f 100644 --- a/scripts/deadcode-exports.baseline.mjs +++ b/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", diff --git a/scripts/generate-npm-shrinkwrap.d.mts b/scripts/generate-npm-shrinkwrap.d.mts index de707cfb17a..5f222636770 100644 --- a/scripts/generate-npm-shrinkwrap.d.mts +++ b/scripts/generate-npm-shrinkwrap.d.mts @@ -48,11 +48,6 @@ export function resolvePackageDirs(args: string[]): { jobs: number; packageDirs: unknown[]; }; -export function runBoundedTasks( - items: Item[], - jobs: number, - runTask: (item: Item) => Promise, -): Promise; export function resolveShrinkwrapJobs( rawValue: unknown, env?: NodeJS.ProcessEnv, diff --git a/scripts/generate-npm-shrinkwrap.mjs b/scripts/generate-npm-shrinkwrap.mjs index 5e7638f0e1a..ef61a707158 100644 --- a/scripts/generate-npm-shrinkwrap.mjs +++ b/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) { diff --git a/scripts/lib/extension-import-boundary-checker.mjs b/scripts/lib/extension-import-boundary-checker.mjs index c09c19e53d2..faa33fb50df 100644 --- a/scripts/lib/extension-import-boundary-checker.mjs +++ b/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); }); diff --git a/scripts/lib/source-file-scan-cache.d.mts b/scripts/lib/source-file-scan-cache.d.mts index c265958eb95..a42d2d06b80 100644 --- a/scripts/lib/source-file-scan-cache.d.mts +++ b/scripts/lib/source-file-scan-cache.d.mts @@ -1,9 +1,3 @@ -export function mapWithConcurrency( - items: Item[], - concurrency: number, - mapper: (item: Item, index: number) => Promise, -): Promise; - export function collectSourceFileContents(params: { repoRoot: string; scanRoots: string[]; diff --git a/scripts/lib/source-file-scan-cache.mjs b/scripts/lib/source-file-scan-cache.mjs index f6991224348..66aff1e3dc9 100644 --- a/scripts/lib/source-file-scan-cache.mjs +++ b/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, + }, ); })(); diff --git a/scripts/native-app-i18n.ts b/scripts/native-app-i18n.ts index fe606eaca9d..f5847d89ab0 100644 --- a/scripts/native-app-i18n.ts +++ b/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( - values: readonly T[], - limit: number, - run: (value: T) => Promise, -): Promise { - const results = Array(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 { @@ -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; diff --git a/scripts/profile-extension-memory.mjs b/scripts/profile-extension-memory.mjs index d937be5f53f..d18531c7ba4 100644 --- a/scripts/profile-extension-memory.mjs +++ b/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)); diff --git a/scripts/update-clawtributors.ts b/scripts/update-clawtributors.ts index e8392e5cc1a..cf2b870b269 100644 --- a/scripts/update-clawtributors.ts +++ b/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, ): Promise { - 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( - items: T[], - limit: number, - mapper: (item: T, index: number) => Promise, -): Promise { - 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); diff --git a/scripts/write-cli-startup-metadata.ts b/scripts/write-cli-startup-metadata.ts index 871676d8966..4bb2d7b7d92 100644 --- a/scripts/write-cli-startup-metadata.ts +++ b/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( - values: readonly T[], - limit: number, - run: (value: T) => Promise, -): Promise { - 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 { - 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, }; diff --git a/src/agents/model-scan.ts b/src/agents/model-scan.ts index c8454d4dc51..d7a83eca05f 100644 --- a/src/agents/model-scan.ts +++ b/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( - items: T[], - concurrency: number, - fn: (item: T, index: number) => Promise, - opts?: { onProgress?: (completed: number, total: number) => void }, -): Promise { - 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 { @@ -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 }, ); } diff --git a/src/agents/sessions/session-manager.ts b/src/agents/sessions/session-manager.ts index 58961ecead9..6ec1d9f0786 100644 --- a/src/agents/sessions/session-manager.ts +++ b/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>(); - let nextIndex = 0; - - const startNext = (): void => { - const index = nextIndex++; - const file = files[index]; - if (!file) { - return; - } - const task: Promise = 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) { diff --git a/src/agents/tools/sessions-list-tool.ts b/src/agents/tools/sessions-list-tool.ts index 292eca6efbf..954dd136f6b 100644 --- a/src/agents/tools/sessions-list-tool.ts +++ b/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 }>({ 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 = diff --git a/src/commands/models/list.probe.ts b/src/commands/models/list.probe.ts index 31d9cee8283..ab44c5e1cf7 100644 --- a/src/commands/models/list.probe.ts +++ b/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 = 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. */ diff --git a/src/media-understanding/apply-capability.ts b/src/media-understanding/apply-capability.ts new file mode 100644 index 00000000000..d1d6f557085 --- /dev/null +++ b/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[0], +): Promise> | undefined> { + try { + return await runCapability(params); + } catch (err) { + if (shouldLogVerbose()) { + logVerbose(`Media understanding task failed: ${String(err)}`); + } + return undefined; + } +} diff --git a/src/media-understanding/apply.ts b/src/media-understanding/apply.ts index 4d5f1fd020c..b0c3163c461 100644 --- a/src/media-understanding/apply.ts +++ b/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) { diff --git a/src/media-understanding/concurrency.ts b/src/media-understanding/concurrency.ts deleted file mode 100644 index 0751f3ec03a..00000000000 --- a/src/media-understanding/concurrency.ts +++ /dev/null @@ -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( - tasks: Array<() => Promise>, - limit: number, -): Promise { - 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; -} diff --git a/src/utils/run-with-concurrency.test.ts b/src/utils/run-with-concurrency.test.ts index 9bcd1150c86..dbe331f32be 100644 --- a/src/utils/run-with-concurrency.test.ts +++ b/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(); diff --git a/src/utils/run-with-concurrency.ts b/src/utils/run-with-concurrency.ts index dad82129d1a..d7ab87a218c 100644 --- a/src/utils/run-with-concurrency.ts +++ b/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( 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 }; } diff --git a/test/scripts/generate-npm-shrinkwrap.test.ts b/test/scripts/generate-npm-shrinkwrap.test.ts index 5eac9289a24..55af543dd26 100644 --- a/test/scripts/generate-npm-shrinkwrap.test.ts +++ b/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((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); diff --git a/test/scripts/write-cli-startup-metadata.test.ts b/test/scripts/write-cli-startup-metadata.test.ts index 2ac02bfc2bc..513ab8dcb4d 100644 --- a/test/scripts/write-cli-startup-metadata.test.ts +++ b/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))"], {