test: inject thread-safe base seams

src/auto-reply/reply/abort.test.ts

@@ -1,10 +1,11 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import type { SubagentRunRecord } from "../../agents/subagent-registry.js";
 import type { OpenClawConfig } from "../../config/config.js";
 import {
+  __testing as abortTesting,
   getAbortMemory,
   getAbortMemorySizeForTest,
   isAbortRequestText,
@@ -17,6 +18,7 @@ import {
   tryFastAbortFromMessage,
 } from "./abort.js";
 import { enqueueFollowupRun, getFollowupQueueDepth, type FollowupRun } from "./queue.js";
+import { __testing as queueCleanupTesting } from "./queue/cleanup.js";
 import { initSessionState } from "./session.js";
 import { buildTestCtx } from "./test-ctx.js";
 
@@ -26,7 +28,7 @@ vi.mock("../../agents/pi-embedded.js", () => ({
 }));
 
 const commandQueueMocks = vi.hoisted(() => ({
-  clearCommandLane: vi.fn(),
+  clearCommandLane: vi.fn(() => 1),
 }));
 
 vi.mock("../../process/command-queue.js", () => commandQueueMocks);
@@ -162,8 +164,29 @@ describe("abort detection", () => {
     expect(commandQueueMocks.clearCommandLane).toHaveBeenCalledWith(`session:${sessionKey}`);
   }
 
+  beforeEach(() => {
+    abortTesting.setDepsForTests({
+      getAcpSessionManager: (() =>
+        ({
+          resolveSession: acpManagerMocks.resolveSession,
+          cancelSession: acpManagerMocks.cancelSession,
+        }) as never) as never,
+      abortEmbeddedPiRun: () => true,
+      listSubagentRunsForController: subagentRegistryMocks.listSubagentRunsForRequester,
+      markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
+    });
+    queueCleanupTesting.setDepsForTests({
+      resolveEmbeddedSessionLane: (key) => `session:${key.trim() || "main"}`,
+      clearCommandLane: commandQueueMocks.clearCommandLane,
+    });
+    commandQueueMocks.clearCommandLane.mockClear().mockReturnValue(1);
+  });
+
   afterEach(() => {
     resetAbortMemoryForTest();
+    abortTesting.resetDepsForTests();
+    queueCleanupTesting.resetDepsForTests();
+    commandQueueMocks.clearCommandLane.mockClear().mockReturnValue(1);
     acpManagerMocks.resolveSession.mockReset().mockReturnValue({ kind: "none" });
     acpManagerMocks.cancelSession.mockReset().mockResolvedValue(undefined);
   });

src/auto-reply/reply/abort.ts

@@ -47,6 +47,35 @@ export {
   setAbortMemory,
 };
 
+const defaultAbortDeps = {
+  getAcpSessionManager,
+  abortEmbeddedPiRun,
+  listSubagentRunsForController,
+  markSubagentRunTerminated,
+};
+
+const abortDeps = {
+  ...defaultAbortDeps,
+};
+
+export const __testing = {
+  setDepsForTests(deps: Partial<typeof defaultAbortDeps> | undefined): void {
+    abortDeps.getAcpSessionManager =
+      deps?.getAcpSessionManager ?? defaultAbortDeps.getAcpSessionManager;
+    abortDeps.abortEmbeddedPiRun = deps?.abortEmbeddedPiRun ?? defaultAbortDeps.abortEmbeddedPiRun;
+    abortDeps.listSubagentRunsForController =
+      deps?.listSubagentRunsForController ?? defaultAbortDeps.listSubagentRunsForController;
+    abortDeps.markSubagentRunTerminated =
+      deps?.markSubagentRunTerminated ?? defaultAbortDeps.markSubagentRunTerminated;
+  },
+  resetDepsForTests(): void {
+    abortDeps.getAcpSessionManager = defaultAbortDeps.getAcpSessionManager;
+    abortDeps.abortEmbeddedPiRun = defaultAbortDeps.abortEmbeddedPiRun;
+    abortDeps.listSubagentRunsForController = defaultAbortDeps.listSubagentRunsForController;
+    abortDeps.markSubagentRunTerminated = defaultAbortDeps.markSubagentRunTerminated;
+  },
+};
+
 export function formatAbortReplyText(stoppedSubagents?: number): string {
   if (typeof stoppedSubagents !== "number" || stoppedSubagents <= 0) {
     return "⚙️ Agent was aborted.";
@@ -107,7 +136,7 @@ export function stopSubagentsForRequester(params: {
   if (!requesterKey) {
     return { stopped: 0 };
   }
-  const runs = listSubagentRunsForController(requesterKey);
+  const runs = abortDeps.listSubagentRunsForController(requesterKey);
   if (runs.length === 0) {
     return { stopped: 0 };
   }
@@ -134,9 +163,9 @@ export function stopSubagentsForRequester(params: {
       }
       const entry = store[childKey];
       const sessionId = entry?.sessionId;
-      const aborted = sessionId ? abortEmbeddedPiRun(sessionId) : false;
+      const aborted = sessionId ? abortDeps.abortEmbeddedPiRun(sessionId) : false;
       const markedTerminated =
-        markSubagentRunTerminated({
+        abortDeps.markSubagentRunTerminated({
           runId: run.runId,
           childSessionKey: childKey,
           reason: "killed",
@@ -198,7 +227,7 @@ export async function tryFastAbortFromMessage(params: {
     const store = loadSessionStore(storePath);
     const { entry, key, legacyKeys } = resolveSessionEntryForKey(store, targetKey);
     const resolvedTargetKey = key ?? targetKey;
-    const acpManager = getAcpSessionManager();
+    const acpManager = abortDeps.getAcpSessionManager();
     const acpResolution = acpManager.resolveSession({
       cfg,
       sessionKey: resolvedTargetKey,
@@ -217,7 +246,7 @@ export async function tryFastAbortFromMessage(params: {
       }
     }
     const sessionId = entry?.sessionId;
-    const aborted = sessionId ? abortEmbeddedPiRun(sessionId) : false;
+    const aborted = sessionId ? abortDeps.abortEmbeddedPiRun(sessionId) : false;
     const cleared = clearSessionQueues([resolvedTargetKey, sessionId]);
     if (cleared.followupCleared > 0 || cleared.laneCleared > 0) {
       logVerbose(

src/auto-reply/reply/queue/cleanup.ts

@@ -9,6 +9,29 @@ export type ClearSessionQueueResult = {
   keys: string[];
 };
 
+const defaultQueueCleanupDeps = {
+  resolveEmbeddedSessionLane,
+  clearCommandLane,
+};
+
+const queueCleanupDeps = {
+  ...defaultQueueCleanupDeps,
+};
+
+export const __testing = {
+  setDepsForTests(deps: Partial<typeof defaultQueueCleanupDeps> | undefined): void {
+    queueCleanupDeps.resolveEmbeddedSessionLane =
+      deps?.resolveEmbeddedSessionLane ?? defaultQueueCleanupDeps.resolveEmbeddedSessionLane;
+    queueCleanupDeps.clearCommandLane =
+      deps?.clearCommandLane ?? defaultQueueCleanupDeps.clearCommandLane;
+  },
+  resetDepsForTests(): void {
+    queueCleanupDeps.resolveEmbeddedSessionLane =
+      defaultQueueCleanupDeps.resolveEmbeddedSessionLane;
+    queueCleanupDeps.clearCommandLane = defaultQueueCleanupDeps.clearCommandLane;
+  },
+};
+
 export function clearSessionQueues(keys: Array<string | undefined>): ClearSessionQueueResult {
   const seen = new Set<string>();
   let followupCleared = 0;
@@ -24,7 +47,9 @@ export function clearSessionQueues(keys: Array<string | undefined>): ClearSessio
     clearedKeys.push(cleaned);
     followupCleared += clearFollowupQueue(cleaned);
     clearFollowupDrainCallback(cleaned);
-    laneCleared += clearCommandLane(resolveEmbeddedSessionLane(cleaned));
+    laneCleared += queueCleanupDeps.clearCommandLane(
+      queueCleanupDeps.resolveEmbeddedSessionLane(cleaned),
+    );
   }
 
   return { followupCleared, laneCleared, keys: clearedKeys };