diff --git a/src/cli/debug/cache-audit.ts b/src/cli/debug/cache-audit.ts index 419600a3db8..03be6ccd490 100644 --- a/src/cli/debug/cache-audit.ts +++ b/src/cli/debug/cache-audit.ts @@ -31,7 +31,7 @@ export async function cacheAuditCommand( session: ReturnType = resolveReplaySessionDir(workspaceId) ): Promise { const { sessionDir, historyService } = session; - const journal = new DurableEventJournal(sessionDir); + const journal = new DurableEventJournal(sessionDir, historyService.locks); const events = await journal.read(); // Final envelope per sequence only: fallback turns emit a superseding row. const envelopes = collapseEnvelopesToFinalPerSequence( diff --git a/src/node/services/refinement/refineService.ts b/src/node/services/refinement/refineService.ts index 2e27d1c35bb..133d1bc3757 100644 --- a/src/node/services/refinement/refineService.ts +++ b/src/node/services/refinement/refineService.ts @@ -54,7 +54,7 @@ import { isDurableContextResetBoundaryMarker, sliceMessagesForProviderFromLatestContextBoundary, } from "@/common/utils/messages/compactionBoundary"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import type { ProcessFileLock } from "@/node/utils/concurrency/fileLock"; import { isWorkspaceRemovalTombstoned, refineApplyLockPath, @@ -432,9 +432,10 @@ export class RefineService { // two processes could both capture an empty attempted set and double- // apply a non-idempotent edit. Short acquisition timeout: a held lock // means another apply is running, mirror the in-process rejection. - let applyLock: Awaited>; + let applyLock: ProcessFileLock; try { - applyLock = await acquireProcessFileLock({ + // The history service's SessionLocks (phase 4): the same layer as removal's refine lock. + applyLock = await this.historyService.locks.acquireProcessFileLock({ // r66: session-dir-external (see refineApplyLockPath) — removal holds // this same lock across its tombstone+delete critical section, so an // in-flight apply completes before the deletion and a late one @@ -1097,9 +1098,9 @@ export class RefineService { // overwritten by that stale spread — the new proposal row would remain // in chat with a hash that no longer matches the file, losing the new // edits and failing later applies closed. - let stagingLock: Awaited>; + let stagingLock: ProcessFileLock; try { - stagingLock = await acquireProcessFileLock({ + stagingLock = await this.historyService.locks.acquireProcessFileLock({ // r66: session-dir-external (see refineApplyLockPath). lockPath: refineApplyLockPath(this.config.rootDir, workspaceId), timeoutMs: this.options.applyLockTimeoutMs ?? REFINE_APPLY_CROSS_PROCESS_LOCK_TIMEOUT_MS, diff --git a/src/node/services/replay/replayFixture.ts b/src/node/services/replay/replayFixture.ts index 927637e40a1..c94c7806a9a 100644 --- a/src/node/services/replay/replayFixture.ts +++ b/src/node/services/replay/replayFixture.ts @@ -133,16 +133,18 @@ export function createReplayFixtureSessionContext( sessionDir: string, workspaceId: string = REPLAY_FIXTURE_WORKSPACE_ID ): ReplayFixtureSessionContext { + const historyService = new HistoryService({ + getSessionDir: () => sessionDir, + // Fixture writes take the history write lock under `/locks`; + // lockfiles are transient (removed on release). + rootDir: path.dirname(sessionDir), + }); return { sessionDir, workspaceId, - historyService: new HistoryService({ - getSessionDir: () => sessionDir, - // Fixture writes take the history write lock under `/locks`; - // lockfiles are transient (removed on release). - rootDir: path.dirname(sessionDir), - }), - journal: new DurableEventJournal(sessionDir), + historyService, + // The journal's locks are the history service's (phase 4: SessionLocks). + journal: new DurableEventJournal(sessionDir, historyService.locks), devtoolsLines: [], turnCounter: 0, }; diff --git a/src/node/services/workspaceRemoval.ts b/src/node/services/workspaceRemoval.ts index 85439f3da41..388f40a272f 100644 --- a/src/node/services/workspaceRemoval.ts +++ b/src/node/services/workspaceRemoval.ts @@ -40,7 +40,8 @@ import { } from "@/node/services/refinement/targetMutationLocks"; import { hasErrorCode } from "@/node/services/tools/skillFileUtils"; import { SESSION_MEMORY_DIR_NAME } from "@/common/constants/memory"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; /** * Removal failed WITHOUT a durable tombstone (r63). Callers must abort @@ -189,6 +190,8 @@ export async function removeSessionDirUnderMemoryLocks(args: { * (see SharedMemoryRemovalAbortedError). */ beforeTombstone?: () => Promise; + /** The refine and history locks (phase 4: SessionLocks). Absent: the Node layer. */ + locks?: SessionLocks; }): Promise { assert(args.sessionDir.length > 0, "removeSessionDirUnderMemoryLocks requires a session dir"); // Crash clearly on a malformed config (test stubs, future refactors): an @@ -264,6 +267,8 @@ export async function sealSubAgentForRemovalUnderMemoryLocks(args: { attemptId: string; sharedWorkspaceMemorySessionDir: string; beforeTombstone: () => Promise; + /** The refine and history locks (phase 4: SessionLocks). Absent: the Node layer. */ + locks?: SessionLocks; }): Promise { try { await withRemovalLocks(args, async () => { @@ -300,9 +305,12 @@ async function withRemovalLocks( sessionDir: string; workspaceId: string; sharedWorkspaceMemorySessionDir?: string; + locks?: SessionLocks; }, body: () => Promise ): Promise { + // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree + const locks = args.locks ?? nodeSessionLocks; // Same key derivations as MemoryService.storeLockKey: the workspace store // root lives inside the session directory; global/project mutations hold // the coarse `/memory` key while journaling into this session dir. @@ -346,7 +354,7 @@ async function withRemovalLocks( // RefineService. Fail-closed like the other locks: a long apply blocks // removal into the orphan path rather than having the directory deleted // out from under its writes. - await using _refineLock = await acquireProcessFileLock({ + await using _refineLock = await locks.acquireProcessFileLock({ lockPath: refineApplyLockPath(args.rootDir, args.workspaceId), timeoutMs: 10_000, label: "refine serialization lock (removal)", @@ -360,7 +368,7 @@ async function withRemovalLocks( // that same (session-dir-external) lock here means the append either // commits before the deletion below or starts afterwards and hits // HistoryService's in-lock tombstone gate. - await using _historyLock = await acquireProcessFileLock({ + await using _historyLock = await locks.acquireProcessFileLock({ lockPath: historyWriteLockPath(args.rootDir, args.workspaceId), timeoutMs: 10_000, label: "history write lock (removal)", diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 22210515dd5..d91f3027ede 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -452,7 +452,6 @@ import { type BashMonitorRegistryRecord, } from "@/node/services/bashMonitorRegistryStore"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; import { getSelfIdentity } from "@/node/utils/concurrency/processLiveness"; import { judgeLifecycleMarkerOwner, @@ -4484,7 +4483,7 @@ export class WorkspaceService ): Promise> { try { return Ok( - await acquireProcessFileLock({ + await this.historyService.locks.acquireProcessFileLock({ // r66: session-dir-external (see refineApplyLockPath) — acquiring // the old in-session lockfile after removal recreated the deleted // directory via the lock's own mkdir. @@ -8311,6 +8310,7 @@ export class WorkspaceService ), beforeTombstone: () => this.lockedSharedMemoryHandover(workspaceId, sharedMemoryOwnerId, force), + locks: this.historyService.locks, }); sealedForRemoval = true; tombstonePublished = true; @@ -8780,6 +8780,7 @@ export class WorkspaceService ownerSessionDir === undefined || sealedForRemoval ? undefined : () => this.lockedSharedMemoryHandover(workspaceId, memoryOwnerId, force), + locks: this.historyService.locks, }); tombstonePublished = true; } catch (error) { diff --git a/src/node/utils/journal/durableEventJournal.ts b/src/node/utils/journal/durableEventJournal.ts index 038e44fcc33..1dddb1f755d 100644 --- a/src/node/utils/journal/durableEventJournal.ts +++ b/src/node/utils/journal/durableEventJournal.ts @@ -24,7 +24,9 @@ import { type DurableEventDraft, } from "@/common/types/durableEvent"; import { AsyncMutex } from "@/node/utils/concurrency/asyncMutex"; -import { acquireProcessFileLock, type ProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import type { ProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { Journal } from "./journal"; import { BlobStore } from "./blobStore"; @@ -141,11 +143,20 @@ export class DurableEventJournal { */ private mentionEpoch = 0; - constructor(sessionDir: string) { + /** + * `locks` takes the blob lock and the journal's append lock (phase 4: SessionLocks). Callers with + * a history service pass its locks. The process-wide registry (sharedDurableEventJournal) and + * replayVerifySession have none, so they keep the Node default. + */ + constructor( + sessionDir: string, + private readonly locks: SessionLocks = nodeSessionLocks + ) { this.journalFilePath = path.join(sessionDir, DURABLE_EVENTS_FILE_NAME); this.blobLockPath = path.join(sessionDir, BLOB_LOCK_FILE_NAME); this.journal = new Journal({ filePath: this.journalFilePath, + locks, schema: DurableEventSchema, getSeq: (row) => row.seq, getId: (row) => row.id, @@ -210,7 +221,7 @@ export class DurableEventJournal { */ async withBlobLock(fn: () => Promise): Promise { await using _mutex = await this.blobLock.acquire(); - await using fileLock = await acquireProcessFileLock({ + await using fileLock = await this.locks.acquireProcessFileLock({ lockPath: this.blobLockPath, timeoutMs: BLOB_LOCK_TIMEOUT_MS, label: "blob lock",