From 6443c2a55232063b650bfa320dbf23e5b2da0f91 Mon Sep 17 00:00:00 2001 From: Vincent Koc <25068+vincentkoc@users.noreply.github.com> Date: Sat, 4 Jul 2026 07:30:57 -0700 Subject: [PATCH] fix(agents): honor aborts during session lock acquisition --- .../run/attempt.abort-race.test.ts | 65 +++++++++++++++++++ .../run/attempt.session-lock.ts | 6 +- .../embedded-agent-runner/run/attempt.ts | 4 ++ src/agents/session-write-lock.test.ts | 23 +++++++ src/agents/session-write-lock.ts | 28 +++++++- 5 files changed, 123 insertions(+), 3 deletions(-) create mode 100644 src/agents/embedded-agent-runner/run/attempt.abort-race.test.ts diff --git a/src/agents/embedded-agent-runner/run/attempt.abort-race.test.ts b/src/agents/embedded-agent-runner/run/attempt.abort-race.test.ts new file mode 100644 index 000000000000..9ae421734c15 --- /dev/null +++ b/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((resolve) => { + markLockRequested = resolve; + }); + hoisted.acquireSessionWriteLockMock.mockImplementationOnce(async (params) => { + observedSignal = params.signal; + markLockRequested(); + await new Promise((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); + }); +}); diff --git a/src/agents/embedded-agent-runner/run/attempt.session-lock.ts b/src/agents/embedded-agent-runner/run/attempt.session-lock.ts index d35b409d136f..82f736a81a79 100644 --- a/src/agents/embedded-agent-runner/run/attempt.session-lock.ts +++ b/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; reloadPromptReleasedSessionFile?: () => Promise | void; }): Promise { - const acquireLock = async (): Promise => + const acquireLock = async (signal?: AbortSignal): Promise => 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(); let ownedPublicationQueue: Promise = Promise.resolve(); let fenceFingerprint: SessionFileFingerprint | undefined; diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 690184d5774f..a57d1e3157c9 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -2146,6 +2146,7 @@ export async function runEmbeddedAttempt( let sessionManager: ReturnType | 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>["session"] | undefined; let removeToolResultContextGuard: (() => void) | undefined; diff --git a/src/agents/session-write-lock.test.ts b/src/agents/session-write-lock.test.ts index 15dbffb6db05..facefb24674c 100644 --- a/src/agents/session-write-lock.test.ts +++ b/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((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 }); diff --git a/src/agents/session-write-lock.ts b/src/agents/session-write-lock.ts index b4fd492cf494..6199badfa973 100644 --- a/src/agents/session-write-lock.ts +++ b/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; }> { + 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);