🤖 refactor: removal, refine and the durable journal take their locks from SessionLocks

src/cli/debug/cache-audit.ts

@@ -31,7 +31,7 @@ export async function cacheAuditCommand(
   session: ReturnType<typeof resolveReplaySessionDir> = resolveReplaySessionDir(workspaceId)
 ): Promise<void> {
   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(

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<ReturnType<typeof acquireProcessFileLock>>;
+    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<ReturnType<typeof acquireProcessFileLock>>;
+      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,

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 `<rootDir>/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 `<rootDir>/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,
   };

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<void>;
+  /** The refine and history locks (phase 4: SessionLocks). Absent: the Node layer. */
+  locks?: SessionLocks;
 }): Promise<void> {
   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<void>;
+  /** The refine and history locks (phase 4: SessionLocks). Absent: the Node layer. */
+  locks?: SessionLocks;
 }): Promise<void> {
   try {
     await withRemovalLocks(args, async () => {
@@ -300,9 +305,12 @@ async function withRemovalLocks(
     sessionDir: string;
     workspaceId: string;
     sharedWorkspaceMemorySessionDir?: string;
+    locks?: SessionLocks;
   },
   body: () => Promise<void>
 ): Promise<void> {
+  // 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 `<rootDir>/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)",

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<Result<AsyncDisposable>> {
     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) {

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<DurableEvent>({
       filePath: this.journalFilePath,
+      locks,
       schema: DurableEventSchema,
       getSeq: (row) => row.seq,
       getId: (row) => row.id,
@@ -210,7 +221,7 @@ export class DurableEventJournal {
    */
   async withBlobLock<T>(fn: () => Promise<T>): Promise<T> {
     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",