diff --git a/src/agents/embedded-agent-runner/compact.hooks.harness.ts b/src/agents/embedded-agent-runner/compact.hooks.harness.ts index 10cdb88de653..c21980df69bc 100644 --- a/src/agents/embedded-agent-runner/compact.hooks.harness.ts +++ b/src/agents/embedded-agent-runner/compact.hooks.harness.ts @@ -48,6 +48,9 @@ export const hookRunner = { }; export const ensureRuntimePluginsLoaded: Mock<(params?: unknown) => void> = vi.fn(); +export const acquireSessionWriteLockMock = vi.fn(async (_params?: unknown) => ({ + release: vi.fn(async () => {}), +})); export const resolveContextEngineMock = vi.fn(async () => ({ info: { ownsCompaction: true as boolean }, compact: contextEngineCompactMock, @@ -521,6 +524,7 @@ export function resetCompactHooksHarnessMocks(): void { hookRunner.runAfterCompaction.mockResolvedValue(undefined); ensureRuntimePluginsLoaded.mockReset(); + acquireSessionWriteLockMock.mockClear(); resolveContextEngineMock.mockReset(); resolveContextEngineMock.mockResolvedValue({ @@ -587,6 +591,7 @@ export async function loadCompactHooksHarness(): Promise<{ compactEmbeddedAgentSessionDirect: typeof import("./compact.js").compactEmbeddedAgentSessionDirect; compactEmbeddedAgentSession: typeof import("./compact.queued.js").compactEmbeddedAgentSession; testing: typeof import("./compact.js").testing; + withOwnedSessionTranscriptWrites: typeof import("../../config/sessions/transcript-write-context.js").withOwnedSessionTranscriptWrites; onSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onSessionTranscriptUpdate; onInternalSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onInternalSessionTranscriptUpdate; }> { @@ -744,7 +749,7 @@ export async function loadCompactHooksHarness(): Promise<{ })); vi.doMock("../session-write-lock.js", () => ({ - acquireSessionWriteLock: vi.fn(async () => ({ release: vi.fn(async () => {}) })), + acquireSessionWriteLock: acquireSessionWriteLockMock, resolveSessionLockMaxHoldFromTimeout: vi.fn(() => 0), resolveSessionWriteLockAcquireTimeoutMs: vi.fn(() => 60_000), resolveSessionWriteLockOptions: vi.fn(() => ({ @@ -1049,17 +1054,20 @@ export async function loadCompactHooksHarness(): Promise<{ }; }); - const [compactModule, compactQueuedModule, transcriptEvents] = await Promise.all([ - import("./compact.js"), - import("./compact.queued.js"), - import("../../sessions/transcript-events.js"), - ]); + const [compactModule, compactQueuedModule, transcriptEvents, transcriptWriteContext] = + await Promise.all([ + import("./compact.js"), + import("./compact.queued.js"), + import("../../sessions/transcript-events.js"), + import("../../config/sessions/transcript-write-context.js"), + ]); return { ...compactModule, compactEmbeddedAgentSession: compactQueuedModule.compactEmbeddedAgentSession, onSessionTranscriptUpdate: transcriptEvents.onSessionTranscriptUpdate, onInternalSessionTranscriptUpdate: transcriptEvents.onInternalSessionTranscriptUpdate, + withOwnedSessionTranscriptWrites: transcriptWriteContext.withOwnedSessionTranscriptWrites, }; } /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/agents/embedded-agent-runner/compact.hooks.test.ts b/src/agents/embedded-agent-runner/compact.hooks.test.ts index 5fca669df660..1c622998c100 100644 --- a/src/agents/embedded-agent-runner/compact.hooks.test.ts +++ b/src/agents/embedded-agent-runner/compact.hooks.test.ts @@ -5,6 +5,7 @@ import type { AgentMessage } from "openclaw/plugin-sdk/agent-core"; import { beforeAll, beforeEach, describe, expect, it, vi, type Mock } from "vitest"; import { createReplyOperation } from "../../auto-reply/reply/reply-run-registry.js"; import { + acquireSessionWriteLockMock, applyExtraParamsToAgentMock, applyAgentCompactionSettingsFromConfigMock, buildAgentRuntimePlanMock, @@ -61,6 +62,7 @@ let compactEmbeddedAgentSession: typeof import("./compact.queued.js").compactEmb let compactTesting: typeof import("./compact.js").testing; let onSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onSessionTranscriptUpdate; let onInternalSessionTranscriptUpdate: typeof import("../../sessions/transcript-events.js").onInternalSessionTranscriptUpdate; +let withOwnedSessionTranscriptWrites: typeof import("../../config/sessions/transcript-write-context.js").withOwnedSessionTranscriptWrites; const TEST_SESSION_ID = "session-1"; const TEST_SESSION_KEY = "agent:main:session-1"; @@ -307,6 +309,7 @@ beforeAll(async () => { compactTesting = loaded.testing; onSessionTranscriptUpdate = loaded.onSessionTranscriptUpdate; onInternalSessionTranscriptUpdate = loaded.onInternalSessionTranscriptUpdate; + withOwnedSessionTranscriptWrites = loaded.withOwnedSessionTranscriptWrites; }); beforeEach(() => { @@ -331,6 +334,36 @@ describe("compactEmbeddedAgentSessionDirect hooks", () => { resetCompactSessionStateMocks(); }); + it("acquires the normal session lock without process-wide reentry", async () => { + const result = await compactEmbeddedAgentSessionDirect(wrappedCompactionArgs()); + + expect(result).toMatchObject({ ok: true, compacted: true }); + const lockOptions = mockCallArg(acquireSessionWriteLockMock); + expect(lockOptions).toMatchObject({ sessionFile: TEST_SESSION_FILE }); + expect(lockOptions).not.toHaveProperty("allowReentrant"); + }); + + it("reuses the matching logical writer lock during direct compaction", async () => { + const withSessionWriteLockCall = vi.fn(); + const withSessionWriteLock = async (run: () => Promise | T): Promise => { + withSessionWriteLockCall(); + return await run(); + }; + + const result = await withOwnedSessionTranscriptWrites( + { + sessionFile: TEST_SESSION_FILE, + sessionKey: TEST_SESSION_KEY, + withSessionWriteLock, + }, + async () => await compactEmbeddedAgentSessionDirect(wrappedCompactionArgs()), + ); + + expect(result).toMatchObject({ ok: true, compacted: true }); + expect(withSessionWriteLockCall).toHaveBeenCalledOnce(); + expect(acquireSessionWriteLockMock).not.toHaveBeenCalled(); + }); + it("fails closed before generic compaction for a model-locked native session", async () => { const result = await compactEmbeddedAgentSessionDirect({ sessionId: "session-1", diff --git a/src/agents/embedded-agent-runner/compact.ts b/src/agents/embedded-agent-runner/compact.ts index af9d38f3156a..6d63dd246928 100644 --- a/src/agents/embedded-agent-runner/compact.ts +++ b/src/agents/embedded-agent-runner/compact.ts @@ -7,6 +7,7 @@ import { isAcpRuntimeSpawnAvailable } from "../../acp/runtime/availability.js"; import type { ThinkLevel } from "../../auto-reply/thinking.js"; import { resolveAgentModelFallbackValues } from "../../config/model-input.js"; import { parseSqliteSessionFileMarker } from "../../config/sessions/sqlite-marker.js"; +import { acquireOwnedSessionTranscriptWriteLock } from "../../config/sessions/transcript-write-context.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { createFileBackedCompactionCheckpointStore, @@ -1376,14 +1377,19 @@ async function compactEmbeddedAgentSessionDirectOnce( }; const compactionTimeoutMs = resolveCompactionTimeoutMs(params.config); - const sessionLock = await acquireSessionWriteLock({ - sessionFile: params.sessionFile, - ...resolveSessionWriteLockOptions(params.config, { - maxHoldMsFallback: resolveSessionLockMaxHoldFromTimeout({ - timeoutMs: compactionTimeoutMs, + const sessionLock = + (await acquireOwnedSessionTranscriptWriteLock({ + sessionFile: params.sessionFile, + sessionKey: params.sessionKey, + })) ?? + (await acquireSessionWriteLock({ + sessionFile: params.sessionFile, + ...resolveSessionWriteLockOptions(params.config, { + maxHoldMsFallback: resolveSessionLockMaxHoldFromTimeout({ + timeoutMs: compactionTimeoutMs, + }), }), - }), - }); + })); try { if (!isSqliteSessionTranscript) { await repairSessionFileIfNeeded({ diff --git a/src/config/sessions/transcript-write-context.ts b/src/config/sessions/transcript-write-context.ts index 453d903639b3..fe406390eb1e 100644 --- a/src/config/sessions/transcript-write-context.ts +++ b/src/config/sessions/transcript-write-context.ts @@ -87,6 +87,47 @@ export async function runWithOwnedSessionTranscriptWriteLock( return await runWithOwnedSessionTranscriptWriteContext(params, run); } +export async function acquireOwnedSessionTranscriptWriteLock(params: { + sessionFile?: string; + sessionKey?: string; +}): Promise<{ release: () => Promise } | undefined> { + const context = ownedTranscriptWriteContext.getStore(); + if (!context || !contextMatches({ context, ...params })) { + return undefined; + } + + // Keep the owner callback pending until release so release-shaped callers + // cannot outlive the logical writer lock or leak a tracked nested operation. + let markAcquired!: () => void; + let rejectAcquire!: (error: unknown) => void; + const acquired = new Promise((resolve, reject) => { + markAcquired = resolve; + rejectAcquire = reject; + }); + let releaseOperation!: () => void; + const releaseRequested = new Promise((resolve) => { + releaseOperation = resolve; + }); + const operation = context.withSessionWriteLock(async () => { + markAcquired(); + await releaseRequested; + }); + void operation.catch(rejectAcquire); + await acquired; + + let released = false; + return { + release: async () => { + if (released) { + return; + } + released = true; + releaseOperation(); + await operation; + }, + }; +} + export function canAdvanceOwnedSessionEntryCache(params: { sessionFile?: string; sessionKey?: string;