From df1851d10af1e901594297d324b60fd70fda69b8 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 22 Oct 2026 22:18:42 +0000 Subject: [PATCH 01/19] refactor: pinned files, artifact index, workflow runs and journal appends take their locks through SessionLocks --- src/node/services/artifactVersionStore.ts | 17 +++++++---- src/node/services/pinnedArtifactFiles.ts | 28 +++++++++++++----- .../services/workflows/WorkflowRunStore.ts | 29 ++++++++++++++----- src/node/utils/journal/journal.ts | 9 ++++-- 4 files changed, 61 insertions(+), 22 deletions(-) diff --git a/src/node/services/artifactVersionStore.ts b/src/node/services/artifactVersionStore.ts index e513f93da80..1465fde416b 100644 --- a/src/node/services/artifactVersionStore.ts +++ b/src/node/services/artifactVersionStore.ts @@ -13,7 +13,8 @@ import { assert } from "@/common/utils/assert"; import { getErrorMessage } from "@/common/utils/errors"; import { log } from "@/node/services/log"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; -import { acquireCrossProcessLock } from "@/node/utils/main/crossProcessLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import writeFileAtomic from "@/node/utils/writeFileAtomic"; /** @@ -117,7 +118,8 @@ export interface RecordArtifactVersionParams { pin?: ArtifactPin | null; nowMs?: number; /** Checked under the index lock: an aborted publish writes nothing. */ - abortSignal?: AbortSignal; + abortSignal?: AbortSignal; /** The session locks (phase 4); callers outside the session lock owners keep the Node layer. */ + locks?: SessionLocks; } export const ARTIFACT_PUBLISH_INTERRUPTED = "Publish was interrupted"; @@ -137,7 +139,11 @@ export async function recordArtifactVersion( const artifactId = getArtifactId(params.relPath); const dir = artifactDir(params.sessionDir, artifactId); return indexLocks.withLock(dir, async () => { - await using _fileLock = await acquireIndexFileLock(dir, params.abortSignal); + await using _fileLock = await acquireIndexFileLock( + dir, + params.abortSignal, + params.locks ?? nodeSessionLocks + ); const existing = await readArtifactIndex(params.sessionDir, artifactId); const index: ArtifactVersionIndex = existing ?? { id: artifactId, @@ -203,12 +209,13 @@ export async function recordArtifactVersion( */ async function acquireIndexFileLock( dir: string, - abortSignal: AbortSignal | undefined + abortSignal: AbortSignal | undefined, + locks: SessionLocks ): Promise { // Checked first: acquiring creates the artifact dir, and a cancelled publish writes nothing. if (abortSignal?.aborted) throw new Error(ARTIFACT_PUBLISH_INTERRUPTED); try { - const release = await acquireCrossProcessLock({ + const release = await locks.acquireCrossProcessLock({ lockPath: path.join(dir, INDEX_LOCK_FILE_NAME), acquireTimeoutMs: 30_000, staleMs: 60_000, diff --git a/src/node/services/pinnedArtifactFiles.ts b/src/node/services/pinnedArtifactFiles.ts index ef26715cdda..f546314245b 100644 --- a/src/node/services/pinnedArtifactFiles.ts +++ b/src/node/services/pinnedArtifactFiles.ts @@ -14,7 +14,8 @@ import { } from "@/node/runtime/runtimeHelpers"; import { log } from "@/node/services/log"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; -import { acquireCrossProcessLock } from "@/node/utils/main/crossProcessLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import writeFileAtomic from "@/node/utils/writeFileAtomic"; import { readArtifactBytesOnRuntime } from "./artifactRuntimeStore"; import { @@ -153,13 +154,14 @@ export async function readPinnedPaths(sessionDir: string): Promise { async function updatePinnedPaths( sessionDir: string, - update: (paths: string[]) => string[] + update: (paths: string[]) => string[], + locks: SessionLocks ): Promise { await storeLocks.withLock(sessionDir, async () => { await fs.mkdir(sessionDir, { recursive: true }); // The in-process queue orders this backend's writers; the file lock excludes another backend // sharing the Xum home (a desktop app alongside `xum server`), whose update could be lost. - const release = await acquireCrossProcessLock({ + const release = await locks.acquireCrossProcessLock({ lockPath: `${storePath(sessionDir)}.lock`, acquireTimeoutMs: 30_000, staleMs: 60_000, @@ -171,14 +173,24 @@ async function updatePinnedPaths( }); } -export async function addPinnedPath(sessionDir: string, relPath: string): Promise { - await updatePinnedPaths(sessionDir, (paths) => - paths.includes(relPath) ? paths : [...paths, relPath] +export async function addPinnedPath( + sessionDir: string, + relPath: string, + locks: SessionLocks = nodeSessionLocks +): Promise { + await updatePinnedPaths( + sessionDir, + (paths) => (paths.includes(relPath) ? paths : [...paths, relPath]), + locks ); } -export async function removePinnedPath(sessionDir: string, relPath: string): Promise { - await updatePinnedPaths(sessionDir, (paths) => paths.filter((p) => p !== relPath)); +export async function removePinnedPath( + sessionDir: string, + relPath: string, + locks: SessionLocks = nodeSessionLocks +): Promise { + await updatePinnedPaths(sessionDir, (paths) => paths.filter((p) => p !== relPath), locks); } async function statPinned( diff --git a/src/node/services/workflows/WorkflowRunStore.ts b/src/node/services/workflows/WorkflowRunStore.ts index a0b5df25cec..aae2570ed1d 100644 --- a/src/node/services/workflows/WorkflowRunStore.ts +++ b/src/node/services/workflows/WorkflowRunStore.ts @@ -34,10 +34,9 @@ import { parseProcessIdentity, type ProcessIdentity, } from "@/node/utils/concurrency/processLiveness"; -import { - acquireCrossProcessLock, - CrossProcessLockTimeoutError, -} from "@/node/utils/main/crossProcessLock"; +import { CrossProcessLockTimeoutError } from "@/node/utils/concurrency/lockErrors"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { workflowRunStreamHub } from "@/node/services/workflows/workflowRunStreamHub"; import { KeyedFifoLock } from "@/node/services/workflows/keyedFifoLock"; @@ -114,6 +113,8 @@ export interface WorkflowRunStoreOptions { staleLeaseMs?: number; /** Test seam: how long a journal/lease mutation waits for its locks before timing out. */ mutationLockWaitTimeoutMs?: number; + /** The session locks (phase 4); every caller today keeps the Node layer. */ + locks?: SessionLocks; } export interface CreateWorkflowRunInput { @@ -219,6 +220,7 @@ export class WorkflowRunStore { private readonly sessionDir: string; private readonly staleLeaseMs: number; private readonly mutationLockWaitTimeoutMs: number | undefined; + private readonly locks: SessionLocks; /** Live draining-owner capabilities by run id; see WorkflowCancellationSettlement. */ private readonly cancellationSettlements = new Map(); @@ -231,6 +233,7 @@ export class WorkflowRunStore { "WorkflowRunStore: mutationLockWaitTimeoutMs must be positive" ); this.mutationLockWaitTimeoutMs = options.mutationLockWaitTimeoutMs; + this.locks = options.locks ?? nodeSessionLocks; } async createRun(input: CreateWorkflowRunInput): Promise { @@ -1107,6 +1110,7 @@ export class WorkflowRunStore { const leaseFile = this.leaseFile(runId); // Single attempt, as before: a busy lease lock reads as "another runner is active". const attempt = await tryWithWorkflowFileLock( + this.locks, this.leaseLockFile(runId), this.leaseMutationLockStaleMs(), async () => { @@ -1144,10 +1148,11 @@ export class WorkflowRunStore { return this.mutationLockWaitTimeoutMs ?? Math.max(30_000, this.leaseMutationLockStaleMs() * 4); } - private workflowLockOptions(): { staleMs: number; acquireTimeoutMs: number } { + private workflowLockOptions(): WorkflowLockOptions { return { staleMs: this.leaseMutationLockStaleMs(), acquireTimeoutMs: this.leaseMutationWaitTimeoutMs(), + locks: this.locks, }; } @@ -1724,12 +1729,13 @@ const workflowLockQueue = new KeyedFifoLock(); * acquire itself fail leaves no window between an existence check and the publication. */ async function acquireWorkflowCrossProcessLock( + locks: SessionLocks, lockPath: string, acquireTimeoutMs: number, staleMs: number, timeoutMessage: string ): Promise<() => Promise> { - return await acquireCrossProcessLock({ + return await locks.acquireCrossProcessLock({ lockPath, acquireTimeoutMs, staleMs, @@ -1744,9 +1750,15 @@ async function acquireWorkflowCrossProcessLock( * finally removed the lock unconditionally, deleting a successor's. crossProcessLock never takes a * lock from a live holder and its release removes only its own record. */ +interface WorkflowLockOptions { + staleMs: number; + acquireTimeoutMs: number; + locks: SessionLocks; +} + async function withWorkflowFileLock( lockPath: string, - options: { staleMs: number; acquireTimeoutMs: number }, + options: WorkflowLockOptions, operation: () => Promise ): Promise { // One budget covers the in-process queue and the cross-process acquire. A waiter behind a hung @@ -1760,6 +1772,7 @@ async function withWorkflowFileLock( } try { const release = await acquireWorkflowCrossProcessLock( + options.locks, lockPath, Math.max(0, deadline - Date.now()), options.staleMs, @@ -1777,6 +1790,7 @@ async function withWorkflowFileLock( /** One attempt, in-process and cross-process; `{ acquired: false }` while anyone holds the lock. */ async function tryWithWorkflowFileLock( + locks: SessionLocks, lockPath: string, staleMs: number, operation: () => Promise @@ -1789,6 +1803,7 @@ async function tryWithWorkflowFileLock( let release: () => Promise; try { release = await acquireWorkflowCrossProcessLock( + locks, lockPath, 0, staleMs, diff --git a/src/node/utils/journal/journal.ts b/src/node/utils/journal/journal.ts index ccc8c6cce84..c2851124ffd 100644 --- a/src/node/utils/journal/journal.ts +++ b/src/node/utils/journal/journal.ts @@ -19,7 +19,8 @@ import assert from "node:assert"; import * as fs from "fs/promises"; import * as path from "path"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { log } from "@/node/services/log"; /** Default bound on waiting for the append lock (see JournalOptions). */ @@ -61,6 +62,8 @@ export interface JournalOptions { * test). */ testOnlyBeforeAppendWrite?: () => Promise; + /** The session locks (phase 4); the default is the Node layer. */ + locks?: SessionLocks; } export class Journal { @@ -72,6 +75,7 @@ export class Journal { private readonly appendLockTimeoutMs: number; private readonly onAppended?: JournalOptions["onAppended"]; private readonly testOnlyBeforeAppendWrite?: () => Promise; + private readonly locks: SessionLocks; /** Next sequence to assign; null until the file has been scanned once. */ private nextSeq: number | null = null; /** @@ -95,6 +99,7 @@ export class Journal { assert(this.appendLockTimeoutMs > 0, "Journal appendLockTimeoutMs must be positive"); this.onAppended = options.onAppended; this.testOnlyBeforeAppendWrite = options.testOnlyBeforeAppendWrite; + this.locks = options.locks ?? nodeSessionLocks; } /** @@ -108,7 +113,7 @@ export class Journal { // Cross-process serialization: seq derivation and the write must be one // exclusive unit, or a concurrent writer in another process (debug CLI // vs live backend) could assign the same sequence number. - await using lock = await acquireProcessFileLock({ + await using lock = await this.locks.acquireProcessFileLock({ lockPath: this.lockPath, timeoutMs: this.appendLockTimeoutMs, label: "append lock", From 7e6bb099adba8ab06e016786f89f80cc677add5e Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 22 Oct 2026 22:19:14 +0000 Subject: [PATCH 02/19] refactor: keep recordArtifactVersion's complexity (default in acquireIndexFileLock) --- src/node/services/artifactVersionStore.ts | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/src/node/services/artifactVersionStore.ts b/src/node/services/artifactVersionStore.ts index 1465fde416b..d31da2edf7f 100644 --- a/src/node/services/artifactVersionStore.ts +++ b/src/node/services/artifactVersionStore.ts @@ -139,11 +139,7 @@ export async function recordArtifactVersion( const artifactId = getArtifactId(params.relPath); const dir = artifactDir(params.sessionDir, artifactId); return indexLocks.withLock(dir, async () => { - await using _fileLock = await acquireIndexFileLock( - dir, - params.abortSignal, - params.locks ?? nodeSessionLocks - ); + await using _fileLock = await acquireIndexFileLock(dir, params.abortSignal, params.locks); const existing = await readArtifactIndex(params.sessionDir, artifactId); const index: ArtifactVersionIndex = existing ?? { id: artifactId, @@ -210,7 +206,7 @@ export async function recordArtifactVersion( async function acquireIndexFileLock( dir: string, abortSignal: AbortSignal | undefined, - locks: SessionLocks + locks: SessionLocks = nodeSessionLocks ): Promise { // Checked first: acquiring creates the artifact dir, and a cancelled publish writes nothing. if (abortSignal?.aborted) throw new Error(ARTIFACT_PUBLISH_INTERRUPTED); From 46e53636e22c68aa61ba07fc2cdf2a9428f1d3d7 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 14:22:41 +0000 Subject: [PATCH 03/19] refactor: keep each doc comment on its declaration --- src/node/services/artifactVersionStore.ts | 3 ++- src/node/services/workflows/WorkflowRunStore.ts | 12 ++++++------ 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/src/node/services/artifactVersionStore.ts b/src/node/services/artifactVersionStore.ts index d31da2edf7f..97a68a0572c 100644 --- a/src/node/services/artifactVersionStore.ts +++ b/src/node/services/artifactVersionStore.ts @@ -118,7 +118,8 @@ export interface RecordArtifactVersionParams { pin?: ArtifactPin | null; nowMs?: number; /** Checked under the index lock: an aborted publish writes nothing. */ - abortSignal?: AbortSignal; /** The session locks (phase 4); callers outside the session lock owners keep the Node layer. */ + abortSignal?: AbortSignal; + /** The session locks (phase 4); every caller today keeps the Node layer. */ locks?: SessionLocks; } diff --git a/src/node/services/workflows/WorkflowRunStore.ts b/src/node/services/workflows/WorkflowRunStore.ts index aae2570ed1d..b07e4179165 100644 --- a/src/node/services/workflows/WorkflowRunStore.ts +++ b/src/node/services/workflows/WorkflowRunStore.ts @@ -1744,18 +1744,18 @@ async function acquireWorkflowCrossProcessLock( }); } -/** - * #4452 gap 1: the former mkdir locks were reclaimed once their mtime aged past the stale window, - * which a live holder stalled inside its owner-check-then-write section never refreshed, and every - * finally removed the lock unconditionally, deleting a successor's. crossProcessLock never takes a - * lock from a live holder and its release removes only its own record. - */ interface WorkflowLockOptions { staleMs: number; acquireTimeoutMs: number; locks: SessionLocks; } +/** + * #4452 gap 1: the former mkdir locks were reclaimed once their mtime aged past the stale window, + * which a live holder stalled inside its owner-check-then-write section never refreshed, and every + * finally removed the lock unconditionally, deleting a successor's. crossProcessLock never takes a + * lock from a live holder and its release removes only its own record. + */ async function withWorkflowFileLock( lockPath: string, options: WorkflowLockOptions, From 8d7daf617a19901f3274b6f3b3f3d6940574719c Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 15:02:06 +0000 Subject: [PATCH 04/19] refactor: mark the two equivalent default-locks mutants --- src/node/services/workflows/WorkflowRunStore.ts | 1 + src/node/utils/journal/journal.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/src/node/services/workflows/WorkflowRunStore.ts b/src/node/services/workflows/WorkflowRunStore.ts index b07e4179165..2ceb110dec6 100644 --- a/src/node/services/workflows/WorkflowRunStore.ts +++ b/src/node/services/workflows/WorkflowRunStore.ts @@ -233,6 +233,7 @@ export class WorkflowRunStore { "WorkflowRunStore: mutationLockWaitTimeoutMs must be positive" ); this.mutationLockWaitTimeoutMs = options.mutationLockWaitTimeoutMs; + // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree this.locks = options.locks ?? nodeSessionLocks; } diff --git a/src/node/utils/journal/journal.ts b/src/node/utils/journal/journal.ts index c2851124ffd..71ba84d7a34 100644 --- a/src/node/utils/journal/journal.ts +++ b/src/node/utils/journal/journal.ts @@ -99,6 +99,7 @@ export class Journal { assert(this.appendLockTimeoutMs > 0, "Journal appendLockTimeoutMs must be positive"); this.onAppended = options.onAppended; this.testOnlyBeforeAppendWrite = options.testOnlyBeforeAppendWrite; + // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree this.locks = options.locks ?? nodeSessionLocks; } From 89231013cbab57cf587a40e8f82ca1a55de51f9e Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 14:21:53 +0000 Subject: [PATCH 05/19] tests: put the session lock modules under local/no-host-imports with their 5 pairs --- .oxlintrc.json | 24 +++++++++++++++++++++--- eslint.config.mjs | 29 ++++++++++++++++++++++++++--- scripts/noHostImports.test.ts | 5 +++++ 3 files changed, 52 insertions(+), 6 deletions(-) diff --git a/.oxlintrc.json b/.oxlintrc.json index 05a52ae6b61..cdcf233f22d 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1264,10 +1264,19 @@ { "module": "@/common/constants/paths", "names": ["cleanupObsoleteXumBinArtifacts", "getXumExtensionMetadataPath", "getXumHome", "getXumLogsDir", "getXumPerfCapturesDir", "getXumPerfReportsDir", "getXumPerfTapesDir", "getXumProjectsDir", "getXumSessionsDir"] - } + }, + "@/node/utils/concurrency/workspaceFileLocks", + "@/node/utils/concurrency/fileLock", + "@/node/utils/concurrency/processLiveness", + "@/node/utils/main/crossProcessLock", + "@/node/services/workspaceUseLeases" ], "allow": { - "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers", "fs/promises"], + "src/node/services/agentSession.ts": [ + "@/node/runtime/runtimeHelpers", + "@/node/services/workspaceUseLeases", + "fs/promises" + ], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", @@ -1278,10 +1287,15 @@ "src/node/services/contextManagement/strategies/tokenBudget.ts": [ "@/node/runtime/runtimeHelpers" ], + "src/node/services/historyService.ts": [ + "@/node/utils/concurrency/fileLock", + "@/node/utils/concurrency/workspaceFileLocks" + ], "src/node/services/tools/agent_skill_read_file.ts": ["@/node/runtime/RemoteRuntime"], "src/node/services/tools/bash.ts": ["@/node/runtime/LocalBaseRuntime", "fs"], "src/node/services/tools/notify.ts": ["electron"], "src/node/services/tools/review_pane.ts": [ + "@/node/utils/concurrency/workspaceFileLocks", "@/node/utils/writeFileAtomic", "fs/promises" ], @@ -1290,7 +1304,11 @@ "fs/promises" ], "src/node/services/tools/skillFileHint.ts": ["@/node/runtime/RemoteRuntime"], - "src/node/services/tools/todo.ts": ["@/node/utils/writeFileAtomic", "fs/promises"], + "src/node/services/tools/todo.ts": [ + "@/node/utils/concurrency/workspaceFileLocks", + "@/node/utils/writeFileAtomic", + "fs/promises" + ], "src/node/services/tools/xum_agents_path.ts": ["fs/promises"], "src/node/services/tools/xum_agents_read.ts": ["fs/promises"], "src/node/services/tools/xum_agents_write.ts": ["fs/promises"] diff --git a/eslint.config.mjs b/eslint.config.mjs index e6c85fa40e3..bb4620abad1 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -173,10 +173,21 @@ const HOST_MODULES = [ "getXumSessionsDir", ], }, + // The session lock modules (phase 4): loop code takes its locks through SessionLocks. Not + // mutexMap: tools/file_edit_operation.ts keeps a pure in-memory map. + "@/node/utils/concurrency/workspaceFileLocks", + "@/node/utils/concurrency/fileLock", + "@/node/utils/concurrency/processLiveness", + "@/node/utils/main/crossProcessLock", + "@/node/services/workspaceUseLeases", ]; // File -> the host modules it still imports. Phase plans list the pairs they add or remove. const HOST_IMPORT_ALLOW = { - "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers", "fs/promises"], + "src/node/services/agentSession.ts": [ + "@/node/runtime/runtimeHelpers", + "@/node/services/workspaceUseLeases", + "fs/promises", + ], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", @@ -187,16 +198,28 @@ const HOST_IMPORT_ALLOW = { "src/node/services/contextManagement/strategies/tokenBudget.ts": [ "@/node/runtime/runtimeHelpers", ], + "src/node/services/historyService.ts": [ + "@/node/utils/concurrency/fileLock", + "@/node/utils/concurrency/workspaceFileLocks", + ], "src/node/services/tools/agent_skill_read_file.ts": ["@/node/runtime/RemoteRuntime"], "src/node/services/tools/bash.ts": ["@/node/runtime/LocalBaseRuntime", "fs"], "src/node/services/tools/notify.ts": ["electron"], - "src/node/services/tools/review_pane.ts": ["@/node/utils/writeFileAtomic", "fs/promises"], + "src/node/services/tools/review_pane.ts": [ + "@/node/utils/concurrency/workspaceFileLocks", + "@/node/utils/writeFileAtomic", + "fs/promises", + ], "src/node/services/tools/shared/configReadWrite.ts": [ "@/node/utils/writeFileAtomic", "fs/promises", ], "src/node/services/tools/skillFileHint.ts": ["@/node/runtime/RemoteRuntime"], - "src/node/services/tools/todo.ts": ["@/node/utils/writeFileAtomic", "fs/promises"], + "src/node/services/tools/todo.ts": [ + "@/node/utils/concurrency/workspaceFileLocks", + "@/node/utils/writeFileAtomic", + "fs/promises", + ], "src/node/services/tools/xum_agents_path.ts": ["fs/promises"], "src/node/services/tools/xum_agents_read.ts": ["fs/promises"], "src/node/services/tools/xum_agents_write.ts": ["fs/promises"], diff --git a/scripts/noHostImports.test.ts b/scripts/noHostImports.test.ts index 92b5b137093..dfd05cfa0e4 100644 --- a/scripts/noHostImports.test.ts +++ b/scripts/noHostImports.test.ts @@ -121,6 +121,7 @@ test("allows listed modules and fails stale entries", () => { // Phases 2 to 5 remove pairs until the map is empty. const FROZEN_ALLOW_PAIRS = [ "src/node/services/agentSession.ts @/node/runtime/runtimeHelpers", + "src/node/services/agentSession.ts @/node/services/workspaceUseLeases", "src/node/services/agentSession.ts fs/promises", "src/node/services/aiService.ts @/node/runtime/multiProjectRuntime", "src/node/services/aiService.ts @/node/runtime/runtimeFactory", @@ -128,15 +129,19 @@ const FROZEN_ALLOW_PAIRS = [ "src/node/services/aiService.ts @/node/utils/fs", "src/node/services/aiService.ts fs/promises", "src/node/services/contextManagement/strategies/tokenBudget.ts @/node/runtime/runtimeHelpers", + "src/node/services/historyService.ts @/node/utils/concurrency/fileLock", + "src/node/services/historyService.ts @/node/utils/concurrency/workspaceFileLocks", "src/node/services/tools/agent_skill_read_file.ts @/node/runtime/RemoteRuntime", "src/node/services/tools/bash.ts @/node/runtime/LocalBaseRuntime", "src/node/services/tools/bash.ts fs", "src/node/services/tools/notify.ts electron", + "src/node/services/tools/review_pane.ts @/node/utils/concurrency/workspaceFileLocks", "src/node/services/tools/review_pane.ts @/node/utils/writeFileAtomic", "src/node/services/tools/review_pane.ts fs/promises", "src/node/services/tools/shared/configReadWrite.ts @/node/utils/writeFileAtomic", "src/node/services/tools/shared/configReadWrite.ts fs/promises", "src/node/services/tools/skillFileHint.ts @/node/runtime/RemoteRuntime", + "src/node/services/tools/todo.ts @/node/utils/concurrency/workspaceFileLocks", "src/node/services/tools/todo.ts @/node/utils/writeFileAtomic", "src/node/services/tools/todo.ts fs/promises", "src/node/services/tools/xum_agents_path.ts fs/promises", From e3cfd1ca5f1288dde47bc0cf02ee00652da62cec Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 17:19:28 +0000 Subject: [PATCH 06/19] tests: add the memory and foreign-holder SessionLocks layers to the contract SessionLocksMemory keeps lock state in a table per build and waits on the HostClock, so a TestClock test moves time instead of waiting. withForeignHolders plays other backends: a live holder refuses at once with today's text, and a dead one is reclaimed. The contract now runs on node, memory, node + foreign and memory + foreign. --- .../foreignHolders.testHarness.ts | 174 +++++++++++++ .../memorySessionLocks.testHarness.ts | 246 ++++++++++++++++++ .../sessionLocks.contract.test.ts | 217 ++++++++++++++- 3 files changed, 631 insertions(+), 6 deletions(-) create mode 100644 src/node/services/sessionLocks/foreignHolders.testHarness.ts create mode 100644 src/node/services/sessionLocks/memorySessionLocks.testHarness.ts diff --git a/src/node/services/sessionLocks/foreignHolders.testHarness.ts b/src/node/services/sessionLocks/foreignHolders.testHarness.ts new file mode 100644 index 00000000000..25583d7ab10 --- /dev/null +++ b/src/node/services/sessionLocks/foreignHolders.testHarness.ts @@ -0,0 +1,174 @@ +/** + * A SessionLocks layer that plays other backends: a test puts a live or dead + * foreign holder on a lock path, and the wrapped layer refuses or reclaims as + * the Node locks would, with today's text. A live holder refuses at once (the + * Node locks would wait the whole timeout first), so tests need no real wait. + */ +import assert from "node:assert"; +import { Effect, Layer } from "effect"; +import { SessionLocksTag } from "@/node/services/di/tags"; +import { + workspaceMutationInProgressMessage, + workspaceMutationLockPath, +} from "@/node/services/workspaceUseLeases"; +import { fileLockTimeoutMessage, type FileLockBlocker } from "@/node/utils/concurrency/fileLock"; +import { + CrossProcessLockTimeoutError, + WorkspaceMutationInProgressError, +} from "@/node/utils/concurrency/lockErrors"; +import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; +import { + crossProcessLockTimeoutMessage, + describeHolder, + type CrossProcessLockHolder, +} from "@/node/utils/main/crossProcessLock"; +import type { SessionLocks } from "./sessionLocks"; + +/** The Node locks' reason for a live holder in this PID domain (processLiveness.judgeHolder). */ +const LIVE_WHY = process.platform === "linux" ? "that process is running" : "that pid is running"; + +/** Why the Node locks refuse a live holder with this pid: the blocker of their timeout text. */ +export function liveBlocker(pid: number, identity?: ProcessIdentity): FileLockBlocker { + return { holder: { pid, identity }, why: LIVE_WHY }; +} + +/** The record a live foreign holder would have written: describeHolder reads pid and identity. */ +export function foreignRecord(pid: number, identity?: ProcessIdentity): CrossProcessLockHolder { + return { pid, token: `foreign-${pid}`, acquiredAt: 0, identity }; +} + +/** The handle that the test keeps. It holds the foreign holders, not the locks. */ +export interface ForeignHolders { + /** + * Another backend holds lockPath. live: refuse at once. + * dead: the next acquire reclaims it, and inspect says "dead" until then. + */ + hold(lockPath: string, state: "live" | "dead", holder?: { pid: number }): void; + /** The foreign holder of lockPath released it cleanly. */ + release(lockPath: string): void; + /** Paths whose dead holder an acquire reclaimed, in order. */ + readonly reclaimed: readonly string[]; +} + +interface Holder { + state: "live" | "dead"; + pid: number; +} + +/** The test's side of makeForeignHolders: wrapWithForeignHolders reads the same map. */ +const holdersOf = new WeakMap>(); + +/** A pid no test process has: the fake holder must not be this process. */ +const DEFAULT_FOREIGN_PID = 4242; + +export function makeForeignHolders(): ForeignHolders { + const table = new Map(); + const reclaimed: string[] = []; + const holders: ForeignHolders = { + hold: (lockPath, state, holder) => { + const pid = holder?.pid ?? DEFAULT_FOREIGN_PID; + assert(pid !== process.pid, "a foreign holder must not be this process"); + table.set(lockPath, { state, pid }); + }, + release: (lockPath) => { + assert(table.delete(lockPath), `no foreign holder on ${lockPath}`); + }, + reclaimed, + }; + holdersOf.set(holders, table); + return holders; +} + +/** + * The locks that check the foreign holders before each call. A dead holder is + * reclaimed by the acquire that meets it: it leaves the table, the path goes to + * `reclaimed`, and the call goes on to the inner locks. + */ +export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHolders): SessionLocks { + const table = holdersOf.get(holders); + assert(table !== undefined, "holders must come from makeForeignHolders"); + const reclaimed = holders.reclaimed as string[]; + /** The live holder of lockPath, after it reclaims a dead one. */ + const liveHolder = (lockPath: string): Holder | undefined => { + const holder = table.get(lockPath); + if (holder?.state === "dead") { + table.delete(lockPath); + reclaimed.push(lockPath); + return undefined; + } + return holder; + }; + // One object per config, as workspaceUseLeasesFor gives. + const leasesByConfig = new WeakMap>(); + return { + workspaceFiles: inner.workspaceFiles, + acquireProcessFileLock: async (options) => { + const holder = liveHolder(options.lockPath); + if (holder !== undefined) { + throw new Error( + fileLockTimeoutMessage( + options.label, + options.lockPath, + options.timeoutMs, + liveBlocker(holder.pid) + ) + ); + } + return await inner.acquireProcessFileLock(options); + }, + acquireCrossProcessLock: async (options) => { + const holder = liveHolder(options.lockPath); + if (holder !== undefined) { + throw new CrossProcessLockTimeoutError( + crossProcessLockTimeoutMessage(options.timeoutMessage, options.lockPath, { + holder: foreignRecord(holder.pid), + why: LIVE_WHY, + }) + ); + } + return await inner.acquireCrossProcessLock(options); + }, + // A probe never reclaims: it reports what an acquire would do. + inspectCrossProcessLock: async (lockPath) => { + const holder = table.get(lockPath); + if (holder === undefined) return await inner.inspectCrossProcessLock(lockPath); + if (holder.state === "dead") return { state: "dead" }; + return { state: "held", holder: describeHolder(foreignRecord(holder.pid)), why: LIVE_WHY }; + }, + useLeasesFor: (config) => { + const cached = leasesByConfig.get(config); + if (cached !== undefined) return cached; + const leases = inner.useLeasesFor(config); + const wrapped: ReturnType = { + // A live foreign mutator holds the gate. The inner hold reads the gate on + // disk, where a fake holder leaves no file, so the check comes first. + hold: async (workspaceId, kind) => { + const gate = table.get(workspaceMutationLockPath(config.rootDir, workspaceId)); + if (gate?.state === "live") { + throw new WorkspaceMutationInProgressError( + workspaceMutationInProgressMessage(workspaceId, { + holder: describeHolder(foreignRecord(gate.pid)), + why: LIVE_WHY, + }) + ); + } + return await leases.hold(workspaceId, kind); + }, + isHeld: (workspaceId, kind) => leases.isHeld(workspaceId, kind), + }; + leasesByConfig.set(config, wrapped); + return wrapped; + }, + }; +} + +/** A layer that builds inner, then checks the foreign holders before each call. */ +export function withForeignHolders( + inner: Layer.Layer, + holders: ForeignHolders +): Layer.Layer { + return Layer.effect( + SessionLocksTag, + Effect.map(SessionLocksTag, (locks) => wrapWithForeignHolders(locks, holders)) + ).pipe(Layer.provide(inner)); +} diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts new file mode 100644 index 00000000000..e05228ba517 --- /dev/null +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -0,0 +1,246 @@ +/** + * SessionLocks in memory: the lock files and records live in a table, and a + * waiter waits on the HostClock, so under a TestClock a test moves time + * instead of waiting. Refusals use today's text (fileLockTimeoutMessage, + * crossProcessLockTimeoutMessage). Two parts stay on the Node locks: + * - Init-record locks: InitStateManager lists their folder (initStateManager.ts + * isRecordLockHeld), so a holder in memory would leave no file, and a live + * init would look abandoned. + * - The use leases: the mutation gate, isHeld and taskService read other + * backends' lease files. Phase 5 moves them. + */ +import assert from "node:assert"; +import * as path from "node:path"; +import { Effect, Layer } from "effect"; +import { SessionLocksTag } from "@/node/services/di/tags"; +import { HostClockTag } from "@/node/services/di/hostClockLive"; +import type { HostClock } from "@/node/utils/hostClock"; +import type { ProcessFileLock } from "@/node/utils/concurrency/fileLock"; +import { fileLockTimeoutMessage } from "@/node/utils/concurrency/fileLock"; +import { CrossProcessLockTimeoutError } from "@/node/utils/concurrency/lockErrors"; +import { getSelfIdentity, type ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; +import { + crossProcessLockTimeoutMessage, + describeHolder, + type CrossProcessLockState, +} from "@/node/utils/main/crossProcessLock"; +import { nodeSessionLocks } from "./nodeSessionLocks"; +import type { SessionLocks } from "./sessionLocks"; +import { foreignRecord, liveBlocker } from "./foreignHolders.testHarness"; + +/** Who holds a lock in the table. */ +interface Entry { + pid: number; + identity?: ProcessIdentity; + /** A foreign holder whose process is gone: the next acquire reclaims it. */ + dead: boolean; +} + +/** Why an acquire did not get the lock: the blocker of the timeout text. */ +interface Blocker { + pid: number; + identity?: ProcessIdentity; + why: string; +} + +interface Waiter { + take(entry: Entry): void; +} + +/** The lock state of one build of the memory layer. */ +export class MemoryLockTable { + private readonly held = new Map(); + /** FIFO per lock path: a release hands the lock to the first waiter. */ + private readonly waiters = new Map(); + + /** Another backend holds lockPath (pid 4242 unless given). */ + hold(lockPath: string, state: "live" | "dead", holder?: { pid: number }): void { + assert(!this.held.has(lockPath), `${lockPath} is already held`); + const pid = holder?.pid ?? 4242; + assert(pid !== process.pid, "a foreign holder must not be this process"); + this.held.set(lockPath, { pid, dead: state === "dead" }); + } + + /** The foreign holder of lockPath released it cleanly. */ + release(lockPath: string): void { + const entry = this.held.get(lockPath); + assert(entry !== undefined && entry.pid !== process.pid, `no foreign holder on ${lockPath}`); + this.free(lockPath, entry); + } + + owner(lockPath: string): Entry | undefined { + return this.held.get(lockPath); + } + + /** Takes a free or dead lock now. Else null, and the caller waits or refuses. */ + tryTake(lockPath: string): Entry | null { + const entry = this.held.get(lockPath); + if (entry !== undefined && !entry.dead) return null; + const own: Entry = { pid: process.pid, identity: getSelfIdentity(), dead: false }; + this.held.set(lockPath, own); + return own; + } + + /** Queues a waiter. The returned function takes it out again (timeout, abort). */ + wait(lockPath: string, waiter: Waiter): () => void { + const queue = this.waiters.get(lockPath) ?? []; + queue.push(waiter); + this.waiters.set(lockPath, queue); + return () => { + const index = queue.indexOf(waiter); + if (index >= 0) queue.splice(index, 1); + if (queue.length === 0 && this.waiters.get(lockPath) === queue) this.waiters.delete(lockPath); + }; + } + + /** Frees lockPath if entry still holds it, and hands it to the first waiter. */ + free(lockPath: string, entry: Entry): void { + if (this.held.get(lockPath) !== entry) return; + this.held.delete(lockPath); + const queue = this.waiters.get(lockPath); + const next = queue?.shift(); + if (queue?.length === 0) this.waiters.delete(lockPath); + if (next === undefined) return; + const own = this.tryTake(lockPath); + assert(own !== null, "a freed lock is free"); + next.take(own); + } +} + +/** The Node locks' reason for a holder (processLiveness.judgeHolder). */ +function blockerOf(entry: Entry): Blocker { + if (entry.pid === process.pid) { + return { pid: entry.pid, identity: entry.identity, why: "it is held by this process" }; + } + const { holder, why } = liveBlocker(entry.pid, entry.identity); + assert(holder !== undefined); + return { pid: holder.pid, identity: holder.identity, why }; +} + +/** + * Takes lockPath, or waits until a release hands it over, or refuses with + * refusal(blocker) once timeoutMs passes on the clock (at once for 0). + */ +function acquireInMemory( + table: MemoryLockTable, + lockPath: string, + timeoutMs: number, + clock: HostClock, + refusal: (blocker: Blocker) => Error, + signal?: AbortSignal +): Promise { + const aborted = () => new Error("Lock acquisition was aborted"); + if (signal?.aborted) return Promise.reject(aborted()); + const taken = table.tryTake(lockPath); + if (taken !== null) return Promise.resolve(taken); + const holder = table.owner(lockPath); + assert(holder !== undefined, "a lock that cannot be taken has a holder"); + if (timeoutMs <= 0) return Promise.reject(refusal(blockerOf(holder))); + return new Promise((resolve, reject) => { + const settle = () => { + leave(); + cancelTimer(); + signal?.removeEventListener("abort", onAbort); + }; + const onAbort = () => { + settle(); + reject(aborted()); + }; + const leave = table.wait(lockPath, { + take: (entry) => { + settle(); + resolve(entry); + }, + }); + const cancelTimer = clock.setTimer(() => { + settle(); + const current = table.owner(lockPath); + assert(current !== undefined, "a waiter waits only on a held lock"); + reject(refusal(blockerOf(current))); + }, timeoutMs); + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +function inspectInMemory(table: MemoryLockTable, lockPath: string): CrossProcessLockState { + const entry = table.owner(lockPath); + if (entry === undefined) return { state: "absent" }; + if (entry.dead) return { state: "dead" }; + const { pid, identity, why } = blockerOf(entry); + return { state: "held", holder: describeHolder(foreignRecord(pid, identity)), why }; +} + +/** An init-record lock: InitStateManager finds these by listing their folder. */ +export function foundByReaddir(lockPath: string): boolean { + return lockPath.split(path.sep).join("/").includes("/locks/init-record/"); +} + +/** The memory locks over one table. SessionLocksMemory gives each build its own. */ +export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): SessionLocks { + return { + // Pure memory already: every session-file service shares this one FIFO. + workspaceFiles: nodeSessionLocks.workspaceFiles, + acquireProcessFileLock: async (options): Promise => { + const { lockPath, label, timeoutMs } = options; + // The same argument checks as fileLock.ts acquireProcessFileLock. + assert(lockPath.length > 0, "acquireProcessFileLock requires a lock path"); + assert(timeoutMs > 0, "acquireProcessFileLock timeoutMs must be positive"); + const entry = await acquireInMemory(table, lockPath, timeoutMs, clock, (blocker) => { + const { why, ...holder } = blocker; + return new Error(fileLockTimeoutMessage(label, lockPath, timeoutMs, { holder, why })); + }); + return { + assertStillOwned: () => { + if (table.owner(lockPath) === entry) return Promise.resolve(); + // fileLock.ts assertStillOwned + return Promise.reject( + new Error( + `${label} ${lockPath} is no longer owned by this holder (displaced or reclaimed); ` + + `aborting before mutation` + ) + ); + }, + [Symbol.asyncDispose]: () => { + table.free(lockPath, entry); + return Promise.resolve(); + }, + }; + }, + acquireCrossProcessLock: async (options) => { + if (foundByReaddir(options.lockPath)) { + return await nodeSessionLocks.acquireCrossProcessLock(options); + } + const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; + const refusal = (blocker: Blocker) => + new CrossProcessLockTimeoutError( + crossProcessLockTimeoutMessage(timeoutMessage, lockPath, { + holder: foreignRecord(blocker.pid, blocker.identity), + why: blocker.why, + }) + ); + const entry = await acquireInMemory( + table, + lockPath, + acquireTimeoutMs, + clock, + refusal, + signal + ); + return () => { + table.free(lockPath, entry); + return Promise.resolve(); + }; + }, + inspectCrossProcessLock: async (lockPath) => + foundByReaddir(lockPath) + ? await nodeSessionLocks.inspectCrossProcessLock(lockPath) + : inspectInMemory(table, lockPath), + useLeasesFor: (config) => nodeSessionLocks.useLeasesFor(config), + }; +} + +/** Each build gets fresh lock tables: no lock state outlives a test's layer. */ +export const SessionLocksMemory: Layer.Layer = Layer.effect( + SessionLocksTag, + Effect.map(HostClockTag, (clock) => memorySessionLocks(clock, new MemoryLockTable())) +); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index f9143a4680c..2789da584e7 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -2,8 +2,29 @@ import { afterEach, expect, test } from "bun:test"; import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; +import { Layer } from "effect"; +import { SessionLocksTag } from "@/node/services/di/tags"; +import { HostClockTag } from "@/node/services/di/hostClockLive"; +import { makeTestEffectRunner } from "@/node/services/di/testEffectRunner"; +import { + WorkspaceMutationInProgressError, + workspaceMutationInProgressMessage, + workspaceMutationLockPath, +} from "@/node/services/workspaceUseLeases"; import { fileLockTimeoutMessage } from "@/node/utils/concurrency/fileLock"; import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; +import { describeHolder } from "@/node/utils/main/crossProcessLock"; +import { + foreignRecord, + liveBlocker, + makeForeignHolders, + withForeignHolders, +} from "./foreignHolders.testHarness"; +import { + MemoryLockTable, + memorySessionLocks, + SessionLocksMemory, +} from "./memorySessionLocks.testHarness"; import { SessionLocksNode } from "./nodeSessionLocks"; import { buildSessionLocks, @@ -11,6 +32,7 @@ import { type ForeignHolder, type LocksUnderTest, } from "./sessionLocks.contract.testHarness"; +import type { SessionLocks } from "./sessionLocks"; const CHILD = path.join(import.meta.dir, "sessionLocks.testChild.ts"); const children: Array> = []; @@ -75,7 +97,54 @@ async function nodeLocks(): Promise { }; } +/** The memory layer over a table the test can reach, on a fresh TestClock. */ +async function memoryLocks(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "xum-session-locks-")); + const clock = makeTestEffectRunner(); + const table = new MemoryLockTable(); + return { + layer: Layer.succeed(SessionLocksTag)(memorySessionLocks(clock.hostClock, table)), + dir, + foreignHolder: (lockPath, state) => { + table.hold(lockPath, state, { pid: 4242 }); + return Promise.resolve({ + pid: 4242, + identity: null, + release: () => Promise.resolve(table.release(lockPath)), + }); + }, + advance: (ms) => clock.adjust(ms), + cleanup: async () => { + await clock.dispose(); + await fs.rm(dir, { recursive: true, force: true }); + }, + }; +} + +/** Wraps a layer under test: its foreign holders come from makeForeignHolders. */ +function withForeign(make: () => Promise): () => Promise { + return async () => { + const t = await make(); + const holders = makeForeignHolders(); + return { + ...t, + layer: withForeignHolders(t.layer, holders), + foreignHolder: (lockPath, state) => { + holders.hold(lockPath, state, { pid: 4343 }); + return Promise.resolve({ + pid: 4343, + identity: null, + release: () => Promise.resolve(holders.release(lockPath)), + }); + }, + }; + }; +} + describeSessionLocksContract("node", nodeLocks); +describeSessionLocksContract("memory", memoryLocks); +describeSessionLocksContract("node + foreign", withForeign(nodeLocks)); +describeSessionLocksContract("memory + foreign", withForeign(memoryLocks)); test("node: the timeout text for a live child holder is fileLockTimeoutMessage", async () => { const t = await nodeLocks(); @@ -89,16 +158,152 @@ test("node: the timeout text for a live child holder is fileLockTimeoutMessage", () => null, (caught: unknown) => caught ); - // The fake layers build their refusal with the same function, so both texts match. - const why = process.platform === "linux" ? "that process is running" : "that pid is running"; + // The fake layers refuse with liveBlocker, so the fake text equals the Node text. expect((error as Error).message).toBe( - fileLockTimeoutMessage("contract lock", lockPath, 200, { - holder: { pid: holder.pid, identity: holder.identity ?? undefined }, - why, - }) + fileLockTimeoutMessage( + "contract lock", + lockPath, + 200, + liveBlocker(holder.pid, holder.identity ?? undefined) + ) ); await holder.release(); } finally { await t.cleanup(); } }); + +/** Runs fn on the memory layer of a fresh memoryLocks(), then cleans up. */ +async function onMemory( + fn: (locks: SessionLocks, t: LocksUnderTest) => Promise +): Promise { + const t = await memoryLocks(); + try { + await fn(await buildSessionLocks(t.layer), t); + } finally { + await t.cleanup(); + } +} + +test("memory: a 60 s timeout passes on TestClock.adjust, with no real wait", () => + onMemory(async (locks, t) => { + const lockPath = path.join(t.dir, "long.lock"); + await t.foreignHolder(lockPath, "live", "file"); + const refused = locks + .acquireProcessFileLock({ lockPath, timeoutMs: 60_000, label: "contract lock" }) + .then( + () => null, + (error: unknown) => error + ); + await t.advance(59_999); + let settled = false; + void refused.then(() => (settled = true)); + await Promise.resolve(); + expect(settled).toBe(false); + await t.advance(1); + expect(((await refused) as Error).message).toBe( + fileLockTimeoutMessage("contract lock", lockPath, 60_000, liveBlocker(4242)) + ); + })); + +test("memory: a release hands the file lock to the waiters in FIFO order", () => + onMemory(async (locks, t) => { + const options = { lockPath: path.join(t.dir, "fifo.lock"), timeoutMs: 1_000, label: "l" }; + const first = await locks.acquireProcessFileLock(options); + const order: string[] = []; + const waiter = (name: string) => + locks.acquireProcessFileLock(options).then(async (lock) => { + order.push(name); + await lock[Symbol.asyncDispose](); + }); + const both = Promise.all([waiter("b"), waiter("c")]); + await first[Symbol.asyncDispose](); + await both; + expect(order).toEqual(["b", "c"]); + await expect(first.assertStillOwned()).rejects.toThrow("is no longer owned by this holder"); + })); + +test("memory: an init-record lock makes a real file, and inspect sees it", () => + onMemory(async (locks, t) => { + const lockPath = path.join(t.dir, "locks", "init-record", "ws", "backend.lock"); + const release = await locks.acquireCrossProcessLock({ + lockPath, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "init record is held.", + }); + expect((await fs.stat(lockPath)).isFile()).toBe(true); + expect(await locks.inspectCrossProcessLock(lockPath)).toMatchObject({ state: "held" }); + await release(); + expect(await locks.inspectCrossProcessLock(lockPath)).toEqual({ state: "absent" }); + })); + +test("memory: each build of SessionLocksMemory has its own lock table", async () => { + const clock = makeTestEffectRunner(); + try { + const layer = SessionLocksMemory.pipe( + Layer.provide(Layer.succeed(HostClockTag)(clock.hostClock)) + ); + const options = { lockPath: "/memory/one.lock", timeoutMs: 200, label: "l" }; + const a = await (await buildSessionLocks(layer)).acquireProcessFileLock(options); + const b = await (await buildSessionLocks(layer)).acquireProcessFileLock(options); + await a[Symbol.asyncDispose](); + await b[Symbol.asyncDispose](); + } finally { + await clock.dispose(); + } +}); + +test("foreign: dead holders are reclaimed in order, and a probe reclaims none", async () => { + const t = await nodeLocks(); + try { + const holders = makeForeignHolders(); + const locks = await buildSessionLocks(withForeignHolders(t.layer, holders)); + const file = path.join(t.dir, "a.lock"); + const record = path.join(t.dir, "b.lock"); + holders.hold(record, "dead"); + holders.hold(file, "dead"); + expect(await locks.inspectCrossProcessLock(record)).toEqual({ state: "dead" }); + const lock = await locks.acquireProcessFileLock({ lockPath: file, timeoutMs: 200, label: "l" }); + const release = await locks.acquireCrossProcessLock({ + lockPath: record, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "held.", + }); + expect(holders.reclaimed).toEqual([file, record]); + await lock[Symbol.asyncDispose](); + await release(); + } finally { + await t.cleanup(); + } +}); + +test("foreign: a live foreign mutator makes hold() refuse with today's text", async () => { + const t = await nodeLocks(); + try { + const holders = makeForeignHolders(); + const locks = await buildSessionLocks(withForeignHolders(t.layer, holders)); + const config = { rootDir: t.dir }; + const leases = locks.useLeasesFor(config); + expect(locks.useLeasesFor(config)).toBe(leases); + holders.hold(workspaceMutationLockPath(t.dir, "ws-gate"), "live", { pid: 4444 }); + const error = await leases.hold("ws-gate", "turn").then( + () => null, + (caught: unknown) => caught + ); + expect(error).toBeInstanceOf(WorkspaceMutationInProgressError); + expect((error as Error).message).toBe( + workspaceMutationInProgressMessage("ws-gate", { + holder: describeHolder(foreignRecord(4444)), + why: liveBlocker(4444).why, + }) + ); + expect(await leases.isHeld("ws-gate", "turn")).toBe(false); + holders.release(workspaceMutationLockPath(t.dir, "ws-gate")); + const lease = await leases.hold("ws-gate", "turn"); + await lease.release(); + } finally { + await t.cleanup(); + } +}); From 30fe6b1db7d170c2c5ea6951f5d56bc7a3a4855f Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 18:24:47 +0000 Subject: [PATCH 07/19] tests: match the Node locks in the fake layers on abort, missing parents, gate order and holder text Codex round 1 on #7367: an aborted signal wins over a fake live holder, the memory layer rejects with ENOENT when createParentDirectory is false and the parent is gone, the fake gate is probed after the lease is published, fake holders record an identity so the text equals the Node text for a current record, and fake pids are offsets from process.pid. --- .../foreignHolders.testHarness.ts | 53 ++++++++++++++---- .../memorySessionLocks.testHarness.ts | 23 +++++--- .../sessionLocks.contract.test.ts | 55 +++++++++++++++---- .../sessionLocks.contract.testHarness.ts | 38 +++++++++++++ 4 files changed, 141 insertions(+), 28 deletions(-) diff --git a/src/node/services/sessionLocks/foreignHolders.testHarness.ts b/src/node/services/sessionLocks/foreignHolders.testHarness.ts index 25583d7ab10..e00729bd1e3 100644 --- a/src/node/services/sessionLocks/foreignHolders.testHarness.ts +++ b/src/node/services/sessionLocks/foreignHolders.testHarness.ts @@ -16,7 +16,7 @@ import { CrossProcessLockTimeoutError, WorkspaceMutationInProgressError, } from "@/node/utils/concurrency/lockErrors"; -import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; +import { getSelfIdentity, type ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; import { crossProcessLockTimeoutMessage, describeHolder, @@ -24,19 +24,49 @@ import { } from "@/node/utils/main/crossProcessLock"; import type { SessionLocks } from "./sessionLocks"; -/** The Node locks' reason for a live holder in this PID domain (processLiveness.judgeHolder). */ +/** + * The Node locks' reason for a live holder with a current record in this PID + * domain (processLiveness.judgeHolder). A record without an identity takes the + * legacy branch, which words it otherwise, so every fake holder has one. + */ const LIVE_WHY = process.platform === "linux" ? "that process is running" : "that pid is running"; +/** + * The identity a fake foreign holder records: another backend on this host, so + * the same platform, boot and PID namespace. It borrows this process's start + * time, which only describeHolder prints. + */ +export function foreignIdentity(): ProcessIdentity { + return getSelfIdentity(); +} + +/** A fake holder pid: n > 0 apart from this process, so it is never this process. */ +export function foreignPid(n = 1): number { + assert(Number.isInteger(n) && n > 0, "foreignPid needs a positive integer"); + return process.pid + n; +} + /** Why the Node locks refuse a live holder with this pid: the blocker of their timeout text. */ -export function liveBlocker(pid: number, identity?: ProcessIdentity): FileLockBlocker { +export function liveBlocker( + pid: number, + identity: ProcessIdentity = foreignIdentity() +): FileLockBlocker { return { holder: { pid, identity }, why: LIVE_WHY }; } /** The record a live foreign holder would have written: describeHolder reads pid and identity. */ -export function foreignRecord(pid: number, identity?: ProcessIdentity): CrossProcessLockHolder { +export function foreignRecord( + pid: number, + identity: ProcessIdentity = foreignIdentity() +): CrossProcessLockHolder { return { pid, token: `foreign-${pid}`, acquiredAt: 0, identity }; } +/** The error of a Node acquire whose signal is aborted: it checks the signal before any attempt. */ +export function abortedAcquireError(): Error { + return new Error("Lock acquisition was aborted"); +} + /** The handle that the test keeps. It holds the foreign holders, not the locks. */ export interface ForeignHolders { /** @@ -58,15 +88,12 @@ interface Holder { /** The test's side of makeForeignHolders: wrapWithForeignHolders reads the same map. */ const holdersOf = new WeakMap>(); -/** A pid no test process has: the fake holder must not be this process. */ -const DEFAULT_FOREIGN_PID = 4242; - export function makeForeignHolders(): ForeignHolders { const table = new Map(); const reclaimed: string[] = []; const holders: ForeignHolders = { hold: (lockPath, state, holder) => { - const pid = holder?.pid ?? DEFAULT_FOREIGN_PID; + const pid = holder?.pid ?? foreignPid(); assert(pid !== process.pid, "a foreign holder must not be this process"); table.set(lockPath, { state, pid }); }, @@ -117,6 +144,8 @@ export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHold return await inner.acquireProcessFileLock(options); }, acquireCrossProcessLock: async (options) => { + // crossProcessLock.ts checks the signal before each attempt: abort wins over a holder. + if (options.signal?.aborted) throw abortedAcquireError(); const holder = liveHolder(options.lockPath); if (holder !== undefined) { throw new CrossProcessLockTimeoutError( @@ -141,10 +170,14 @@ export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHold const leases = inner.useLeasesFor(config); const wrapped: ReturnType = { // A live foreign mutator holds the gate. The inner hold reads the gate on - // disk, where a fake holder leaves no file, so the check comes first. + // disk, where a fake holder leaves no file. Like WorkspaceUseLeases.hold, the + // lease is published first and the gate probed second, so a mutator that + // takes the gate meanwhile is not missed. hold: async (workspaceId, kind) => { + const lease = await leases.hold(workspaceId, kind); const gate = table.get(workspaceMutationLockPath(config.rootDir, workspaceId)); if (gate?.state === "live") { + await lease.release(); throw new WorkspaceMutationInProgressError( workspaceMutationInProgressMessage(workspaceId, { holder: describeHolder(foreignRecord(gate.pid)), @@ -152,7 +185,7 @@ export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHold }) ); } - return await leases.hold(workspaceId, kind); + return lease; }, isHeld: (workspaceId, kind) => leases.isHeld(workspaceId, kind), }; diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index e05228ba517..2ba03b0fb63 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -10,6 +10,7 @@ * backends' lease files. Phase 5 moves them. */ import assert from "node:assert"; +import * as fs from "node:fs/promises"; import * as path from "node:path"; import { Effect, Layer } from "effect"; import { SessionLocksTag } from "@/node/services/di/tags"; @@ -26,7 +27,13 @@ import { } from "@/node/utils/main/crossProcessLock"; import { nodeSessionLocks } from "./nodeSessionLocks"; import type { SessionLocks } from "./sessionLocks"; -import { foreignRecord, liveBlocker } from "./foreignHolders.testHarness"; +import { + abortedAcquireError, + foreignIdentity, + foreignPid, + foreignRecord, + liveBlocker, +} from "./foreignHolders.testHarness"; /** Who holds a lock in the table. */ interface Entry { @@ -53,12 +60,12 @@ export class MemoryLockTable { /** FIFO per lock path: a release hands the lock to the first waiter. */ private readonly waiters = new Map(); - /** Another backend holds lockPath (pid 4242 unless given). */ + /** Another backend holds lockPath (pid foreignPid() unless given). */ hold(lockPath: string, state: "live" | "dead", holder?: { pid: number }): void { assert(!this.held.has(lockPath), `${lockPath} is already held`); - const pid = holder?.pid ?? 4242; + const pid = holder?.pid ?? foreignPid(); assert(pid !== process.pid, "a foreign holder must not be this process"); - this.held.set(lockPath, { pid, dead: state === "dead" }); + this.held.set(lockPath, { pid, identity: foreignIdentity(), dead: state === "dead" }); } /** The foreign holder of lockPath released it cleanly. */ @@ -129,8 +136,7 @@ function acquireInMemory( refusal: (blocker: Blocker) => Error, signal?: AbortSignal ): Promise { - const aborted = () => new Error("Lock acquisition was aborted"); - if (signal?.aborted) return Promise.reject(aborted()); + if (signal?.aborted) return Promise.reject(abortedAcquireError()); const taken = table.tryTake(lockPath); if (taken !== null) return Promise.resolve(taken); const holder = table.owner(lockPath); @@ -144,7 +150,7 @@ function acquireInMemory( }; const onAbort = () => { settle(); - reject(aborted()); + reject(abortedAcquireError()); }; const leave = table.wait(lockPath, { take: (entry) => { @@ -211,6 +217,9 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se return await nodeSessionLocks.acquireCrossProcessLock(options); } const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; + // Node rejects with ENOENT when the parent is gone and may not be recreated + // (a removed workflow run directory). Callers rely on that refusal. + if (options.createParentDirectory === false) await fs.stat(path.dirname(lockPath)); const refusal = (blocker: Blocker) => new CrossProcessLockTimeoutError( crossProcessLockTimeoutMessage(timeoutMessage, lockPath, { diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index 2789da584e7..fca6993622f 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -15,17 +15,20 @@ import { fileLockTimeoutMessage } from "@/node/utils/concurrency/fileLock"; import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; import { describeHolder } from "@/node/utils/main/crossProcessLock"; import { + foreignIdentity, + foreignPid, foreignRecord, liveBlocker, makeForeignHolders, withForeignHolders, + wrapWithForeignHolders, } from "./foreignHolders.testHarness"; import { MemoryLockTable, memorySessionLocks, SessionLocksMemory, } from "./memorySessionLocks.testHarness"; -import { SessionLocksNode } from "./nodeSessionLocks"; +import { nodeSessionLocks, SessionLocksNode } from "./nodeSessionLocks"; import { buildSessionLocks, describeSessionLocksContract, @@ -106,10 +109,10 @@ async function memoryLocks(): Promise { layer: Layer.succeed(SessionLocksTag)(memorySessionLocks(clock.hostClock, table)), dir, foreignHolder: (lockPath, state) => { - table.hold(lockPath, state, { pid: 4242 }); + table.hold(lockPath, state, { pid: foreignPid(1) }); return Promise.resolve({ - pid: 4242, - identity: null, + pid: foreignPid(1), + identity: foreignIdentity(), release: () => Promise.resolve(table.release(lockPath)), }); }, @@ -130,10 +133,10 @@ function withForeign(make: () => Promise): () => Promise { - holders.hold(lockPath, state, { pid: 4343 }); + holders.hold(lockPath, state, { pid: foreignPid(2) }); return Promise.resolve({ - pid: 4343, - identity: null, + pid: foreignPid(2), + identity: foreignIdentity(), release: () => Promise.resolve(holders.release(lockPath)), }); }, @@ -202,7 +205,7 @@ test("memory: a 60 s timeout passes on TestClock.adjust, with no real wait", () expect(settled).toBe(false); await t.advance(1); expect(((await refused) as Error).message).toBe( - fileLockTimeoutMessage("contract lock", lockPath, 60_000, liveBlocker(4242)) + fileLockTimeoutMessage("contract lock", lockPath, 60_000, liveBlocker(foreignPid(1))) ); })); @@ -287,7 +290,7 @@ test("foreign: a live foreign mutator makes hold() refuse with today's text", as const config = { rootDir: t.dir }; const leases = locks.useLeasesFor(config); expect(locks.useLeasesFor(config)).toBe(leases); - holders.hold(workspaceMutationLockPath(t.dir, "ws-gate"), "live", { pid: 4444 }); + holders.hold(workspaceMutationLockPath(t.dir, "ws-gate"), "live", { pid: foreignPid(3) }); const error = await leases.hold("ws-gate", "turn").then( () => null, (caught: unknown) => caught @@ -295,8 +298,8 @@ test("foreign: a live foreign mutator makes hold() refuse with today's text", as expect(error).toBeInstanceOf(WorkspaceMutationInProgressError); expect((error as Error).message).toBe( workspaceMutationInProgressMessage("ws-gate", { - holder: describeHolder(foreignRecord(4444)), - why: liveBlocker(4444).why, + holder: describeHolder(foreignRecord(foreignPid(3))), + why: liveBlocker(foreignPid(3)).why, }) ); expect(await leases.isHeld("ws-gate", "turn")).toBe(false); @@ -307,3 +310,33 @@ test("foreign: a live foreign mutator makes hold() refuse with today's text", as await t.cleanup(); } }); + +test("foreign: hold() publishes its lease before it probes the gate", async () => { + const t = await nodeLocks(); + try { + const holders = makeForeignHolders(); + const gate = workspaceMutationLockPath(t.dir, "ws-order"); + // A mutator takes the gate while the inner hold publishes the lease. + const inner = { + ...nodeSessionLocks, + useLeasesFor: (config: { readonly rootDir: string }) => { + const leases = nodeSessionLocks.useLeasesFor(config); + return { + hold: async (...args: Parameters) => { + const lease = await leases.hold(...args); + holders.hold(gate, "live", { pid: foreignPid(4) }); + return lease; + }, + isHeld: (...args: Parameters) => leases.isHeld(...args), + }; + }, + }; + const leases = wrapWithForeignHolders(inner, holders).useLeasesFor({ rootDir: t.dir }); + await expect(leases.hold("ws-order", "turn")).rejects.toBeInstanceOf( + WorkspaceMutationInProgressError + ); + expect(await leases.isHeld("ws-order", "turn")).toBe(false); + } finally { + await t.cleanup(); + } +}); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts index c6fa7f73436..2a2d39de503 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts @@ -152,6 +152,44 @@ export function describeSessionLocksContract( await release(); })); + test("an aborted signal wins over a live holder of a record", () => + withLayer(async (locks, t) => { + const lockPath = path.join(t.dir, "aborted.lock"); + const holder = await t.foreignHolder(lockPath, "live", "record"); + const error = await locks + .acquireCrossProcessLock({ + lockPath, + acquireTimeoutMs: SHORT_MS, + staleMs: 60_000, + timeoutMessage: "contract record is held.", + signal: AbortSignal.abort(), + }) + .then( + () => null, + (caught: unknown) => caught + ); + expect(error).not.toBeInstanceOf(CrossProcessLockTimeoutError); + expect((error as Error).message).toBe("Lock acquisition was aborted"); + await holder.release(); + })); + + test("createParentDirectory false on a missing parent rejects with ENOENT", () => + withLayer(async (locks, t) => { + const error = await locks + .acquireCrossProcessLock({ + lockPath: path.join(t.dir, "removed-run", "run.lock"), + createParentDirectory: false, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "contract record is held.", + }) + .then( + () => null, + (caught: unknown) => caught + ); + expect((error as NodeJS.ErrnoException).code).toBe("ENOENT"); + })); + test("useLeasesFor gives one object per config: hold, isHeld, release", () => withLayer(async (locks, t) => { const config = { rootDir: t.dir }; From 9675e47d03b74357983ae5f652484408d6f1a56e Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 18:33:42 +0000 Subject: [PATCH 08/19] tests: check the signal first and the parent again after the take in the memory record lock Codex round 2 on #7367: an aborted signal wins over a missing parent, and a parent removed after the first check (or during the wait) refuses with ENOENT and frees the lock, as Node's write would. --- .../memorySessionLocks.testHarness.ts | 23 +++++++++++-- .../sessionLocks.contract.test.ts | 33 +++++++++++++++++++ .../sessionLocks.contract.testHarness.ts | 18 ++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index 2ba03b0fb63..77f042a3b0a 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -217,9 +217,18 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se return await nodeSessionLocks.acquireCrossProcessLock(options); } const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; + // crossProcessLock.ts checks the signal before each attempt, so abort wins. + if (signal?.aborted) throw abortedAcquireError(); // Node rejects with ENOENT when the parent is gone and may not be recreated - // (a removed workflow run directory). Callers rely on that refusal. - if (options.createParentDirectory === false) await fs.stat(path.dirname(lockPath)); + // (a removed workflow run directory): each attempt's first write fails, before + // any holder is judged. Callers rely on that refusal. + const checkParent = + options.createParentDirectory === false + ? async () => { + await fs.stat(path.dirname(lockPath)); + } + : null; + await checkParent?.(); const refusal = (blocker: Blocker) => new CrossProcessLockTimeoutError( crossProcessLockTimeoutMessage(timeoutMessage, lockPath, { @@ -235,6 +244,16 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se refusal, signal ); + // Check again once the lock is taken: a parent removed after the first check + // (or while this waited) must refuse here too, as Node's write would. + if (checkParent !== null) { + try { + await checkParent(); + } catch (error) { + table.free(lockPath, entry); + throw error; + } + } return () => { table.free(lockPath, entry); return Promise.resolve(); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index fca6993622f..d7589997c86 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -226,6 +226,39 @@ test("memory: a release hands the file lock to the waiters in FIFO order", () => await expect(first.assertStillOwned()).rejects.toThrow("is no longer owned by this holder"); })); +test("memory: a parent removed while the acquire waits refuses with ENOENT", () => + onMemory(async (locks, t) => { + const runDir = path.join(t.dir, "run"); + await fs.mkdir(runDir); + const lockPath = path.join(runDir, "run.lock"); + const holder = await t.foreignHolder(lockPath, "live", "record"); + const acquire = locks + .acquireCrossProcessLock({ + lockPath, + createParentDirectory: false, + acquireTimeoutMs: 1_000, + staleMs: 60_000, + timeoutMessage: "run lock is held.", + }) + .then( + () => null, + (caught: unknown) => caught + ); + await fs.rm(runDir, { recursive: true }); + await holder.release(); + expect(((await acquire) as NodeJS.ErrnoException).code).toBe("ENOENT"); + // The refusal frees the lock: a later acquire takes it at once. + await fs.mkdir(runDir); + const release = await locks.acquireCrossProcessLock({ + lockPath, + createParentDirectory: false, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "run lock is held.", + }); + await release(); + })); + test("memory: an init-record lock makes a real file, and inspect sees it", () => onMemory(async (locks, t) => { const lockPath = path.join(t.dir, "locks", "init-record", "ws", "backend.lock"); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts index 2a2d39de503..8526e200d8d 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts @@ -190,6 +190,24 @@ export function describeSessionLocksContract( expect((error as NodeJS.ErrnoException).code).toBe("ENOENT"); })); + test("an aborted signal wins over a missing parent", () => + withLayer(async (locks, t) => { + const error = await locks + .acquireCrossProcessLock({ + lockPath: path.join(t.dir, "removed-run", "run.lock"), + createParentDirectory: false, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "contract record is held.", + signal: AbortSignal.abort(), + }) + .then( + () => null, + (caught: unknown) => caught + ); + expect((error as Error).message).toBe("Lock acquisition was aborted"); + })); + test("useLeasesFor gives one object per config: hold, isHeld, release", () => withLayer(async (locks, t) => { const config = { rootDir: t.dir }; From f7525029ffa159734a3200406e96814d91c0d30f Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 18:44:37 +0000 Subject: [PATCH 09/19] tests: leave non-creatable record locks on the Node locks and count fake leases in isHeld Codex round 3 on #7367. The memory layer's parent checks did not converge, so a record lock with createParentDirectory false now goes to the Node locks, whose own write gives WorkflowRunStore the ENOENT it needs. Other record locks create their parent as Node does. The foreign layer's isHeld counts a live fake lease in the workspace's lease folder. --- .../foreignHolders.testHarness.ts | 13 ++++- .../memorySessionLocks.testHarness.ts | 34 ++++-------- .../sessionLocks.contract.test.ts | 52 +++++++------------ .../sessionLocks.contract.testHarness.ts | 14 +++++ 4 files changed, 54 insertions(+), 59 deletions(-) diff --git a/src/node/services/sessionLocks/foreignHolders.testHarness.ts b/src/node/services/sessionLocks/foreignHolders.testHarness.ts index e00729bd1e3..540ec9add82 100644 --- a/src/node/services/sessionLocks/foreignHolders.testHarness.ts +++ b/src/node/services/sessionLocks/foreignHolders.testHarness.ts @@ -5,11 +5,13 @@ * Node locks would wait the whole timeout first), so tests need no real wait. */ import assert from "node:assert"; +import * as path from "node:path"; import { Effect, Layer } from "effect"; import { SessionLocksTag } from "@/node/services/di/tags"; import { workspaceMutationInProgressMessage, workspaceMutationLockPath, + workspaceUseLockDir, } from "@/node/services/workspaceUseLeases"; import { fileLockTimeoutMessage, type FileLockBlocker } from "@/node/utils/concurrency/fileLock"; import { @@ -187,7 +189,16 @@ export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHold } return lease; }, - isHeld: (workspaceId, kind) => leases.isHeld(workspaceId, kind), + // WorkspaceUseLeases.isHeld lists every backend's "..lock" in the + // workspace's lease folder. A live fake there is another backend's lease. + isHeld: async (workspaceId, kind) => { + const dir = workspaceUseLockDir(config.rootDir, workspaceId); + for (const [lockPath, holder] of table) { + const lease = path.dirname(lockPath) === dir && lockPath.endsWith(`.${kind}.lock`); + if (lease && holder.state === "live") return true; + } + return await leases.isHeld(workspaceId, kind); + }, }; leasesByConfig.set(config, wrapped); return wrapped; diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index 77f042a3b0a..10db7f2c606 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -2,10 +2,13 @@ * SessionLocks in memory: the lock files and records live in a table, and a * waiter waits on the HostClock, so under a TestClock a test moves time * instead of waiting. Refusals use today's text (fileLockTimeoutMessage, - * crossProcessLockTimeoutMessage). Two parts stay on the Node locks: + * crossProcessLockTimeoutMessage). Three parts stay on the Node locks: * - Init-record locks: InitStateManager lists their folder (initStateManager.ts * isRecordLockHeld), so a holder in memory would leave no file, and a live * init would look abandoned. + * - Record locks with createParentDirectory false: WorkflowRunStore needs their + * acquire to fail with ENOENT once the run directory is gone, with no window + * between a check and the take. Only the Node lock's own write gives that. * - The use leases: the mutation gate, isHeld and taskService read other * backends' lease files. Phase 5 moves them. */ @@ -213,22 +216,12 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se }; }, acquireCrossProcessLock: async (options) => { - if (foundByReaddir(options.lockPath)) { + if (foundByReaddir(options.lockPath) || options.createParentDirectory === false) { return await nodeSessionLocks.acquireCrossProcessLock(options); } const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; - // crossProcessLock.ts checks the signal before each attempt, so abort wins. - if (signal?.aborted) throw abortedAcquireError(); - // Node rejects with ENOENT when the parent is gone and may not be recreated - // (a removed workflow run directory): each attempt's first write fails, before - // any holder is judged. Callers rely on that refusal. - const checkParent = - options.createParentDirectory === false - ? async () => { - await fs.stat(path.dirname(lockPath)); - } - : null; - await checkParent?.(); + // As crossProcessLock.ts does first: a caller may write into the folder once it holds the lock. + await fs.mkdir(path.dirname(lockPath), { recursive: true }); const refusal = (blocker: Blocker) => new CrossProcessLockTimeoutError( crossProcessLockTimeoutMessage(timeoutMessage, lockPath, { @@ -244,23 +237,14 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se refusal, signal ); - // Check again once the lock is taken: a parent removed after the first check - // (or while this waited) must refuse here too, as Node's write would. - if (checkParent !== null) { - try { - await checkParent(); - } catch (error) { - table.free(lockPath, entry); - throw error; - } - } return () => { table.free(lockPath, entry); return Promise.resolve(); }; }, + // A path the table does not hold may be a Node lock (see the module doc). inspectCrossProcessLock: async (lockPath) => - foundByReaddir(lockPath) + table.owner(lockPath) === undefined ? await nodeSessionLocks.inspectCrossProcessLock(lockPath) : inspectInMemory(table, lockPath), useLeasesFor: (config) => nodeSessionLocks.useLeasesFor(config), diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index d7589997c86..055b4df3d9d 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -10,6 +10,7 @@ import { WorkspaceMutationInProgressError, workspaceMutationInProgressMessage, workspaceMutationLockPath, + workspaceUseLockDir, } from "@/node/services/workspaceUseLeases"; import { fileLockTimeoutMessage } from "@/node/utils/concurrency/fileLock"; import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; @@ -226,39 +227,6 @@ test("memory: a release hands the file lock to the waiters in FIFO order", () => await expect(first.assertStillOwned()).rejects.toThrow("is no longer owned by this holder"); })); -test("memory: a parent removed while the acquire waits refuses with ENOENT", () => - onMemory(async (locks, t) => { - const runDir = path.join(t.dir, "run"); - await fs.mkdir(runDir); - const lockPath = path.join(runDir, "run.lock"); - const holder = await t.foreignHolder(lockPath, "live", "record"); - const acquire = locks - .acquireCrossProcessLock({ - lockPath, - createParentDirectory: false, - acquireTimeoutMs: 1_000, - staleMs: 60_000, - timeoutMessage: "run lock is held.", - }) - .then( - () => null, - (caught: unknown) => caught - ); - await fs.rm(runDir, { recursive: true }); - await holder.release(); - expect(((await acquire) as NodeJS.ErrnoException).code).toBe("ENOENT"); - // The refusal frees the lock: a later acquire takes it at once. - await fs.mkdir(runDir); - const release = await locks.acquireCrossProcessLock({ - lockPath, - createParentDirectory: false, - acquireTimeoutMs: 0, - staleMs: 60_000, - timeoutMessage: "run lock is held.", - }); - await release(); - })); - test("memory: an init-record lock makes a real file, and inspect sees it", () => onMemory(async (locks, t) => { const lockPath = path.join(t.dir, "locks", "init-record", "ws", "backend.lock"); @@ -373,3 +341,21 @@ test("foreign: hold() publishes its lease before it probes the gate", async () = await t.cleanup(); } }); + +test("foreign: a live fake lease of another backend makes isHeld true", async () => { + const t = await nodeLocks(); + try { + const holders = makeForeignHolders(); + const leases = (await buildSessionLocks(withForeignHolders(t.layer, holders))).useLeasesFor({ + rootDir: t.dir, + }); + const dir = workspaceUseLockDir(t.dir, "ws-leases"); + holders.hold(path.join(dir, "other-backend.turn.lock"), "live"); + holders.hold(path.join(dir, "other-backend.exec.lock"), "dead"); + expect(await leases.isHeld("ws-leases", "turn")).toBe(true); + expect(await leases.isHeld("ws-leases", "exec")).toBe(false); + expect(await leases.isHeld("ws-leases", "terminal")).toBe(false); + } finally { + await t.cleanup(); + } +}); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts index 8526e200d8d..63778070a99 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts @@ -4,6 +4,7 @@ * case builds a fresh layer: no holders, clock or lock state is shared. */ import { describe, expect, test } from "bun:test"; +import * as fs from "node:fs/promises"; import * as path from "node:path"; import { Effect, type Layer } from "effect"; import { SessionLocksTag } from "@/node/services/di/tags"; @@ -190,6 +191,19 @@ export function describeSessionLocksContract( expect((error as NodeJS.ErrnoException).code).toBe("ENOENT"); })); + test("a record lock creates its missing parent by default", () => + withLayer(async (locks, t) => { + const lockPath = path.join(t.dir, "new-dir", "record.lock"); + const release = await locks.acquireCrossProcessLock({ + lockPath, + acquireTimeoutMs: 0, + staleMs: 60_000, + timeoutMessage: "contract record is held.", + }); + expect((await fs.stat(path.dirname(lockPath))).isDirectory()).toBe(true); + await release(); + })); + test("an aborted signal wins over a missing parent", () => withLayer(async (locks, t) => { const error = await locks From 423f813f4229b734bd69265e3384aa4734495343 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 18:53:56 +0000 Subject: [PATCH 10/19] tests: assert the two rejections without awaiting a void matcher @typescript-eslint/await-thenable (CI lint-full): bun types rejects.toThrow and rejects.toBeInstanceOf as void. --- .../sessionLocks/sessionLocks.contract.test.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index 055b4df3d9d..47f71203a46 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -224,7 +224,11 @@ test("memory: a release hands the file lock to the waiters in FIFO order", () => await first[Symbol.asyncDispose](); await both; expect(order).toEqual(["b", "c"]); - await expect(first.assertStillOwned()).rejects.toThrow("is no longer owned by this holder"); + const stale = await first.assertStillOwned().then( + () => null, + (error: unknown) => error + ); + expect((stale as Error).message).toContain("is no longer owned by this holder"); })); test("memory: an init-record lock makes a real file, and inspect sees it", () => @@ -333,9 +337,11 @@ test("foreign: hold() publishes its lease before it probes the gate", async () = }, }; const leases = wrapWithForeignHolders(inner, holders).useLeasesFor({ rootDir: t.dir }); - await expect(leases.hold("ws-order", "turn")).rejects.toBeInstanceOf( - WorkspaceMutationInProgressError + const refused = await leases.hold("ws-order", "turn").then( + () => null, + (error: unknown) => error ); + expect(refused).toBeInstanceOf(WorkspaceMutationInProgressError); expect(await leases.isHeld("ws-order", "turn")).toBe(false); } finally { await t.cleanup(); From e9272f9ea01f73a11723e783d09e47de36211093 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 19:08:53 +0000 Subject: [PATCH 11/19] tests: give the memory file lock fileLock's folder side effect and build SessionLocks without bun:test The memory acquireProcessFileLock creates the lock's parent as fileLock.ts does, and writes no lock file (#7326 follow-up from #7367 round 4). buildSessionLocks moves to its own module so testHistoryService.ts, which the Jest integration tests load, can build a layer. --- .../buildSessionLocks.testHarness.ts | 19 +++++++ .../memorySessionLocks.testHarness.ts | 3 + .../sessionLocks.contract.test.ts | 55 ++++++++++++------- .../sessionLocks.contract.testHarness.ts | 30 +++++----- 4 files changed, 72 insertions(+), 35 deletions(-) create mode 100644 src/node/services/sessionLocks/buildSessionLocks.testHarness.ts diff --git a/src/node/services/sessionLocks/buildSessionLocks.testHarness.ts b/src/node/services/sessionLocks/buildSessionLocks.testHarness.ts new file mode 100644 index 00000000000..0ab57a34bbf --- /dev/null +++ b/src/node/services/sessionLocks/buildSessionLocks.testHarness.ts @@ -0,0 +1,19 @@ +/** + * Builds one SessionLocks from a layer. Every test builds its layer here. + * + * Kept apart from the contract harness, which imports bun:test: testHistoryService.ts calls this, + * and the Jest integration tests (tests/ipc) load testHistoryService.ts, where bun:test does not + * resolve. + */ +import { Effect, type Layer } from "effect"; +import { SessionLocksTag } from "@/node/services/di/tags"; +import type { SessionLocks } from "./sessionLocks"; + +/** The layers keep no scope, so a synchronous build is enough. */ +export function buildSessionLocks(layer: Layer.Layer): SessionLocks { + return Effect.runSync( + Effect.gen(function* () { + return yield* SessionLocksTag; + }).pipe(Effect.provide(layer)) + ); +} diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index 10db7f2c606..4ba8760ec9f 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -194,6 +194,9 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se // The same argument checks as fileLock.ts acquireProcessFileLock. assert(lockPath.length > 0, "acquireProcessFileLock requires a lock path"); assert(timeoutMs > 0, "acquireProcessFileLock timeoutMs must be positive"); + // As fileLock.ts does first: a caller may write into the folder once it holds the lock + // (branchSummary and refinementRollback lock inside the session folder). No lock file. + await fs.mkdir(path.dirname(lockPath), { recursive: true }); const entry = await acquireInMemory(table, lockPath, timeoutMs, clock, (blocker) => { const { why, ...holder } = blocker; return new Error(fileLockTimeoutMessage(label, lockPath, timeoutMs, { holder, why })); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.test.ts b/src/node/services/sessionLocks/sessionLocks.contract.test.ts index 47f71203a46..5be0624dac2 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.test.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.test.ts @@ -30,8 +30,8 @@ import { SessionLocksMemory, } from "./memorySessionLocks.testHarness"; import { nodeSessionLocks, SessionLocksNode } from "./nodeSessionLocks"; +import { buildSessionLocks } from "./buildSessionLocks.testHarness"; import { - buildSessionLocks, describeSessionLocksContract, type ForeignHolder, type LocksUnderTest, @@ -153,7 +153,7 @@ describeSessionLocksContract("memory + foreign", withForeign(memoryLocks)); test("node: the timeout text for a live child holder is fileLockTimeoutMessage", async () => { const t = await nodeLocks(); try { - const locks = await buildSessionLocks(t.layer); + const locks = buildSessionLocks(t.layer); const lockPath = path.join(t.dir, "text.lock"); const holder = await t.foreignHolder(lockPath, "live", "file"); const error = await locks @@ -183,7 +183,7 @@ async function onMemory( ): Promise { const t = await memoryLocks(); try { - await fn(await buildSessionLocks(t.layer), t); + await fn(buildSessionLocks(t.layer), t); } finally { await t.cleanup(); } @@ -231,6 +231,18 @@ test("memory: a release hands the file lock to the waiters in FIFO order", () => expect((stale as Error).message).toContain("is no longer owned by this holder"); })); +test("memory: a held file lock leaves no file at its path", () => + onMemory(async (locks, t) => { + const lockPath = path.join(t.dir, "session", "held.lock"); + const lock = await locks.acquireProcessFileLock({ lockPath, timeoutMs: 200, label: "l" }); + const found = await fs.stat(lockPath).then( + () => true, + () => false + ); + expect(found).toBe(false); + await lock[Symbol.asyncDispose](); + })); + test("memory: an init-record lock makes a real file, and inspect sees it", () => onMemory(async (locks, t) => { const lockPath = path.join(t.dir, "locks", "init-record", "ws", "backend.lock"); @@ -246,27 +258,28 @@ test("memory: an init-record lock makes a real file, and inspect sees it", () => expect(await locks.inspectCrossProcessLock(lockPath)).toEqual({ state: "absent" }); })); -test("memory: each build of SessionLocksMemory has its own lock table", async () => { - const clock = makeTestEffectRunner(); - try { - const layer = SessionLocksMemory.pipe( - Layer.provide(Layer.succeed(HostClockTag)(clock.hostClock)) - ); - const options = { lockPath: "/memory/one.lock", timeoutMs: 200, label: "l" }; - const a = await (await buildSessionLocks(layer)).acquireProcessFileLock(options); - const b = await (await buildSessionLocks(layer)).acquireProcessFileLock(options); - await a[Symbol.asyncDispose](); - await b[Symbol.asyncDispose](); - } finally { - await clock.dispose(); - } -}); +test("memory: each build of SessionLocksMemory has its own lock table", () => + onMemory(async (_locks, t) => { + const clock = makeTestEffectRunner(); + try { + const layer = SessionLocksMemory.pipe( + Layer.provide(Layer.succeed(HostClockTag)(clock.hostClock)) + ); + const options = { lockPath: path.join(t.dir, "one.lock"), timeoutMs: 200, label: "l" }; + const a = await buildSessionLocks(layer).acquireProcessFileLock(options); + const b = await buildSessionLocks(layer).acquireProcessFileLock(options); + await a[Symbol.asyncDispose](); + await b[Symbol.asyncDispose](); + } finally { + await clock.dispose(); + } + })); test("foreign: dead holders are reclaimed in order, and a probe reclaims none", async () => { const t = await nodeLocks(); try { const holders = makeForeignHolders(); - const locks = await buildSessionLocks(withForeignHolders(t.layer, holders)); + const locks = buildSessionLocks(withForeignHolders(t.layer, holders)); const file = path.join(t.dir, "a.lock"); const record = path.join(t.dir, "b.lock"); holders.hold(record, "dead"); @@ -291,7 +304,7 @@ test("foreign: a live foreign mutator makes hold() refuse with today's text", as const t = await nodeLocks(); try { const holders = makeForeignHolders(); - const locks = await buildSessionLocks(withForeignHolders(t.layer, holders)); + const locks = buildSessionLocks(withForeignHolders(t.layer, holders)); const config = { rootDir: t.dir }; const leases = locks.useLeasesFor(config); expect(locks.useLeasesFor(config)).toBe(leases); @@ -352,7 +365,7 @@ test("foreign: a live fake lease of another backend makes isHeld true", async () const t = await nodeLocks(); try { const holders = makeForeignHolders(); - const leases = (await buildSessionLocks(withForeignHolders(t.layer, holders))).useLeasesFor({ + const leases = buildSessionLocks(withForeignHolders(t.layer, holders)).useLeasesFor({ rootDir: t.dir, }); const dir = workspaceUseLockDir(t.dir, "ws-leases"); diff --git a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts index 63778070a99..33cd22740d5 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts @@ -6,23 +6,13 @@ import { describe, expect, test } from "bun:test"; import * as fs from "node:fs/promises"; import * as path from "node:path"; -import { Effect, type Layer } from "effect"; -import { SessionLocksTag } from "@/node/services/di/tags"; +import type { Layer } from "effect"; +import type { SessionLocksTag } from "@/node/services/di/tags"; import type { ProcessIdentity } from "@/node/utils/concurrency/processLiveness"; import { CrossProcessLockTimeoutError } from "@/node/utils/concurrency/lockErrors"; +import { buildSessionLocks } from "./buildSessionLocks.testHarness"; import type { SessionLocks } from "./sessionLocks"; -/** Builds a SessionLocks layer into its service. The layers keep no scope. */ -export async function buildSessionLocks( - layer: Layer.Layer -): Promise { - return await Effect.runPromise( - Effect.gen(function* () { - return yield* SessionLocksTag; - }).pipe(Effect.provide(layer)) - ); -} - /** Another backend that holds a lock, in the medium of the layer under test. */ export interface ForeignHolder { pid: number; @@ -57,7 +47,7 @@ export function describeSessionLocksContract( const withLayer = async (fn: (locks: SessionLocks, t: LocksUnderTest) => Promise) => { const t = await make(); try { - await fn(await buildSessionLocks(t.layer), t); + await fn(buildSessionLocks(t.layer), t); } finally { await t.cleanup(); } @@ -191,6 +181,18 @@ export function describeSessionLocksContract( expect((error as NodeJS.ErrnoException).code).toBe("ENOENT"); })); + test("a file lock creates its missing parent", () => + withLayer(async (locks, t) => { + const lockPath = path.join(t.dir, "new-file-dir", "file.lock"); + const lock = await locks.acquireProcessFileLock({ + lockPath, + timeoutMs: SHORT_MS, + label: "contract lock", + }); + expect((await fs.stat(path.dirname(lockPath))).isDirectory()).toBe(true); + await lock[Symbol.asyncDispose](); + })); + test("a record lock creates its missing parent by default", () => withLayer(async (locks, t) => { const lockPath = path.join(t.dir, "new-dir", "record.lock"); From 23b694c3d420a7c9f2b0019ad152e5881ed66e6f Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 19:09:21 +0000 Subject: [PATCH 12/19] refactor: HistoryService takes its locks from SessionLocks HistoryService gets a third constructor argument, locks (default nodeSessionLocks), and exposes it as readonly locks. The workspace FIFO and the history write lock go through it, so historyService.ts imports no lock module. HistoryLive yields SessionLocksTag, and createTestHistoryService takes a locks layer. historyService.foreignHolder.test.ts plays a live and a dead foreign writer, and the golden leak test also runs on SessionLocksMemory. --- src/node/services/di/layers/core.ts | 7 ++- .../historyService.foreignHolder.test.ts | 60 +++++++++++++++++++ src/node/services/historyService.ts | 21 +++++-- .../sessionFiles/historyFiles.golden.test.ts | 26 +++++++- src/node/services/testHistoryService.ts | 12 +++- 5 files changed, 115 insertions(+), 11 deletions(-) create mode 100644 src/node/services/historyService.foreignHolder.test.ts diff --git a/src/node/services/di/layers/core.ts b/src/node/services/di/layers/core.ts index d5c18e3da14..26882ae7246 100644 --- a/src/node/services/di/layers/core.ts +++ b/src/node/services/di/layers/core.ts @@ -50,6 +50,7 @@ import { SecretsStoreTag, SessionFilesTag, SessionLocatorTag, + SessionLocksTag, SessionUsage, StreamManagerTag, Task, @@ -176,7 +177,11 @@ export const WorkspaceMcpOverridesDefaultLive: Layer.Layer< export const HistoryLive = Layer.effect( History, Effect.gen(function* () { - return new HistoryService(yield* SessionLocatorTag, yield* SessionFilesTag); + return new HistoryService( + yield* SessionLocatorTag, + yield* SessionFilesTag, + yield* SessionLocksTag + ); }) ); diff --git a/src/node/services/historyService.foreignHolder.test.ts b/src/node/services/historyService.foreignHolder.test.ts new file mode 100644 index 00000000000..45c22210dcc --- /dev/null +++ b/src/node/services/historyService.foreignHolder.test.ts @@ -0,0 +1,60 @@ +/** + * Another backend holds this workspace's history write lock (phase 4). With the + * foreign-holder layer the refusal comes at once, with the text that the Node + * lock gives after HISTORY_WRITE_LOCK_TIMEOUT_MS today. + */ +import { expect, test } from "bun:test"; +import { createMuxMessage } from "@/common/types/message"; +import { fileLockTimeoutMessage } from "@/node/utils/concurrency/fileLock"; +import { + foreignPid, + liveBlocker, + makeForeignHolders, + withForeignHolders, +} from "./sessionLocks/foreignHolders.testHarness"; +import { SessionLocksNode } from "./sessionLocks/nodeSessionLocks"; +import { createTestHistoryService } from "./testHistoryService"; +import { historyWriteLockPath } from "./workspaceRemoval"; + +const WORKSPACE_ID = "ws-foreign-writer"; +/** HISTORY_WRITE_LOCK_TIMEOUT_MS in historyService.ts: the timeout the refusal names. */ +const HISTORY_WRITE_LOCK_TIMEOUT_MS = 10_000; + +test("a live foreign writer refuses at once with today's text, a dead one is reclaimed", async () => { + const holders = makeForeignHolders(); + const { historyService, config, cleanup } = await createTestHistoryService({ + locks: withForeignHolders(SessionLocksNode, holders), + }); + try { + const lockPath = historyWriteLockPath(config.rootDir, WORKSPACE_ID); + holders.hold(lockPath, "live", { pid: foreignPid() }); + const refused = await historyService.appendToHistory( + WORKSPACE_ID, + createMuxMessage("u1", "user", "hello") + ); + expect(refused.success ? "" : refused.error).toContain( + fileLockTimeoutMessage( + "history write lock", + lockPath, + HISTORY_WRITE_LOCK_TIMEOUT_MS, + liveBlocker(foreignPid()) + ) + ); + expect(await historyService.getLastMessages(WORKSPACE_ID, 10)).toEqual({ + success: true, + data: [], + }); + + // The holder's process died: the next write reclaims the lock and goes on. + holders.release(lockPath); + holders.hold(lockPath, "dead"); + const appended = await historyService.appendToHistory( + WORKSPACE_ID, + createMuxMessage("u2", "user", "again") + ); + expect(appended.success).toBe(true); + expect(holders.reclaimed).toEqual([lockPath]); + } finally { + await cleanup(); + } +}); diff --git a/src/node/services/historyService.ts b/src/node/services/historyService.ts index 9cc5745a20b..e6cfd1a44bc 100644 --- a/src/node/services/historyService.ts +++ b/src/node/services/historyService.ts @@ -85,9 +85,10 @@ import type { AIService } from "@/node/services/aiService"; import type { TaskService } from "@/node/services/taskService"; import { nodeSessionFiles } from "@/node/services/sessionFiles/nodeSessionFiles"; import type { SessionFiles } from "@/node/services/sessionFiles/sessionFiles"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { isErrnoWithCode } from "@/node/utils/fs"; import { isPathInsideDir } from "@/node/utils/pathUtils"; -import { workspaceFileLocks } from "@/node/utils/concurrency/workspaceFileLocks"; import { unlockedHistoryScans } from "./unlockedHistoryScans"; import { decideSendIdsFromHistory, @@ -119,7 +120,6 @@ import { import { isRefusalFinishReason } from "@/common/utils/messages/refusalFinishReason"; import { getErrorMessage } from "@/common/utils/errors"; import { isNonNegativeInteger, isPositiveInteger } from "@/common/utils/numbers"; -import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock"; import { historyWriteLockPath, isWorkspaceRemovalTombstoned, @@ -864,14 +864,25 @@ export class HistoryService { private sealedRotationChecked = new Set(); // Shared file operation lock across all workspace file services // This prevents deadlocks when operations compose while touching the same workspace files. - private readonly fileLocks = workspaceFileLocks; + private readonly fileLocks: SessionLocks["workspaceFiles"]; private readonly config: HistorySessionLocation; /** Every host file call of this class goes through here (node:fs in the app). */ readonly files: SessionFiles; + /** + * Every session lock of this class goes through here (today's lock functions in the app). + * Phase 4: the other owners of session locks take them from here. + */ + readonly locks: SessionLocks; - constructor(config: HistorySessionLocation, files: SessionFiles = nodeSessionFiles) { + constructor( + config: HistorySessionLocation, + files: SessionFiles = nodeSessionFiles, + locks: SessionLocks = nodeSessionLocks + ) { this.config = config; this.files = files; + this.locks = locks; + this.fileLocks = locks.workspaceFiles; } private readonly emptyErroredReplyDeletedListeners = new Set< @@ -4115,7 +4126,7 @@ export class HistoryService { workspaceId: string, operation: (assertStillOwned: () => Promise) => Promise ): Promise { - await using _lock = await acquireProcessFileLock({ + await using _lock = await this.locks.acquireProcessFileLock({ lockPath: historyWriteLockPath(this.config.rootDir, workspaceId), timeoutMs: HISTORY_WRITE_LOCK_TIMEOUT_MS, label: "history write lock", diff --git a/src/node/services/sessionFiles/historyFiles.golden.test.ts b/src/node/services/sessionFiles/historyFiles.golden.test.ts index d59ec2adefb..1cc6b19d69d 100644 --- a/src/node/services/sessionFiles/historyFiles.golden.test.ts +++ b/src/node/services/sessionFiles/historyFiles.golden.test.ts @@ -8,7 +8,11 @@ import { expect, test } from "bun:test"; import * as fs from "node:fs/promises"; import * as path from "node:path"; import * as prettier from "prettier"; +import { Layer } from "effect"; +import { HostClockTag } from "@/node/services/di/hostClockLive"; +import { SessionLocksMemory } from "@/node/services/sessionLocks/memorySessionLocks.testHarness"; import { createTestHistoryService } from "@/node/services/testHistoryService"; +import { nodeClock } from "@/node/utils/hostClock"; import { SessionFilesMemory } from "./memorySessionFiles.testHarness"; import { runGoldenScript, STEPS, type GoldenSnapshot } from "./goldenScript.testHarness"; @@ -50,16 +54,34 @@ async function listHostEntries(root: string): Promise { .sort(); } -// A history write that skips SessionFiles lands in the real temp folder instead of the volume. -test("the memory layer writes the golden bytes and nothing on disk", async () => { +/** Every file (not folder) under root on the host disk, relative to root. */ +async function listHostFiles(root: string): Promise { + const entries = await fs + .readdir(root, { recursive: true, withFileTypes: true }) + .catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return []; + throw error; + }); + return entries + .filter((entry) => entry.isFile()) + .map((entry) => path.relative(root, path.join(entry.parentPath, entry.name))) + .sort(); +} + +// A history write that skips SessionFiles lands in the real temp folder instead of the volume, +// and a lock taken around SessionLocks leaves a lock file under /locks. +test("the memory layers write the golden bytes and nothing on disk", async () => { const { historyService, config, files, tempDir, cleanup } = await createTestHistoryService({ files: SessionFilesMemory, + locks: SessionLocksMemory.pipe(Layer.provide(Layer.succeed(HostClockTag)(nodeClock))), }); try { const snapshots = await runGoldenScript(historyService, config, STEPS, files); const fixture = JSON.parse(await fs.readFile(FIXTURE, "utf8")) as GoldenSnapshot[]; expect(snapshots).toEqual(fixture); expect(await listHostEntries(path.join(tempDir, "sessions"))).toEqual([]); + // The memory locks create the lock folder as the Node locks do, but no lock file. + expect(await listHostFiles(path.join(tempDir, "locks"))).toEqual([]); } finally { await cleanup(); } diff --git a/src/node/services/testHistoryService.ts b/src/node/services/testHistoryService.ts index 30bf1f0524a..88e6315da3d 100644 --- a/src/node/services/testHistoryService.ts +++ b/src/node/services/testHistoryService.ts @@ -9,17 +9,21 @@ * spyOn(historyService, "appendToHistory").mockRejectedValueOnce(...) * * `files` picks the SessionFiles layer of the history files (default: the Node layer). With - * SessionFilesMemory the session files stay in memory; config, tempDir and the locks stay on disk. + * SessionFilesMemory the session files stay in memory; config and tempDir stay on disk. + * `locks` picks the SessionLocks layer (default: the Node layer). With SessionLocksMemory the + * history locks stay in memory too; withForeignHolders plays other backends' lock holders. * * Test-only: tsconfig.main.json excludes this file by path so it stays out of the * app build (AGENTS.md documents this path, so it keeps its name). */ import { HistoryService } from "@/node/services/historyService"; import { Config } from "@/node/config"; -import type { SessionFilesTag } from "@/node/services/di/tags"; +import type { SessionFilesTag, SessionLocksTag } from "@/node/services/di/tags"; import { SessionFilesNode } from "@/node/services/sessionFiles/nodeSessionFiles"; import { buildSessionFiles } from "@/node/services/sessionFiles/buildSessionFiles.testHarness"; import type { SessionFiles } from "@/node/services/sessionFiles/sessionFiles"; +import { buildSessionLocks } from "@/node/services/sessionLocks/buildSessionLocks.testHarness"; +import { SessionLocksNode } from "@/node/services/sessionLocks/nodeSessionLocks"; import type { Layer } from "effect"; import * as fs from "fs/promises"; import * as os from "os"; @@ -27,6 +31,7 @@ import * as path from "path"; export async function createTestHistoryService(options?: { files?: Layer.Layer; + locks?: Layer.Layer; }): Promise<{ historyService: HistoryService; /** The SessionFiles that historyService uses, for reads in assertions. */ @@ -42,7 +47,8 @@ export async function createTestHistoryService(options?: { await fs.mkdir(tempDir, { recursive: true }); const config = new Config(tempDir); const files = buildSessionFiles(options?.files ?? SessionFilesNode); - const historyService = new HistoryService(config, files); + const locks = buildSessionLocks(options?.locks ?? SessionLocksNode); + const historyService = new HistoryService(config, files, locks); return { historyService, files, From 1c8ad130aefc6f5d29a6c0cff2a9b74b5f5287cf Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 20:08:45 +0000 Subject: [PATCH 13/19] refactor: usage, goal and timing services take the workspace FIFO from SessionLocks SessionUsageService and WorkspaceGoalService read historyService.locks.workspaceFiles at each use. SessionTimingService takes an optional locks argument (default nodeSessionLocks), and desktop.ts keeps the default. The three files import no lock module. --- src/node/services/sessionTimingService.ts | 13 ++++++++++--- src/node/services/sessionUsageService.ts | 11 +++++++++-- src/node/services/workspaceGoalService.ts | 11 +++++++++-- 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/src/node/services/sessionTimingService.ts b/src/node/services/sessionTimingService.ts index c22fb2ba35e..4d16397bac5 100644 --- a/src/node/services/sessionTimingService.ts +++ b/src/node/services/sessionTimingService.ts @@ -4,7 +4,8 @@ import * as fs from "fs/promises"; import { EventEmitter } from "events"; import writeFileAtomic from "@/node/utils/writeFileAtomic"; import type { Config } from "@/node/config"; -import { workspaceFileLocks } from "@/node/utils/concurrency/workspaceFileLocks"; +import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { normalizeToCanonical } from "@/common/utils/ai/models"; import type { AgentMode } from "@/common/types/mode"; import { @@ -163,7 +164,8 @@ function validateTiming(params: { export class SessionTimingService { private readonly config: Config; private readonly telemetryService: TelemetryService; - private readonly fileLocks = workspaceFileLocks; + /** The one workspace FIFO (phase 4: from SessionLocks; desktop.ts keeps the Node default). */ + private readonly fileLocks: SessionLocks["workspaceFiles"]; private readonly activeStreams = new Map(); private readonly timingFileCache = new Map(); @@ -181,9 +183,14 @@ export class SessionTimingService { { lastEmitTimeMs: number; timer?: ReturnType } >(); - constructor(config: Config, telemetryService: TelemetryService) { + constructor( + config: Config, + telemetryService: TelemetryService, + locks: SessionLocks = nodeSessionLocks + ) { this.config = config; this.telemetryService = telemetryService; + this.fileLocks = locks.workspaceFiles; } addSubscriber(workspaceId: string): void { diff --git a/src/node/services/sessionUsageService.ts b/src/node/services/sessionUsageService.ts index 19439d12f47..a417cf76520 100644 --- a/src/node/services/sessionUsageService.ts +++ b/src/node/services/sessionUsageService.ts @@ -8,7 +8,7 @@ import type { Config } from "@/node/config"; import type { HistoryService } from "./historyService"; import { isWorkspaceRemovalTombstoned } from "@/node/services/workspaceRemoval"; import { withTargetMutationLock } from "@/node/services/refinement/targetMutationLocks"; -import { workspaceFileLocks } from "@/node/utils/concurrency/workspaceFileLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import type { ChatUsageDisplay } from "@/common/utils/tokens/usageAggregator"; import { sumUsageHistory } from "@/common/utils/tokens/usageAggregator"; import { createDisplayUsage } from "@/common/utils/tokens/displayUsage"; @@ -227,7 +227,14 @@ function pruneSettledInFlightKeys( */ export class SessionUsageService { private readonly SESSION_USAGE_FILE = "session-usage.json"; - private readonly fileLocks = workspaceFileLocks; + /** + * The one workspace FIFO, from the HistoryService's SessionLocks (phase 4), so a + * memory or foreign-holder layer reaches these writes too. Read at each use, not + * at construction: tests that pass a partial HistoryService keep working. + */ + private get fileLocks(): SessionLocks["workspaceFiles"] { + return this.historyService.locks.workspaceFiles; + } private readonly config: Config; private readonly historyService: HistoryService; private readonly getProvidersConfig: () => ProvidersConfigMap | null; diff --git a/src/node/services/workspaceGoalService.ts b/src/node/services/workspaceGoalService.ts index 7bee933813c..d29bf3d5e53 100644 --- a/src/node/services/workspaceGoalService.ts +++ b/src/node/services/workspaceGoalService.ts @@ -44,7 +44,7 @@ import type { SendMessageError } from "@/common/types/errors"; import { ProvidersConfigStore, type Config } from "@/node/config"; import type { HistoryService } from "@/node/services/historyService"; import type { ExtensionMetadataService } from "@/node/services/ExtensionMetadataService"; -import { workspaceFileLocks } from "@/node/utils/concurrency/workspaceFileLocks"; +import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks"; import { DEFAULT_GOAL_CONTINUATION_COOLDOWN_MS, GOAL_STREAM_ERROR_RESUME_MAX_ATTEMPTS, @@ -609,7 +609,14 @@ export interface WorkspaceGoalServiceOptions { } export class WorkspaceGoalService { - private readonly fileLocks = workspaceFileLocks; + /** + * The one workspace FIFO, from the HistoryService's SessionLocks (phase 4), so a + * memory or foreign-holder layer reaches these writes too. Read at each use, not + * at construction: tests that pass a partial HistoryService keep working. + */ + private get fileLocks(): SessionLocks["workspaceFiles"] { + return this.historyService.locks.workspaceFiles; + } private readonly continuationCooldownMs: number; private readonly allowUserOriginBudgetWrapup: boolean; private readonly suppressKickoffContinuation: boolean; From 6353fe43d194dea6e4219b585c9ccd2594ab7331 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 20:50:45 +0000 Subject: [PATCH 14/19] refactor: AgentSession takes its turn use lease from SessionLocks The turn use lease now comes from this.historyService.locks.useLeasesFor(config), so agentSession.ts makes no value import of workspaceUseLeases. The Node layer returns the same per-Config object as workspaceUseLeasesFor, so the lease files and the refusal text do not change. agentSession.useLease.test.ts gains a live foreign gate case: the foreign-holder layer plays another backend's mutation gate with no file on disk, and the turn is refused at once with today's text and never reaches the provider. --- src/node/services/agentSession.ts | 12 ++-- .../services/agentSession.useLease.test.ts | 57 +++++++++++++++++++ 2 files changed, 63 insertions(+), 6 deletions(-) diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index f2a2d270cc4..ad2f7ba701e 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -176,11 +176,8 @@ import { } from "@/node/services/utils/fileChangeTracker"; import type { Result } from "@/common/types/result"; import { Ok, Err } from "@/common/types/result"; -import { - WorkspaceMutationInProgressError, - workspaceUseLeasesFor, - type WorkspaceUseLease, -} from "@/node/services/workspaceUseLeases"; +import { WorkspaceMutationInProgressError } from "@/node/services/sessionLocks/sessionLocks"; +import type { WorkspaceUseLease } from "@/node/services/workspaceUseLeases"; import { coerceOpenAIReasoningMode, coerceThinkingLevel, @@ -4707,7 +4704,10 @@ export class AgentSession { private beginTurnUseLease(): void { // A refusal (or an unexpected lock error) resolves to the error, so a preparation that ends // before streamWithHistory confirms the lease leaves no unhandled rejection. - this.turnUseLease ??= workspaceUseLeasesFor(this.config) + // The lease comes from the history service's SessionLocks (phase 4), so a test layer can + // play another backend's mutation gate. + this.turnUseLease ??= this.historyService.locks + .useLeasesFor(this.config) .hold(this.workspaceId, "turn") .catch((error: unknown) => (error instanceof Error ? error : new Error(String(error)))); } diff --git a/src/node/services/agentSession.useLease.test.ts b/src/node/services/agentSession.useLease.test.ts index 50b56f7b85a..2ff406485d9 100644 --- a/src/node/services/agentSession.useLease.test.ts +++ b/src/node/services/agentSession.useLease.test.ts @@ -2,12 +2,24 @@ import { describe, expect, mock, test } from "bun:test"; import { Err, Ok } from "@/common/types/result"; import type { StreamEndEvent } from "@/common/types/stream"; +import { describeHolder } from "@/node/utils/main/crossProcessLock"; import type { AgentSessionAIService } from "./agentSession"; import { createAgentSessionHarness } from "./agentSession.testHarness"; +import { + foreignPid, + foreignRecord, + liveBlocker, + makeForeignHolders, + withForeignHolders, +} from "./sessionLocks/foreignHolders.testHarness"; +import { SessionLocksNode } from "./sessionLocks/nodeSessionLocks"; import type { TurnCompletion } from "./streamManager"; import { createTestConfig } from "./taskService.testHarness"; +import { createTestHistoryService } from "./testHistoryService"; import { WorkspaceBusyError, + workspaceMutationInProgressMessage, + workspaceMutationLockPath, workspaceUseLeasesFor, type WorkspaceUseLeases, } from "./workspaceUseLeases"; @@ -154,6 +166,51 @@ describe("AgentSession turn use lease across two backends on one root", () => { } }); + test("a live foreign mutation gate refuses B's turn at once with today's text", async () => { + // Phase 4: the lease comes from the history service's SessionLocks, so a fake layer plays + // the other backend's gate. It leaves no gate file on disk, so only that path refuses. + const holders = makeForeignHolders(); + const history = await createTestHistoryService({ + locks: withForeignHolders(SessionLocksNode, holders), + }); + const streamMessage = mock(() => + Promise.resolve(Err({ type: "unknown", raw: "provider down" })) + ); + const b = await createAgentSessionHarness({ + workspaceId, + config: history.config, + historyService: history.historyService, + aiServiceOverrides: { streamMessage }, + }); + try { + const gatePath = workspaceMutationLockPath(history.config.rootDir, workspaceId); + const pid = foreignPid(); + holders.hold(gatePath, "live", { pid }); + + const result = await b.session.sendMessage("blocked", sendOptions); + expect(result.success).toBe(false); + // The refusal is the lease error's own text, not the fail-closed wrapper. + expect(!result.success && result.error.type === "unknown" && result.error.raw).toBe( + workspaceMutationInProgressMessage(workspaceId, { + holder: describeHolder(foreignRecord(pid)), + why: liveBlocker(pid).why, + }) + ); + expect(streamMessage).not.toHaveBeenCalled(); + expect(b.session.isBusy()).toBe(false); + expect(workspaceUseLeasesFor(b.config).heldCount(workspaceId)).toBe(0); + + // The other backend finished: B's next turn is admitted again. + holders.release(gatePath); + await b.session.sendMessage("admitted", sendOptions); + expect(streamMessage).toHaveBeenCalledTimes(1); + } finally { + await b.session.dispose(); + await b.cleanup(); + await history.cleanup(); + } + }); + test("a lifecycle listener that throws at idle does not keep the lease", async () => { const reached = Promise.withResolvers(); const completion = Promise.withResolvers(); From c0e84fa1d0de751a076630db638e308e8ab459f9 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 22:33:02 +0000 Subject: [PATCH 15/19] tests: queue memory lock waiters in call order The memory layer's folder side effect (e9272f9ea0) awaited fs.mkdir before a waiter joined the FIFO, so a later caller could queue first. The contract case "a release hands the file lock to the waiters in FIFO order" failed 14 of 1500 reruns that way, and once in make test. A synchronous mkdir keeps the call order: 0 of 1500 fail. --- .../sessionLocks/memorySessionLocks.testHarness.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index 4ba8760ec9f..a0e04f57b02 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -13,7 +13,7 @@ * backends' lease files. Phase 5 moves them. */ import assert from "node:assert"; -import * as fs from "node:fs/promises"; +import * as fs from "node:fs"; import * as path from "node:path"; import { Effect, Layer } from "effect"; import { SessionLocksTag } from "@/node/services/di/tags"; @@ -196,7 +196,9 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se assert(timeoutMs > 0, "acquireProcessFileLock timeoutMs must be positive"); // As fileLock.ts does first: a caller may write into the folder once it holds the lock // (branchSummary and refinementRollback lock inside the session folder). No lock file. - await fs.mkdir(path.dirname(lockPath), { recursive: true }); + // Synchronous, so waiters queue in call order: with an awaited mkdir, a later caller's + // mkdir could finish first and take its place in the FIFO. + fs.mkdirSync(path.dirname(lockPath), { recursive: true }); const entry = await acquireInMemory(table, lockPath, timeoutMs, clock, (blocker) => { const { why, ...holder } = blocker; return new Error(fileLockTimeoutMessage(label, lockPath, timeoutMs, { holder, why })); @@ -224,7 +226,8 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se } const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; // As crossProcessLock.ts does first: a caller may write into the folder once it holds the lock. - await fs.mkdir(path.dirname(lockPath), { recursive: true }); + // Synchronous for the same FIFO reason as the file lock above. + fs.mkdirSync(path.dirname(lockPath), { recursive: true }); const refusal = (blocker: Blocker) => new CrossProcessLockTimeoutError( crossProcessLockTimeoutMessage(timeoutMessage, lockPath, { From 6c5629c648739ec39f1f964a5ef610fb0a7169b3 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 23:55:15 +0000 Subject: [PATCH 16/19] tests: allow the two FIFO-keeping mkdirSync calls past no-sync-fs-methods c0e84fa1d0 made both memory-layer folder writes synchronous so waiters queue in call order. The local no-sync-fs-methods rule refuses mkdirSync, so each call gets a disable comment with that reason, as serverCrashLogging.ts does. --- .../services/sessionLocks/memorySessionLocks.testHarness.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts index a0e04f57b02..52357fb8e71 100644 --- a/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts +++ b/src/node/services/sessionLocks/memorySessionLocks.testHarness.ts @@ -198,6 +198,7 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se // (branchSummary and refinementRollback lock inside the session folder). No lock file. // Synchronous, so waiters queue in call order: with an awaited mkdir, a later caller's // mkdir could finish first and take its place in the FIFO. + // eslint-disable-next-line local/no-sync-fs-methods -- keeps the waiters' FIFO order (above). fs.mkdirSync(path.dirname(lockPath), { recursive: true }); const entry = await acquireInMemory(table, lockPath, timeoutMs, clock, (blocker) => { const { why, ...holder } = blocker; @@ -227,6 +228,7 @@ export function memorySessionLocks(clock: HostClock, table: MemoryLockTable): Se const { lockPath, acquireTimeoutMs, timeoutMessage, signal } = options; // As crossProcessLock.ts does first: a caller may write into the folder once it holds the lock. // Synchronous for the same FIFO reason as the file lock above. + // eslint-disable-next-line local/no-sync-fs-methods -- keeps the waiters' FIFO order (above). fs.mkdirSync(path.dirname(lockPath), { recursive: true }); const refusal = (blocker: Blocker) => new CrossProcessLockTimeoutError( From b3905b112fb7f5d76a0f397b7a918f4e2c2ba4a9 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 05:59:56 +0000 Subject: [PATCH 17/19] tests: drop agentSession.ts's workspaceUseLeases pair, which AgentSession no longer imports --- .oxlintrc.json | 5 +---- eslint.config.mjs | 5 +---- scripts/noHostImports.test.ts | 1 - 3 files changed, 2 insertions(+), 9 deletions(-) diff --git a/.oxlintrc.json b/.oxlintrc.json index 8a9a82a13c6..077f405ae60 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1282,10 +1282,7 @@ "@/node/services/workspaceUseLeases" ], "allow": { - "src/node/services/agentSession.ts": [ - "@/node/runtime/runtimeHelpers", - "@/node/services/workspaceUseLeases" - ], + "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers"], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", diff --git a/eslint.config.mjs b/eslint.config.mjs index ac8303118e3..8794929ecca 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -183,10 +183,7 @@ const HOST_MODULES = [ ]; // File -> the host modules it still imports. Phase plans list the pairs they add or remove. const HOST_IMPORT_ALLOW = { - "src/node/services/agentSession.ts": [ - "@/node/runtime/runtimeHelpers", - "@/node/services/workspaceUseLeases", - ], + "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers"], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", diff --git a/scripts/noHostImports.test.ts b/scripts/noHostImports.test.ts index d86908ecf6c..4d90f846c52 100644 --- a/scripts/noHostImports.test.ts +++ b/scripts/noHostImports.test.ts @@ -121,7 +121,6 @@ test("allows listed modules and fails stale entries", () => { // Phases 2 to 5 remove pairs until the map is empty. const FROZEN_ALLOW_PAIRS = [ "src/node/services/agentSession.ts @/node/runtime/runtimeHelpers", - "src/node/services/agentSession.ts @/node/services/workspaceUseLeases", "src/node/services/aiService.ts @/node/runtime/multiProjectRuntime", "src/node/services/aiService.ts @/node/runtime/runtimeFactory", "src/node/services/aiService.ts @/node/runtime/runtimeHelpers", From 3c8d5b7ade6e48fce9eecafd88af9f22b6518af3 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 23 Oct 2026 21:46:45 +0000 Subject: [PATCH 18/19] refactor: the workspace-turn manager and its handle store take their locks from SessionLocks WorkspaceTurnManager takes the live-owner lock and its read-only probe from this.historyService.locks, and passes the same locks to its handle store. TaskHandleStore takes the publication lock from an optional locks argument (Node default), so its other callers do not change. Paths, timeouts and messages stay the same. Tests: - workspaceTurnManager.foreignOwner.test.ts writes no lock file. A dead owner is a foreign-holder fake that backend B reclaims, and a new case shows a fake live owner keeps the handle through keepIfOwnedElsewhere. - The harness gains a locks option, through historyServiceWithLocks in testHistoryService.ts. - The SessionLocks contract gains an inspect case on all four layers: an own record reads held, and a probe never reclaims a dead one. --- .../sessionLocks.contract.testHarness.ts | 20 +++++ src/node/services/taskHandleStore.ts | 15 +++- src/node/services/testHistoryService.ts | 11 +++ .../workspaceTurnManager.foreignOwner.test.ts | 84 ++++++++++++++++--- .../workspaceTurnManager.testHarness.ts | 16 +++- src/node/services/workspaceTurnManager.ts | 22 +++-- 6 files changed, 145 insertions(+), 23 deletions(-) diff --git a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts index c9c43cf2c08..6d9ced564e5 100644 --- a/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts +++ b/src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts @@ -130,6 +130,26 @@ export function describeSessionLocksContract( expect(await locks.inspectCrossProcessLock(dead)).toEqual({ state: "dead" }); })); + // The turn owner's probe (workspaceTurnManager keepIfOwnedElsewhere) relies on both halves: + // a record this backend holds reads as held, and a probe never takes or reclaims a lock. + test("inspect never takes a lock: an own record reads held, a dead one stays until an acquire", () => + withLayer(async (locks, t) => { + const own = path.join(t.dir, "own.lock"); + const options = { acquireTimeoutMs: 0, staleMs: 60_000, timeoutMessage: "contract." }; + const release = await locks.acquireCrossProcessLock({ lockPath: own, ...options }); + expect(await locks.inspectCrossProcessLock(own)).toMatchObject({ state: "held" }); + await release(); + expect(await locks.inspectCrossProcessLock(own)).toEqual({ state: "absent" }); + + const dead = path.join(t.dir, "dead-record.lock"); + await t.foreignHolder(dead, "dead", "record"); + expect(await locks.inspectCrossProcessLock(dead)).toEqual({ state: "dead" }); + expect(await locks.inspectCrossProcessLock(dead)).toEqual({ state: "dead" }); + const reclaimed = await locks.acquireCrossProcessLock({ lockPath: dead, ...options }); + expect(await locks.inspectCrossProcessLock(dead)).toMatchObject({ state: "held" }); + await reclaimed(); + })); + test("a dead foreign holder is reclaimed", () => withLayer(async (locks, t) => { const file = path.join(t.dir, "dead-file.lock"); diff --git a/src/node/services/taskHandleStore.ts b/src/node/services/taskHandleStore.ts index 0fcd2c7304e..5b5be887688 100644 --- a/src/node/services/taskHandleStore.ts +++ b/src/node/services/taskHandleStore.ts @@ -19,7 +19,9 @@ import { import { log } from "@/node/services/log"; import { isErrnoWithCode } from "@/node/utils/fs"; import writeFileAtomic from "@/node/utils/writeFileAtomic"; -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"; export type { WorkspaceTurnFinalMessageRef }; @@ -150,7 +152,14 @@ function assertValidWorkspaceTurnTaskId(handleId: string): void { } export class TaskHandleStore { - constructor(private readonly config: Config) {} + /** + * `locks` takes the publication lock (phase 4: SessionLocks). The workspace-turn manager passes + * its history service's locks; other callers keep the Node default. + */ + constructor( + private readonly config: Config, + private readonly locks: SessionLocks = nodeSessionLocks + ) {} async upsertWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { this.assertValidRecord(record); @@ -203,7 +212,7 @@ export class TaskHandleStore { fn: (lock: ProcessFileLock) => Promise ): Promise { assertValidWorkspaceTurnTaskId(handleId); - await using lock = await acquireProcessFileLock({ + await using lock = await this.locks.acquireProcessFileLock({ lockPath: path.join( this.config.rootDir, "locks", diff --git a/src/node/services/testHistoryService.ts b/src/node/services/testHistoryService.ts index 88e6315da3d..e50533f46de 100644 --- a/src/node/services/testHistoryService.ts +++ b/src/node/services/testHistoryService.ts @@ -57,3 +57,14 @@ export async function createTestHistoryService(options?: { cleanup: () => fs.rm(tempDir, { recursive: true, force: true }), }; } + +/** + * A HistoryService on an existing Config's root (no temp dir) whose locks come from `locks`: a + * test that runs several backends on one root gives one of them withForeignHolders. + */ +export function historyServiceWithLocks( + config: Config, + locks: Layer.Layer +): HistoryService { + return new HistoryService(config, buildSessionFiles(SessionFilesNode), buildSessionLocks(locks)); +} diff --git a/src/node/services/workspaceTurnManager.foreignOwner.test.ts b/src/node/services/workspaceTurnManager.foreignOwner.test.ts index cd764bc3c3a..d6238a7b400 100644 --- a/src/node/services/workspaceTurnManager.foreignOwner.test.ts +++ b/src/node/services/workspaceTurnManager.foreignOwner.test.ts @@ -16,6 +16,12 @@ import { workspaceTurnOwnerLockPath, type WorkspaceTurnManager, } from "@/node/services/workspaceTurnManager"; +import { + makeForeignHolders, + withForeignHolders, + type ForeignHolders, +} from "@/node/services/sessionLocks/foreignHolders.testHarness"; +import { SessionLocksNode } from "@/node/services/sessionLocks/nodeSessionLocks"; import { registerLiveWorkspaceTurnHandle } from "@/node/services/taskService.shared.testHarness"; import { createAIServiceMocks, @@ -61,6 +67,7 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { manager as unknown as { taskHandleStore: TaskHandleStore; countActiveWorkspaceTurns: () => Promise; + turnOwnerLocks: Map Promise>; }; const exists = (file: string) => @@ -69,11 +76,33 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { () => false ); - /** The owner process died: its lock record names a token no live process holds. */ - async function markTurnOwnerDead(handleId: string): Promise { + /** + * Backend A stops holding the handle's live-owner lock without settling it. Its handle stays + * "running" on disk, and no lock file remains: the test puts a fake owner on the path instead. + */ + async function dropOwnerLock(backendA: WorkspaceTurnManager, handleId: string): Promise { + const owned = internals(backendA).turnOwnerLocks; + const release = owned.get(handleId); + expect(release).toBeDefined(); + owned.delete(handleId); + await release!(); const lockPath = workspaceTurnOwnerLockPath(rootDir, handleId); - const record = JSON.parse(await fsPromises.readFile(lockPath, "utf-8")) as object; - await fsPromises.writeFile(lockPath, JSON.stringify({ ...record, token: "dead-owner" })); + expect(await exists(lockPath)).toBe(false); + return lockPath; + } + + /** + * The owner process died: backend B's locks (phase 4: SessionLocks with foreign holders) see a + * dead holder on the owner lock, which B's next acquire reclaims. Writes no lock file. + */ + async function markTurnOwnerDead( + backendA: WorkspaceTurnManager, + handleId: string + ): Promise<{ holders: ForeignHolders; lockPath: string }> { + const lockPath = await dropOwnerLock(backendA, handleId); + const holders = makeForeignHolders(); + holders.hold(lockPath, "dead"); + return { holders, lockPath }; } test("another backend's count never settles a handle its live owner runs", async () => { @@ -96,10 +125,11 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { }); test("another backend settles a handle whose owner died, as after a restart", async () => { - const { parentId } = await startWorkspaceTurnForTest(rootDir); - await markTurnOwnerDead("wst_handle"); + const { parentId, taskService: backendA } = await startWorkspaceTurnForTest(rootDir); + const { holders, lockPath } = await markTurnOwnerDead(backendA, "wst_handle"); const { taskService: backendB } = createWorkspaceTurnManagerHarness( - await createTestConfig(rootDir) + await createTestConfig(rootDir), + { locks: withForeignHolders(SessionLocksNode, holders) } ); expect(await internals(backendB).countActiveWorkspaceTurns()).toBe(0); @@ -108,7 +138,8 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { status: "interrupted", error: "Workspace turn interrupted after restart", }); - expect(await exists(workspaceTurnOwnerLockPath(rootDir, "wst_handle"))).toBe(false); + expect(holders.reclaimed).toEqual([lockPath]); + expect(await exists(lockPath)).toBe(false); }); test("the owner releases its lock once the handle settles", async () => { @@ -320,11 +351,12 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { }); test("a dead owner's handle is still skipped, and left for a later settlement", async () => { - const { config, parentId } = await startWorkspaceTurnForTest(rootDir); - await markTurnOwnerDead("wst_handle"); + const { config, parentId, taskService: backendA } = await startWorkspaceTurnForTest(rootDir); + const { holders } = await markTurnOwnerDead(backendA, "wst_handle"); const { taskService: backendB } = createWorkspaceTurnManagerHarness( await createTestConfig(rootDir), { + locks: withForeignHolders(SessionLocksNode, holders), workspaceService: createWorkspaceServiceMocks({ acquireIdleTurnExclusion: mock(() => Err(new Error("admission in flight"))), }).workspaceService, @@ -336,6 +368,38 @@ describe("workspace-turn handles owned by another live backend (#4446)", () => { expect( await new TaskHandleStore(config).getWorkspaceTurn(parentId, "wst_handle") ).toMatchObject({ status: "running" }); + // The probe only reads: the dead owner stays for the later settlement to reclaim. + expect(holders.reclaimed).toEqual([]); + }); + + test("a fake live owner on the lock keeps the handle, with no lock file", async () => { + const { config, parentId, taskService: backendA } = await startWorkspaceTurnForTest(rootDir); + const lockPath = await dropOwnerLock(backendA, "wst_handle"); + const holders = makeForeignHolders(); + holders.hold(lockPath, "live"); + const { taskService: backendB } = createWorkspaceTurnManagerHarness( + await createTestConfig(rootDir), + { + locks: withForeignHolders(SessionLocksNode, holders), + workspaceService: createWorkspaceServiceMocks({ + acquireIdleTurnExclusion: mock(() => Err(new Error("admission in flight"))), + }).workspaceService, + } + ); + + // keepIfOwnedElsewhere sees the fake owner as another live backend. + expect(await internals(backendB).countActiveWorkspaceTurns()).toBe(1); + expect(await backendB.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).toEqual([ + "wst_handle", + ]); + expect(await exists(lockPath)).toBe(false); + + // Control: once that owner releases, the same probe no longer keeps the handle. + holders.release(lockPath); + expect(await internals(backendB).countActiveWorkspaceTurns()).toBe(0); + expect( + await new TaskHandleStore(config).getWorkspaceTurn(parentId, "wst_handle") + ).toMatchObject({ status: "running" }); }); }); diff --git a/src/node/services/workspaceTurnManager.testHarness.ts b/src/node/services/workspaceTurnManager.testHarness.ts index b35e12fee6c..a431adf0063 100644 --- a/src/node/services/workspaceTurnManager.testHarness.ts +++ b/src/node/services/workspaceTurnManager.testHarness.ts @@ -1,8 +1,12 @@ +import assert from "node:assert"; import * as path from "path"; +import type { Layer } from "effect"; import type { Config } from "@/node/config"; import type { AIService } from "@/node/services/aiService"; import { isActiveWorkflowRunStatus } from "@/common/types/workflow"; +import type { SessionLocksTag } from "@/node/services/di/tags"; import { HistoryService } from "@/node/services/historyService"; +import { historyServiceWithLocks } from "@/node/services/testHistoryService"; import { buildAgentTaskIndex, countActiveAgentTasks, @@ -187,6 +191,8 @@ export function createWorkspaceTurnManagerHarness( config: Config, overrides?: { historyService?: HistoryService; + /** The SessionLocks layer of the default history service, e.g. withForeignHolders. */ + locks?: Layer.Layer; aiService?: AIService; workspaceService?: WorkspaceHost; initStateManager?: InitStateManager; @@ -200,7 +206,15 @@ export function createWorkspaceTurnManagerHarness( workspaceService: WorkspaceHost; initStateManager: InitStateManager; } { - const historyService = overrides?.historyService ?? new HistoryService(config); + assert( + overrides?.historyService == null || overrides.locks == null, + "createWorkspaceTurnManagerHarness: locks only configures the default history service" + ); + const historyService = + overrides?.historyService ?? + (overrides?.locks != null + ? historyServiceWithLocks(config, overrides.locks) + : new HistoryService(config)); const aiService = overrides?.aiService ?? createAIServiceMocks(config).aiService; const workspaceService = overrides?.workspaceService ?? createWorkspaceServiceMocks().workspaceService; diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index 60fa585df38..7a41d1595aa 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -3,10 +3,9 @@ import * as fsPromises from "node:fs/promises"; import * as path from "node:path"; import { DesktopInputCoordinator } from "@/node/services/desktop/DesktopInputCoordinator"; import { - acquireCrossProcessLock, CrossProcessLockTimeoutError, - inspectCrossProcessLock, -} from "@/node/utils/main/crossProcessLock"; + type SessionLocks, +} from "@/node/services/sessionLocks/sessionLocks"; import assert from "node:assert/strict"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; import { isModelHiddenMessage } from "@/common/utils/messages/modelHiddenMessages"; @@ -446,9 +445,10 @@ const WORKSPACE_TURN_OWNER_LOCK_STALE_MS = 5 * 60 * 1000; class OwnedWorkspaceTurnHandleStore extends TaskHandleStore { constructor( config: Config, - private readonly afterUpsert: (record: WorkspaceTurnTaskHandleRecord) => Promise + private readonly afterUpsert: (record: WorkspaceTurnTaskHandleRecord) => Promise, + locks: SessionLocks ) { - super(config); + super(config, locks); } override async upsertWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { @@ -713,8 +713,12 @@ export class WorkspaceTurnManager { private readonly streamManager?: StreamManager, private readonly desktopInputCoordinator = new DesktopInputCoordinator(config) ) { - this.taskHandleStore = new OwnedWorkspaceTurnHandleStore(config, (record) => - this.afterHandleWrite(record) + // Phase 4: every lock of this manager (publication, live owner, its probe) comes from the + // history service's SessionLocks, so a test layer can play another backend's owner. + this.taskHandleStore = new OwnedWorkspaceTurnHandleStore( + config, + (record) => this.afterHandleWrite(record), + historyService.locks ); } @@ -737,7 +741,7 @@ export class WorkspaceTurnManager { private async acquireTurnOwnerLock(handleId: string): Promise<"held" | "foreign-live"> { if (this.turnOwnerLocks.has(handleId)) return "held"; try { - const release = await acquireCrossProcessLock({ + const release = await this.historyService.locks.acquireCrossProcessLock({ lockPath: workspaceTurnOwnerLockPath(this.config.rootDir, handleId), acquireTimeoutMs: 0, staleMs: WORKSPACE_TURN_OWNER_LOCK_STALE_MS, @@ -4851,7 +4855,7 @@ export class WorkspaceTurnManager { */ private async keepIfOwnedElsewhere(handleId: string): Promise<"keep" | undefined> { if (this.turnOwnerLocks.has(handleId)) return undefined; - const lock = await inspectCrossProcessLock( + const lock = await this.historyService.locks.inspectCrossProcessLock( workspaceTurnOwnerLockPath(this.config.rootDir, handleId) ); return lock.state === "held" ? "keep" : undefined; From fcda462f51e723080da5d4f5a0c4afb50891d8aa Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 24 Oct 2026 06:53:09 +0000 Subject: [PATCH 19/19] tests: drop the agentSession.ts workspaceUseLeases pair again, which the merge of #7355's squash restored --- .oxlintrc.json | 5 +---- eslint.config.mjs | 5 +---- scripts/noHostImports.test.ts | 1 - 3 files changed, 2 insertions(+), 9 deletions(-) diff --git a/.oxlintrc.json b/.oxlintrc.json index 8a9a82a13c6..077f405ae60 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -1282,10 +1282,7 @@ "@/node/services/workspaceUseLeases" ], "allow": { - "src/node/services/agentSession.ts": [ - "@/node/runtime/runtimeHelpers", - "@/node/services/workspaceUseLeases" - ], + "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers"], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", diff --git a/eslint.config.mjs b/eslint.config.mjs index ac8303118e3..8794929ecca 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -183,10 +183,7 @@ const HOST_MODULES = [ ]; // File -> the host modules it still imports. Phase plans list the pairs they add or remove. const HOST_IMPORT_ALLOW = { - "src/node/services/agentSession.ts": [ - "@/node/runtime/runtimeHelpers", - "@/node/services/workspaceUseLeases", - ], + "src/node/services/agentSession.ts": ["@/node/runtime/runtimeHelpers"], "src/node/services/aiService.ts": [ "@/node/runtime/multiProjectRuntime", "@/node/runtime/runtimeFactory", diff --git a/scripts/noHostImports.test.ts b/scripts/noHostImports.test.ts index d86908ecf6c..4d90f846c52 100644 --- a/scripts/noHostImports.test.ts +++ b/scripts/noHostImports.test.ts @@ -121,7 +121,6 @@ test("allows listed modules and fails stale entries", () => { // Phases 2 to 5 remove pairs until the map is empty. const FROZEN_ALLOW_PAIRS = [ "src/node/services/agentSession.ts @/node/runtime/runtimeHelpers", - "src/node/services/agentSession.ts @/node/services/workspaceUseLeases", "src/node/services/aiService.ts @/node/runtime/multiProjectRuntime", "src/node/services/aiService.ts @/node/runtime/runtimeFactory", "src/node/services/aiService.ts @/node/runtime/runtimeHelpers",