r51 batch: three review fixes across history locking and branch summary

- historyService: every chat.jsonl mutation (appends, updateHistory, deletes,
  truncations, boundary persistence, migration) now runs under the
  cross-process history write lock via withRecoveredHistoryWriteResultLock;
  reads stay lock-free (atomic-rename replacement)
- historyService: append paths refresh the cached sequence counter from
  durable history under the lock so foreign backends' rows can never receive
  duplicate sequences (delete/truncate keep their own post-mutation counter
  recompute)
- branchSummary: the deadline drain (reader.cancel + consume) is bounded by
  BRANCH_SUMMARY_CANCEL_DRAIN_MS so a provider wedged in its cancel path
  cannot hold edit-resend or workspace removal

Author: ThomasK33

src/constants/branchSummary.ts

@@ -42,6 +42,15 @@ export const BRANCH_SUMMARY_MAX_OUTPUT_TOKENS = 512;
  */
 export const BRANCH_SUMMARY_TIMEOUT_MS = 6_000;
 
+/**
+ * Bounded cleanup window for draining a deadline-cancelled summary stream
+ * (reader.cancel + consumer settlement). Cancellation normally settles in
+ * milliseconds; a provider wedged in its own cancel path must not hold the
+ * synchronous edit-resend wait or workspace removal past the deadline the
+ * drain exists to serve — after this window the consumer is detached.
+ */
+export const BRANCH_SUMMARY_CANCEL_DRAIN_MS = 2_000;
+
 /**
  * Hard cap on characters accumulated from the summary stream. Purely
  * defensive: BRANCH_SUMMARY_MAX_OUTPUT_TOKENS already bounds well-behaved

src/node/services/branchSummary.test.ts

@@ -800,6 +800,39 @@ describe("maybeAppendAbandonedBranchSummary", () => {
     }
   });
 
+  test("a provider wedged in its cancel path cannot hold the deadline drain (r51)", async () => {
+    const { historyService, cleanup } = await createTestHistoryService();
+    try {
+      // Never produces chunks AND never settles its cancel: the deadline
+      // drain (reader.cancel + consume) must be bounded, or the synchronous
+      // edit-resend wait blocks indefinitely on exactly the wedged provider
+      // the deadline exists to cap.
+      const wedgedCancel = new MockLanguageModelV3({
+        doStream: () =>
+          Promise.resolve({
+            stream: new ReadableStream<LanguageModelV3StreamPart>({
+              pull: () => new Promise<never>(() => undefined),
+              cancel: () => new Promise<never>(() => undefined),
+            }),
+          }),
+      });
+      const startedAt = Date.now();
+      const appended = await maybeAppendAbandonedBranchSummary({
+        historyService,
+        aiService: fakeAiService(wedgedCancel),
+        workspaceId: "ws-wedged-cancel",
+        abandonedMessages: meatyExchange("wedged-cancel"),
+        experiments: RLM_ON,
+        timeoutMs: 100,
+      });
+      expect(appended).toBeNull();
+      // Bounded: deadline + drain window, well under the suite cap.
+      expect(Date.now() - startedAt).toBeLessThan(5_000);
+    } finally {
+      await cleanup();
+    }
+  });
+
   test("wedged model creation is cut off by the shared deadline (r50)", async () => {
     const { historyService, cleanup } = await createTestHistoryService();
     try {

src/node/services/branchSummary.ts

@@ -29,6 +29,7 @@ import { getErrorMessage } from "@/common/utils/errors";
 import { estimateMuxMessageTokens } from "@/common/utils/messages/keepRecentTail";
 import { acquireProcessFileLock } from "@/node/utils/concurrency/fileLock";
 import {
+  BRANCH_SUMMARY_CANCEL_DRAIN_MS,
   BRANCH_SUMMARY_MAX_ACCUMULATED_CHARS,
   BRANCH_SUMMARY_MAX_OUTPUT_TOKENS,
   BRANCH_SUMMARY_MAX_TRANSCRIPT_CHARS,
@@ -471,14 +472,25 @@ async function generateAbandonedBranchSummaryText(input: {
         // Actively cancel the losing consumer: a wedged provider leaves it
         // pinned in read() (the loop's aborted check only runs when a delta
         // arrives), and cancel resolves that pending read so the reader is
-        // released promptly. AWAITED, then the consume task drained (r50,
-        // mirroring the refine runner's deadline path): returning while
+        // released promptly. Drained before cleanup (r50): returning while
         // cancellation is still in flight would run the finally's
         // runLanguageModelCleanup underneath a provider whose asynchronous
         // stream teardown had not settled, keeping network/runtime resources
-        // alive past workspace removal.
-        await reader.cancel().catch(() => undefined);
-        await consume;
+        // alive past workspace removal. The drain itself is BOUNDED (r51):
+        // a provider wedged in its own cancel path would otherwise hold the
+        // synchronous edit-resend wait or workspace removal indefinitely —
+        // exactly the wedged-provider case the deadline exists to cap. After
+        // the window the consumer is detached; nothing observable depends on
+        // it (the salvage snapshot below is taken from `accumulated`, and
+        // the raced-away task can only settle into an abandoned stream).
+        const drained = (async () => {
+          await reader.cancel().catch(() => undefined);
+          await consume;
+        })();
+        await Promise.race([
+          drained,
+          new Promise<void>((resolve) => setTimeout(resolve, BRANCH_SUMMARY_CANCEL_DRAIN_MS)),
+        ]);
         // Deadline hit. Salvage whole sentences already streamed — a missed
         // deadline should still buy a (shorter) summary when tokens flowed.
         const salvaged = trimSummaryToBoundary(accumulated);

src/node/services/historyService.test.ts

@@ -384,7 +384,7 @@ describe("HistoryService", () => {
       // replacement — built from contents read before the foreign append —
       // would silently delete the foreign row.
       const foreign = await acquireProcessFileLock({
-        lockPath: path.join(config.getSessionDir(workspaceId), "history-append.lock"),
+        lockPath: path.join(config.getSessionDir(workspaceId), "history.lock"),
         timeoutMs: 5_000,
         label: "test foreign backend",
       });
@@ -406,6 +406,39 @@ describe("HistoryService", () => {
       const messages = await collectFullHistory(service, workspaceId);
       expect(messages.map((m) => m.id)).toEqual(["msg1", "payload-1", "trigger-1"]);
     });
+
+    it("advances the sequence counter past foreign rows under the write lock (r51)", async () => {
+      const workspaceId = "workspace1";
+      // Cache a counter in this instance (msg1 takes sequence 0, counter -> 1).
+      const seeded = await service.appendToHistory(
+        workspaceId,
+        createMuxMessage("msg1", "user", "Hello")
+      );
+      expect(seeded.success).toBe(true);
+      // A foreign backend (XUM_ALLOW_MULTIPLE_INSTANCES=1) appends a row with
+      // a higher sequence from its own counter.
+      const foreignLine = messageLine(
+        workspaceId,
+        createMuxMessage("foreign-1", "assistant", "foreign row", { historySequence: 7 })
+      );
+      await fs.appendFile(
+        path.join(config.getSessionDir(workspaceId), "chat.jsonl"),
+        foreignLine + "\n"
+      );
+      // Without the in-lock counter refresh this batch would assign stale
+      // sequences from the cached counter; updateHistory replaces the FIRST
+      // row matching a sequence, so a duplicate would let a later stream
+      // finalization overwrite an unrelated foreign row.
+      const result = await service.appendManyToHistory(workspaceId, [
+        createMuxMessage("payload-1", "assistant", "family payload"),
+        createMuxMessage("trigger-1", "user", "family trigger"),
+      ]);
+      expect(result.success).toBe(true);
+      const messages = await collectFullHistory(service, workspaceId);
+      const seqById = new Map(messages.map((m) => [m.id, m.metadata?.historySequence]));
+      expect(seqById.get("payload-1")).toBe(8);
+      expect(seqById.get("trigger-1")).toBe(9);
+    });
   });
 
   describe("updateHistory", () => {