perf: speed up test parallelism

src/config/sessions/store.ts

@@ -159,6 +159,26 @@ export function clearSessionStoreCacheForTest(): void {
   LOCK_QUEUES.clear();
 }
 
+export async function drainSessionStoreLockQueuesForTest(): Promise<void> {
+  while (LOCK_QUEUES.size > 0) {
+    const queues = [...LOCK_QUEUES.values()];
+    for (const queue of queues) {
+      for (const task of queue.pending) {
+        task.reject(new Error("session store queue cleared for test"));
+      }
+      queue.pending.length = 0;
+    }
+    const activeDrains = queues.flatMap((queue) =>
+      queue.drainPromise ? [queue.drainPromise] : [],
+    );
+    if (activeDrains.length === 0) {
+      LOCK_QUEUES.clear();
+      return;
+    }
+    await Promise.allSettled(activeDrains);
+  }
+}
+
 /** Expose lock queue size for tests. */
 export function getSessionStoreLockQueueSizeForTest(): number {
   return LOCK_QUEUES.size;
@@ -602,6 +622,7 @@ type SessionStoreLockTask = {
 type SessionStoreLockQueue = {
   running: boolean;
   pending: SessionStoreLockTask[];
+  drainPromise: Promise<void> | null;
 };
 
 const LOCK_QUEUES = new Map<string, SessionStoreLockQueue>();
@@ -686,63 +707,71 @@ function getOrCreateLockQueue(storePath: string): SessionStoreLockQueue {
   if (existing) {
     return existing;
   }
-  const created: SessionStoreLockQueue = { running: false, pending: [] };
+  const created: SessionStoreLockQueue = { running: false, pending: [], drainPromise: null };
   LOCK_QUEUES.set(storePath, created);
   return created;
 }
 
 async function drainSessionStoreLockQueue(storePath: string): Promise<void> {
   const queue = LOCK_QUEUES.get(storePath);
-  if (!queue || queue.running) {
+  if (!queue) {
+    return;
+  }
+  if (queue.drainPromise) {
+    await queue.drainPromise;
     return;
   }
   queue.running = true;
-  try {
-    while (queue.pending.length > 0) {
-      const task = queue.pending.shift();
-      if (!task) {
-        continue;
-      }
+  queue.drainPromise = (async () => {
+    try {
+      while (queue.pending.length > 0) {
+        const task = queue.pending.shift();
+        if (!task) {
+          continue;
+        }
 
-      const remainingTimeoutMs = task.timeoutMs ?? Number.POSITIVE_INFINITY;
-      if (task.timeoutMs != null && remainingTimeoutMs <= 0) {
-        task.reject(lockTimeoutError(storePath));
-        continue;
-      }
+        const remainingTimeoutMs = task.timeoutMs ?? Number.POSITIVE_INFINITY;
+        if (task.timeoutMs != null && remainingTimeoutMs <= 0) {
+          task.reject(lockTimeoutError(storePath));
+          continue;
+        }
 
-      let lock: { release: () => Promise<void> } | undefined;
-      let result: unknown;
-      let failed: unknown;
-      let hasFailure = false;
-      try {
-        lock = await acquireSessionWriteLock({
-          sessionFile: storePath,
-          timeoutMs: remainingTimeoutMs,
-          staleMs: task.staleMs,
+        let lock: { release: () => Promise<void> } | undefined;
+        let result: unknown;
+        let failed: unknown;
+        let hasFailure = false;
+        try {
+          lock = await acquireSessionWriteLock({
+            sessionFile: storePath,
+            timeoutMs: remainingTimeoutMs,
+            staleMs: task.staleMs,
+          });
+          result = await task.fn();
+        } catch (err) {
+          hasFailure = true;
+          failed = err;
+        } finally {
+          await lock?.release().catch(() => undefined);
+        }
+        if (hasFailure) {
+          task.reject(failed);
+          continue;
+        }
+        task.resolve(result);
+      }
+    } finally {
+      queue.running = false;
+      queue.drainPromise = null;
+      if (queue.pending.length === 0) {
+        LOCK_QUEUES.delete(storePath);
+      } else {
+        queueMicrotask(() => {
+          void drainSessionStoreLockQueue(storePath);
         });
-        result = await task.fn();
-      } catch (err) {
-        hasFailure = true;
-        failed = err;
-      } finally {
-        await lock?.release().catch(() => undefined);
       }
-      if (hasFailure) {
-        task.reject(failed);
-        continue;
-      }
-      task.resolve(result);
     }
-  } finally {
-    queue.running = false;
-    if (queue.pending.length === 0) {
-      LOCK_QUEUES.delete(storePath);
-    } else {
-      queueMicrotask(() => {
-        void drainSessionStoreLockQueue(storePath);
-      });
-    }
-  }
+  })();
+  await queue.drainPromise;
 }
 
 async function withSessionStoreLock<T>(

src/test-utils/session-state-cleanup.test.ts

@@ -0,0 +1,58 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import {
+  getSessionStoreLockQueueSizeForTest,
+  withSessionStoreLockForTest,
+} from "../config/sessions/store.js";
+import { cleanupSessionStateForTest } from "./session-state-cleanup.js";
+
+function createDeferred<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((nextResolve, nextReject) => {
+    resolve = nextResolve;
+    reject = nextReject;
+  });
+  return { promise, resolve, reject };
+}
+
+describe("cleanupSessionStateForTest", () => {
+  afterEach(async () => {
+    await cleanupSessionStateForTest();
+  });
+
+  it("waits for in-flight session store locks before clearing test state", async () => {
+    const fixtureRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-cleanup-"));
+    const storePath = path.join(fixtureRoot, "openclaw-sessions.json");
+    const started = createDeferred<void>();
+    const release = createDeferred<void>();
+    try {
+      const running = withSessionStoreLockForTest(storePath, async () => {
+        started.resolve();
+        await release.promise;
+      });
+
+      await started.promise;
+      expect(getSessionStoreLockQueueSizeForTest()).toBe(1);
+
+      let settled = false;
+      const cleanupPromise = cleanupSessionStateForTest().then(() => {
+        settled = true;
+      });
+
+      await new Promise((resolve) => setTimeout(resolve, 25));
+      expect(settled).toBe(false);
+
+      release.resolve();
+      await running;
+      await cleanupPromise;
+
+      expect(getSessionStoreLockQueueSizeForTest()).toBe(0);
+    } finally {
+      release.resolve();
+      await fs.rm(fixtureRoot, { recursive: true, force: true });
+    }
+  });
+});

src/test-utils/session-state-cleanup.ts

@@ -1,8 +1,12 @@
 import { drainSessionWriteLockStateForTest } from "../agents/session-write-lock.js";
-import { clearSessionStoreCacheForTest } from "../config/sessions/store.js";
+import {
+  clearSessionStoreCacheForTest,
+  drainSessionStoreLockQueuesForTest,
+} from "../config/sessions/store.js";
 import { drainFileLockStateForTest } from "../infra/file-lock.js";
 
 export async function cleanupSessionStateForTest(): Promise<void> {
+  await drainSessionStoreLockQueuesForTest();
   clearSessionStoreCacheForTest();
   await drainFileLockStateForTest();
   await drainSessionWriteLockStateForTest();