🤖 refactor: the branch-summary pending marker takes its lock and a read-only probe from SessionLocks

src/node/services/agentSession.ts

@@ -4896,22 +4896,22 @@
     return outcome;
   }
 
   private async prepareMessage(
     message: string,
     options: (SendMessageOptions & { fileParts?: FilePart[] }) | undefined,
     internal: SendMessageInternalOptions | undefined,
     attempt: PreparationAttempt
   ): Promise<AgentSessionResult<void>> {
     assert(typeof message === "string", "sendMessage requires a string message");
 
     const isManualUserMessage = internal?.synthetic !== true;
     const manualReplacement = attempt.acceptanceOrigin === "manual";
     // Idempotent sends: a manual-origin publication stamps its send ids, and its row is
     // irrevocable once written (replacementCommitted below). An automatic send carries only ids
     // its caller supplies (a task launch's brief, see TaskService TaskLaunchStart): its row stays
     // rollback-eligible, and a rollback deletes the id with the row, so a row on disk is still
     // the only acceptance evidence.
     const sendIdentities = internal?.sendIdentities ?? [];
     // A rejected input's record row (preserveRejectedManualSend) is stamped for manual sends only:
     // an automatic caller keeps its rejected input (the launch keeps taskPrompt) and sends it again.
     const rejectedSendIdentities = manualReplacement ? sendIdentities : [];
@@ -4925,9 +4925,9 @@
     // Single admission-staleness predicate for all three turn-admission gates below.
     const isAdmissionStale = () =>
       attempt.compactionAdmissionStale() ||
       internal?.admissionEpochStale?.() === true ||
       internal?.admissionStale?.() === true ||
       !this.coordinator.isCurrentTurn(attempt.owner ?? attempt.expectedTurn) ||
       this.coordinator.editBlocked(attempt.editReservation?.id);
     // An edit's truncation is irreversible: past it, only the replacement row records the user's
     // input, so shutdown must let that row land (the PREPARING gate then refuses with rows
@@ -4936,7 +4936,7 @@
     const shutdownRefusesBeforePersist = (): boolean =>
       this.coordinator.closing && !editTailTruncated;
 
     const cancelSignal = internal?.cancelSignal;
     const persistedCancelableMessageIds: string[] = [];
     const stagedPrefixes: MuxMessage[] = [];
     let replacementCapture: CompactionReplacementCapture | undefined;
@@ -4957,7 +4957,7 @@
       }
       const replacesCancellation = manualReplacement || automaticReplacement;
       const batch = [...stagedPrefixes, ...messages];
       attempt.inputPublication = messages.at(-1);
       assert(replacementCapture, "Publication requires its admission capture");
       // Plan-review feedback (direct, or nested in an on-send compaction request) is admitted
       // only while its snapshot and threads are still in history, checked under the history
@@ -5439,7 +5439,8 @@
       this.workspaceId,
       // Session dir enables the cross-process pending-marker wait (r48): a
       // fork registered in another backend has no entry in this process.
-      path.join(this.config.sessionsDir, this.workspaceId)
+      path.join(this.config.sessionsDir, this.workspaceId),
+      this.historyService.locks
     );
     // Workspace removal disposes the session and cancels the summary writer
     // while this send is parked on the await above; every append between here

src/node/services/branchSummary.test.ts

@@ -5,7 +5,6 @@ import {
 import type { PlanReviewRecord } from "@/common/utils/planReview/planReviewRecord";
 import { describe, expect, spyOn, test } from "bun:test";
 
-import * as fs from "node:fs/promises";
 import * as path from "node:path";
 import { MockLanguageModelV3, simulateReadableStream } from "ai/test";
 import type { LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-sdk/provider";
@@ -44,6 +43,12 @@ import {
   type SideChannelMetadata,
 } from "./branchSummary";
 import { createTestHistoryService } from "./testHistoryService";
+import { Layer } from "effect";
+import type { SessionLocksTag } from "@/node/services/di/tags";
+import { HostClockTag } from "@/node/services/di/hostClockLive";
+import { createMemoryHost } from "@/node/services/hostFiles/memoryHostFiles.testHarness";
+import { SessionLocksMemory } from "@/node/services/sessionLocks/memorySessionLocks.testHarness";
+import { nodeClock } from "@/node/utils/hostClock";
 
 function finishChunk(unified: "stop" | "length" = "stop"): LanguageModelV3StreamPart {
   return {
@@ -1347,7 +1352,9 @@ describe("branch summary placement on fork/truncate flows", () => {
     }
   });
 
-  test("a send in another process waits on the pending marker before proceeding (r48)", async () => {
+  // The Node layer and the memory layer: the reader must find the writer's marker through the
+  // same SessionLocks, since a memory lock leaves no lock file on the disk.
+  const crossProcessMarkerWait = async (locks?: Layer.Layer<SessionLocksTag>) => {
     // The registration map is process-local: with XUM_ALLOW_MULTIPLE_INSTANCES=1
     // a fork created by backend A is invisible to backend B, whose first send
     // would append its user row immediately and advance the guarded tail —
@@ -1356,7 +1363,7 @@ describe("branch summary placement on fork/truncate flows", () => {
     // send that finds NO local registration must wait on that marker.
     // Simulated here with a foreign workspace id (no local map entry)
     // sharing the session dir.
-    const { historyService, config, cleanup } = await createTestHistoryService();
+    const { historyService, config, cleanup } = await createTestHistoryService({ locks });
     try {
       const ws = "ws-cross-process-marker";
       const branchPoint = createMuxMessage("xp-1", "assistant", "branch point", { timestamp: 1 });
@@ -1397,23 +1404,23 @@ describe("branch summary placement on fork/truncate flows", () => {
         ],
         isExperimentEnabled: RLM_ON,
         guardTailMessageId: "xp-1",
+        locks: historyService.locks,
       });
 
       // r55: the starter resolves only after the marker is stat-visible —
       // the fork IPC must not return before a foreign backend's immediate
       // first send could observe it. No polling: a regression to detached
       // acquisition fails this assertion outright.
       const lockPath = path.join(sessionDir, "branch-summary.lock");
-      expect(
-        await fs.stat(lockPath).then(
-          () => true,
-          () => false
-        )
-      ).toBe(true);
+      expect(await historyService.locks.processFileLockExists(lockPath)).toBe(true);
 
       // Foreign send: no local registration under this id, marker exists —
       // it must BLOCK until the writer settles, not return immediately.
-      const foreignWait = awaitPendingBranchSummary("ws-foreign-process", sessionDir);
+      const foreignWait = awaitPendingBranchSummary(
+        "ws-foreign-process",
+        sessionDir,
+        historyService.locks
+      );
       const sentinel = Symbol("still-pending");
       expect(
         await Promise.race([
@@ -1438,7 +1445,18 @@ describe("branch summary placement on fork/truncate flows", () => {
     } finally {
       await cleanup();
     }
-  });
+  };
+
+  test("a send in another process waits on the pending marker before proceeding (r48)", () =>
+    crossProcessMarkerWait());
+
+  test("a send in another process sees a marker that the memory locks hold (r48)", () =>
+    crossProcessMarkerWait(
+      SessionLocksMemory.pipe(
+        Layer.provide(Layer.succeed(HostClockTag)(nodeClock)),
+        Layer.provide(createMemoryHost().layer)
+      )
+    ));
 
   test("removal cancels an inline edit-resend summary through clearPendingBranchSummary (r57 P1)", async () => {
     // The edit-resend path awaits its summary synchronously — no first-send

src/node/services/branchSummary.ts

@@ -27,14 +27,14 @@ import {
   anthropicThinksUnlessDisabled,
 } from "@/common/types/thinking";
 import type { WorkspaceMetadata } from "@/common/types/workspace";
-import * as fs from "node:fs/promises";
 import * as path from "node:path";
 import assert from "@/common/utils/assert";
 import { buildProviderOptions } from "@/common/utils/ai/providerOptions";
 import { getErrorMessage } from "@/common/utils/errors";
 import { estimateMuxMessageTokens } from "@/common/utils/messages/keepRecentTail";
 import { pluginHookContextTexts } from "@/common/utils/messages/pluginHookContext";
-import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock";
+import { nodeSessionLocks } from "@/node/services/sessionLocks/nodeSessionLocks";
+import type { SessionLocks } from "@/node/services/sessionLocks/sessionLocks";
 import {
   BRANCH_SUMMARY_MAX_ACCUMULATED_CHARS,
   BRANCH_SUMMARY_MAX_OUTPUT_TOKENS,
@@ -953,7 +953,12 @@ export async function runInlineAbandonedBranchSummary(
  * observe the marker (r55).
  */
 export async function startAbandonedBranchSummaryInBackground(
-  input: AbandonedBranchSummaryInput & { guardTailMessageId: string; sessionDir?: string }
+  input: AbandonedBranchSummaryInput & {
+    guardTailMessageId: string;
+    sessionDir?: string;
+    /** The pending marker's lock (phase 4: SessionLocks). Absent: the Node layer. */
+    locks?: SessionLocks;
+  }
 ): Promise<void> {
   const controller = new AbortController();
   // r55: the returned promise resolves only after the cross-process pending
@@ -980,7 +985,8 @@ export async function startAbandonedBranchSummaryInBackground(
     let lock: AsyncDisposable | null = null;
     if (input.sessionDir !== undefined) {
       try {
-        lock = await acquireProcessFileLock({
+        // mutation-equivalent: locks is a SessionLocks object or undefined, so ?? and || agree
+        lock = await (input.locks ?? nodeSessionLocks).acquireProcessFileLock({
           lockPath: path.join(input.sessionDir, BRANCH_SUMMARY_LOCK_FILENAME),
           timeoutMs: BRANCH_SUMMARY_LOCK_ACQUIRE_TIMEOUT_MS,
           label: "branch summary pending marker",
@@ -1033,7 +1039,9 @@ export async function startAbandonedBranchSummaryInBackground(
  */
 export async function awaitPendingBranchSummary(
   workspaceId: string,
-  sessionDir?: string
+  sessionDir?: string,
+  /** The pending marker's lock (phase 4: SessionLocks). */
+  locks: SessionLocks = nodeSessionLocks
 ): Promise<MuxMessage | null> {
   const entry = pendingBranchSummaries.get(workspaceId);
   if (!entry) {
@@ -1047,16 +1055,13 @@ export async function awaitPendingBranchSummary(
     // produced — is durable before the marker releases, so this send's
     // request assembly reads it from history; only the foreign process can
     // emit it to its renderer. The ENOENT fast path keeps ordinary sends at
-    // one stat of a nonexistent file.
+    // one stat of a nonexistent file. The probe comes from the same locks as
+    // the writer's acquire: a memory layer leaves no lock file to stat.
     if (sessionDir !== undefined) {
       const lockPath = path.join(sessionDir, BRANCH_SUMMARY_LOCK_FILENAME);
-      const markerExists = await fs.stat(lockPath).then(
-        () => true,
-        () => false
-      );
-      if (markerExists) {
+      if (await locks.processFileLockExists(lockPath)) {
         try {
-          const lock = await acquireProcessFileLock({
+          const lock = await locks.acquireProcessFileLock({
             lockPath,
             timeoutMs: BRANCH_SUMMARY_LOCK_WAIT_TIMEOUT_MS,
             label: "branch summary pending marker",

src/node/services/sessionLocks/foreignHolders.testHarness.ts

@@ -145,6 +145,9 @@ export function wrapWithForeignHolders(inner: SessionLocks, holders: ForeignHold
       }
       return await inner.acquireProcessFileLock(options);
     },
+    // A probe never reclaims: a dead holder's lock exists until an acquire meets it.
+    processFileLockExists: async (lockPath) =>
+      table.has(lockPath) || (await inner.processFileLockExists(lockPath)),
     acquireCrossProcessLock: async (options) => {
       // crossProcessLock.ts checks the signal before each attempt: abort wins over a holder.
       if (options.signal?.aborted) throw abortedAcquireError();

src/node/services/sessionLocks/memorySessionLocks.testHarness.ts

@@ -355,6 +355,12 @@ export function memorySessionLocks(
         },
       };
     },
+    // A memory file lock leaves no file, so the table answers: a dead holder stays until an
+    // acquire reclaims it, as its lock file would on the disk.
+    processFileLockExists: (lockPath) => {
+      assert(lockPath.length > 0, "processFileLockExists requires a lock path");
+      return Promise.resolve(table.owner(lockPath) !== undefined);
+    },
     acquireCrossProcessLock: async (options) => {
       if (foundByReaddir(options.lockPath) || options.createParentDirectory === false) {
         return await nodeSessionLocks.acquireCrossProcessLock(options);

src/node/services/sessionLocks/nodeSessionLocks.ts

@@ -1,7 +1,7 @@
 import { Layer } from "effect";
 import { SessionLocksTag } from "@/node/services/di/tags";
 import { workspaceUseLeasesFor } from "@/node/services/workspaceUseLeases";
-import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock";
+import { acquireProcessFileLock, processFileLockExists } from "@/node/utils/concurrency/fileLock";
 import { workspaceFileLocks } from "@/node/utils/concurrency/workspaceFileLocks";
 import {
   acquireCrossProcessLock,
@@ -19,6 +19,7 @@ export const nodeSessionLocks: SessionLocks = {
   // let two writers of one workspace run at once.
   workspaceFiles: workspaceFileLocks,
   acquireProcessFileLock: (options) => acquireProcessFileLock(options),
+  processFileLockExists: (lockPath) => processFileLockExists(lockPath),
   acquireCrossProcessLock: (options) => acquireCrossProcessLock(options),
   inspectCrossProcessLock: (lockPath) => inspectCrossProcessLock(lockPath),
   useLeasesFor: (config) => workspaceUseLeasesFor(config),

src/node/services/sessionLocks/sessionLocks.contract.testHarness.ts

@@ -150,6 +150,31 @@ export function describeSessionLocksContract(
         await reclaimed();
       }));
 
+    // branchSummary's marker reader relies on it: the probe sees the writer in every layer, and
+    // it never takes or reclaims the lock.
+    test("processFileLockExists sees an own or a foreign file lock, live or dead, and takes none", () =>
+      withLayer(async (locks, t) => {
+        const own = path.join(t.dir, "probe-own.lock");
+        const options = { lockPath: own, timeoutMs: SHORT_MS, label: "contract lock" };
+        expect(await locks.processFileLockExists(own)).toBe(false);
+        const lock = await locks.acquireProcessFileLock(options);
+        expect(await locks.processFileLockExists(own)).toBe(true);
+        await lock[Symbol.asyncDispose]();
+        expect(await locks.processFileLockExists(own)).toBe(false);
+        const live = path.join(t.dir, "probe-live.lock");
+        const holder = await t.foreignHolder(live, "live", "file");
+        expect(await locks.processFileLockExists(live)).toBe(true);
+        await holder.release();
+        expect(await locks.processFileLockExists(live)).toBe(false);
+        const dead = path.join(t.dir, "probe-dead.lock");
+        await t.foreignHolder(dead, "dead", "file");
+        expect(await locks.processFileLockExists(dead)).toBe(true);
+        expect(await locks.processFileLockExists(dead)).toBe(true);
+        const reclaimed = await locks.acquireProcessFileLock({ ...options, lockPath: dead });
+        expect(await locks.processFileLockExists(dead)).toBe(true);
+        await reclaimed[Symbol.asyncDispose]();
+        expect(await locks.processFileLockExists(dead)).toBe(false);
+      }));
     test("a dead foreign holder is reclaimed", () =>
       withLayer(async (locks, t) => {
         const file = path.join(t.dir, "dead-file.lock");

src/node/services/sessionLocks/sessionLocks.ts

@@ -18,6 +18,11 @@ export interface SessionLocks {
   readonly workspaceFiles: Pick<MutexMap<string>, "withLock">;
   /** A cross-process lock file. Node: acquireProcessFileLock. */
   acquireProcessFileLock(options: ProcessFileLockOptions): Promise<ProcessFileLock>;
+  /**
+   * A read-only probe of a lock file: true while any holder (this process or another, live or
+   * dead) has it. It never takes or reclaims the lock. Node: processFileLockExists, one stat.
+   */
+  processFileLockExists(lockPath: string): Promise<boolean>;
   /** A cross-process lock record. Node: acquireCrossProcessLock. */
   acquireCrossProcessLock(options: CrossProcessLockOptions): Promise<() => Promise<void>>;
   /** A read-only probe of a lock record: absent, dead or held. Node: inspectCrossProcessLock. */

src/node/services/workspaceService.ts

@@ -15269,6 +15269,7 @@ export class WorkspaceService
           // Cross-process pending marker home (r48): lets a first send served
           // by another backend wait for the in-flight summary.
           sessionDir: path.join(this.config.sessionsDir, newWorkspaceId),
+          locks: this.historyService.locks,
           abandonedMessages: abandonedBranchMessages,
           isExperimentEnabled: (experimentId) => this.isExperimentEnabled(experimentId),
           guardTailMessageId: sourceMessageId,

src/node/utils/concurrency/fileLock.ts

@@ -253,6 +253,19 @@ function parseLockToken(raw: string): HolderEvidence | null {
   return null;
 }
 
+/**
+ * Whether lockPath's lock file exists: a holder, live or dead, has it. One stat, so a caller
+ * that only needs to know if a holder may exist (branchSummary's marker reader) neither creates
+ * the lock's folder nor contends with the holder.
+ */
+export async function processFileLockExists(lockPath: string): Promise<boolean> {
+  assert(lockPath.length > 0, "processFileLockExists requires a lock path");
+  return await fs.stat(lockPath).then(
+    () => true,
+    () => false
+  );
+}
+
 export async function acquireProcessFileLock(
   options: ProcessFileLockOptions
 ): Promise<ProcessFileLock> {