feat: bound transcript reads across replacements (#110374)

2026-07-21 02:06:43 +00:00 · 2026-07-18 11:41:39 -07:00
parent 1ca6b6ff67
commit 453444636c
32 changed files with 1855 additions and 37 deletions
@@ -106,6 +106,7 @@ const rootEntries = [
  "scripts/repro/tool-surface-live-bench.ts!",
  // Workflow/package-script entrypoints are not imported from production modules.
  "scripts/openclaw-cross-os-release-checks.ts!",
+  "scripts/bench-transcript-cursors.ts!",
  "scripts/bench-sqlite-reliability.ts!",
  // Docker/manual E2E executables and their nested assertion/probe entrypoints.
  "scripts/e2e/*.{js,mjs,ts}!",
@@ -276,7 +276,7 @@ e9c4398b04d04fead0e46ec618589d8f597a0b6087805ab132138e941b76e190  module/sandbox
 f07839f5b8929a179857a0b4b89f8448864078cd5972dd53f9a30e50bf55408d  module/session-key-runtime
 f59099aa2d536246d4b1297f01bcea66796351a9259debdd45909afeb7a42d86  module/session-store-runtime
 b1d0a76337122cb9dbb0c89fbb8bfacc1a88ce689270d3d684019c9a03ce669a  module/session-transcript-hit
-480e6b526b788bee458ef6e023a8bb53dd3b5a09c11278f8e3424629e43852f5  module/session-transcript-runtime
+04c19558a5f27380ea13cadee1e3238729788c18a919fb078b24d2d67d006ecf  module/session-transcript-runtime
 a8fb87c7c39ec6e874926a311d2306b49843767ee5659e006024c59af822466e  module/session-visibility
 695971d31b3e16f0bf9b643acc30df312fe94b3c0dfb27a2e6156c8a00b5e261  module/setup
 575fe05081bf9edd1c6d886143e4835c218fb8d0641ce148a4a69d3cefcea2ef  module/setup-adapter-runtime
@@ -200,7 +200,11 @@ two-party event loops that do not go through the shared inbound reply runner.

    `formatSqliteSessionFileMarker(...)`, `parseSqliteSessionFileMarker(...)`, and `sqliteSessionFileMarkerMatchesSession(...)` are transitional helpers for code that still receives a legacy field named `sessionFile`. A parsed SQLite marker identifies a live SQLite transcript target; it is not a filesystem path. New APIs should carry typed session identity instead of marker strings.

-    For transcript reads and writes, import `openclaw/plugin-sdk/session-transcript-runtime` and use `resolveSessionTranscriptIdentity(...)`, `resolveSessionTranscriptTarget(...)`, `readSessionTranscriptEvents(...)`, `readVisibleSessionTranscriptMessageEntries(...)`, `appendSessionTranscriptMessageByIdentity(...)`, `publishSessionTranscriptUpdateByIdentity(...)`, or `withSessionTranscriptWriteLock(...)` with `{ agentId, sessionKey, sessionId }`. These APIs let plugins identify a transcript, read raw events or visible branch-safe message entries, append messages, publish updates, and run related operations under the same transcript write lock without depending on active transcript file paths. `readVisibleSessionTranscriptMessageEntries(...)` returns ordered read metadata; its `seq` field is not a resumable cursor.
+    For transcript reads and writes, import `openclaw/plugin-sdk/session-transcript-runtime` and use `resolveSessionTranscriptIdentity(...)`, `resolveSessionTranscriptTarget(...)`, `readSessionTranscriptEvents(...)`, `readSessionTranscriptRawDelta(...)`, `readSessionTranscriptVisibleMessageDelta(...)`, `readVisibleSessionTranscriptMessageEntries(...)`, `appendSessionTranscriptMessageByIdentity(...)`, `publishSessionTranscriptUpdateByIdentity(...)`, or `withSessionTranscriptWriteLock(...)` with `{ agentId, sessionKey, sessionId }`. These APIs let plugins identify a transcript, read raw events or visible branch-safe message entries, append messages, publish updates, and run related operations under the same transcript write lock without depending on active transcript file paths. `readVisibleSessionTranscriptMessageEntries(...)` returns ordered read metadata; its `seq` field is not a resumable cursor.
+
+    `readSessionTranscriptRawDelta(...)` returns a bounded `page`, `reset`, or `missing` result. Pass the opaque `page.cursor` into the next call. Pure appends preserve the cursor, while transcript replacement returns `reset` with a new bootstrap cursor. Pages default to 1,000 events and 1,000,000 serialized bytes; callers may request up to 10,000 events and 64 MiB. When the next event alone exceeds `maxBytes`, the page is empty and reports `requiredBytes`; retry with at least that byte limit when it is no greater than 64 MiB. Larger individual events require the complete-read API. A cursor identifies position only and never grants access to another session.
+
+    `readSessionTranscriptVisibleMessageDelta(...)` provides the same bounded bootstrap-and-resume shape over the host-owned active message projection. It returns messages from oldest to newest, so context engines can drain initial history and persist the opaque cursor as their watermark. Store and return the cursor unchanged; it is a continuation hint, not an authorization credential. Linear appends resume after the last returned message. Transcript replacement, a cursor whose anchor left or moved within the active branch, malformed cursors, and cross-session cursors return `reset` with a fresh bootstrap cursor. The count and byte defaults and caps match the raw delta API. While the active projection is rebuilding after a branch change, the result is `unavailable` with reason `projection_rebuilding`; retry later rather than falling back to an active transcript file.

    The legacy whole-store and active transcript file helpers are no longer exported from the plugin SDK. Use the scoped entry helpers for session metadata and the transcript identity helpers for active transcript operations. Archive/support workflows that need file artifacts should use their dedicated archive surfaces instead of active session runtime APIs.

@@ -275,7 +275,7 @@ usage endpoint failed or returned no usable usage data.
    | `plugin-sdk/reply-reference` | `createReplyReferencePlanner` |
    | `plugin-sdk/reply-chunking` | Narrow text/markdown chunking helpers |
    | `plugin-sdk/session-store-runtime` | Session workflow helpers (`getSessionEntry`, `listSessionEntries`, `patchSessionEntry`, `upsertSessionEntry`), repair/lifecycle helpers (`deleteSessionEntry`, `cleanupSessionLifecycleArtifacts`, `resolveSessionStoreBackupPaths`), marker helpers for transitional `sessionFile` values, bounded recent user/assistant transcript text reads by session identity, session store path/session-key helpers, and updated-at reads, without broad config writes/maintenance imports |
-    | `plugin-sdk/session-transcript-runtime` | Transcript identity, scoped target/read/write helpers, visible message-entry projection, update publishing, write locks, and transcript memory hit keys |
+    | `plugin-sdk/session-transcript-runtime` | Transcript identity, bounded raw and visible cursors, scoped target/read/write helpers, visible message-entry projection, update publishing, write locks, and transcript memory hit keys |
    | `plugin-sdk/sqlite-runtime` | Focused SQLite agent-schema, path, and transaction helpers for first-party runtime, without database lifecycle controls |
    | `plugin-sdk/cron-store-runtime` | Cron store path/load/save helpers |
    | `plugin-sdk/state-paths` | State/OAuth dir path helpers |
@@ -52,6 +52,7 @@ const hoisted = vi.hoisted(() => {
    closeActiveMemorySearchManager: vi.fn(async () => {}),
    cleanupSessionLifecycleArtifacts: vi.fn(),
    patchSessionEntry: vi.fn(),
+    rawDeltaReads: [] as Array<{ maxBytes?: number; maxEvents?: number; sessionId: string }>,
    runtimeTranscriptFiles: {} as Record<string, string>,
    sessionStore,
    updateSessionStore: vi.fn(
@@ -87,6 +88,52 @@ vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async () => {
  >("openclaw/plugin-sdk/session-transcript-runtime");
  return {
    ...actual,
+    readSessionTranscriptRawDelta: async (
+      params: Parameters<typeof actual.readSessionTranscriptRawDelta>[0],
+    ) => {
+      hoisted.rawDeltaReads.push({
+        maxBytes: params.maxBytes,
+        maxEvents: params.maxEvents,
+        sessionId: params.sessionId,
+      });
+      const testSessionFile = hoisted.runtimeTranscriptFiles[params.sessionId];
+      if (!testSessionFile) {
+        return await actual.readSessionTranscriptRawDelta(params);
+      }
+      try {
+        const lines = (await fs.readFile(testSessionFile, "utf8"))
+          .split("\n")
+          .map((line) => line.trim())
+          .filter(Boolean);
+        const maxEvents = params.maxEvents ?? lines.length;
+        const maxBytes = params.maxBytes ?? Number.MAX_SAFE_INTEGER;
+        const events: Array<{ event: unknown; seq: number }> = [];
+        let serializedBytes = 0;
+        for (const [seq, line] of lines.entries()) {
+          const nextBytes = Buffer.byteLength(`${line}\n`, "utf8");
+          if (events.length >= maxEvents || serializedBytes + nextBytes > maxBytes) {
+            break;
+          }
+          events.push({ event: JSON.parse(line) as unknown, seq });
+          serializedBytes += nextBytes;
+        }
+        const requiredBytes =
+          events.length === 0 && lines[0] ? Buffer.byteLength(`${lines[0]}\n`, "utf8") : undefined;
+        return {
+          kind: "page" as const,
+          cursor: `test-runtime:${String(events.length)}`,
+          events,
+          hasMore: events.length < lines.length,
+          ...(requiredBytes !== undefined ? { requiredBytes } : {}),
+          serializedBytes,
+        };
+      } catch (error) {
+        if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
+          throw error;
+        }
+      }
+      return await actual.readSessionTranscriptRawDelta(params);
+    },
    readSessionTranscriptEvents: async (
      target: Parameters<typeof actual.readSessionTranscriptEvents>[0],
    ) => {
@@ -538,6 +585,7 @@ describe("active-memory plugin", () => {
    for (const key of Object.keys(hoisted.runtimeTranscriptFiles)) {
      delete hoisted.runtimeTranscriptFiles[key];
    }
+    hoisted.rawDeltaReads.length = 0;
    hoisted.sessionStore["agent:main:main"] = {
      sessionId: "s-main",
      updatedAt: 0,
@@ -3783,6 +3831,12 @@ describe("active-memory plugin", () => {
    );

    expectPrependContextContains(result, "sqlite partial recall summary");
+    expect(hoisted.rawDeltaReads).toContainEqual(
+      expect.objectContaining({
+        maxBytes: 50 * 1024 * 1024,
+        maxEvents: 2_000,
+      }),
+    );
    if (artifactSessionFile) {
      await expectPathMissing(artifactSessionFile);
    }
@@ -3,7 +3,7 @@ import fs from "node:fs/promises";
 import * as readline from "node:readline";
 import { parseSqliteSessionFileMarker } from "openclaw/plugin-sdk/session-store-runtime";
 import {
-  readSessionTranscriptEvents,
+  readSessionTranscriptRawDelta,
  type SessionTranscriptTargetParams,
 } from "openclaw/plugin-sdk/session-transcript-runtime";
 import {
@@ -127,39 +127,26 @@ function transcriptSourceFromReturnedSessionFile(params: {
  };
 }

-function estimateTranscriptEventsBytes(events: readonly unknown[]): number {
-  let total = 0;
-  for (const event of events) {
-    try {
-      total += Buffer.byteLength(`${JSON.stringify(event)}\n`, "utf8");
-    } catch {
-      total += 1;
-    }
-  }
-  return total;
-}
-
 async function streamRuntimeTranscriptEvents(params: {
  target: SessionTranscriptTargetParams;
  limits?: TranscriptReadLimits;
  onRecord: (record: unknown) => boolean | void;
 }): Promise<void> {
  const limits = resolveTranscriptReadLimits(params.limits);
-  let events: readonly unknown[];
+  let page: Awaited<ReturnType<typeof readSessionTranscriptRawDelta>>;
  try {
-    events = await readSessionTranscriptEvents(params.target);
+    page = await readSessionTranscriptRawDelta({
+      ...params.target,
+      maxBytes: limits.maxBytes,
+      maxEvents: limits.maxLines,
+    });
  } catch {
    return;
  }
-  if (estimateTranscriptEventsBytes(events) > limits.maxBytes) {
+  if (page.kind !== "page") {
    return;
  }
-  let seenLines = 0;
-  for (const event of events) {
-    seenLines += 1;
-    if (seenLines > limits.maxLines) {
-      break;
-    }
+  for (const { event } of page.events) {
    try {
      if (params.onRecord(event)) {
        break;
@@ -4,7 +4,7 @@
  "openclaw": {
    "schemaVersions": {
      "state": 4,
-      "agent": 12
+      "agent": 13
    }
  },
  "description": "Multi-channel AI gateway with extensible messaging integrations",
@@ -0,0 +1,186 @@
+// Benchmarks generation-aware raw transcript reads against a 100k-event session.
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import {
+  loadTranscriptEventsSync,
+  readTranscriptRawDelta,
+} from "../src/config/sessions/session-accessor.js";
+import {
+  closeOpenClawAgentDatabasesForTest,
+  openOpenClawAgentDatabase,
+} from "../src/state/openclaw-agent-db.js";
+import { closeOpenClawStateDatabaseForTest } from "../src/state/openclaw-state-db.js";
+
+const EVENT_COUNT = 100_000;
+const DELTA_COUNT = 20;
+const RUNS = 12;
+
+type Timing = {
+  p50Ms: number;
+  p95Ms: number;
+};
+
+function percentile(values: readonly number[], fraction: number): number {
+  const sorted = values.toSorted((left, right) => left - right);
+  const index = Math.min(sorted.length - 1, Math.floor(sorted.length * fraction));
+  return Number((sorted[index] ?? 0).toFixed(3));
+}
+
+function measure(operation: () => void): Timing {
+  const values: number[] = [];
+  for (let index = 0; index < RUNS; index += 1) {
+    const startedAt = performance.now();
+    operation();
+    values.push(performance.now() - startedAt);
+  }
+  return { p50Ms: percentile(values, 0.5), p95Ms: percentile(values, 0.95) };
+}
+
+function seedTranscript(
+  database: ReturnType<typeof openOpenClawAgentDatabase>,
+  sessionId: string,
+  sessionKey: string,
+): void {
+  database.db.exec("BEGIN IMMEDIATE");
+  try {
+    const now = Date.now();
+    database.db
+      .prepare(
+        `INSERT INTO sessions (session_id, session_key, session_scope, created_at, updated_at)
+         VALUES (?, ?, 'conversation', ?, ?)`,
+      )
+      .run(sessionId, sessionKey, now, now);
+    database.db
+      .prepare(
+        `INSERT INTO session_transcript_generations (session_id, generation, updated_at)
+         VALUES (?, 'benchmark-generation', ?)`,
+      )
+      .run(sessionId, now);
+    const insert = database.db.prepare(
+      `INSERT INTO transcript_events (session_id, seq, event_json, created_at)
+       VALUES (?, ?, ?, ?)`,
+    );
+    for (let seq = 0; seq < EVENT_COUNT; seq += 1) {
+      insert.run(
+        sessionId,
+        seq,
+        JSON.stringify({ id: `event-${seq}`, text: "x".repeat(64), type: "custom" }),
+        now + seq,
+      );
+    }
+    database.db.exec("COMMIT");
+  } catch (error) {
+    database.db.exec("ROLLBACK");
+    throw error;
+  }
+}
+
+function appendDelta(
+  database: ReturnType<typeof openOpenClawAgentDatabase>,
+  sessionId: string,
+): void {
+  const insert = database.db.prepare(
+    `INSERT INTO transcript_events (session_id, seq, event_json, created_at)
+     VALUES (?, ?, ?, ?)`,
+  );
+  database.db.exec("BEGIN IMMEDIATE");
+  try {
+    for (let offset = 0; offset < DELTA_COUNT; offset += 1) {
+      const seq = EVENT_COUNT + offset;
+      insert.run(
+        sessionId,
+        seq,
+        JSON.stringify({ id: `delta-${offset}`, text: "delta", type: "custom" }),
+        Date.now() + offset,
+      );
+    }
+    database.db.exec("COMMIT");
+  } catch (error) {
+    database.db.exec("ROLLBACK");
+    throw error;
+  }
+}
+
+function readFrontierCursor(scope: Parameters<typeof readTranscriptRawDelta>[0]): string {
+  let cursor: string | undefined;
+  for (;;) {
+    const page = readTranscriptRawDelta(scope, {
+      ...(cursor ? { cursor } : {}),
+      maxBytes: 64 * 1024 * 1024,
+      maxEvents: 10_000,
+    });
+    if (page.kind !== "page") {
+      throw new Error(`unexpected cursor bootstrap result: ${page.kind}`);
+    }
+    cursor = page.cursor;
+    if (!page.hasMore) {
+      return cursor;
+    }
+  }
+}
+
+function main(): void {
+  const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-transcript-bench-"));
+  const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
+  const agentId = "benchmark";
+  const sessionId = "cursor-benchmark";
+  const sessionKey = `agent:${agentId}:cursor-benchmark`;
+  try {
+    const database = openOpenClawAgentDatabase({ agentId, env });
+    seedTranscript(database, sessionId, sessionKey);
+    const scope = { agentId, sessionId, sessionKey, storePath: database.path };
+
+    const bootstrap = measure(() => {
+      const page = readTranscriptRawDelta(scope, { maxBytes: 1_000_000, maxEvents: 1_000 });
+      if (page.kind !== "page" || page.events.length !== 1_000) {
+        throw new Error("raw bootstrap did not return 1,000 events");
+      }
+    });
+    const activeMemoryBounded = measure(() => {
+      const page = readTranscriptRawDelta(scope, {
+        maxBytes: 50 * 1024 * 1024,
+        maxEvents: 2_000,
+      });
+      if (page.kind !== "page" || page.events.length !== 2_000) {
+        throw new Error("active-memory bounds did not return 2,000 events");
+      }
+    });
+    const frontierCursor = readFrontierCursor(scope);
+    appendDelta(database, sessionId);
+    const shortDelta = measure(() => {
+      const page = readTranscriptRawDelta(scope, {
+        cursor: frontierCursor,
+        maxBytes: 1_000_000,
+        maxEvents: 100,
+      });
+      if (page.kind !== "page" || page.events.length !== DELTA_COUNT || page.hasMore) {
+        throw new Error("raw delta did not return the appended frontier");
+      }
+    });
+    const fullRead = measure(() => {
+      if (loadTranscriptEventsSync(scope).length !== EVENT_COUNT + DELTA_COUNT) {
+        throw new Error("full transcript read returned the wrong row count");
+      }
+    });
+
+    console.log(
+      JSON.stringify(
+        {
+          events: EVENT_COUNT,
+          appendedEvents: DELTA_COUNT,
+          runs: RUNS,
+          timings: { activeMemoryBounded, bootstrap, fullRead, shortDelta },
+        },
+        null,
+        2,
+      ),
+    );
+  } finally {
+    closeOpenClawAgentDatabasesForTest();
+    closeOpenClawStateDatabaseForTest();
+    fs.rmSync(stateDir, { force: true, recursive: true });
+  }
+}
+
+main();
@@ -271,7 +271,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
      // +4: gateway-backed harness question runner, claim/cancel helpers, and caller type.
      // Harvest: internal question runtime exports -2.
      // +1: ingress-effect-once factory.
-      8158,
+      // +3: bounded raw transcript cursor request, result, and reader.
+      // +3: bounded visible transcript cursor request, result, and reader.
+      8164,
      env,
    ),
    publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
@@ -312,7 +314,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
      // +3: gateway-backed harness question runner and claim/cancel helpers.
      // Harvest: internal question runtime callable -1.
      // +1: ingress-effect-once factory.
-      4540,
+      // +1: bounded raw transcript cursor reader.
+      // +1: bounded visible transcript cursor reader.
+      4542,
      env,
    ),
    publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv(
@@ -0,0 +1,177 @@
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import type { AgentMessage } from "openclaw/plugin-sdk/agent-core";
+import {
+  readSessionTranscriptVisibleMessageDelta,
+  type SessionTranscriptTargetParams,
+} from "openclaw/plugin-sdk/session-transcript-runtime";
+import { describe, expect, it } from "vitest";
+import {
+  appendTranscriptMessage,
+  replaceTranscriptEvents,
+  upsertSessionEntry,
+} from "../../config/sessions/session-accessor.js";
+import type { ContextEngine, ContextEngineSessionTarget } from "../../context-engine/types.js";
+import {
+  bootstrapHarnessContextEngine,
+  finalizeHarnessContextEngineTurn,
+} from "./context-engine-lifecycle.js";
+
+function requireTranscriptTarget(params: {
+  sessionId: string;
+  sessionKey?: string;
+  sessionTarget?: ContextEngineSessionTarget;
+}): SessionTranscriptTargetParams {
+  const sessionId = params.sessionTarget?.sessionId ?? params.sessionId;
+  const sessionKey = params.sessionTarget?.sessionKey ?? params.sessionKey;
+  if (!sessionKey) {
+    throw new Error("context engine transcript reads require a session key");
+  }
+  return {
+    sessionId,
+    sessionKey,
+    ...(params.sessionTarget?.agentId ? { agentId: params.sessionTarget.agentId } : {}),
+    ...(params.sessionTarget?.storePath ? { storePath: params.sessionTarget.storePath } : {}),
+    ...(params.sessionTarget?.threadId !== undefined
+      ? { threadId: params.sessionTarget.threadId }
+      : {}),
+  };
+}
+
+function readMessageContent(message: AgentMessage): unknown {
+  return "content" in message ? message.content : undefined;
+}
+
+describe("context engine transcript cursor contract", () => {
+  it("bootstraps, resumes appends, and rebuilds after replacement through the public SDK", async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-context-engine-cursor-"));
+    const storePath = path.join(tempDir, "sessions.json");
+    const target = {
+      agentId: "main",
+      sessionId: "context-engine-cursor",
+      sessionKey: "agent:main:context-engine-cursor",
+      storePath,
+    };
+    const projectedMessages: AgentMessage[] = [];
+    let cursor: string | undefined;
+    let resetCount = 0;
+
+    const consumeVisibleTranscript = async (params: {
+      sessionId: string;
+      sessionKey?: string;
+      sessionTarget?: ContextEngineSessionTarget;
+    }) => {
+      const readTarget = requireTranscriptTarget(params);
+      for (;;) {
+        const result = await readSessionTranscriptVisibleMessageDelta({
+          ...readTarget,
+          ...(cursor ? { cursor } : {}),
+          maxBytes: 10_000,
+          maxMessages: 1,
+        });
+        if (result.kind === "missing" || result.kind === "unavailable") {
+          return;
+        }
+        if (result.kind === "reset") {
+          projectedMessages.length = 0;
+          cursor = result.cursor;
+          resetCount += 1;
+          continue;
+        }
+        projectedMessages.push(...result.entries.map((entry) => entry.message));
+        cursor = result.cursor;
+        if (!result.hasMore) {
+          return;
+        }
+      }
+    };
+    const engine: ContextEngine = {
+      info: { id: "cursor-proof", name: "Cursor proof" },
+      bootstrap: async (params) => {
+        await consumeVisibleTranscript(params);
+        return { bootstrapped: true, importedMessages: projectedMessages.length };
+      },
+      ingest: async () => ({ ingested: true }),
+      assemble: async (params) => ({ messages: params.messages, estimatedTokens: 0 }),
+      compact: async () => ({ ok: true, compacted: false }),
+      afterTurn: consumeVisibleTranscript,
+    };
+    const skipMaintenance: NonNullable<
+      Parameters<typeof bootstrapHarnessContextEngine>[0]["runMaintenance"]
+    > = async () => undefined;
+
+    try {
+      await upsertSessionEntry(target, { sessionId: target.sessionId, updatedAt: 10 });
+      const first = await appendTranscriptMessage(target, {
+        message: { role: "user", content: "first" },
+        now: 1_000,
+      });
+      await appendTranscriptMessage(target, {
+        message: { role: "assistant", content: "second" },
+        parentId: first?.messageId,
+        now: 2_000,
+      });
+
+      await bootstrapHarnessContextEngine({
+        hadSessionFile: true,
+        contextEngine: engine,
+        sessionId: target.sessionId,
+        sessionKey: target.sessionKey,
+        sessionTarget: target,
+        sessionFile: "sqlite://context-engine-cursor",
+        runMaintenance: skipMaintenance,
+        warn: () => {},
+      });
+      expect(projectedMessages.map(readMessageContent)).toEqual(["first", "second"]);
+
+      await appendTranscriptMessage(target, {
+        message: { role: "user", content: "third" },
+        now: 3_000,
+      });
+      await finalizeHarnessContextEngineTurn({
+        contextEngine: engine,
+        promptError: false,
+        aborted: false,
+        yieldAborted: false,
+        sessionIdUsed: target.sessionId,
+        sessionKey: target.sessionKey,
+        sessionTarget: target,
+        sessionFile: "sqlite://context-engine-cursor",
+        messagesSnapshot: [],
+        prePromptMessageCount: 0,
+        runMaintenance: skipMaintenance,
+        warn: () => {},
+      });
+      expect(projectedMessages.map(readMessageContent)).toEqual(["first", "second", "third"]);
+
+      await replaceTranscriptEvents(target, [
+        {
+          type: "message",
+          id: "replacement",
+          parentId: null,
+          timestamp: "1970-01-01T00:00:04.000Z",
+          message: { role: "user", content: "replacement" },
+        },
+      ]);
+      await finalizeHarnessContextEngineTurn({
+        contextEngine: engine,
+        promptError: false,
+        aborted: false,
+        yieldAborted: false,
+        sessionIdUsed: target.sessionId,
+        sessionKey: target.sessionKey,
+        sessionTarget: target,
+        sessionFile: "sqlite://context-engine-cursor",
+        messagesSnapshot: [],
+        prePromptMessageCount: 0,
+        runMaintenance: skipMaintenance,
+        warn: () => {},
+      });
+      expect(resetCount).toBe(1);
+      expect(projectedMessages.map(readMessageContent)).toEqual(["replacement"]);
+    } finally {
+      fs.rmSync(tempDir, { force: true, recursive: true });
+    }
+  });
+});
@@ -308,6 +308,26 @@ describe("runDoctorSessionSqlite", () => {
      message?: { content?: unknown };
    };
    expect(message?.message?.content).toEqual([{ type: "text", text: "legacy string" }]);
+    closeOpenClawAgentDatabasesForTest();
+    const sqlite = nodeSqlite.requireNodeSqlite();
+    const migrated = new sqlite.DatabaseSync(
+      resolveOpenClawAgentSqlitePath({ agentId: "main", env: store.env }),
+      { readOnly: true },
+    );
+    try {
+      expect(migrated.prepare("PRAGMA user_version").get()).toEqual({
+        user_version: OPENCLAW_AGENT_SCHEMA_VERSION,
+      });
+      expect(
+        migrated
+          .prepare(
+            "SELECT session_id, length(generation) AS generation_length FROM session_transcript_generations",
+          )
+          .all(),
+      ).toEqual([{ generation_length: 32, session_id: "session-1" }]);
+    } finally {
+      migrated.close();
+    }
  });

  it("preserves the legacy transcript mtime as the SQLite mutation watermark", async () => {
@@ -1,5 +1,6 @@
 // Bounded reads over the materialized active transcript path. Dirty paths
 // schedule maintenance and fail fast; clean reads deserialize selected rows.
+import { sql } from "kysely";
 import {
  executeSqliteQuerySync,
  executeSqliteQueryTakeFirstSync,
@@ -12,6 +13,8 @@ import {
  type OpenClawAgentDatabase,
 } from "../../state/openclaw-agent-db.js";
 import type {
+  SessionTranscriptVisibleMessageDeltaLimits,
+  SessionTranscriptVisibleMessageDeltaResult,
  SessionTranscriptReadScope,
  TranscriptEvent,
 } from "./session-accessor.sqlite-contract.js";
@@ -25,11 +28,27 @@ import { startSessionTranscriptIndexReconcile } from "./session-transcript-recon
 type ActiveTranscriptDatabase = Pick<
  OpenClawAgentKyselyDatabase,
  | "session_transcript_active_events"
+  | "session_transcript_generations"
  | "session_transcript_index_state"
  | "transcript_event_identities"
  | "transcript_events"
 >;

+const VISIBLE_MESSAGE_CURSOR_VERSION = 1;
+const DEFAULT_VISIBLE_MESSAGE_MAX_MESSAGES = 1_000;
+const DEFAULT_VISIBLE_MESSAGE_MAX_BYTES = 1_000_000;
+const MAX_VISIBLE_MESSAGE_MAX_MESSAGES = 10_000;
+const MAX_VISIBLE_MESSAGE_MAX_BYTES = 64 * 1024 * 1024;
+
+type VisibleMessageCursor = {
+  agentId: string;
+  generation: string;
+  lastEventSeq: number;
+  lastMessagePosition: number;
+  sessionId: string;
+  version: typeof VISIBLE_MESSAGE_CURSOR_VERSION;
+};
+
 export type SessionTranscriptMessageEvent = {
  event: TranscriptEvent;
  seq: number;
@@ -77,6 +96,66 @@ function getActiveTranscriptKysely(database: OpenClawAgentDatabase) {
  return getNodeSqliteKysely<ActiveTranscriptDatabase>(database.db);
 }

+function normalizeVisibleMessageLimit(
+  value: number | undefined,
+  fallback: number,
+  maximum: number,
+  name: string,
+): number {
+  const resolved = value ?? fallback;
+  if (!Number.isInteger(resolved) || resolved < 1 || resolved > maximum) {
+    throw new RangeError(`${name} must be an integer between 1 and ${String(maximum)}`);
+  }
+  return resolved;
+}
+
+function encodeVisibleMessageCursor(cursor: VisibleMessageCursor): string {
+  return Buffer.from(JSON.stringify(cursor), "utf8").toString("base64url");
+}
+
+function parseVisibleMessageCursor(value: string): VisibleMessageCursor | undefined {
+  // The cursor is a continuation hint, not an authorization token. Every field
+  // is revalidated against the current scope, generation, and projection.
+  if (value.length > 4_096) {
+    return undefined;
+  }
+  try {
+    const parsed = JSON.parse(
+      Buffer.from(value, "base64url").toString("utf8"),
+    ) as Partial<VisibleMessageCursor>;
+    if (
+      parsed.version !== VISIBLE_MESSAGE_CURSOR_VERSION ||
+      typeof parsed.agentId !== "string" ||
+      typeof parsed.sessionId !== "string" ||
+      typeof parsed.generation !== "string" ||
+      !Number.isSafeInteger(parsed.lastEventSeq) ||
+      (parsed.lastEventSeq ?? -2) < -1 ||
+      !Number.isSafeInteger(parsed.lastMessagePosition) ||
+      (parsed.lastMessagePosition ?? -2) < -1 ||
+      (parsed.lastEventSeq === -1) !== (parsed.lastMessagePosition === -1)
+    ) {
+      return undefined;
+    }
+    return parsed as VisibleMessageCursor;
+  } catch {
+    return undefined;
+  }
+}
+
+function bootstrapVisibleMessageCursor(
+  projection: CurrentProjection,
+  generation: string,
+): VisibleMessageCursor {
+  return {
+    agentId: projection.resolved.agentId,
+    generation,
+    lastEventSeq: -1,
+    lastMessagePosition: -1,
+    sessionId: projection.resolved.sessionId,
+    version: VISIBLE_MESSAGE_CURSOR_VERSION,
+  };
+}
+
 function readProjectionSnapshot(
  database: OpenClawAgentDatabase,
  sessionId: string,
@@ -214,6 +293,171 @@ export function readSessionTranscriptMessageEvents(
  );
 }

+/** Reads one append-stable forward page from the materialized active-message projection. */
+export function readSessionTranscriptVisibleMessageDelta(
+  scope: SessionTranscriptReadScope,
+  limits: SessionTranscriptVisibleMessageDeltaLimits = {},
+): SessionTranscriptVisibleMessageDeltaResult {
+  const maxMessages = normalizeVisibleMessageLimit(
+    limits.maxMessages,
+    DEFAULT_VISIBLE_MESSAGE_MAX_MESSAGES,
+    MAX_VISIBLE_MESSAGE_MAX_MESSAGES,
+    "maxMessages",
+  );
+  const maxBytes = normalizeVisibleMessageLimit(
+    limits.maxBytes,
+    DEFAULT_VISIBLE_MESSAGE_MAX_BYTES,
+    MAX_VISIBLE_MESSAGE_MAX_BYTES,
+    "maxBytes",
+  );
+  return withCurrentProjectionSnapshot(scope, (projection) => {
+    const db = getActiveTranscriptKysely(projection.database);
+    const generation = executeSqliteQueryTakeFirstSync(
+      projection.database.db,
+      db
+        .selectFrom("session_transcript_generations")
+        .select("generation")
+        .where("session_id", "=", projection.resolved.sessionId),
+    )?.generation;
+    if (!generation) {
+      return { kind: "missing" };
+    }
+
+    const initialCursor = bootstrapVisibleMessageCursor(projection, generation);
+    const reset = (
+      reason: Extract<SessionTranscriptVisibleMessageDeltaResult, { kind: "reset" }>["reason"],
+    ) => ({
+      kind: "reset" as const,
+      cursor: encodeVisibleMessageCursor(initialCursor),
+      reason,
+    });
+    const cursor =
+      limits.cursor !== undefined ? parseVisibleMessageCursor(limits.cursor) : initialCursor;
+    if (!cursor) {
+      return reset("invalid_cursor");
+    }
+    if (
+      cursor.agentId !== projection.resolved.agentId ||
+      cursor.sessionId !== projection.resolved.sessionId
+    ) {
+      return reset("scope_mismatch");
+    }
+    if (cursor.generation !== generation) {
+      return reset("generation_mismatch");
+    }
+
+    let startPosition = 0;
+    if (cursor.lastEventSeq >= 0) {
+      const anchor = executeSqliteQueryTakeFirstSync(
+        projection.database.db,
+        db
+          .selectFrom("session_transcript_active_events")
+          .select("message_position")
+          .where("session_id", "=", projection.resolved.sessionId)
+          .where("event_seq", "=", cursor.lastEventSeq)
+          .where("message_position", "is not", null),
+      );
+      if (anchor?.message_position === null || anchor?.message_position === undefined) {
+        return reset("anchor_missing");
+      }
+      if (anchor.message_position !== cursor.lastMessagePosition) {
+        return reset("anchor_moved");
+      }
+      startPosition = anchor.message_position + 1;
+    }
+
+    const metadata = executeSqliteQuerySync(
+      projection.database.db,
+      db
+        .selectFrom("session_transcript_active_events as active")
+        .innerJoin("transcript_events as event", (join) =>
+          join
+            .onRef("event.session_id", "=", "active.session_id")
+            .onRef("event.seq", "=", "active.event_seq"),
+        )
+        .select([
+          "active.event_seq",
+          "active.message_position",
+          /* kysely-allow-raw: SQLite byte length avoids fetching or parsing excluded JSON. */
+          sql<number>`LENGTH(CAST(event.event_json AS BLOB)) + 1`.as("serialized_bytes"),
+        ])
+        .where("active.session_id", "=", projection.resolved.sessionId)
+        .where("active.message_position", "is not", null)
+        .where("active.message_position", ">=", startPosition)
+        .orderBy("active.message_position", "asc")
+        .limit(maxMessages + 1),
+    ).rows;
+
+    let serializedBytes = 0;
+    let selectedCount = 0;
+    for (const row of metadata) {
+      if (selectedCount >= maxMessages || serializedBytes + row.serialized_bytes > maxBytes) {
+        break;
+      }
+      serializedBytes += row.serialized_bytes;
+      selectedCount += 1;
+    }
+    const selected = metadata.slice(0, selectedCount);
+    const lastEventSeq = selected.at(-1)?.event_seq ?? cursor.lastEventSeq;
+    const lastMessagePosition = selected.at(-1)?.message_position ?? cursor.lastMessagePosition;
+    const rows =
+      selectedCount === 0
+        ? []
+        : executeSqliteQuerySync(
+            projection.database.db,
+            db
+              .selectFrom("session_transcript_active_events as active")
+              .innerJoin("transcript_events as event", (join) =>
+                join
+                  .onRef("event.session_id", "=", "active.session_id")
+                  .onRef("event.seq", "=", "active.event_seq"),
+              )
+              .leftJoin("session_transcript_active_events as parent_active", (join) =>
+                join
+                  .onRef("parent_active.session_id", "=", "active.session_id")
+                  .on((eb) =>
+                    eb("parent_active.active_position", "=", eb("active.active_position", "-", 1)),
+                  ),
+              )
+              .leftJoin("transcript_event_identities as parent_identity", (join) =>
+                join
+                  .onRef("parent_identity.session_id", "=", "parent_active.session_id")
+                  .onRef("parent_identity.seq", "=", "parent_active.event_seq"),
+              )
+              .select([
+                "active.event_seq",
+                "active.message_position",
+                "event.event_json",
+                "parent_identity.event_id as parent_id",
+              ])
+              .where("active.session_id", "=", projection.resolved.sessionId)
+              .where("active.message_position", ">=", startPosition)
+              .where("active.message_position", "<=", lastMessagePosition)
+              .orderBy("active.message_position", "asc"),
+          ).rows.map((row) => {
+            if (row.message_position === null) {
+              throw new Error("Active transcript message row is missing its message position");
+            }
+            return {
+              event: JSON.parse(row.event_json) as TranscriptEvent,
+              eventSeq: row.event_seq,
+              parentId: row.parent_id,
+              seq: row.message_position + 1,
+            };
+          });
+    const requiredBytes =
+      selectedCount === 0 && metadata[0] ? metadata[0].serialized_bytes : undefined;
+    return {
+      kind: "page",
+      cursor: encodeVisibleMessageCursor({ ...cursor, lastEventSeq, lastMessagePosition }),
+      events: rows,
+      hasMore: selectedCount < metadata.length,
+      ...(requiredBytes !== undefined ? { requiredBytes } : {}),
+      serializedBytes,
+    };
+  });
+}
+
 /** Reads a bounded active-path tail while preserving transcript line and byte caps. */
 export function readRecentSessionTranscriptMessageEvents(
  scope: SessionTranscriptReadScope,
@@ -86,6 +86,13 @@ export type SessionTranscriptEventRow = {
  seq: number;
 };

+export type {
+  SessionTranscriptRawDeltaLimits,
+  SessionTranscriptRawDeltaResult,
+  SessionTranscriptVisibleMessageDeltaLimits,
+  SessionTranscriptVisibleMessageDeltaResult,
+} from "./session-accessor.types.js";
+
 export type TranscriptMessageAppendOptions<TMessage> = {
  config?: OpenClawConfig;
  cwd?: string;
@@ -0,0 +1,228 @@
+import { sql } from "kysely";
+import {
+  executeSqliteQuerySync,
+  executeSqliteQueryTakeFirstSync,
+} from "../../infra/kysely-sync.js";
+import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js";
+import { openOpenClawAgentDatabase } from "../../state/openclaw-agent-db.js";
+import type {
+  SessionTranscriptRawDeltaLimits,
+  SessionTranscriptRawDeltaResult,
+  SessionTranscriptReadScope,
+  TranscriptEvent,
+} from "./session-accessor.sqlite-contract.js";
+import { normalizeSqliteNumber } from "./session-accessor.sqlite-normalize.js";
+import {
+  getSessionKysely,
+  resolveSqliteTranscriptReadScope,
+  toDatabaseOptions,
+} from "./session-accessor.sqlite-scope.js";
+
+const RAW_TRANSCRIPT_CURSOR_VERSION = 1;
+const DEFAULT_RAW_TRANSCRIPT_MAX_EVENTS = 1_000;
+const DEFAULT_RAW_TRANSCRIPT_MAX_BYTES = 1_000_000;
+const MAX_RAW_TRANSCRIPT_EVENTS = 10_000;
+const MAX_RAW_TRANSCRIPT_BYTES = 64 * 1024 * 1024;
+
+type RawTranscriptCursor = {
+  agentId: string;
+  generation: string;
+  lastSeq: number;
+  sessionId: string;
+  version: typeof RAW_TRANSCRIPT_CURSOR_VERSION;
+};
+
+type ResolvedTranscriptReadScope = ReturnType<typeof resolveSqliteTranscriptReadScope>;
+
+function normalizeRawDeltaLimit(
+  value: number | undefined,
+  fallback: number,
+  maximum: number,
+  name: string,
+): number {
+  const resolved = value ?? fallback;
+  if (!Number.isInteger(resolved) || resolved < 1 || resolved > maximum) {
+    throw new RangeError(`${name} must be an integer between 1 and ${String(maximum)}`);
+  }
+  return resolved;
+}
+
+function encodeRawTranscriptCursor(cursor: RawTranscriptCursor): string {
+  return Buffer.from(JSON.stringify(cursor), "utf8").toString("base64url");
+}
+
+function parseRawTranscriptCursor(value: string): RawTranscriptCursor | undefined {
+  if (value.length > 4_096) {
+    return undefined;
+  }
+  try {
+    const parsed = JSON.parse(
+      Buffer.from(value, "base64url").toString("utf8"),
+    ) as Partial<RawTranscriptCursor>;
+    if (
+      parsed.version !== RAW_TRANSCRIPT_CURSOR_VERSION ||
+      typeof parsed.agentId !== "string" ||
+      typeof parsed.sessionId !== "string" ||
+      typeof parsed.generation !== "string" ||
+      !Number.isSafeInteger(parsed.lastSeq) ||
+      (parsed.lastSeq ?? -2) < -1
+    ) {
+      return undefined;
+    }
+    return parsed as RawTranscriptCursor;
+  } catch {
+    return undefined;
+  }
+}
+
+function bootstrapCursor(
+  scope: ResolvedTranscriptReadScope,
+  generation: string,
+): RawTranscriptCursor {
+  return {
+    agentId: scope.agentId,
+    generation,
+    lastSeq: -1,
+    sessionId: scope.sessionId,
+    version: RAW_TRANSCRIPT_CURSOR_VERSION,
+  };
+}
+
+/** Read one generation-consistent raw transcript page without parsing excluded payload rows. */
+export function readSqliteTranscriptRawDelta(
+  scope: SessionTranscriptReadScope,
+  limits: SessionTranscriptRawDeltaLimits = {},
+): SessionTranscriptRawDeltaResult {
+  const resolved = resolveSqliteTranscriptReadScope(scope);
+  const maxEvents = normalizeRawDeltaLimit(
+    limits.maxEvents,
+    DEFAULT_RAW_TRANSCRIPT_MAX_EVENTS,
+    MAX_RAW_TRANSCRIPT_EVENTS,
+    "maxEvents",
+  );
+  const maxBytes = normalizeRawDeltaLimit(
+    limits.maxBytes,
+    DEFAULT_RAW_TRANSCRIPT_MAX_BYTES,
+    MAX_RAW_TRANSCRIPT_BYTES,
+    "maxBytes",
+  );
+  const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved));
+  return runSqliteDeferredTransactionSync(
+    database.db,
+    () => readRawDeltaInTransaction(database.db, resolved, limits.cursor, maxEvents, maxBytes),
+    {
+      databaseLabel: database.path,
+      operationLabel: "session transcript raw delta",
+    },
+  );
+}
+
+function readRawDeltaInTransaction(
+  database: import("node:sqlite").DatabaseSync,
+  scope: ResolvedTranscriptReadScope,
+  encodedCursor: string | undefined,
+  maxEvents: number,
+  maxBytes: number,
+): SessionTranscriptRawDeltaResult {
+  const db = getSessionKysely(database);
+  const state = executeSqliteQueryTakeFirstSync(
+    database,
+    db
+      .selectFrom("session_transcript_generations")
+      .select("generation")
+      .where("session_id", "=", scope.sessionId),
+  );
+  if (!state) {
+    return { kind: "missing" };
+  }
+
+  const initialCursor = bootstrapCursor(scope, state.generation);
+  const reset = (
+    reason: Extract<SessionTranscriptRawDeltaResult, { kind: "reset" }>["reason"],
+  ) => ({
+    kind: "reset" as const,
+    cursor: encodeRawTranscriptCursor(initialCursor),
+    reason,
+  });
+  const cursor =
+    encodedCursor !== undefined ? parseRawTranscriptCursor(encodedCursor) : initialCursor;
+  if (!cursor) {
+    return reset("invalid_cursor");
+  }
+  if (cursor.agentId !== scope.agentId || cursor.sessionId !== scope.sessionId) {
+    return reset("scope_mismatch");
+  }
+  if (cursor.generation !== state.generation) {
+    return reset("generation_mismatch");
+  }
+  const frontier = executeSqliteQueryTakeFirstSync(
+    database,
+    db
+      .selectFrom("transcript_events")
+      .select("seq")
+      .where("session_id", "=", scope.sessionId)
+      .orderBy("seq", "desc")
+      .limit(1),
+  );
+  const maxSeq = frontier ? normalizeSqliteNumber(frontier.seq) : -1;
+  if (cursor.lastSeq > maxSeq) {
+    return reset("invalid_cursor");
+  }
+
+  const metadata = executeSqliteQuerySync(
+    database,
+    db
+      .selectFrom("transcript_events")
+      .select([
+        "seq",
+        /* kysely-allow-raw: SQLite byte length avoids fetching or parsing excluded JSON. */
+        sql<number>`LENGTH(CAST(event_json AS BLOB)) + 1`.as("serialized_bytes"),
+      ])
+      .where("session_id", "=", scope.sessionId)
+      .where("seq", ">", cursor.lastSeq)
+      .orderBy("seq", "asc")
+      .limit(maxEvents + 1),
+  ).rows.map((row) => ({
+    seq: normalizeSqliteNumber(row.seq),
+    serializedBytes: normalizeSqliteNumber(row.serialized_bytes),
+  }));
+
+  let serializedBytes = 0;
+  let selectedCount = 0;
+  for (const row of metadata) {
+    if (selectedCount >= maxEvents || serializedBytes + row.serializedBytes > maxBytes) {
+      break;
+    }
+    serializedBytes += row.serializedBytes;
+    selectedCount += 1;
+  }
+  const selectedMetadata = metadata.slice(0, selectedCount);
+  const lastSeq = selectedMetadata.at(-1)?.seq ?? cursor.lastSeq;
+  const rows =
+    selectedCount === 0
+      ? []
+      : executeSqliteQuerySync(
+          database,
+          db
+            .selectFrom("transcript_events")
+            .select(["event_json", "seq"])
+            .where("session_id", "=", scope.sessionId)
+            .where("seq", ">", cursor.lastSeq)
+            .where("seq", "<=", lastSeq)
+            .orderBy("seq", "asc"),
+        ).rows.map((row) => ({
+          event: JSON.parse(row.event_json) as TranscriptEvent,
+          seq: normalizeSqliteNumber(row.seq),
+        }));
+  const nextCursor = encodeRawTranscriptCursor({ ...cursor, lastSeq });
+  const requiredBytes =
+    selectedCount === 0 && metadata[0] ? metadata[0].serializedBytes : undefined;
+  return {
+    kind: "page",
+    cursor: nextCursor,
+    events: rows,
+    hasMore: selectedCount < metadata.length,
+    ...(requiredBytes !== undefined ? { requiredBytes } : {}),
+    serializedBytes,
+  };
+}
@@ -30,6 +30,7 @@ type SessionSqliteDatabase = Pick<
  | "session_conversations"
  | "session_entries"
  | "session_routes"
+  | "session_transcript_generations"
  | "sessions"
  | "trajectory_runtime_events"
  | "transcript_event_identities"
@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import {
  executeSqliteQuerySync,
  executeSqliteQueryTakeFirstSync,
@@ -7,6 +8,61 @@ import { normalizeSqliteNumber } from "./session-accessor.sqlite-normalize.js";
 import { getSessionKysely, type ResolvedTranscriptScope } from "./session-accessor.sqlite-scope.js";
 import { deleteSessionTranscriptIndexInTransaction } from "./session-transcript-index.js";

+function createTranscriptGeneration(): string {
+  return randomUUID().replaceAll("-", "");
+}
+
+/** Read the current raw transcript generation inside the caller's transaction. */
+export function readTranscriptGenerationInTransaction(
+  database: OpenClawAgentDatabase,
+  sessionId: string,
+): string | undefined {
+  const db = getSessionKysely(database.db);
+  return executeSqliteQueryTakeFirstSync(
+    database.db,
+    db
+      .selectFrom("session_transcript_generations")
+      .select("generation")
+      .where("session_id", "=", sessionId),
+  )?.generation;
+}
+
+/** Materialize a generation once; pure appends must preserve an existing token. */
+export function ensureTranscriptGenerationInTransaction(
+  database: OpenClawAgentDatabase,
+  sessionId: string,
+): string {
+  const db = getSessionKysely(database.db);
+  const generation = createTranscriptGeneration();
+  executeSqliteQuerySync(
+    database.db,
+    db
+      .insertInto("session_transcript_generations")
+      .values({ session_id: sessionId, generation, updated_at: Date.now() })
+      .onConflict((conflict) => conflict.column("session_id").doNothing()),
+  );
+  return readTranscriptGenerationInTransaction(database, sessionId) ?? generation;
+}
+
+/** Rotate the watermark in the same transaction as destructive transcript replacement. */
+export function rotateTranscriptGenerationInTransaction(
+  database: OpenClawAgentDatabase,
+  sessionId: string,
+): string {
+  const db = getSessionKysely(database.db);
+  const generation = createTranscriptGeneration();
+  executeSqliteQuerySync(
+    database.db,
+    db
+      .insertInto("session_transcript_generations")
+      .values({ session_id: sessionId, generation, updated_at: Date.now() })
+      .onConflict((conflict) =>
+        conflict.column("session_id").doUpdateSet({ generation, updated_at: Date.now() }),
+      ),
+  );
+  return generation;
+}
+
 export function ensureTranscriptSessionRoot(
  database: OpenClawAgentDatabase,
  scope: ResolvedTranscriptScope,
@@ -19,8 +19,11 @@ import {
 import { getSessionKysely, type ResolvedTranscriptScope } from "./session-accessor.sqlite-scope.js";
 import {
  deleteSqliteTranscriptEventsInTransaction,
+  ensureTranscriptGenerationInTransaction,
  ensureTranscriptSessionRoot,
+  readTranscriptGenerationInTransaction,
  readNextTranscriptSeq,
+  rotateTranscriptGenerationInTransaction,
  touchTranscriptMutationInTransaction,
 } from "./session-accessor.sqlite-transcript-state.js";
 import {
@@ -49,6 +52,7 @@ export function appendTranscriptEventInTransaction(
  const db = getSessionKysely(database.db);
  const createdAt = readEventTimestamp(event) ?? Date.now();
  ensureTranscriptSessionRoot(database, scope, createdAt);
+  ensureTranscriptGenerationInTransaction(database, scope.sessionId);
  const identity = readTranscriptEventIdentity(event);
  if (identity && readTranscriptIdentityByEventId(database, scope.sessionId, identity.eventId)) {
    return false;
@@ -310,14 +314,21 @@ export function replaceSqliteTranscriptEventsInTransaction(
  resolved: ResolvedTranscriptScope,
  events: readonly TranscriptEvent[],
 ): void {
+  const previousGeneration = readTranscriptGenerationInTransaction(database, resolved.sessionId);
  const deleted = deleteSqliteTranscriptEventsInTransaction(database, resolved.sessionId);
  if (events.length === 0) {
-    if (deleted) {
+    if (deleted || previousGeneration) {
+      rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
      touchTranscriptMutationInTransaction(database, resolved.sessionId);
    }
    return;
  }
  ensureTranscriptSessionRoot(database, resolved, readEventTimestamp(events[0]) ?? Date.now());
+  if (deleted || previousGeneration) {
+    rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
+  } else {
+    ensureTranscriptGenerationInTransaction(database, resolved.sessionId);
+  }
  let seq = 0;
  const seenEventIds = new Set<string>();
  const seenMessageIdempotencyKeys = new Set<string>();
@@ -55,6 +55,7 @@ export {
  withSqliteTranscriptWriteTransaction,
 } from "./session-accessor.sqlite-transcript-write.js";
 export { publishSqliteTranscriptUpdate } from "./session-accessor.sqlite-events.js";
+export { readSqliteTranscriptRawDelta } from "./session-accessor.sqlite-delta.js";
 export { previewSqliteSessionDiskBudget } from "./session-accessor.sqlite-maintenance.js";
 export {
  findSqliteTranscriptEvent,
@@ -3044,6 +3044,45 @@ describe("session accessor seam", () => {
    expect(cleared.lastMutationAtMs).toBeGreaterThan(imported.lastMutationAtMs ?? 0);
  });

+  it("preserves transcript generation on append and rotates it on replacement", async () => {
+    const scope = {
+      agentId: "main",
+      sessionId: "generation-session",
+      sessionKey: "agent:main:generation-session",
+      storePath,
+    };
+    const databasePath = resolveSqliteTargetFromSessionStorePath(storePath, {
+      agentId: scope.agentId,
+    }).path;
+    expect(databasePath).toBeDefined();
+    const readGeneration = () =>
+      openOpenClawAgentDatabase({ agentId: scope.agentId, path: databasePath })
+        .db.prepare("SELECT generation FROM session_transcript_generations WHERE session_id = ?")
+        .get(scope.sessionId) as { generation: string } | undefined;
+
+    await appendTranscriptMessage(scope, {
+      message: { role: "user", content: "first" },
+    });
+    const first = readGeneration()?.generation;
+    expect(first).toMatch(/^[0-9a-f]{32}$/);
+
+    await appendTranscriptMessage(scope, {
+      message: { role: "assistant", content: "second" },
+    });
+    expect(readGeneration()?.generation).toBe(first);
+
+    await replaceSqliteTranscriptEvents(scope, [
+      { sessionId: scope.sessionId, type: "session" },
+      { id: "replacement", parentId: null, type: "custom" },
+    ]);
+    const replaced = readGeneration()?.generation;
+    expect(replaced).toMatch(/^[0-9a-f]{32}$/);
+    expect(replaced).not.toBe(first);
+
+    await replaceSqliteTranscriptEvents(scope, []);
+    expect(readGeneration()?.generation).not.toBe(replaced);
+  });
+
  it("resolves an explicit read transcript file without agent identity", () => {
    const explicitSessionFile = path.join(tempDir, "explicit-read-session.jsonl");

@@ -14,6 +14,7 @@ import {
  loadSqliteTranscriptEventsSync,
  readSqliteTranscriptStatsSync,
  readSqliteTranscriptEventAtSeqSync,
+  readSqliteTranscriptRawDelta,
  publishSqliteTranscriptUpdate,
  replaceSqliteTranscriptEvents,
  replaceSqliteTranscriptEventsSync,
@@ -29,6 +30,8 @@ import type {
  TranscriptEvent,
  SessionTranscriptStats,
  SessionTranscriptEventRow,
+  SessionTranscriptRawDeltaLimits,
+  SessionTranscriptRawDeltaResult,
  TranscriptMessageAppendOptions,
  TranscriptMessageAppendResult,
  TranscriptUpdatePayload,
@@ -78,6 +81,14 @@ export async function loadTranscriptEvents(
  return await loadSqliteTranscriptEvents(scope);
 }

+/** Reads one bounded raw transcript page using an opaque generation-aware cursor. */
+export function readTranscriptRawDelta(
+  scope: SessionTranscriptReadScope,
+  limits: SessionTranscriptRawDeltaLimits = {},
+): SessionTranscriptRawDeltaResult {
+  return readSqliteTranscriptRawDelta(scope, limits);
+}
+
 /** Replaces all transcript records for one SQLite-backed transcript. */
 export async function replaceTranscriptEvents(
  scope: SessionTranscriptAccessScope,
@@ -78,6 +78,8 @@ export type {
  SessionTranscriptManualTrimResult,
  SessionTranscriptReadScope,
  SessionTranscriptReadTarget,
+  SessionTranscriptRawDeltaLimits,
+  SessionTranscriptRawDeltaResult,
  SessionTranscriptRuntimeScope,
  SessionTranscriptRuntimeTarget,
  SessionTranscriptStats,
@@ -86,6 +88,9 @@ export type {
  SessionTranscriptTurnPersistResult,
  SessionTranscriptTurnUpdateMode,
  SessionTranscriptTurnWriteContext,
+  SessionTranscriptVisibleMessageDeltaLimits,
+  SessionTranscriptVisibleMessageDeltaResult,
+  SessionTranscriptVisibleMessageEventRow,
  SessionTranscriptWriteLockAccessorContext,
  SessionTranscriptWriteScope,
  SessionTranscriptWriteTransactionContext,
@@ -175,6 +180,7 @@ export {
  publishTranscriptUpdate,
  readLatestTranscriptAssistantText,
  readTranscriptEventAtSeqSync,
+  readTranscriptRawDelta,
  readTranscriptStatsSync,
  replaceTranscriptEvents,
  replaceTranscriptEventsSync,
@@ -192,6 +198,7 @@ export {
  readSessionTranscriptMessageEventCount,
  readSessionTranscriptMessageEventPage,
  readSessionTranscriptMessageEvents,
+  readSessionTranscriptVisibleMessageDelta,
  SessionTranscriptProjectionUnavailableError,
 } from "./session-accessor.sqlite-active-events.js";
 export type {
@@ -204,6 +204,87 @@ export type SessionTranscriptEventRow = {
  seq: number;
 };

+/** Count, byte, and continuation bounds for one raw transcript page. */
+export type SessionTranscriptRawDeltaLimits = {
+  /** Opaque cursor returned by a prior page or reset result. */
+  cursor?: string;
+  /** Maximum serialized JSONL bytes returned by this page. */
+  maxBytes?: number;
+  /** Maximum number of events returned by this page. */
+  maxEvents?: number;
+};
+
+/** Generation-aware outcome for one bounded raw transcript read. */
+export type SessionTranscriptRawDeltaResult =
+  | {
+      kind: "page";
+      /** Cursor positioned after the last returned event. */
+      cursor: string;
+      /** Ordered raw transcript events selected for this page. */
+      events: SessionTranscriptEventRow[];
+      /** True when another event remains after this page. */
+      hasMore: boolean;
+      /** First unread event size when it cannot fit under maxBytes. */
+      requiredBytes?: number;
+      /** Stored JSONL bytes represented by events. */
+      serializedBytes: number;
+    }
+  | {
+      kind: "reset";
+      /** Fresh bootstrap cursor for the current generation. */
+      cursor: string;
+      /** Stable discontinuity that invalidated the supplied cursor. */
+      reason: "generation_mismatch" | "invalid_cursor" | "scope_mismatch";
+    }
+  | { kind: "missing" };
+
+/** Count, byte, and continuation bounds for one visible-message page. */
+export type SessionTranscriptVisibleMessageDeltaLimits = {
+  /** Opaque continuation cursor; store and return it unchanged. */
+  cursor?: string;
+  /** Maximum serialized JSONL bytes returned by this page. */
+  maxBytes?: number;
+  /** Maximum number of visible messages returned by this page. */
+  maxMessages?: number;
+};
+
+/** One active-path message row selected from the materialized projection. */
+export type SessionTranscriptVisibleMessageEventRow = SessionTranscriptEventRow & {
+  /** Raw transcript sequence used only by the opaque continuation cursor. */
+  eventSeq: number;
+  /** Parent id after active-branch normalization; null for the visible root. */
+  parentId: string | null;
+};
+
+/** Generation-aware outcome for one bounded visible-message read. */
+export type SessionTranscriptVisibleMessageDeltaResult =
+  | {
+      kind: "page";
+      /** Cursor positioned after the last returned visible message. */
+      cursor: string;
+      /** Ordered active-path message events selected for this page. */
+      events: SessionTranscriptVisibleMessageEventRow[];
+      /** True when another visible message remains after this page. */
+      hasMore: boolean;
+      /** First unread event size when it cannot fit under maxBytes. */
+      requiredBytes?: number;
+      /** Stored JSONL bytes represented by events. */
+      serializedBytes: number;
+    }
+  | {
+      kind: "reset";
+      /** Fresh bootstrap cursor for the current visible generation. */
+      cursor: string;
+      /** Stable discontinuity that invalidated the supplied cursor. */
+      reason:
+        | "anchor_missing"
+        | "anchor_moved"
+        | "generation_mismatch"
+        | "invalid_cursor"
+        | "scope_mismatch";
+    }
+  | { kind: "missing" };
+
 export type TranscriptMessageAppendOptions<TMessage> = {
  /** Runtime config used for message redaction and transcript header metadata. */
  config?: OpenClawConfig;
@@ -0,0 +1,259 @@
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { appendTranscriptEvent, upsertSessionEntry } from "../config/sessions/session-accessor.js";
+import {
+  resolveSqliteTranscriptReadScope,
+  toDatabaseOptions,
+} from "../config/sessions/session-accessor.sqlite-scope.js";
+import { waitForSessionTranscriptIndexReconcile } from "../config/sessions/session-transcript-reconcile.js";
+import {
+  appendSessionTranscriptMessageByIdentity,
+  readSessionTranscriptVisibleMessageDelta,
+} from "./session-transcript-runtime.js";
+
+describe("session transcript visible cursor SDK", () => {
+  let tempDir: string;
+  let storePath: string;
+
+  beforeEach(() => {
+    tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sdk-visible-transcript-"));
+    storePath = path.join(tempDir, "sessions.json");
+  });
+
+  afterEach(() => {
+    fs.rmSync(tempDir, { force: true, recursive: true });
+  });
+
+  it("pages appends and resets when the active branch changes", async () => {
+    const scope = {
+      agentId: "main",
+      sessionId: "visible-delta-session",
+      sessionKey: "agent:main:visible-delta",
+      storePath,
+    };
+    await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
+    const root = await appendSessionTranscriptMessageByIdentity({
+      ...scope,
+      message: { role: "user", content: "root" },
+      now: 1_000,
+    });
+    const firstBranch = await appendSessionTranscriptMessageByIdentity({
+      ...scope,
+      message: { role: "assistant", content: "first branch" },
+      parentId: root?.messageId,
+      now: 2_000,
+    });
+    if (!root || !firstBranch) {
+      throw new Error("expected visible delta setup messages");
+    }
+
+    const first = await readSessionTranscriptVisibleMessageDelta({
+      ...scope,
+      maxBytes: 10_000,
+      maxMessages: 1,
+    });
+    expect(first).toMatchObject({
+      kind: "page",
+      entries: [
+        {
+          entryId: root.messageId,
+          message: { role: "user", content: "root" },
+          parentId: null,
+        },
+      ],
+      hasMore: true,
+    });
+    if (first.kind !== "page") {
+      throw new Error("expected first visible transcript page");
+    }
+
+    const second = await readSessionTranscriptVisibleMessageDelta({
+      ...scope,
+      cursor: first.cursor,
+      maxBytes: 10_000,
+      maxMessages: 1,
+    });
+    expect(second).toMatchObject({
+      kind: "page",
+      entries: [
+        {
+          entryId: firstBranch.messageId,
+          message: { role: "assistant", content: "first branch" },
+          parentId: root.messageId,
+        },
+      ],
+      hasMore: false,
+    });
+    if (second.kind !== "page") {
+      throw new Error("expected second visible transcript page");
+    }
+    const movedAnchorCursor = Buffer.from(
+      JSON.stringify({
+        ...(JSON.parse(Buffer.from(second.cursor, "base64url").toString("utf8")) as object),
+        lastMessagePosition: 0,
+      }),
+      "utf8",
+    ).toString("base64url");
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: movedAnchorCursor,
+        maxBytes: 10_000,
+        maxMessages: 1,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "anchor_moved" });
+
+    const appended = await appendSessionTranscriptMessageByIdentity({
+      ...scope,
+      message: { role: "user", content: "linear append" },
+      parentId: firstBranch.messageId,
+      now: 3_000,
+    });
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: second.cursor,
+        maxBytes: 10_000,
+        maxMessages: 1,
+      }),
+    ).resolves.toMatchObject({
+      kind: "page",
+      entries: [{ entryId: appended?.messageId, message: { content: "linear append" } }],
+      hasMore: false,
+    });
+
+    await appendTranscriptEvent(scope, {
+      type: "leaf",
+      id: "select-replacement-branch",
+      parentId: appended?.messageId,
+      targetId: root.messageId,
+    });
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: second.cursor,
+        maxBytes: 10_000,
+        maxMessages: 1,
+      }),
+    ).resolves.toEqual({ kind: "unavailable", reason: "projection_rebuilding" });
+    await waitForSessionTranscriptIndexReconcile(
+      toDatabaseOptions(resolveSqliteTranscriptReadScope(scope)),
+    );
+    await appendSessionTranscriptMessageByIdentity({
+      ...scope,
+      eventId: "replacement-branch",
+      parentId: "select-replacement-branch",
+      message: { role: "assistant", content: "replacement branch" },
+      now: 4_000,
+    });
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: second.cursor,
+        maxBytes: 10_000,
+        maxMessages: 1,
+      }),
+    ).resolves.toEqual({ kind: "unavailable", reason: "projection_rebuilding" });
+    await waitForSessionTranscriptIndexReconcile(
+      toDatabaseOptions(resolveSqliteTranscriptReadScope(scope)),
+    );
+    const reset = await readSessionTranscriptVisibleMessageDelta({
+      ...scope,
+      cursor: second.cursor,
+      maxBytes: 10_000,
+      maxMessages: 1,
+    });
+    expect(reset).toMatchObject({ kind: "reset", reason: "anchor_missing" });
+    if (reset.kind !== "reset") {
+      throw new Error("expected visible branch reset");
+    }
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: reset.cursor,
+        maxBytes: 10_000,
+        maxMessages: 10,
+      }),
+    ).resolves.toMatchObject({
+      kind: "page",
+      entries: [
+        { entryId: root.messageId, parentId: null },
+        { entryId: "replacement-branch", parentId: root.messageId },
+      ],
+      hasMore: false,
+    });
+  });
+
+  it("bounds pages before parsing oversized entries", async () => {
+    const scope = {
+      agentId: "main",
+      sessionId: "visible-delta-bounds",
+      sessionKey: "agent:main:visible-delta-bounds",
+      storePath,
+    };
+    const content = "x".repeat(200);
+    await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
+    await appendSessionTranscriptMessageByIdentity({
+      ...scope,
+      message: { role: "user", content },
+      now: 1_000,
+    });
+
+    const invalid = await readSessionTranscriptVisibleMessageDelta({
+      ...scope,
+      cursor: "not-a-cursor",
+      maxBytes: 10,
+      maxMessages: 1,
+    });
+    expect(invalid).toMatchObject({ kind: "reset", reason: "invalid_cursor" });
+    if (invalid.kind !== "reset") {
+      throw new Error("expected invalid visible cursor reset");
+    }
+    const inconsistentBootstrapCursor = Buffer.from(
+      JSON.stringify({
+        ...(JSON.parse(Buffer.from(invalid.cursor, "base64url").toString("utf8")) as object),
+        lastMessagePosition: 0,
+      }),
+      "utf8",
+    ).toString("base64url");
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: inconsistentBootstrapCursor,
+        maxBytes: 10,
+        maxMessages: 1,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "invalid_cursor" });
+    const bounded = await readSessionTranscriptVisibleMessageDelta({
+      ...scope,
+      cursor: invalid.cursor,
+      maxBytes: 10,
+      maxMessages: 1,
+    });
+    expect(bounded).toMatchObject({
+      kind: "page",
+      entries: [],
+      hasMore: true,
+      requiredBytes: expect.any(Number),
+      serializedBytes: 0,
+    });
+    if (bounded.kind !== "page" || bounded.requiredBytes === undefined) {
+      throw new Error("expected oversized visible entry metadata");
+    }
+    await expect(
+      readSessionTranscriptVisibleMessageDelta({
+        ...scope,
+        cursor: bounded.cursor,
+        maxBytes: bounded.requiredBytes,
+        maxMessages: 1,
+      }),
+    ).resolves.toMatchObject({
+      kind: "page",
+      entries: [{ message: { role: "user", content } }],
+      hasMore: false,
+      serializedBytes: bounded.requiredBytes,
+    });
+  });
+});
@@ -6,6 +6,7 @@ import {
  appendTranscriptEvent,
  listSessionEntries,
  loadSessionEntry,
+  replaceTranscriptEvents,
  upsertSessionEntry,
 } from "../config/sessions/session-accessor.js";
 import * as transcriptEvents from "../sessions/transcript-events.js";
@@ -16,6 +17,7 @@ import {
  parseSessionTranscriptMemoryHitKey,
  publishSessionTranscriptUpdateByIdentity,
  readLatestAssistantTextByIdentity,
+  readSessionTranscriptRawDelta,
  readSessionTranscriptEvents,
  readVisibleSessionTranscriptMessageEntries,
  resolveSessionTranscriptIdentity,
@@ -79,6 +81,163 @@ describe("session transcript runtime SDK", () => {
    expect(loadSessionEntry(scope)?.sessionFile).toBeUndefined();
  });

+  it("pages raw events across appends and resets after replacement", async () => {
+    const scope = {
+      agentId: "main",
+      sessionId: "raw-delta-session",
+      sessionKey: "agent:main:raw-delta",
+      storePath,
+    };
+    await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
+    for (const id of ["event-1", "event-2", "event-3"]) {
+      await appendTranscriptEvent(scope, { id, type: "custom" });
+    }
+
+    const first = await readSessionTranscriptRawDelta({
+      ...scope,
+      maxBytes: 10_000,
+      maxEvents: 2,
+    });
+    expect(first).toMatchObject({
+      kind: "page",
+      events: [
+        { event: { id: "event-1", type: "custom" }, seq: 0 },
+        { event: { id: "event-2", type: "custom" }, seq: 1 },
+      ],
+      hasMore: true,
+    });
+    if (first.kind !== "page") {
+      throw new Error("expected the first raw transcript page");
+    }
+
+    await appendTranscriptEvent(scope, { id: "event-4", type: "custom" });
+    const second = await readSessionTranscriptRawDelta({
+      ...scope,
+      cursor: first.cursor,
+      maxBytes: 10_000,
+      maxEvents: 2,
+    });
+    expect(second).toMatchObject({
+      kind: "page",
+      events: [
+        { event: { id: "event-3", type: "custom" }, seq: 2 },
+        { event: { id: "event-4", type: "custom" }, seq: 3 },
+      ],
+      hasMore: false,
+    });
+
+    await replaceTranscriptEvents(scope, [{ id: "replacement", type: "custom" }]);
+    await expect(
+      readSessionTranscriptRawDelta({
+        ...scope,
+        cursor: first.cursor,
+        maxBytes: 10_000,
+        maxEvents: 2,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "generation_mismatch" });
+  });
+
+  it("bounds raw pages before parsing an oversized event", async () => {
+    const missingScope = {
+      agentId: "main",
+      sessionId: "missing-raw-delta",
+      sessionKey: "agent:main:missing-raw-delta",
+      storePath,
+    };
+    await upsertSessionEntry(missingScope, { sessionId: missingScope.sessionId, updatedAt: 10 });
+    await expect(
+      readSessionTranscriptRawDelta({ ...missingScope, maxBytes: 10, maxEvents: 1 }),
+    ).resolves.toEqual({ kind: "missing" });
+
+    const scope = {
+      ...missingScope,
+      sessionId: "oversized-raw-delta",
+      sessionKey: "agent:main:oversized-raw-delta",
+    };
+    await appendTranscriptEvent(scope, { id: "large", text: "x".repeat(200), type: "custom" });
+    const blocked = await readSessionTranscriptRawDelta({
+      ...scope,
+      cursor: "not-a-cursor",
+      maxBytes: 10,
+      maxEvents: 1,
+    });
+    expect(blocked).toMatchObject({ kind: "reset", reason: "invalid_cursor" });
+    await expect(
+      readSessionTranscriptRawDelta({
+        ...scope,
+        cursor: "",
+        maxBytes: 10,
+        maxEvents: 1,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "invalid_cursor" });
+    if (blocked.kind !== "reset") {
+      throw new Error("expected invalid cursor reset");
+    }
+    const unsafeCursor = Buffer.from(
+      JSON.stringify({
+        agentId: scope.agentId,
+        generation: "untrusted",
+        lastSeq: 1e100,
+        sessionId: scope.sessionId,
+        version: 1,
+      }),
+      "utf8",
+    ).toString("base64url");
+    await expect(
+      readSessionTranscriptRawDelta({
+        ...scope,
+        cursor: unsafeCursor,
+        maxBytes: 10,
+        maxEvents: 1,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "invalid_cursor" });
+
+    const bounded = await readSessionTranscriptRawDelta({
+      ...scope,
+      cursor: blocked.cursor,
+      maxBytes: 10,
+      maxEvents: 1,
+    });
+    expect(bounded).toMatchObject({
+      kind: "page",
+      events: [],
+      hasMore: true,
+      requiredBytes: expect.any(Number),
+      serializedBytes: 0,
+    });
+    if (bounded.kind !== "page" || bounded.requiredBytes === undefined) {
+      throw new Error("expected oversized event metadata");
+    }
+
+    const retried = await readSessionTranscriptRawDelta({
+      ...scope,
+      cursor: bounded.cursor,
+      maxBytes: bounded.requiredBytes,
+      maxEvents: 1,
+    });
+    expect(retried).toMatchObject({
+      kind: "page",
+      events: [{ event: { id: "large", text: "x".repeat(200), type: "custom" }, seq: 0 }],
+      hasMore: false,
+      serializedBytes: bounded.requiredBytes,
+    });
+
+    const otherScope = {
+      ...scope,
+      sessionId: "other-raw-delta",
+      sessionKey: "agent:main:other-raw-delta",
+    };
+    await appendTranscriptEvent(otherScope, { id: "other", type: "custom" });
+    await expect(
+      readSessionTranscriptRawDelta({
+        ...otherScope,
+        cursor: retried.kind === "page" ? retried.cursor : "",
+        maxBytes: 1_000,
+        maxEvents: 1,
+      }),
+    ).resolves.toMatchObject({ kind: "reset", reason: "scope_mismatch" });
+  });
+
  it("projects only visible transcript message entries with read-order metadata", async () => {
    const scope = {
      agentId: "main",
@@ -1,9 +1,12 @@
 import { redactTranscriptMessage } from "../agents/transcript-redact.js";
 import {
  appendTranscriptMessage,
+  isSessionTranscriptProjectionUnavailableError,
  loadSessionEntry,
  loadTranscriptEvents,
  publishTranscriptUpdate,
+  readTranscriptRawDelta,
+  readSessionTranscriptVisibleMessageDelta as readVisibleMessageDelta,
  readLatestTranscriptAssistantText,
  resolveSessionTranscriptRuntimeReadTarget,
  resolveSessionTranscriptRuntimeTarget,
@@ -11,6 +14,9 @@ import {
  type TranscriptMessageAppendOptions,
  type TranscriptMessageAppendResult,
  type TranscriptUpdatePayload,
+  type SessionTranscriptRawDeltaLimits,
+  type SessionTranscriptRawDeltaResult,
+  type SessionTranscriptVisibleMessageDeltaLimits,
 } from "../config/sessions/session-accessor.js";
 import { resolveMirroredTranscriptText } from "../config/sessions/transcript-mirror.js";
 import {
@@ -58,6 +64,45 @@ export type SessionTranscriptEvent = unknown;

 export type SessionTranscriptTargetParams = SessionTranscriptReadParams;

+/** Scoped target and bounds for one raw generation-aware transcript page. */
+export type SessionTranscriptRawDeltaParams = SessionTranscriptTargetParams &
+  SessionTranscriptRawDeltaLimits;
+export type { SessionTranscriptRawDeltaResult };
+
+/** Scoped target and bounds for one active-path visible-message page. */
+export type SessionTranscriptVisibleMessageDeltaParams = SessionTranscriptTargetParams &
+  SessionTranscriptVisibleMessageDeltaLimits;
+
+/** Generation-aware outcome for one bounded visible-message read. */
+export type SessionTranscriptVisibleMessageDeltaResult =
+  | {
+      kind: "page";
+      /** Opaque cursor positioned after the last returned visible message. */
+      cursor: string;
+      /** Ordered active-path message entries selected for this page. */
+      entries: SessionTranscriptMessageEntry[];
+      /** True when another visible message remains after this page. */
+      hasMore: boolean;
+      /** First unread event size when it cannot fit under maxBytes. */
+      requiredBytes?: number;
+      /** Stored JSONL bytes represented by entries. */
+      serializedBytes: number;
+    }
+  | {
+      kind: "reset";
+      /** Fresh opaque bootstrap cursor for the current visible generation. */
+      cursor: string;
+      /** Stable discontinuity that invalidated the supplied cursor. */
+      reason:
+        | "anchor_missing"
+        | "anchor_moved"
+        | "generation_mismatch"
+        | "invalid_cursor"
+        | "scope_mismatch";
+    }
+  | { kind: "unavailable"; reason: "projection_rebuilding" }
+  | { kind: "missing" };
+
 export type SessionTranscriptMessageEntry = {
  /** Stable transcript event id for this message entry. */
  entryId: string;
@@ -147,6 +192,52 @@ export async function readSessionTranscriptEvents(
  return await loadTranscriptEvents(params);
 }

+/** Reads one bounded raw page; the opaque cursor survives append and resets after replacement. */
+export async function readSessionTranscriptRawDelta(
+  params: SessionTranscriptRawDeltaParams,
+): Promise<SessionTranscriptRawDeltaResult> {
+  const { cursor, maxBytes, maxEvents, ...target } = params;
+  return readTranscriptRawDelta(target, {
+    ...(cursor !== undefined ? { cursor } : {}),
+    ...(maxBytes !== undefined ? { maxBytes } : {}),
+    ...(maxEvents !== undefined ? { maxEvents } : {}),
+  });
+}
+
+/** Reads one bounded active-path page that resumes appends and resets after discontinuities. */
+export async function readSessionTranscriptVisibleMessageDelta(
+  params: SessionTranscriptVisibleMessageDeltaParams,
+): Promise<SessionTranscriptVisibleMessageDeltaResult> {
+  const { cursor, maxBytes, maxMessages, ...target } = params;
+  let result: ReturnType<typeof readVisibleMessageDelta>;
+  try {
+    result = readVisibleMessageDelta(target, {
+      ...(cursor !== undefined ? { cursor } : {}),
+      ...(maxBytes !== undefined ? { maxBytes } : {}),
+      ...(maxMessages !== undefined ? { maxMessages } : {}),
+    });
+  } catch (error) {
+    if (isSessionTranscriptProjectionUnavailableError(error)) {
+      return { kind: "unavailable", reason: "projection_rebuilding" };
+    }
+    throw error;
+  }
+  if (result.kind !== "page") {
+    return result;
+  }
+  const { events, ...page } = result;
+  return {
+    ...page,
+    entries: events.flatMap((entry) =>
+      projectVisibleMessageEntry({
+        event: entry.event,
+        parentId: entry.parentId,
+        seq: entry.seq,
+      }),
+    ),
+  };
+}
+
 /**
 * Reads visible transcript message entries by scoped identity.
 *
@@ -2,6 +2,7 @@ import type { DatabaseSync } from "node:sqlite";
 import type { SqliteWalMaintenance } from "../infra/sqlite-wal.js";
 import type { OpenClawStateDatabaseOptions } from "./openclaw-state-db.js";

+// v13 = one durable generation token per raw session transcript.
 // v12 = session-owned ACP parent-stream events.
 // v11 = agent-scoped runtime leases, durable delivery operations, canonical
 // external conversation addresses, and bounded per-session heartbeat outcome context.
@@ -13,7 +14,7 @@ import type { OpenClawStateDatabaseOptions } from "./openclaw-state-db.js";
 // The v4 session/transcript flip and main's v2 memory-identity
 // change is folded in structure-gated migrations, so v2 main DBs and
 // pre-merge v4 flip DBs both converge on this schema.
-export const OPENCLAW_AGENT_SCHEMA_VERSION = 12;
+export const OPENCLAW_AGENT_SCHEMA_VERSION = 13;

 /** Open per-agent SQLite database handle plus lifecycle maintenance. */
 export type OpenClawAgentDatabase = {
@@ -251,6 +251,19 @@ function migrateOpenClawAgentSchema(db: DatabaseSync): void {
  backfillTranscriptMutationWatermarks(db);
 }

+/** Backfill one generation token without copying or rewriting transcript rows. */
+function migrateSessionTranscriptGenerations(db: DatabaseSync, previousVersion: number): void {
+  if (previousVersion >= 13) {
+    return;
+  }
+  db.prepare(
+    `INSERT OR IGNORE INTO session_transcript_generations (session_id, generation, updated_at)
+     SELECT session_id, lower(hex(randomblob(16))), ?
+     FROM transcript_events
+     GROUP BY session_id`,
+  ).run(Date.now());
+}
+
 function migrateSessionTranscriptActiveProjection(db: DatabaseSync, previousVersion: number): void {
  if (previousVersion >= 10) {
    return;
@@ -500,6 +513,7 @@ function ensureAgentSchema(db: DatabaseSync, agentId: string, pathname: string):
      migrateMemoryIndexSourcesIdentity(db);
      migrateOpenClawAgentSchema(db);
      db.exec(OPENCLAW_AGENT_SCHEMA_SQL);
+      migrateSessionTranscriptGenerations(db, previousVersion);
      migrateConversationDeliveryTargetColumn(db);
      migrateSessionTranscriptActiveProjection(db, previousVersion);
      if (previousVersion < 11) {
@@ -212,6 +212,12 @@ export interface SessionTranscriptFtsIdx {
  term: string;
 }

+export interface SessionTranscriptGenerations {
+  generation: string;
+  session_id: string;
+  updated_at: number;
+}
+
 export interface SessionTranscriptIndexState {
  active_event_count: Generated<number>;
  active_message_count: Generated<number>;
@@ -309,6 +315,7 @@ export interface DB {
  session_transcript_fts_data: SessionTranscriptFtsData;
  session_transcript_fts_docsize: SessionTranscriptFtsDocsize;
  session_transcript_fts_idx: SessionTranscriptFtsIdx;
+  session_transcript_generations: SessionTranscriptGenerations;
  session_transcript_index_state: SessionTranscriptIndexState;
  sessions: Sessions;
  state_leases: StateLeases;
@@ -568,8 +568,8 @@ describe("openclaw agent database", () => {
    expect(registered?.sizeBytes).toBeGreaterThan(0);
  });

-  it("keeps additive heartbeat repair while upgrading schema version 11", () => {
-    expect(OPENCLAW_AGENT_SCHEMA_VERSION).toBe(12);
+  it("keeps additive heartbeat repair while upgrading schema version 12", () => {
+    expect(OPENCLAW_AGENT_SCHEMA_VERSION).toBe(13);
    const stateDir = createTempStateDir();
    const env = { OPENCLAW_STATE_DIR: stateDir };
    const opened = openOpenClawAgentDatabase({ agentId: "worker-1", env });
@@ -577,13 +577,13 @@ describe("openclaw agent database", () => {
    closeOpenClawAgentDatabasesForTest();

    const { DatabaseSync } = requireNodeSqlite();
-    const existingV11 = new DatabaseSync(databasePath);
-    existingV11.exec(`
+    const existingV12 = new DatabaseSync(databasePath);
+    existingV12.exec(`
      DROP TABLE heartbeat_outcomes;
-      PRAGMA user_version = 11;
-      UPDATE schema_meta SET schema_version = 11 WHERE meta_key = 'primary';
+      PRAGMA user_version = 12;
+      UPDATE schema_meta SET schema_version = 12 WHERE meta_key = 'primary';
    `);
-    existingV11.close();
+    existingV12.close();

    const reopened = openOpenClawAgentDatabase({ agentId: "worker-1", env });
    expect(
@@ -599,6 +599,52 @@ describe("openclaw agent database", () => {
    ).toEqual({ schema_version: OPENCLAW_AGENT_SCHEMA_VERSION });
  });

+  it("backfills one generation per existing transcript when upgrading v12", () => {
+    const stateDir = createTempStateDir();
+    const env = { OPENCLAW_STATE_DIR: stateDir };
+    const opened = openOpenClawAgentDatabase({ agentId: "worker-1", env });
+    const databasePath = opened.path;
+    closeOpenClawAgentDatabasesForTest();
+    closeOpenClawStateDatabaseForTest();
+
+    const { DatabaseSync } = requireNodeSqlite();
+    const legacy = new DatabaseSync(databasePath);
+    legacy.exec(`
+      DROP TABLE IF EXISTS session_transcript_generations;
+      INSERT INTO sessions (session_id, session_key, created_at, updated_at)
+      VALUES
+        ('with-transcript', 'agent:worker-1:with-transcript', 10, 20),
+        ('without-transcript', 'agent:worker-1:without-transcript', 10, 20);
+      INSERT INTO transcript_events (session_id, seq, event_json, created_at)
+      VALUES ('with-transcript', 0, '{"type":"session","id":"with-transcript"}', 10);
+      PRAGMA user_version = 12;
+      UPDATE schema_meta SET schema_version = 12 WHERE meta_key = 'primary';
+    `);
+    legacy.close();
+
+    const migrated = openOpenClawAgentDatabase({ agentId: "worker-1", env });
+    const generations = migrated.db
+      .prepare(
+        "SELECT session_id, generation FROM session_transcript_generations ORDER BY session_id",
+      )
+      .all() as Array<{ generation: string; session_id: string }>;
+
+    expect(generations).toHaveLength(1);
+    expect(generations[0]?.session_id).toBe("with-transcript");
+    expect(generations[0]?.generation).toMatch(/^[0-9a-f]{32}$/);
+    expect(
+      migrated.db
+        .prepare("SELECT strict FROM pragma_table_list WHERE name = ?")
+        .get("session_transcript_generations"),
+    ).toEqual({ strict: 1 });
+    expect(readSqliteNumberPragma(migrated.db, "user_version")).toBe(OPENCLAW_AGENT_SCHEMA_VERSION);
+    expect(
+      migrated.db
+        .prepare("SELECT schema_version FROM schema_meta WHERE meta_key = 'primary'")
+        .get(),
+    ).toEqual({ schema_version: OPENCLAW_AGENT_SCHEMA_VERSION });
+  });
+
  it("upgrades version 10 with agent state intact and adds lease storage", () => {
    const stateDir = createTempStateDir();
    const env = { OPENCLAW_STATE_DIR: stateDir };
@@ -662,6 +708,11 @@ describe("openclaw agent database", () => {
    legacy.exec(`
      DROP INDEX idx_agent_acp_parent_stream_run;
      DROP TABLE acp_parent_stream_events;
+      DROP TABLE session_transcript_generations;
+      INSERT INTO sessions (session_id, session_key, created_at, updated_at)
+      VALUES ('with-transcript', 'agent:worker-1:with-transcript', 10, 20);
+      INSERT INTO transcript_events (session_id, seq, event_json, created_at)
+      VALUES ('with-transcript', 0, '{"type":"session","id":"with-transcript"}', 10);
      PRAGMA user_version = 11;
      UPDATE schema_meta SET schema_version = 11 WHERE meta_key = 'primary';
    `);
@@ -683,6 +734,14 @@ describe("openclaw agent database", () => {
        )
        .get(),
    ).toEqual({ name: "acp_parent_stream_events" });
+    expect(
+      migrated.db
+        .prepare("SELECT session_id, generation FROM session_transcript_generations")
+        .get(),
+    ).toMatchObject({
+      session_id: "with-transcript",
+      generation: expect.stringMatching(/^[0-9a-f]{32}$/),
+    });
  });

  it("migrates version 8 tables to STRICT without losing agent state", () => {
@@ -1858,6 +1917,13 @@ describe("openclaw agent database", () => {
        )
        .get(),
    ).toEqual({ name: "session_transcript_active_events" });
+    expect(
+      database.db
+        .prepare(
+          "SELECT length(generation) AS generation_length FROM session_transcript_generations WHERE session_id = ?",
+        )
+        .get("session-1"),
+    ).toEqual({ generation_length: 32 });
    expect(readSqliteNumberPragma(database.db, "user_version")).toBe(OPENCLAW_AGENT_SCHEMA_VERSION);
  });

@@ -208,6 +208,13 @@ CREATE TABLE IF NOT EXISTS transcript_events (
  FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
 ) STRICT;

+CREATE TABLE IF NOT EXISTS session_transcript_generations (
+  session_id TEXT NOT NULL PRIMARY KEY,
+  generation TEXT NOT NULL,
+  updated_at INTEGER NOT NULL,
+  FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
+) STRICT;
+
 CREATE TABLE IF NOT EXISTS trajectory_runtime_events (
  session_id TEXT NOT NULL,
  seq INTEGER NOT NULL,
@@ -203,6 +203,13 @@ CREATE TABLE IF NOT EXISTS transcript_events (
  FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
 ) STRICT;

+CREATE TABLE IF NOT EXISTS session_transcript_generations (
+  session_id TEXT NOT NULL PRIMARY KEY,
+  generation TEXT NOT NULL,
+  updated_at INTEGER NOT NULL,
+  FOREIGN KEY (session_id) REFERENCES sessions(session_id) ON DELETE CASCADE
+) STRICT;
+
 CREATE TABLE IF NOT EXISTS trajectory_runtime_events (
  session_id TEXT NOT NULL,
  seq INTEGER NOT NULL,
@@ -209,6 +209,45 @@ describe("sqlite hot query plans", () => {
    );
    expect(latestMessagePlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");

+    expectPlanIncludes({
+      db: database.db,
+      expected: "sqlite_autoindex_session_transcript_generations_1",
+      params: ["session-1"],
+      sql: `
+        SELECT generation
+          FROM session_transcript_generations
+         WHERE session_id = ?
+      `,
+    });
+    const rawDeltaPlan = explainQueryPlan(
+      database.db,
+      `
+        SELECT seq, LENGTH(CAST(event_json AS BLOB)) + 1 AS serialized_bytes
+          FROM transcript_events
+         WHERE session_id = ? AND seq > ?
+         ORDER BY seq ASC
+         LIMIT 1001
+      `,
+      ["session-1", 90_000],
+    );
+    expect(rawDeltaPlan).toContain("sqlite_autoindex_transcript_events_1");
+    expect(rawDeltaPlan).not.toContain("SCAN transcript_events");
+    expect(rawDeltaPlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
+    const rawFrontierPlan = explainQueryPlan(
+      database.db,
+      `
+        SELECT seq
+          FROM transcript_events
+         WHERE session_id = ?
+         ORDER BY seq DESC
+         LIMIT 1
+      `,
+      ["session-1"],
+    );
+    expect(rawFrontierPlan).toContain("sqlite_autoindex_transcript_events_1");
+    expect(rawFrontierPlan).not.toContain("SCAN transcript_events");
+    expect(rawFrontierPlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
+
    const historyPagePlan = explainQueryPlan(
      database.db,
      `
@@ -228,6 +267,55 @@ describe("sqlite hot query plans", () => {
    expect(historyPagePlan).toContain("sqlite_autoindex_transcript_events_1");
    expect(historyPagePlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");

+    const visibleDeltaPlan = explainQueryPlan(
+      database.db,
+      `
+        SELECT active.event_seq, active.message_position,
+               LENGTH(CAST(event.event_json AS BLOB)) + 1 AS serialized_bytes
+          FROM session_transcript_active_events AS active
+          JOIN transcript_events AS event
+            ON event.session_id = active.session_id AND event.seq = active.event_seq
+         WHERE active.session_id = ?
+           AND active.message_position IS NOT NULL
+           AND active.message_position >= ?
+         ORDER BY active.message_position ASC
+         LIMIT 1001
+      `,
+      ["session-1", 100],
+    );
+    expect(visibleDeltaPlan).toContain("idx_agent_transcript_active_messages");
+    expect(visibleDeltaPlan).toContain("sqlite_autoindex_transcript_events_1");
+    expect(visibleDeltaPlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
+
+    const visibleDeltaPayloadPlan = explainQueryPlan(
+      database.db,
+      `
+        SELECT active.event_seq, active.message_position, event.event_json,
+               parent_identity.event_id AS parent_id
+          FROM session_transcript_active_events AS active
+          JOIN transcript_events AS event
+            ON event.session_id = active.session_id AND event.seq = active.event_seq
+          LEFT JOIN session_transcript_active_events AS parent_active
+            ON parent_active.session_id = active.session_id
+           AND parent_active.active_position = active.active_position - 1
+          LEFT JOIN transcript_event_identities AS parent_identity
+            ON parent_identity.session_id = parent_active.session_id
+           AND parent_identity.seq = parent_active.event_seq
+         WHERE active.session_id = ?
+           AND active.message_position >= ?
+           AND active.message_position < ?
+         ORDER BY active.message_position ASC
+      `,
+      ["session-1", 100, 125],
+    );
+    expect(visibleDeltaPayloadPlan).toContain("idx_agent_transcript_active_messages");
+    expect(visibleDeltaPayloadPlan).toContain("sqlite_autoindex_transcript_events_1");
+    expect(visibleDeltaPayloadPlan).toContain(
+      "sqlite_autoindex_session_transcript_active_events_1",
+    );
+    expect(visibleDeltaPayloadPlan).toContain("idx_agent_transcript_event_sequence");
+    expect(visibleDeltaPayloadPlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
+
    const historyAnchorPlan = explainQueryPlan(
      database.db,
      `