fix: defer locked transcript update publishes

jalehman · jalehman · commit d5f82f4358df · 2026-06-16T07:07:00.000-07:00
diff --git a/src/plugin-sdk/session-transcript-runtime.test.ts b/src/plugin-sdk/session-transcript-runtime.test.ts
@@ -1,10 +1,11 @@
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { appendTranscriptEvent, upsertSessionEntry } from "../config/sessions/session-accessor.js";
 import { loadSessionStore } from "../config/sessions/store.js";
 import { withOwnedSessionTranscriptWrites } from "../config/sessions/transcript-write-context.js";
+import * as transcriptEvents from "../sessions/transcript-events.js";
 import {
   appendSessionTranscriptMessageByIdentity,
   formatSessionTranscriptMemoryHitKey,
@@ -26,6 +27,7 @@ describe("session transcript runtime SDK", () => {
   });
 
   afterEach(() => {
+    vi.restoreAllMocks();
     fs.rmSync(tempDir, { force: true, recursive: true });
   });
 
@@ -221,6 +223,89 @@ describe("session transcript runtime SDK", () => {
     ]);
   });
 
+  it("publishes queued locked updates after callback appends are visible", async () => {
+    const scope = {
+      agentId: "main",
+      sessionFile: path.join(tempDir, "queued-publish-target.jsonl"),
+      sessionId: "queued-publish-session",
+      sessionKey: "agent:main:main",
+      storePath,
+    };
+    const observedUpdates: Array<{
+      callbackCompleted: boolean;
+      fileText: string;
+      update: unknown;
+    }> = [];
+    let callbackCompleted = false;
+    const emitSpy = vi
+      .spyOn(transcriptEvents, "emitSessionTranscriptUpdate")
+      .mockImplementation((update) => {
+        observedUpdates.push({
+          callbackCompleted,
+          fileText: fs.readFileSync(scope.sessionFile, "utf8"),
+          update,
+        });
+      });
+
+    const result = await withSessionTranscriptWriteLock(scope, async (locked) => {
+      await locked.appendMessage({
+        message: {
+          role: "assistant",
+          content: [{ type: "text", text: "queued publish" }],
+          timestamp: 1,
+        },
+      });
+      await locked.publishUpdate({
+        messageId: "message-from-callback",
+        sessionKey: scope.sessionKey,
+      });
+      expect(emitSpy).not.toHaveBeenCalled();
+      callbackCompleted = true;
+      return "complete";
+    });
+
+    expect(result).toBe("complete");
+    expect(emitSpy).toHaveBeenCalledTimes(1);
+    expect(observedUpdates).toEqual([
+      {
+        callbackCompleted: true,
+        fileText: expect.stringContaining("queued publish"),
+        update: expect.objectContaining({
+          messageId: "message-from-callback",
+          sessionFile: scope.sessionFile,
+          sessionKey: scope.sessionKey,
+        }),
+      },
+    ]);
+  });
+
+  it("does not publish queued locked updates when the callback throws", async () => {
+    const scope = {
+      agentId: "main",
+      sessionFile: path.join(tempDir, "failed-queued-publish-target.jsonl"),
+      sessionId: "failed-queued-publish-session",
+      sessionKey: "agent:main:main",
+      storePath,
+    };
+    const emitSpy = vi.spyOn(transcriptEvents, "emitSessionTranscriptUpdate");
+
+    await expect(
+      withSessionTranscriptWriteLock(scope, async (locked) => {
+        await locked.appendMessage({
+          message: {
+            role: "assistant",
+            content: [{ type: "text", text: "durable but failed" }],
+            timestamp: 1,
+          },
+        });
+        await locked.publishUpdate({ sessionKey: scope.sessionKey });
+        throw new Error("stop before commit");
+      }),
+    ).rejects.toThrow("stop before commit");
+    expect(emitSpy).not.toHaveBeenCalled();
+    expect(fs.readFileSync(scope.sessionFile, "utf8")).toContain("durable but failed");
+  });
+
   it("round-trips encoded memory hit keys with opaque session ids", () => {
     const key = formatSessionTranscriptMemoryHitKey({
       agentId: "SECONDARY",
diff --git a/src/plugin-sdk/session-transcript-runtime.ts b/src/plugin-sdk/session-transcript-runtime.ts
@@ -203,7 +203,10 @@ export async function withSessionTranscriptWriteLock<T>(
     ...params,
     sessionFile: storageTarget.sessionFile,
   };
-  return await runSessionTranscriptAppendTransaction(
+  // Treat publishUpdate as a post-commit callback: future transactional stores
+  // must not expose updates when the scoped write callback fails.
+  const queuedUpdates: Array<TranscriptUpdatePayload | undefined> = [];
+  const result = await runSessionTranscriptAppendTransaction(
     {
       config: params.config,
       transcriptPath: storageTarget.sessionFile,
@@ -217,13 +220,18 @@ export async function withSessionTranscriptWriteLock<T>(
             ...options,
             sessionId: params.sessionId,
           }),
-        publishUpdate: (update) =>
-          publishSessionTranscriptUpdateByIdentity({
-            ...boundScope,
-            update,
-          }),
+        publishUpdate: async (update) => {
+          queuedUpdates.push(update ? { ...update } : undefined);
+        },
       }),
   );
+  for (const update of queuedUpdates) {
+    await publishSessionTranscriptUpdateByIdentity({
+      ...boundScope,
+      update,
+    });
+  }
+  return result;
 }
 
 /**
