🤖 refactor: the workspace-turn manager and its handle store take their locks from SessionLocks

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");

src/node/services/taskHandleStore.ts

@@ -19,7 +19,9 @@
 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 @@
 }
 
 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<void> {
     this.assertValidRecord(record);
@@ -203,7 +212,7 @@
     fn: (lock: ProcessFileLock) => Promise<T>
   ): Promise<T> {
     assertValidWorkspaceTurnTaskId(handleId);
-    await using lock = await acquireProcessFileLock({
+    await using lock = await this.locks.acquireProcessFileLock({
       lockPath: path.join(
         this.config.rootDir,
         "locks",

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<SessionLocksTag>
+): HistoryService {
+  return new HistoryService(config, buildSessionFiles(SessionFilesNode), buildSessionLocks(locks));
+}

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<number>;
+      turnOwnerLocks: Map<string, () => Promise<void>>;
     };
 
   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<void> {
+  /**
+   * 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<string> {
+    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" });
     });
   });
 

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<SessionLocksTag>;
     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;

src/node/services/workspaceTurnManager.ts

@@ -3,10 +3,9 @@
 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 @@
 class OwnedWorkspaceTurnHandleStore extends TaskHandleStore {
   constructor(
     config: Config,
-    private readonly afterUpsert: (record: WorkspaceTurnTaskHandleRecord) => Promise<void>
+    private readonly afterUpsert: (record: WorkspaceTurnTaskHandleRecord) => Promise<void>,
+    locks: SessionLocks
   ) {
-    super(config);
+    super(config, locks);
   }
 
   override async upsertWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise<void> {
@@ -713,8 +713,12 @@
     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,9 +741,9 @@
   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,
         timeoutMessage: `Workspace turn ${handleId} is owned by another live Xum backend.`,
       });
@@ -750,7 +754,7 @@
       }
       return "held";
     } catch (error: unknown) {
       if (!(error instanceof CrossProcessLockTimeoutError)) {
         log.warn("Workspace turn live-owner lock unreadable; leaving the handle alone", {
           handleId,
           error: getErrorMessage(error),
@@ -4851,7 +4855,7 @@
    */
   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;