refactor(concurrency): use p-map skip results

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

@@ -6,7 +6,7 @@ import type {
   OpenKeyedStoreOptions,
   PluginStateKeyedStore,
 } from "openclaw/plugin-sdk/plugin-state-runtime";
-import pMap from "p-map";
+import pMap, { pMapSkip } from "p-map";
 
 const LEGACY_IMPORT_RUN_READ_CONCURRENCY = 16;
 
@@ -474,13 +474,12 @@ export async function readLegacyMemoryWikiImportRunRecords(
       }
       throw error;
     });
-  const records = await pMap(
+  return 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);
+      return normalizeMemoryWikiImportRunRecord(JSON.parse(raw) as unknown) ?? pMapSkip;
     },
     { concurrency: LEGACY_IMPORT_RUN_READ_CONCURRENCY, stopOnError: true },
   );
-  return records.filter((entry): entry is ChatGptImportRunRecord => entry !== null);
 }

extensions/memory-wiki/src/query.ts

@@ -18,7 +18,7 @@ import {
   normalizeLowercaseStringOrEmpty,
   uniqueStrings,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
-import pMap from "p-map";
+import pMap, { pMapSkip } from "p-map";
 import type { OpenClawConfig } from "../api.js";
 import { assessClaimFreshness, isClaimContestedStatus } from "./claim-health.js";
 import type { ResolvedMemoryWikiConfig, WikiSearchBackend, WikiSearchCorpus } from "./config.js";
@@ -273,17 +273,16 @@ async function readQueryableWikiPagesByPaths(
   rootDir: string,
   files: string[],
 ): Promise<QueryableWikiPage[]> {
-  const pages = await pMap(
+  return 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;
+      return summary ? { ...summary, raw } : pMapSkip;
     },
     { concurrency: QUERY_PAGE_READ_CONCURRENCY, stopOnError: true },
   );
-  return pages.flatMap((page) => (page ? [page] : []));
 }
 
 function parseClaimsDigest(raw: string): QueryDigestClaim[] {

extensions/slack/src/monitor/media.ts

@@ -8,7 +8,7 @@ import {
   normalizeLowercaseStringOrEmpty,
   normalizeOptionalLowercaseString,
 } from "openclaw/plugin-sdk/string-coerce-runtime";
-import pMap from "p-map";
+import pMap, { pMapSkip } 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";
@@ -321,7 +321,7 @@ export async function resolveSlackMedia(params: {
 
   const resolved = await pMap(
     limitedFiles,
-    async (file) => {
+    async (file): Promise<SlackMediaResult | typeof pMapSkip> => {
       // Audio preflight keys the original event file object so admission can
       // reuse that exact download without turning this into a persistent cache.
       const preloaded = params.preloadedMedia?.get(file);
@@ -331,7 +331,7 @@ export async function resolveSlackMedia(params: {
       const eventUrl = file.url_private_download ?? file.url_private;
       const url = eventUrl ?? (await fetchFreshSlackFileUrl({ file, client: params.client }));
       if (!url) {
-        return null;
+        return pMapSkip;
       }
       const result = await downloadSlackMediaFile({
         file,
@@ -343,14 +343,14 @@ export async function resolveSlackMedia(params: {
         abortSignal: params.abortSignal,
       }).catch(() => null);
       if (result || !eventUrl) {
-        return result;
+        return result ?? pMapSkip;
       }
 
       const freshUrl = await fetchFreshSlackFileUrl({ file, client: params.client });
       if (!freshUrl) {
-        return null;
+        return pMapSkip;
       }
-      return await downloadSlackMediaFile({
+      const retryResult = await downloadSlackMediaFile({
         file,
         url: freshUrl,
         token: params.token,
@@ -359,12 +359,12 @@ export async function resolveSlackMedia(params: {
         totalTimeoutMs: params.totalTimeoutMs,
         abortSignal: params.abortSignal,
       }).catch(() => null);
+      return retryResult ?? pMapSkip;
     },
     { concurrency: MAX_SLACK_MEDIA_CONCURRENCY, stopOnError: true },
   );
 
-  const results = resolved.filter((entry): entry is SlackMediaResult => Boolean(entry));
-  return results.length > 0 ? results : null;
+  return resolved.length > 0 ? resolved : null;
 }
 
 /** Extracts text and media from forwarded-message attachments. Returns null when empty. */

scripts/check-temp-path-guardrails.ts

@@ -2,7 +2,7 @@
 import { execFileSync } from "node:child_process";
 import fs from "node:fs/promises";
 import path from "node:path";
-import pMap from "p-map";
+import pMap, { pMapSkip } from "p-map";
 
 type QuoteChar = "'" | '"' | "`";
 
@@ -226,9 +226,9 @@ async function readRuntimeSourceFiles(
   repoRoot: string,
   absolutePaths: string[],
 ): Promise<RuntimeSourceGuardrailFile[]> {
-  const output = await pMap(
+  return await pMap(
     absolutePaths,
-    async (absolutePath): Promise<RuntimeSourceGuardrailFile | null> => {
+    async (absolutePath) => {
       try {
         return {
           relativePath: path.relative(repoRoot, absolutePath),
@@ -236,12 +236,11 @@ async function readRuntimeSourceFiles(
         };
       } catch {
         // File tracked by git but deleted on disk (e.g. pending deletion).
-        return null;
+        return pMapSkip;
       }
     },
     { concurrency: FILE_READ_CONCURRENCY, stopOnError: false },
   );
-  return output.filter((entry): entry is RuntimeSourceGuardrailFile => entry !== null);
 }
 
 async function main() {

scripts/update-clawtributors.ts

@@ -2,7 +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 pMap, { pMapSkip } from "p-map";
 import { expectDefined } from "../packages/normalization-core/src/expect.js";
 import type { ApiContributor, Entry, MapConfig, User } from "./update-clawtributors.types.js";
 
@@ -626,7 +626,7 @@ async function filterVisibleEntries(
   entriesResult: Entry[],
   hiddenLogins: ReadonlySet<string>,
 ): Promise<Entry[]> {
-  const results = await pMap(
+  return await pMap(
     entriesResult,
     async (entry) => {
       const login = entry.login ?? entry.key;
@@ -635,13 +635,12 @@ async function filterVisibleEntries(
       }
       const normalized = normalizeLogin(login)?.toLowerCase();
       if (normalized && hiddenLogins.has(normalized)) {
-        return null;
+        return pMapSkip;
       }
-      return (await isDefaultGitHubAvatar(login)) ? null : entry;
+      return (await isDefaultGitHubAvatar(login)) ? pMapSkip : entry;
     },
     { concurrency: 8, stopOnError: true },
   );
-  return results.filter((entry): entry is Entry => entry !== null);
 }
 
 function readImageDimensions(buffer: Buffer): { width: number; height: number } | null {

src/agents/sessions/session-manager.ts

@@ -19,7 +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 pMap, { pMapSkip } from "p-map";
 import {
   appendTranscriptEventSync,
   appendTranscriptMessageSync,
@@ -1457,9 +1457,8 @@ async function listSessionsFromDir(
   progressTotal?: number,
   cwd?: string,
 ): Promise<SessionInfo[]> {
-  const sessions: SessionInfo[] = [];
   if (!existsSync(dir)) {
-    return sessions;
+    return [];
   }
 
   try {
@@ -1468,13 +1467,13 @@ async function listSessionsFromDir(
     const total = progressTotal ?? files.length;
 
     let loaded = 0;
-    const results = await pMap(
+    const sessions = await pMap(
       files,
       async (file) => {
         try {
-          return await buildSessionInfo(file);
+          return (await buildSessionInfo(file)) ?? pMapSkip;
         } catch {
-          return null;
+          return pMapSkip;
         } finally {
           loaded++;
           onProgress?.(progressOffset + loaded, total);
@@ -1482,16 +1481,10 @@ async function listSessionsFromDir(
       },
       { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false },
     );
-    for (const info of results) {
-      if (info && sessionInfoMatchesCwd(info, cwd)) {
-        sessions.push(info);
-      }
-    }
+    return sessions.filter((info) => sessionInfoMatchesCwd(info, cwd));
   } catch {
-    // Return empty list on error
+    return [];
   }
-
-  return sessions;
 }
 
 /**
@@ -3290,16 +3283,15 @@ export class SessionManager {
 
       // Process all files with progress tracking
       let loaded = 0;
-      const sessions: SessionInfo[] = [];
       const allFiles = dirFiles.flat();
 
-      const results = await pMap(
+      const sessions = await pMap(
         allFiles,
         async (file) => {
           try {
-            return await buildSessionInfo(file);
+            return (await buildSessionInfo(file)) ?? pMapSkip;
           } catch {
-            return null;
+            return pMapSkip;
           } finally {
             loaded++;
             onProgress?.(loaded, totalFiles);
@@ -3308,12 +3300,6 @@ export class SessionManager {
         { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false },
       );
 
-      for (const info of results) {
-        if (info) {
-          sessions.push(info);
-        }
-      }
-
       sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime());
       return sessions;
     } catch {