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;