fix(agents): honor aborts during session lock acquisition

src/agents/embedded-agent-runner/run/attempt.abort-race.test.ts

@@ -0,0 +1,65 @@
+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  cleanupTempPaths,
+  createContextEngineAttemptRunner,
+  createContextEngineBootstrapAndAssemble,
+  getHoisted,
+  preloadRunEmbeddedAttemptForTests,
+  resetEmbeddedAttemptHarness,
+} from "./attempt.spawn-workspace.test-support.js";
+
+const hoisted = getHoisted();
+const tempPaths: string[] = [];
+
+describe("runEmbeddedAttempt abort races", () => {
+  beforeAll(async () => {
+    await preloadRunEmbeddedAttemptForTests();
+  });
+
+  beforeEach(() => {
+    resetEmbeddedAttemptHarness();
+  });
+
+  afterEach(async () => {
+    await cleanupTempPaths(tempPaths);
+    tempPaths.length = 0;
+  });
+
+  it("stops before session creation when aborted during eager lock acquisition", async () => {
+    const abortController = new AbortController();
+    const prompt = vi.fn(async () => {});
+    const abortError = new Error("stopped during lock acquisition");
+    abortError.name = "AbortError";
+    let markLockRequested!: () => void;
+    let observedSignal: AbortSignal | undefined;
+    const lockRequested = new Promise<void>((resolve) => {
+      markLockRequested = resolve;
+    });
+    hoisted.acquireSessionWriteLockMock.mockImplementationOnce(async (params) => {
+      observedSignal = params.signal;
+      markLockRequested();
+      await new Promise<void>((resolve) => {
+        params.signal?.addEventListener("abort", () => resolve(), { once: true });
+      });
+      throw params.signal?.reason;
+    });
+
+    const attempt = createContextEngineAttemptRunner({
+      contextEngine: createContextEngineBootstrapAndAssemble(),
+      sessionKey: "agent:main:telegram:direct:123",
+      tempPaths,
+      sessionPrompt: prompt,
+      attemptOverrides: {
+        abortSignal: abortController.signal,
+      },
+    });
+    await lockRequested;
+    abortController.abort(abortError);
+
+    await expect(attempt).rejects.toBe(abortError);
+
+    expect(hoisted.createAgentSessionMock).not.toHaveBeenCalled();
+    expect(prompt).not.toHaveBeenCalled();
+    expect(observedSignal).toBe(abortController.signal);
+  });
+});

src/agents/embedded-agent-runner/run/attempt.session-lock.ts

@@ -1149,21 +1149,23 @@ export type EmbeddedAttemptSessionLockController = {
 
 export async function createEmbeddedAttemptSessionLockController(params: {
   acquireSessionWriteLock: AcquireSessionWriteLock;
+  initialAcquireSignal?: AbortSignal;
   lockOptions: LockOptions;
   mergePromptReleasedSessionEntries?: (
     entries: readonly PromptReleasedSessionEntry[],
   ) => Promise<PromptReleasedSessionMergeResult | void> | PromptReleasedSessionMergeResult | void;
   reloadPromptReleasedSessionFile?: () => Promise<void> | void;
 }): Promise<EmbeddedAttemptSessionLockController> {
-  const acquireLock = async (): Promise<SessionLock> =>
+  const acquireLock = async (signal?: AbortSignal): Promise<SessionLock> =>
     await params.acquireSessionWriteLock({
       sessionFile: params.lockOptions.sessionFile,
       timeoutMs: params.lockOptions.timeoutMs,
       staleMs: params.lockOptions.staleMs,
       maxHoldMs: params.lockOptions.maxHoldMs,
+      ...(signal ? { signal } : {}),
     });
 
-  let heldLock: SessionLock | undefined = await acquireLock();
+  let heldLock: SessionLock | undefined = await acquireLock(params.initialAcquireSignal);
   const activeWriteLock = new AsyncLocalStorage<ActiveWriteLockState>();
   let ownedPublicationQueue: Promise<void> = Promise.resolve();
   let fenceFingerprint: SessionFileFingerprint | undefined;

src/agents/embedded-agent-runner/run/attempt.ts

@@ -2146,6 +2146,7 @@ export async function runEmbeddedAttempt(
     let sessionManager: ReturnType<typeof guardSessionManager> | undefined;
     const sessionLockController = await createEmbeddedAttemptSessionLockController({
       acquireSessionWriteLock,
+      initialAcquireSignal: params.abortSignal,
       lockOptions: {
         sessionFile: params.sessionFile,
         ...sessionWriteLockOptions,
@@ -2181,6 +2182,9 @@ export async function runEmbeddedAttempt(
         sessionLockController.withSessionWriteLock(operation),
       );
     armExternalAbortSignal();
+    // The signal can fire while the eager session lock is being acquired.
+    // Recheck after arming so a stopped run never reaches session creation or provider prompt.
+    await throwIfAttemptAbortSignalFiredAfterPrepCleanup();
 
     let session: Awaited<ReturnType<typeof createAgentSession>>["session"] | undefined;
     let removeToolResultContextGuard: (() => void) | undefined;

src/agents/session-write-lock.test.ts

@@ -244,6 +244,29 @@ describe("acquireSessionWriteLock", () => {
     });
   });
 
+  it("cancels a contended infinite acquisition without leaving a lock waiter", async () => {
+    await withTempSessionLockFile(async ({ sessionFile }) => {
+      const heldLock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+      const abortController = new AbortController();
+      const abortError = new Error("stop requested");
+      abortError.name = "AbortError";
+      const pendingLock = acquireSessionWriteLock({
+        sessionFile,
+        timeoutMs: Number.POSITIVE_INFINITY,
+        signal: abortController.signal,
+      });
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 20);
+      });
+      abortController.abort(abortError);
+
+      await expect(pendingLock).rejects.toBe(abortError);
+      await heldLock.release();
+      const nextLock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+      await nextLock.release();
+    });
+  });
+
   it("does not reenter locks by default through symlinked session paths", async () => {
     await withSymlinkedSessionPaths(async ({ sessionReal, sessionLink }) => {
       const lock = await acquireSessionWriteLock({ sessionFile: sessionReal, timeoutMs: 500 });

src/agents/session-write-lock.ts

@@ -50,6 +50,7 @@ const WATCHDOG_STATE_KEY = Symbol.for("openclaw.sessionWriteLockWatchdogState");
 const DEFAULT_SESSION_WRITE_LOCK_STALE_MS = 30 * 60 * 1000;
 const DEFAULT_SESSION_WRITE_LOCK_MAX_HOLD_MS = 5 * 60 * 1000;
 const DEFAULT_SESSION_WRITE_LOCK_ACQUIRE_TIMEOUT_MS = 60_000;
+const ABORTABLE_SESSION_WRITE_LOCK_POLL_MS = 100;
 const DEFAULT_WATCHDOG_INTERVAL_MS = 60_000;
 const DEFAULT_TIMEOUT_GRACE_MS = 2 * 60 * 1000;
 const REPORT_ONLY_STALE_LOCK_REASONS = new Set(["too-old", "hold-exceeded"]);
@@ -888,9 +889,22 @@ export async function acquireSessionWriteLock(params: {
   staleMs?: number;
   maxHoldMs?: number;
   allowReentrant?: boolean;
+  signal?: AbortSignal;
 }): Promise<{
   release: () => Promise<void>;
 }> {
+  const throwIfAborted = () => {
+    if (!params.signal?.aborted) {
+      return;
+    }
+    if (params.signal.reason instanceof Error) {
+      throw params.signal.reason;
+    }
+    const error = new Error("request aborted", { cause: params.signal.reason });
+    error.name = "AbortError";
+    throw error;
+  };
+  throwIfAborted();
   registerCleanupHandlers();
   const allowReentrant = params.allowReentrant ?? false;
   const defaultOptions = resolveSessionWriteLockOptions();
@@ -908,6 +922,7 @@ export async function acquireSessionWriteLock(params: {
   const startedAtMs = Date.now();
 
   while (true) {
+    throwIfAborted();
     const remainingTimeoutMs = resolveRemainingAcquireTimeoutMs(timeoutMs, startedAtMs, Date.now());
     if (remainingTimeoutMs <= 0) {
       const payload = await readLockPayload(lockPath);
@@ -926,9 +941,12 @@ export async function acquireSessionWriteLock(params: {
       throw new SessionWriteLockTimeoutError({ timeoutMs, owner, lockPath });
     }
     try {
+      const acquireAttemptTimeoutMs = params.signal
+        ? Math.min(remainingTimeoutMs, ABORTABLE_SESSION_WRITE_LOCK_POLL_MS)
+        : remainingTimeoutMs;
       const lock = await SESSION_LOCKS.acquire(sessionFile, {
         staleMs,
-        timeoutMs: remainingTimeoutMs,
+        timeoutMs: acquireAttemptTimeoutMs,
         retry: { minTimeout: 50, maxTimeout: 1000, factor: 1 },
         staleRecovery: "remove-if-unchanged",
         allowReentrant,
@@ -992,9 +1010,17 @@ export async function acquireSessionWriteLock(params: {
       });
       return { release: lock.release };
     } catch (err) {
+      throwIfAborted();
       if (!isFileLockError(err, "file_lock_timeout") && !isFileLockError(err, "file_lock_stale")) {
         throw err;
       }
+      if (
+        params.signal &&
+        isFileLockError(err, "file_lock_timeout") &&
+        resolveRemainingAcquireTimeoutMs(timeoutMs, startedAtMs, Date.now()) > 0
+      ) {
+        continue;
+      }
       const errorLockPath = (err as { lockPath?: string }).lockPath ?? lockPath;
       const { payload, missing: lockMissingAtDiagnostics } =
         await readLockPayloadForDiagnostics(errorLockPath);