fix: preserve transcript cursors across embedded turns (#111949)

* fix(agents): preserve transcript generation on flush (clawdbot-85c)

* fix(agents): preserve SQLite transcript parents (clawdbot-85c.5)

* test(agents): prove cursor continuity through Gateway (clawdbot-85c.7)

src/agents/cli-runner.ts

@@ -184,14 +184,6 @@ export async function isCliBindingFlushed(
   return false;
 }
 
-function flushSessionManagerTranscript(sessionManager: SessionManager): void {
-  (
-    sessionManager as unknown as {
-      replacePersistedTranscript?: () => void;
-    }
-  ).replacePersistedTranscript?.();
-}
-
 async function assertSuccessfulCliRuntimeBindingCurrent(
   context: PreparedCliRunContext,
 ): Promise<void> {
@@ -887,7 +879,7 @@ export async function runPreparedCliAgent(
       sessionManager.appendMessage(
         redactedUserMessage as Parameters<typeof sessionManager.appendMessage>[0],
       );
-      flushSessionManagerTranscript(sessionManager);
+      sessionManager.flushPendingPersistence();
     } catch (err) {
       log.warn(
         `before_agent_run block: failed to persist redacted CLI user message: ${formatErrorMessage(

src/agents/embedded-agent-runner/run/attempt-transcript-helpers.ts

@@ -22,11 +22,7 @@ import type { EmbeddedRunAttemptParams } from "./types.js";
 type AttemptSessionManager = ReturnType<typeof guardSessionManager>;
 
 export function flushSessionManagerTranscript(sessionManager: AttemptSessionManager): void {
-  (
-    sessionManager as unknown as {
-      replacePersistedTranscript?: () => void;
-    }
-  ).replacePersistedTranscript?.();
+  sessionManager.flushPendingPersistence();
 }
 
 export function repairAttemptToolUseResultPairing(

src/agents/embedded-agent-runner/run/attempt-transcript-persistence.e2e.test.ts

@@ -0,0 +1,119 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { readSessionTranscriptRawDelta } from "openclaw/plugin-sdk/session-transcript-runtime";
+import { afterEach, describe, expect, it } from "vitest";
+import {
+  appendTranscriptMessage,
+  upsertSessionEntry,
+} from "../../../config/sessions/session-accessor.js";
+import { formatSqliteSessionFileMarker } from "../../../config/sessions/sqlite-marker.js";
+import { SessionManager } from "../../sessions/session-manager.js";
+import { flushSessionManagerTranscript } from "./attempt-transcript-helpers.js";
+
+const tempPaths: string[] = [];
+
+async function makeTempDir(): Promise<string> {
+  const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-attempt-transcript-"));
+  tempPaths.push(dir);
+  return dir;
+}
+
+function buildAssistantMessage(text: string) {
+  return {
+    role: "assistant" as const,
+    content: [{ type: "text" as const, text }],
+    api: "openai-responses" as const,
+    provider: "openai",
+    model: "test-model",
+    usage: {
+      input: 0,
+      output: 0,
+      cacheRead: 0,
+      cacheWrite: 0,
+      totalTokens: 0,
+      cost: {
+        input: 0,
+        output: 0,
+        cacheRead: 0,
+        cacheWrite: 0,
+        total: 0,
+      },
+    },
+    stopReason: "stop" as const,
+    timestamp: Date.now(),
+  };
+}
+
+describe("embedded attempt transcript persistence", () => {
+  afterEach(async () => {
+    await Promise.all(
+      tempPaths.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })),
+    );
+  });
+
+  it("resumes a raw cursor after append-only attempt settlement", async () => {
+    const dir = await makeTempDir();
+    const storePath = path.join(dir, "sessions.json");
+    const target = {
+      agentId: "main",
+      sessionId: "embedded-generation",
+      sessionKey: "agent:main:embedded-generation",
+      storePath,
+    };
+    const marker = formatSqliteSessionFileMarker(target);
+    await upsertSessionEntry(target, {
+      sessionFile: marker,
+      sessionId: target.sessionId,
+      updatedAt: 1,
+    });
+    await appendTranscriptMessage(target, {
+      cwd: dir,
+      eventId: "first-user",
+      message: { role: "user", content: "first turn" },
+      now: 1,
+    });
+
+    const bootstrap = await readSessionTranscriptRawDelta({
+      ...target,
+      maxBytes: 100_000,
+      maxEvents: 100,
+    });
+    expect(bootstrap.kind).toBe("page");
+    if (bootstrap.kind !== "page") {
+      throw new Error(`expected bootstrap page, got ${bootstrap.kind}`);
+    }
+
+    const sessionManager = SessionManager.open(marker, dir, dir);
+    sessionManager.appendMessage({
+      role: "user",
+      content: "second turn",
+      timestamp: Date.now(),
+    });
+    sessionManager.appendMessage(buildAssistantMessage("second answer"));
+
+    // Production settlement invokes this barrier immediately before afterTurn.
+    flushSessionManagerTranscript(sessionManager);
+
+    const resumed = await readSessionTranscriptRawDelta({
+      ...target,
+      cursor: bootstrap.cursor,
+      maxBytes: 100_000,
+      maxEvents: 100,
+    });
+    expect(resumed.kind).toBe("page");
+    if (resumed.kind !== "page") {
+      throw new Error(`expected append page, got ${resumed.kind}`);
+    }
+    expect(
+      resumed.events
+        .map((row) => row.event)
+        .filter((event): event is { message: { content: unknown }; type: "message" } =>
+          Boolean(
+            event && typeof event === "object" && "type" in event && event.type === "message",
+          ),
+        )
+        .map((event) => event.message.content),
+    ).toEqual(["second turn", [{ type: "text", text: "second answer" }]]);
+  });
+});

src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts

@@ -2803,7 +2803,7 @@ describe("runEmbeddedAttempt context engine sessionKey forwarding", () => {
     const afterTurn = vi.fn(async () => {
       events.push("afterTurn");
     });
-    hoisted.sessionManager.replacePersistedTranscript.mockImplementation(() => {
+    hoisted.sessionManager.flushPendingPersistence.mockImplementation(() => {
       events.push("flush");
     });
 

src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts

@@ -74,7 +74,7 @@ type SessionManagerMocks = {
   appendCustomEntry: UnknownMock;
   appendSessionInfo: UnknownMock;
   appendLabelChange: UnknownMock;
-  replacePersistedTranscript: UnknownMock;
+  flushPendingPersistence: UnknownMock;
   flushPendingToolResults: UnknownMock;
   clearPendingToolResults: UnknownMock;
   clearNextUserMessagePersistenceSuppression: UnknownMock;
@@ -242,7 +242,7 @@ const hoisted = vi.hoisted((): AttemptSpawnWorkspaceHoisted => {
     appendCustomEntry: vi.fn(),
     appendSessionInfo: vi.fn(),
     appendLabelChange: vi.fn(),
-    replacePersistedTranscript: vi.fn(),
+    flushPendingPersistence: vi.fn(),
     flushPendingToolResults: vi.fn(),
     clearPendingToolResults: vi.fn(),
     clearNextUserMessagePersistenceSuppression: vi.fn(),
@@ -1126,7 +1126,7 @@ export function resetEmbeddedAttemptHarness(
   hoisted.sessionManager.appendCustomEntry.mockReset();
   hoisted.sessionManager.appendSessionInfo.mockReset();
   hoisted.sessionManager.appendLabelChange.mockReset();
-  hoisted.sessionManager.replacePersistedTranscript.mockReset();
+  hoisted.sessionManager.flushPendingPersistence.mockReset();
   if (params.subscribeImpl) {
     hoisted.subscribeEmbeddedAgentSessionMock.mockImplementation(params.subscribeImpl);
   }

src/agents/sessions/session-manager-core.ts

@@ -582,6 +582,15 @@ export class SessionManagerCore {
     }
   }
 
+  /** Makes pending append-oriented persistence durable without replacing SQLite transcripts. */
+  protected flushPendingPersistence(): void {
+    if (!this.shouldPersist || this.sqlitePersistence || this.flushed || !this.sessionFile) {
+      return;
+    }
+    this.replacePersistedTranscript();
+    this.flushed = true;
+  }
+
   isPersisted(): boolean {
     return this.shouldPersist;
   }

src/agents/sessions/session-manager-persistence.ts

@@ -227,7 +227,7 @@ export class SessionManagerPersistence extends SessionManagerCore {
       appendTranscriptEventSync(scope, entry);
       return;
     }
-    const result = appendTranscriptMessageSync(scope, {
+    const appendOptions = {
       cwd: this.cwd,
       eventId: entry.id,
       ...(options?.config ? { config: options.config } : {}),
@@ -235,7 +235,20 @@ export class SessionManagerPersistence extends SessionManagerCore {
       message: entry.message,
       now: Date.parse(entry.timestamp),
       parentId: entry.parentId,
-    });
+    } satisfies Parameters<typeof appendTranscriptMessageSync>[1];
+    let result = appendTranscriptMessageSync(scope, appendOptions);
+    if (result && !result.appended && result.messageId !== entry.id) {
+      // SessionManager has already adopted this event ID as the next parent. A
+      // pre-persisted user turn may share its idempotency key, but dropping the
+      // canonical node would leave every later descendant dangling in SQLite.
+      result = appendTranscriptMessageSync(scope, {
+        ...appendOptions,
+        idempotencyLookup: "caller-checked",
+      });
+    }
+    if (result && result.messageId !== entry.id) {
+      throw new Error(`Session transcript parent entry was not persisted: ${entry.id}`);
+    }
     if (
       options?.idempotencyLookup === "caller-checked" &&
       (!result?.appended || result.messageId !== entry.id)

src/agents/sessions/session-manager.test.ts

@@ -9,6 +9,7 @@ import {
   appendTranscriptMessage,
   loadSessionEntry,
   loadTranscriptEvents,
+  readTranscriptRawDelta,
   upsertSessionEntry,
 } from "../../config/sessions/session-accessor.js";
 import { formatSqliteSessionFileMarker } from "../../config/sessions/sqlite-marker.js";
@@ -45,6 +46,24 @@ describe("SessionManager.open", () => {
     );
   });
 
+  it("flushes a pending initial file transcript before later appends", async () => {
+    const dir = await makeTempDir();
+    const sessionFile = path.join(dir, "pending-session.jsonl");
+    const sessionManager = SessionManager.open(sessionFile, dir, dir);
+
+    sessionManager.appendMessage({ role: "user", content: "question", timestamp: Date.now() });
+    await expect(fs.stat(sessionFile)).rejects.toMatchObject({ code: "ENOENT" });
+
+    sessionManager.flushPendingPersistence();
+    sessionManager.appendMessage(buildAssistantMessage("answer"));
+
+    expect(
+      loadEntriesFromFile(sessionFile)
+        .filter((entry) => entry.type === "message")
+        .map((entry) => ("content" in entry.message ? entry.message.content : undefined)),
+    ).toEqual(["question", [{ type: "text", text: "answer" }]]);
+  });
+
   it("opens SQLite markers without creating marker-named files and persists assistant replies", async () => {
     const dir = await makeTempDir();
     const storePath = path.join(dir, "sessions.json");
@@ -151,6 +170,53 @@ describe("SessionManager.open", () => {
     );
   });
 
+  it("persists a deduped runtime user entry before its SQLite descendants", async () => {
+    const dir = await makeTempDir();
+    const storePath = path.join(dir, "sessions.json");
+    const sessionId = "sqlite-runtime-user-parent";
+    const sessionKey = "agent:main:dashboard:sqlite-runtime-user-parent";
+    const scope = { agentId: "main", sessionId, sessionKey, storePath };
+    const marker = formatSqliteSessionFileMarker(scope);
+    const userMessage = {
+      role: "user" as const,
+      content: "question",
+      idempotencyKey: "runtime-user-parent:user",
+      timestamp: 1,
+    };
+    await upsertSessionEntry(scope, { sessionFile: marker, sessionId, updatedAt: 1 });
+    await appendTranscriptMessage(scope, {
+      cwd: dir,
+      eventId: "pre-persisted-user",
+      message: userMessage,
+      now: 1,
+    });
+    const bootstrap = readTranscriptRawDelta(scope, { maxBytes: 10_000, maxEvents: 100 });
+    expect(bootstrap.kind).toBe("page");
+    if (bootstrap.kind !== "page") {
+      throw new Error(`expected bootstrap page, got ${bootstrap.kind}`);
+    }
+
+    const sessionManager = SessionManager.open(marker, dir, dir);
+    const runtimeUserId = sessionManager.appendMessage(userMessage);
+    const assistantId = sessionManager.appendMessage(buildAssistantMessage("answer"));
+    const resumed = readTranscriptRawDelta(scope, {
+      cursor: bootstrap.cursor,
+      maxBytes: 10_000,
+      maxEvents: 100,
+    });
+
+    expect(resumed.kind).toBe("page");
+    if (resumed.kind !== "page") {
+      throw new Error(`expected append page, got ${resumed.kind}`);
+    }
+    expect(resumed.events.map((row) => (row.event as { id?: string }).id)).toEqual([
+      runtimeUserId,
+      assistantId,
+    ]);
+    const assistantEvent = resumed.events.at(1)?.event as { parentId?: string } | undefined;
+    expect(assistantEvent?.parentId).toBe(runtimeUserId);
+  });
+
   it("preserves root-to-leaf ordering across session branches", () => {
     const entries = [
       {

src/agents/sessions/session-manager.ts

@@ -96,6 +96,11 @@ export class SessionManager extends SessionManagerBranching {
     super.clearPreservedOpaqueFileEntries();
   }
 
+  /** Makes pending append-oriented persistence durable without rewriting committed entries. */
+  override flushPendingPersistence(): void {
+    super.flushPendingPersistence();
+  }
+
   override isPersisted(): boolean {
     return super.isPersisted();
   }

test/embedded-transcript-cursor.e2e.test.ts

@@ -0,0 +1,289 @@
+// E2E: ordinary embedded Gateway turns preserve raw transcript cursor continuity.
+import { randomUUID } from "node:crypto";
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import path from "node:path";
+import { readSessionTranscriptRawDelta } from "openclaw/plugin-sdk/session-transcript-runtime";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { loadSessionEntry } from "../src/config/sessions/session-accessor.js";
+import type { OpenClawConfig } from "../src/config/types.openclaw.js";
+import { connectGatewayClient, disconnectGatewayClient } from "../src/gateway/test-helpers.e2e.js";
+import {
+  createOpenClawTestInstance,
+  type OpenClawTestInstance,
+} from "./helpers/openclaw-test-instance.js";
+
+const TEST_TIMEOUT_MS = 180_000;
+const MODEL_REF = "cursor-settlement/cursor-settlement";
+const SESSION_KEY = "agent:main:cursor-settlement-e2e";
+const GATEWAY_TOKEN_OPTION = "token";
+
+type MockModelServer = {
+  baseUrl: string;
+  close: () => Promise<void>;
+};
+
+const instances: OpenClawTestInstance[] = [];
+const modelServers: MockModelServer[] = [];
+
+afterEach(async () => {
+  await Promise.allSettled(instances.splice(0).map((instance) => instance.cleanup()));
+  await Promise.allSettled(modelServers.splice(0).map((server) => server.close()));
+});
+
+describe("embedded transcript cursor settlement", () => {
+  it(
+    "resumes a public raw cursor after a real append-only Gateway turn",
+    { timeout: TEST_TIMEOUT_MS },
+    async () => {
+      const modelServer = await startMockModelServer();
+      modelServers.push(modelServer);
+      const instance = await createOpenClawTestInstance({
+        name: "embedded-transcript-cursor",
+        config: createTestConfig(modelServer.baseUrl),
+        env: { OPENCLAW_SKIP_PROVIDERS: undefined },
+      });
+      instances.push(instance);
+      await instance.startGateway();
+
+      const client = await connectGatewayClient({
+        url: instance.url,
+        [GATEWAY_TOKEN_OPTION]: instance.gatewayToken,
+        role: "operator",
+        scopes: ["operator.admin", "operator.read", "operator.write"],
+      });
+      try {
+        await runAgentTurn(client, instance, "first cursor turn");
+        const storePath = path.join(instance.state.sessionsDir("main"), "sessions.json");
+        const sessionId = await waitForSessionId(storePath);
+        const target = { agentId: "main", sessionId, sessionKey: SESSION_KEY, storePath };
+        const bootstrap = await readSessionTranscriptRawDelta({
+          ...target,
+          maxBytes: 1_000_000,
+          maxEvents: 100,
+        });
+        expect(bootstrap.kind, instance.logs()).toBe("page");
+        if (bootstrap.kind !== "page") {
+          throw new Error(`expected bootstrap page, got ${bootstrap.kind}`);
+        }
+        expect(bootstrap.hasMore).toBe(false);
+
+        await runAgentTurn(client, instance, "second cursor turn");
+        const resumed = await readSessionTranscriptRawDelta({
+          ...target,
+          cursor: bootstrap.cursor,
+          maxBytes: 1_000_000,
+          maxEvents: 100,
+        });
+
+        expect(resumed.kind, `${JSON.stringify(resumed)}\n${instance.logs()}`).toBe("page");
+        if (resumed.kind !== "page") {
+          throw new Error(`expected resumed page, got ${resumed.kind}`);
+        }
+        expect(resumed.hasMore).toBe(false);
+        const messageTexts = resumed.events.flatMap((row) => readMessageText(row.event));
+        expect(messageTexts).toContain("second cursor turn");
+        expect(messageTexts).toContain("cursor settlement response 2");
+        expect(messageTexts).not.toContain("first cursor turn");
+        expect(messageTexts).not.toContain("cursor settlement response 1");
+      } finally {
+        await disconnectGatewayClient(client);
+      }
+    },
+  );
+});
+
+function createTestConfig(baseUrl: string): OpenClawConfig {
+  return {
+    plugins: { slots: { memory: "none" } },
+    agents: {
+      defaults: {
+        heartbeat: { every: "0m" },
+        model: { primary: MODEL_REF },
+        models: { [MODEL_REF]: { agentRuntime: { id: "openclaw" } } },
+        skipBootstrap: true,
+        skills: [],
+      },
+    },
+    tools: { profile: "minimal" },
+    models: {
+      mode: "replace",
+      providers: {
+        "cursor-settlement": {
+          baseUrl: `${baseUrl}/v1`,
+          apiKey: "test-token-placeholder",
+          api: "openai-responses",
+          request: { allowPrivateNetwork: true },
+          models: [
+            {
+              id: "cursor-settlement",
+              name: "cursor-settlement",
+              api: "openai-responses",
+              reasoning: false,
+              input: ["text"],
+              cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+              contextWindow: 128_000,
+              maxTokens: 4_096,
+            },
+          ],
+        },
+      },
+    },
+  };
+}
+
+async function runAgentTurn(
+  client: Awaited<ReturnType<typeof connectGatewayClient>>,
+  instance: OpenClawTestInstance,
+  message: string,
+): Promise<void> {
+  const requestedRunId = randomUUID();
+  const started = await client.request<{ runId?: string; status?: string }>("agent", {
+    sessionKey: SESSION_KEY,
+    message,
+    deliver: false,
+    idempotencyKey: requestedRunId,
+  });
+  expect(started.status).toBe("accepted");
+  const completed = await client.request<{ error?: unknown; status?: string }>(
+    "agent.wait",
+    { runId: started.runId ?? requestedRunId, timeoutMs: 120_000 },
+    { timeoutMs: 125_000 },
+  );
+  expect(completed.status, `${JSON.stringify(completed)}\n${instance.logs()}`).toBe("ok");
+}
+
+async function waitForSessionId(storePath: string): Promise<string> {
+  let sessionId: string | undefined;
+  await vi.waitFor(
+    () => {
+      sessionId = loadSessionEntry({
+        agentId: "main",
+        readConsistency: "latest",
+        sessionKey: SESSION_KEY,
+        storePath,
+      })?.sessionId;
+      expect(sessionId).toBeTruthy();
+    },
+    { interval: 20, timeout: 30_000 },
+  );
+  if (!sessionId) {
+    throw new Error(`session id was not persisted for ${SESSION_KEY}`);
+  }
+  return sessionId;
+}
+
+function readMessageText(event: unknown): string[] {
+  if (!event || typeof event !== "object" || (event as { type?: unknown }).type !== "message") {
+    return [];
+  }
+  const message = (event as { message?: { content?: unknown } }).message;
+  if (typeof message?.content === "string") {
+    return [message.content];
+  }
+  if (!Array.isArray(message?.content)) {
+    return [];
+  }
+  return message.content.flatMap((part) =>
+    part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string"
+      ? [(part as { text: string }).text]
+      : [],
+  );
+}
+
+async function startMockModelServer(): Promise<MockModelServer> {
+  let responseCount = 0;
+  const server = createServer((request, response) => {
+    void (async () => {
+      const url = new URL(request.url ?? "/", "http://127.0.0.1");
+      if (request.method === "GET" && url.pathname === "/v1/models") {
+        response.writeHead(200, { "content-type": "application/json" });
+        response.end(JSON.stringify({ data: [{ id: "cursor-settlement", object: "model" }] }));
+        return;
+      }
+      if (request.method !== "POST" || url.pathname !== "/v1/responses") {
+        response.writeHead(404).end();
+        return;
+      }
+      await drainRequest(request);
+      responseCount += 1;
+      writeModelResponse(response, responseCount);
+    })().catch((error) => {
+      response.writeHead(500, { "content-type": "application/json" });
+      response.end(JSON.stringify({ error: { message: String(error) } }));
+    });
+  });
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("cursor settlement model server did not bind");
+  }
+  return {
+    baseUrl: `http://127.0.0.1:${address.port}`,
+    close: async () => {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+    },
+  };
+}
+
+async function drainRequest(request: IncomingMessage): Promise<void> {
+  for await (const chunk of request) {
+    // Consume the body before replying so the embedded transport completes cleanly.
+    void chunk;
+  }
+}
+
+function writeModelResponse(response: ServerResponse, sequence: number): void {
+  const text = `cursor settlement response ${sequence}`;
+  const message = {
+    type: "message",
+    id: `cursor-settlement-message-${sequence}`,
+    role: "assistant",
+    status: "completed",
+    content: [{ type: "output_text", text, annotations: [] }],
+  };
+  const events = [
+    {
+      type: "response.output_item.added",
+      output_index: 0,
+      item: { ...message, status: "in_progress", content: [] },
+    },
+    {
+      type: "response.output_text.delta",
+      item_id: message.id,
+      output_index: 0,
+      content_index: 0,
+      delta: text,
+    },
+    {
+      type: "response.output_text.done",
+      item_id: message.id,
+      output_index: 0,
+      content_index: 0,
+      text,
+    },
+    { type: "response.output_item.done", output_index: 0, item: message },
+    {
+      type: "response.completed",
+      response: {
+        id: `cursor-settlement-response-${sequence}`,
+        status: "completed",
+        output: [message],
+        usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 },
+      },
+    },
+  ];
+  response.writeHead(200, {
+    "content-type": "text/event-stream",
+    "cache-control": "no-store",
+    connection: "keep-alive",
+  });
+  response.end(
+    `${events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("")}data: [DONE]\n\n`,
+  );
+}