From a9b2dbe988a97a280a78bebaf205ae75f63e5580 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 06:09:06 +0000 Subject: [PATCH 1/4] refactor: removal, refine, branch summary and the durable journal take their locks from SessionLocks withRemovalLocks, RefineService's apply and staging locks, the branch-summary pending marker and DurableEventJournal's blob and append locks go through SessionLocks. WorkspaceService and AgentSession pass their history service's locks. RefineService uses its own history service's. The process-wide journal registry and the CLIs keep the Node default. --- src/node/services/agentSession.ts | 3 ++- src/node/services/branchSummary.ts | 19 ++++++++++++++----- src/node/services/refinement/refineService.ts | 11 ++++++----- src/node/services/workspaceRemoval.ts | 14 +++++++++++--- src/node/services/workspaceService.ts | 3 +++ src/node/utils/journal/durableEventJournal.ts | 16 +++++++++++++--- 6 files changed, 49 insertions(+), 17 deletions(-) diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index f24b9fae0f0..957f5dabf95 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -5439,7 +5439,8 @@ export class AgentSession { this.workspaceId, // Session dir enables the cross-process pending-marker wait (r48): a // fork registered in another backend has no entry in this process. - path.join(this.config.sessionsDir, this.workspaceId) + path.join(this.config.sessionsDir, this.workspaceId), + this.historyService.locks ); // Workspace removal disposes the session and cancels the summary writer // while this send is parked on the await above; every append between here diff --git a/src/node/services/branchSummary.ts b/src/node/services/branchSummary.ts index 38cdcd1e302..9016fb99679 100644 --- a/src/node/services/branchSummary.ts +++ b/src/node/services/branchSummary.ts @@ -34,7 +34,8 @@ import { buildProviderOptions } from "@/common/utils/ai/providerOptions"; import { getErrorMessage } from "@/common/utils/errors"; import { estimateMuxMessageTokens } from "@/common/utils/messages/keepRecentTail"; import { pluginHookContextTexts } from "@/common/utils/messages/pluginHookContext"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { BRANCH_SUMMARY_MAX_ACCUMULATED_CHARS, BRANCH_SUMMARY_MAX_OUTPUT_TOKENS, @@ -953,7 +954,12 @@ export async function runInlineAbandonedBranchSummary( * observe the marker (r55). */ export async function startAbandonedBranchSummaryInBackground( - input: AbandonedBranchSummaryInput & { guardTailMessageId: string; sessionDir?: string } + input: AbandonedBranchSummaryInput & { + guardTailMessageId: string; + sessionDir?: string; + /** The pending marker's lock (phase 4: SessionLocks). Absent: the Node layer. */ + locks?: SessionLocks; + } ): Promise { const controller = new AbortController(); // r55: the returned promise resolves only after the cross-process pending @@ -980,7 +986,8 @@ export async function startAbandonedBranchSummaryInBackground( let lock: AsyncDisposable | null = null; if (input.sessionDir !== undefined) { try { - lock = await acquireProcessFileLock({ + // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree + lock = await (input.locks ?? nodeSessionLocks).acquireProcessFileLock({ lockPath: path.join(input.sessionDir, BRANCH_SUMMARY_LOCK_FILENAME), timeoutMs: BRANCH_SUMMARY_LOCK_ACQUIRE_TIMEOUT_MS, label: "branch summary pending marker", @@ -1033,7 +1040,9 @@ export async function startAbandonedBranchSummaryInBackground( */ export async function awaitPendingBranchSummary( workspaceId: string, - sessionDir?: string + sessionDir?: string, + /** The pending marker's lock (phase 4: SessionLocks). */ + locks: SessionLocks = nodeSessionLocks ): Promise { const entry = pendingBranchSummaries.get(workspaceId); if (!entry) { @@ -1056,7 +1065,7 @@ export async function awaitPendingBranchSummary( ); if (markerExists) { try { - const lock = await acquireProcessFileLock({ + const lock = await locks.acquireProcessFileLock({ lockPath, timeoutMs: BRANCH_SUMMARY_LOCK_WAIT_TIMEOUT_MS, label: "branch summary pending marker", 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/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..1e218136623 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -8311,6 +8311,7 @@ export class WorkspaceService ), beforeTombstone: () => this.lockedSharedMemoryHandover(workspaceId, sharedMemoryOwnerId, force), + locks: this.historyService.locks, }); sealedForRemoval = true; tombstonePublished = true; @@ -8780,6 +8781,7 @@ export class WorkspaceService ownerSessionDir === undefined || sealedForRemoval ? undefined : () => this.lockedSharedMemoryHandover(workspaceId, memoryOwnerId, force), + locks: this.historyService.locks, }); tombstonePublished = true; } catch (error) { @@ -15255,6 +15257,7 @@ export class WorkspaceService // Cross-process pending marker home (r48): lets a first send served // by another backend wait for the in-flight summary. sessionDir: path.join(this.config.sessionsDir, newWorkspaceId), + locks: this.historyService.locks, abandonedMessages: abandonedBranchMessages, isExperimentEnabled: (experimentId) => this.isExperimentEnabled(experimentId), guardTailMessageId: sourceMessageId, diff --git a/src/node/utils/journal/durableEventJournal.ts b/src/node/utils/journal/durableEventJournal.ts index 038e44fcc33..84345a28f2e 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,19 @@ export class DurableEventJournal { */ private mentionEpoch = 0; - constructor(sessionDir: string) { + /** + * `locks` takes the blob lock and the journal's append lock (phase 4: SessionLocks). The + * process-wide registry (sharedDurableEventJournal) and the CLIs 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 +220,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", From afd879b13755ddf3d79b4e02970f5b01a4db6c03 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 06:41:01 +0000 Subject: [PATCH 2/4] refactor: the debug CLI's cache audit and the replay fixture pass their history service's locks to the journal --- src/cli/debug/cache-audit.ts | 2 +- src/node/services/replay/replayFixture.ts | 16 +++++++++------- src/node/utils/journal/durableEventJournal.ts | 5 +++-- 3 files changed, 13 insertions(+), 10 deletions(-) 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/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/utils/journal/durableEventJournal.ts b/src/node/utils/journal/durableEventJournal.ts index 84345a28f2e..1dddb1f755d 100644 --- a/src/node/utils/journal/durableEventJournal.ts +++ b/src/node/utils/journal/durableEventJournal.ts @@ -144,8 +144,9 @@ export class DurableEventJournal { private mentionEpoch = 0; /** - * `locks` takes the blob lock and the journal's append lock (phase 4: SessionLocks). The - * process-wide registry (sharedDurableEventJournal) and the CLIs keep the Node default. + * `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, From 88c4b16bd2efe0ca3462b61c370154bf4aefe81d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 11:31:28 +0000 Subject: [PATCH 3/4] revert: branchSummary's pending marker keeps its direct process file lock The marker reader discovers a held marker with fs.stat, which an injected memory SessionLocks never creates. The correct fix needs a held-lock probe on every SessionLocks layer, so this PR drops the marker surface and a follow-up phase 4 PR moves the writer and the reader together. --- src/node/services/agentSession.ts | 3 +-- src/node/services/branchSummary.ts | 19 +++++-------------- src/node/services/workspaceService.ts | 1 - 3 files changed, 6 insertions(+), 17 deletions(-) diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 957f5dabf95..f24b9fae0f0 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -5439,8 +5439,7 @@ export class AgentSession { this.workspaceId, // Session dir enables the cross-process pending-marker wait (r48): a // fork registered in another backend has no entry in this process. - path.join(this.config.sessionsDir, this.workspaceId), - this.historyService.locks + path.join(this.config.sessionsDir, this.workspaceId) ); // Workspace removal disposes the session and cancels the summary writer // while this send is parked on the await above; every append between here diff --git a/src/node/services/branchSummary.ts b/src/node/services/branchSummary.ts index 9016fb99679..38cdcd1e302 100644 --- a/src/node/services/branchSummary.ts +++ b/src/node/services/branchSummary.ts @@ -34,8 +34,7 @@ import { buildProviderOptions } from "@/common/utils/ai/providerOptions"; import { getErrorMessage } from "@/common/utils/errors"; import { estimateMuxMessageTokens } from "@/common/utils/messages/keepRecentTail"; import { pluginHookContextTexts } from "@/common/utils/messages/pluginHookContext"; -import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; -import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; +import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; import { BRANCH_SUMMARY_MAX_ACCUMULATED_CHARS, BRANCH_SUMMARY_MAX_OUTPUT_TOKENS, @@ -954,12 +953,7 @@ export async function runInlineAbandonedBranchSummary( * observe the marker (r55). */ export async function startAbandonedBranchSummaryInBackground( - input: AbandonedBranchSummaryInput & { - guardTailMessageId: string; - sessionDir?: string; - /** The pending marker's lock (phase 4: SessionLocks). Absent: the Node layer. */ - locks?: SessionLocks; - } + input: AbandonedBranchSummaryInput & { guardTailMessageId: string; sessionDir?: string } ): Promise { const controller = new AbortController(); // r55: the returned promise resolves only after the cross-process pending @@ -986,8 +980,7 @@ export async function startAbandonedBranchSummaryInBackground( let lock: AsyncDisposable | null = null; if (input.sessionDir !== undefined) { try { - // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree - lock = await (input.locks ?? nodeSessionLocks).acquireProcessFileLock({ + lock = await acquireProcessFileLock({ lockPath: path.join(input.sessionDir, BRANCH_SUMMARY_LOCK_FILENAME), timeoutMs: BRANCH_SUMMARY_LOCK_ACQUIRE_TIMEOUT_MS, label: "branch summary pending marker", @@ -1040,9 +1033,7 @@ export async function startAbandonedBranchSummaryInBackground( */ export async function awaitPendingBranchSummary( workspaceId: string, - sessionDir?: string, - /** The pending marker's lock (phase 4: SessionLocks). */ - locks: SessionLocks = nodeSessionLocks + sessionDir?: string ): Promise { const entry = pendingBranchSummaries.get(workspaceId); if (!entry) { @@ -1065,7 +1056,7 @@ export async function awaitPendingBranchSummary( ); if (markerExists) { try { - const lock = await locks.acquireProcessFileLock({ + const lock = await acquireProcessFileLock({ lockPath, timeoutMs: BRANCH_SUMMARY_LOCK_WAIT_TIMEOUT_MS, label: "branch summary pending marker", diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 1e218136623..abd074d54de 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -15257,7 +15257,6 @@ export class WorkspaceService // Cross-process pending marker home (r48): lets a first send served // by another backend wait for the in-flight summary. sessionDir: path.join(this.config.sessionsDir, newWorkspaceId), - locks: this.historyService.locks, abandonedMessages: abandonedBranchMessages, isExperimentEnabled: (experimentId) => this.isExperimentEnabled(experimentId), guardTailMessageId: sourceMessageId, From 4eabff0f7b097de8072555ed01e8d132c0e59dfe Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 12:50:41 +0000 Subject: [PATCH 4/4] refactor: WorkspaceService's refine serialization lock takes its file lock from the history service's locks Its partner, RefineService's apply lock on the same path, already does, so a non-Node SessionLocks must see both or the reset and the refine apply stop contending. Moved here from PR 13. --- src/node/services/workspaceService.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index abd074d54de..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.