From 6beaff450e6e9e8980c1bf1142d13d24b8742246 Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Wed, 29 Jul 2026 11:59:14 -0700 Subject: [PATCH] fix(codex): avoid transcript mirror snapshot churn (#115070) --- config/knip.config.ts | 1 + docs/plugins/sdk-subpaths.md | 2 + .../src/app-server/transcript-mirror.test.ts | 32 ++ .../codex/src/app-server/transcript-mirror.ts | 107 ++---- ...transcript-mirror.user-idempotency.test.ts | 44 ++- .../tsconfig.package-boundary.paths.json | 3 + extensions/xai/tsconfig.json | 3 + package.json | 4 + scripts/bench-codex-transcript-mirror.ts | 349 ++++++++++++++++++ scripts/lib/plugin-sdk-entrypoints.json | 1 + ...lugin-sdk-private-local-only-subpaths.json | 1 + scripts/stage-bundled-plugin-runtime.mjs | 1 + .../session-accessor.sqlite-import.ts | 110 ++++++ .../sessions/session-accessor.sqlite-read.ts | 19 +- .../sessions/session-accessor.sqlite-scope.ts | 1 + ...ssion-accessor.sqlite-transcript-mirror.ts | 148 ++++++++ ...ession-accessor.sqlite-transcript-write.ts | 151 +++----- .../sessions/session-accessor.sqlite.ts | 2 +- src/config/sessions/session-accessor.types.ts | 12 + .../codex-session-transcript-runtime.ts | 60 +++ .../session-transcript-lock-runtime.ts | 114 ++++++ .../session-transcript-mirror-runtime.test.ts | 154 ++++++++ .../session-transcript-runtime.test.ts | 2 + src/plugin-sdk/session-transcript-runtime.ts | 38 +- src/plugins/sdk-alias.ts | 6 +- src/state/sqlite-query-plan.test.ts | 28 ++ tsdown.config.ts | 3 + 27 files changed, 1150 insertions(+), 246 deletions(-) create mode 100644 scripts/bench-codex-transcript-mirror.ts create mode 100644 src/config/sessions/session-accessor.sqlite-import.ts create mode 100644 src/config/sessions/session-accessor.sqlite-transcript-mirror.ts create mode 100644 src/plugin-sdk/codex-session-transcript-runtime.ts create mode 100644 src/plugin-sdk/session-transcript-lock-runtime.ts create mode 100644 src/plugin-sdk/session-transcript-mirror-runtime.test.ts diff --git a/config/knip.config.ts b/config/knip.config.ts index 0bd2b89a01b1..e87f55d13462 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -111,6 +111,7 @@ const rootEntries = [ // Workflow/package-script entrypoints are not imported from production modules. "scripts/openclaw-cross-os-release-checks.ts!", "scripts/bench-transcript-cursors.ts!", + "scripts/bench-codex-transcript-mirror.ts!", "scripts/bench-sqlite-reliability.ts!", // Docker/manual E2E executables and their nested assertion/probe entrypoints. "scripts/e2e/*.{js,mjs,ts}!", diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index b5a3c2aff8be..70d02cc66517 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -221,6 +221,7 @@ Use `isLoopbackHost(host)` when a plugin must accept only the local machine. It | `plugin-sdk/browser-config` | Private-local after July 2026; Supported browser config facade for normalized profile/defaults, CDP URL parsing, and browser-control auth helpers | | `plugin-sdk/agent-harness-task-runtime` | Private-local after July 2026; Generic task lifecycle and completion delivery helpers for harness-backed agents using a host-issued task scope | | `plugin-sdk/codex-mcp-projection` | Private-local after July 2026; Reserved bundled Codex helper for projecting user MCP server config into Codex thread config; not for third-party plugins | + | `plugin-sdk/codex-session-transcript-runtime` | Private-local bundled Codex helper for serializing transcript-mirror writes; not for third-party plugins | | `plugin-sdk/channel-runtime-context` | Generic channel runtime-context registration and lookup helpers | | `plugin-sdk/matrix` | Deprecated Matrix compatibility facade for older third-party channel packages; new plugins should import `plugin-sdk/run-command` directly | | `plugin-sdk/runtime-store` | `createPluginRuntimeStore` | @@ -390,6 +391,7 @@ Use `isLoopbackHost(host)` when a plugin must accept only the local machine. It | Subpath | Owner and purpose | | --- | --- | | `plugin-sdk/codex-mcp-projection` | Private-local after July 2026; Bundled Codex plugin helper for projecting user MCP server config into Codex app-server thread config (reserved package export) | + | `plugin-sdk/codex-session-transcript-runtime` | Private-local bundled Codex plugin helper for serializing transcript-mirror writes (reserved package export) | diff --git a/extensions/codex/src/app-server/transcript-mirror.test.ts b/extensions/codex/src/app-server/transcript-mirror.test.ts index 1932a1db7e8c..ff05384f7def 100644 --- a/extensions/codex/src/app-server/transcript-mirror.test.ts +++ b/extensions/codex/src/app-server/transcript-mirror.test.ts @@ -926,6 +926,38 @@ describe("mirrorCodexAppServerTranscript", () => { expect((await readMirrorMessages(target)).filter((message) => message.role)).toHaveLength(2); }); + it("serializes concurrent mirrors with the same supplied identity", async () => { + const target = await createSqliteMirrorTarget("openclaw-codex-mirror-concurrent-"); + const message = attachCodexMirrorIdentity( + makeAgentUserMessage({ + content: [{ type: "text", text: "append once" }], + timestamp: Date.now(), + }), + "turn-1:prompt", + ); + + const results = await Promise.all([ + mirrorCodexAppServerTranscript({ + ...target, + messages: [message], + idempotencyScope: "codex-app-server:thread-1", + }), + mirrorCodexAppServerTranscript({ + ...target, + messages: [message], + idempotencyScope: "codex-app-server:thread-1", + }), + ]); + + expect((await readMirrorMessages(target)).filter((entry) => entry.role)).toEqual([ + { role: "user", text: "append once" }, + ]); + expect(results.map((result) => messageContent(result.userMessagesPresent[0]))).toEqual([ + [{ type: "text", text: "append once" }], + [{ type: "text", text: "append once" }], + ]); + }); + it("reports final assistant ownership for new and idempotent mirrors", async () => { const target = await createSqliteMirrorTarget("openclaw-codex-mirror-assistant-owned-"); const assistantMessage = attachCodexMirrorIdentity( diff --git a/extensions/codex/src/app-server/transcript-mirror.ts b/extensions/codex/src/app-server/transcript-mirror.ts index 04314e70ffc7..6e524eb87e16 100644 --- a/extensions/codex/src/app-server/transcript-mirror.ts +++ b/extensions/codex/src/app-server/transcript-mirror.ts @@ -8,10 +8,10 @@ import { type AgentMessage, type EmbeddedRunAttemptParams, } from "openclaw/plugin-sdk/agent-harness-runtime"; +import { withCodexSessionTranscriptMirrorWriteLock } from "openclaw/plugin-sdk/codex-session-transcript-runtime"; import type { AssistantMessage, Usage } from "openclaw/plugin-sdk/llm"; import { publishSessionTranscriptUpdateByIdentity, - withSessionTranscriptWriteLock, type SessionTranscriptTargetParams, type SessionTranscriptWriteLockParams, } from "openclaw/plugin-sdk/session-transcript-runtime"; @@ -538,34 +538,41 @@ async function mirror(params: { return { assistantMirrorIdentitiesOwned: [], messagesPresent: [], userMessagesPresent: [] }; } + const candidates = messages.map((message) => { + const dedupeIdentity = buildMirrorDedupeIdentity(message); + const sourceFingerprint = fingerprintCodexMirrorSourceMessage(message); + const sourceUserIdempotencyKey = + message.role === "user" + ? normalizeOptionalString( + (message as unknown as { idempotencyKey?: unknown }).idempotencyKey, + ) + : undefined; + // Gateway-owned user keys keep optimistic client rows stable. Other rows use + // the provider mirror identity so retries find the exact logical message. + const idempotencyKey = + sourceUserIdempotencyKey ?? + (params.idempotencyScope ? `${params.idempotencyScope}:${dedupeIdentity}` : undefined); + return { dedupeIdentity, idempotencyKey, message, sourceFingerprint }; + }); + const candidateIdempotencyKeys = candidates.flatMap(({ idempotencyKey }) => + idempotencyKey ? [idempotencyKey] : [], + ); const transcriptTarget = resolveCodexMirrorTranscriptTarget(params); - const mirrorBatch = await withSessionTranscriptWriteLock( + const mirrorBatch = await withCodexSessionTranscriptMirrorWriteLock( { ...transcriptTarget, config: params.config }, async (transcript) => { const nextAppendedUpdates: Array<{ messageId: string; message: AgentMessage; - messageSeq: number; + messageSeq?: number; }> = []; const nextAssistantMirrorIdentitiesOwned = new Set(); const nextMessagesPresent: MirroredAgentMessage[] = []; const nextUserMessagesPresent: MirroredUserMessage[] = []; - const mirrorState = readTranscriptMirrorState(await transcript.readEvents()); - let nextMessageSeq = mirrorState.messageCount; - for (const message of messages) { - const dedupeIdentity = buildMirrorDedupeIdentity(message); - const sourceFingerprint = fingerprintCodexMirrorSourceMessage(message); - const sourceUserIdempotencyKey = - message.role === "user" - ? normalizeOptionalString( - (message as unknown as { idempotencyKey?: unknown }).idempotencyKey, - ) - : undefined; - // The gateway owns user-turn identity. Preserve its key so clients can - // correlate optimistic rows; provider mirror identity is only a fallback. - const idempotencyKey = - sourceUserIdempotencyKey ?? - (params.idempotencyScope ? `${params.idempotencyScope}:${dedupeIdentity}` : undefined); + const mirrorFacts = await transcript.readMessageFacts({ + idempotencyKeys: candidateIdempotencyKeys, + }); + for (const { dedupeIdentity, idempotencyKey, message, sourceFingerprint } of candidates) { const transcriptMessage = { ...(attachCodexMirrorAttestation(message, sourceFingerprint) as unknown as Record< string, @@ -573,15 +580,13 @@ async function mirror(params: { >), ...(idempotencyKey ? { idempotencyKey } : {}), } as AgentMessage; - if (idempotencyKey && mirrorState.idempotencyKeys.has(idempotencyKey)) { - const persistedMessage = mirrorState.messagesByIdempotencyKey.get(idempotencyKey); - if (persistedMessage) { + if (idempotencyKey && mirrorFacts.existingIdempotencyKeys.has(idempotencyKey)) { + const persistedMessage = mirrorFacts.messagesByIdempotencyKey.get(idempotencyKey); + if (persistedMessage && isMirroredAgentMessage(persistedMessage)) { nextMessagesPresent.push(persistedMessage); - } - const persistedUserMessage = - persistedMessage?.role === "user" ? persistedMessage : undefined; - if (persistedUserMessage) { - nextUserMessagesPresent.push(persistedUserMessage); + if (persistedMessage.role === "user") { + nextUserMessagesPresent.push(persistedMessage); + } } if (message.role === "assistant") { nextAssistantMirrorIdentitiesOwned.add(dedupeIdentity); @@ -625,9 +630,11 @@ async function mirror(params: { hidden: (message as { display?: boolean }).display === false, message: messageToAppend, }); - const appended = await transcript.appendMessage({ + const { messageSeq, result: appended } = await transcript.appendMessageWithMessageSequence({ message: messageToAppend, - idempotencyLookup: idempotencyKey && message.role !== "user" ? "caller-checked" : "scan", + // Preliminary facts avoid hooks and payload work on normal retries. + // SQLite repeats this lookup under BEGIN IMMEDIATE for cross-process safety. + idempotencyLookup: "scan", cwd: params.cwd, }); if (!appended) { @@ -637,7 +644,7 @@ async function mirror(params: { if (isMirroredAgentMessage(appendedMessage)) { nextMessagesPresent.push(appendedMessage); if (idempotencyKey) { - mirrorState.messagesByIdempotencyKey.set(idempotencyKey, appendedMessage); + mirrorFacts.messagesByIdempotencyKey.set(idempotencyKey, appendedMessage); } } if (message.role === "assistant") { @@ -646,16 +653,15 @@ async function mirror(params: { if (appendedMessage.role === "user") { nextUserMessagesPresent.push(appendedMessage); } - nextMessageSeq += 1; if (appended.appended) { nextAppendedUpdates.push({ messageId, message: appendedMessage, - messageSeq: nextMessageSeq, + ...(messageSeq !== undefined ? { messageSeq } : {}), }); } if (idempotencyKey) { - mirrorState.idempotencyKeys.add(idempotencyKey); + mirrorFacts.existingIdempotencyKeys.add(idempotencyKey); } } return { @@ -677,7 +683,7 @@ async function mirror(params: { ...(params.agentId ? { agentId: params.agentId } : {}), message: update.message, messageId: update.messageId, - messageSeq: update.messageSeq, + ...(update.messageSeq !== undefined ? { messageSeq: update.messageSeq } : {}), sessionKey: transcriptTarget.sessionKey, }, }); @@ -713,36 +719,3 @@ function resolveCodexMirrorTranscriptTarget(params: { storePath, }; } - -function readTranscriptMirrorState(events: unknown[]): { - idempotencyKeys: Set; - messagesByIdempotencyKey: Map; - messageCount: number; -} { - const idempotencyKeys = new Set(); - const messagesByIdempotencyKey = new Map(); - let messageCount = 0; - for (const event of events) { - if (!event || typeof event !== "object" || Array.isArray(event)) { - continue; - } - const parsed = event as { - message?: AgentMessage & { idempotencyKey?: unknown }; - type?: unknown; - }; - if (parsed.type === "message") { - messageCount += 1; - } - if (typeof parsed.message?.idempotencyKey === "string") { - idempotencyKeys.add(parsed.message.idempotencyKey); - if (isMirroredAgentMessage(parsed.message)) { - messagesByIdempotencyKey.set(parsed.message.idempotencyKey, parsed.message); - } - } - } - return { - idempotencyKeys, - messagesByIdempotencyKey, - messageCount, - }; -} diff --git a/extensions/codex/src/app-server/transcript-mirror.user-idempotency.test.ts b/extensions/codex/src/app-server/transcript-mirror.user-idempotency.test.ts index 625f32e0d315..e1d70e919e16 100644 --- a/extensions/codex/src/app-server/transcript-mirror.user-idempotency.test.ts +++ b/extensions/codex/src/app-server/transcript-mirror.user-idempotency.test.ts @@ -2,16 +2,14 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime"; +import type { CodexSessionTranscriptMirrorWriteLockContext } from "openclaw/plugin-sdk/codex-session-transcript-runtime"; import { initializeGlobalHookRunner, resetGlobalHookRunner, } from "openclaw/plugin-sdk/hook-runtime"; import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime"; import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime"; -import { - readSessionTranscriptEvents, - type SessionTranscriptWriteLockContext, -} from "openclaw/plugin-sdk/session-transcript-runtime"; +import { readSessionTranscriptEvents } from "openclaw/plugin-sdk/session-transcript-runtime"; import { castAgentMessage, makeAgentAssistantMessage, @@ -37,27 +35,37 @@ vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async (importOriginal) return { ...actual, publishSessionTranscriptUpdateByIdentity: transcriptRace.publish, - withSessionTranscriptWriteLock: async ( - params: Parameters[0], - run: Parameters[1], + }; +}); + +vi.mock("openclaw/plugin-sdk/codex-session-transcript-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + withCodexSessionTranscriptMirrorWriteLock: async ( + params: Parameters[0], + run: Parameters[1], ) => - await actual.withSessionTranscriptWriteLock(params, async (locked) => { + await actual.withCodexSessionTranscriptMirrorWriteLock(params, async (locked) => { const competingMessage = transcriptRace.competingMessage; if (!competingMessage) { return await run(locked); } transcriptRace.competingMessage = undefined; - const staleEvents = await locked.readEvents(); - await locked.appendMessage({ - message: competingMessage as AgentMessage, - idempotencyLookup: "scan", - }); - const intercepted: SessionTranscriptWriteLockContext = { + const intercepted: CodexSessionTranscriptMirrorWriteLockContext = { ...locked, - readEvents: async () => staleEvents, - appendMessage: async (options) => { + readMessageFacts: async (factParams) => { + const staleFacts = await locked.readMessageFacts(factParams); + await locked.appendMessage({ + message: competingMessage as AgentMessage, + idempotencyLookup: "scan", + }); + return staleFacts; + }, + appendMessageWithMessageSequence: async (options) => { transcriptRace.lookups.push(options.idempotencyLookup); - return await locked.appendMessage(options); + return await locked.appendMessageWithMessageSequence(options); }, }; return await run(intercepted); @@ -164,7 +172,7 @@ it("adopts a competing indexed user without duplicating writes or slowing assist }), ]); expect(result.assistantMirrorIdentitiesOwned).toEqual(["turn-1:assistant"]); - expect(transcriptRace.lookups).toEqual(["scan", "caller-checked"]); + expect(transcriptRace.lookups).toEqual(["scan", "scan"]); expect(transcriptRace.publish).toHaveBeenCalledTimes(1); expect(transcriptRace.publish).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/extensions/tsconfig.package-boundary.paths.json b/extensions/tsconfig.package-boundary.paths.json index 099a44ce7b64..e44cf06f8cf9 100644 --- a/extensions/tsconfig.package-boundary.paths.json +++ b/extensions/tsconfig.package-boundary.paths.json @@ -125,6 +125,9 @@ "openclaw/plugin-sdk/codex-mcp-projection": [ "../packages/plugin-sdk/dist/src/plugin-sdk/codex-mcp-projection.d.ts" ], + "openclaw/plugin-sdk/codex-session-transcript-runtime": [ + "../packages/plugin-sdk/dist/src/plugin-sdk/codex-session-transcript-runtime.d.ts" + ], "openclaw/plugin-sdk/agent-harness-task-runtime": [ "../packages/plugin-sdk/dist/src/plugin-sdk/agent-harness-task-runtime.d.ts" ], diff --git a/extensions/xai/tsconfig.json b/extensions/xai/tsconfig.json index 072b7680aab5..032ffd063092 100644 --- a/extensions/xai/tsconfig.json +++ b/extensions/xai/tsconfig.json @@ -123,6 +123,9 @@ "openclaw/plugin-sdk/codex-mcp-projection": [ "../../packages/plugin-sdk/dist/src/plugin-sdk/codex-mcp-projection.d.ts" ], + "openclaw/plugin-sdk/codex-session-transcript-runtime": [ + "../../packages/plugin-sdk/dist/src/plugin-sdk/codex-session-transcript-runtime.d.ts" + ], "openclaw/plugin-sdk/agent-harness-task-runtime": [ "../../packages/plugin-sdk/dist/src/plugin-sdk/agent-harness-task-runtime.d.ts" ], diff --git a/package.json b/package.json index 142def61a5c1..20240899d847 100644 --- a/package.json +++ b/package.json @@ -73,6 +73,7 @@ "!dist/plugin-sdk/cli-backend.d.ts", "!dist/plugin-sdk/cli-runtime.d.ts", "!dist/plugin-sdk/codex-mcp-projection.d.ts", + "!dist/plugin-sdk/codex-session-transcript-runtime.d.ts", "!dist/plugin-sdk/command-status-runtime.d.ts", "!dist/plugin-sdk/command-surface.d.ts", "!dist/plugin-sdk/concurrency-runtime.d.ts", @@ -719,6 +720,9 @@ "./plugin-sdk/codex-mcp-projection": { "default": "./dist/plugin-sdk/codex-mcp-projection.js" }, + "./plugin-sdk/codex-session-transcript-runtime": { + "default": "./dist/plugin-sdk/codex-session-transcript-runtime.js" + }, "./plugin-sdk/agent-harness-task-runtime": { "default": "./dist/plugin-sdk/agent-harness-task-runtime.js" }, diff --git a/scripts/bench-codex-transcript-mirror.ts b/scripts/bench-codex-transcript-mirror.ts new file mode 100644 index 000000000000..d0de49bc492a --- /dev/null +++ b/scripts/bench-codex-transcript-mirror.ts @@ -0,0 +1,349 @@ +// Benchmarks the real Codex transcript mirror against a large indexed SQLite transcript. +import { execFileSync } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { SQLInputValue } from "node:sqlite"; +import { codexTranscriptMirrorRuntime } from "../extensions/codex/src/app-server/transcript-mirror.js"; +import { attachCodexMirrorIdentity } from "../extensions/codex/src/app-server/upstream-prompt-provenance.js"; +import { upsertSessionEntry } from "../src/config/sessions/session-accessor.js"; +import type { AgentMessage } from "../src/plugin-sdk/agent-core.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../src/state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../src/state/openclaw-state-db.js"; + +const DEFAULT_EVENT_COUNT = 100_000; +const DEFAULT_PAYLOAD_BYTES = 64; +const DEFAULT_RUNS = 8; +const DEFAULT_WARMUPS = 2; +const NEW_MESSAGES_PER_OPERATION = 2; + +type MirrorTarget = { + agentId: string; + sessionId: string; + sessionKey: string; + storePath: string; +}; + +type WorkCounters = { + fullTranscriptQueries: number; + seededEventJsonParses: number; + selectQueries: number; +}; + +function readIntegerArg(name: string, fallback: number): number { + const raw = process.argv.find((arg) => arg.startsWith(`--${name}=`))?.slice(name.length + 3); + if (raw === undefined) { + return fallback; + } + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < 1) { + throw new Error(`--${name} must be a positive integer`); + } + return value; +} + +function readSourceSha(): string { + const value = process.argv + .find((arg) => arg.startsWith("--source-sha=")) + ?.slice("--source-sha=".length); + if (!value || !/^[a-f0-9]{40}$/u.test(value)) { + throw new Error("benchmark requires --source-sha=<40-character commit SHA>"); + } + const checkoutSha = execFileSync("git", ["rev-parse", "HEAD"], { + cwd: path.resolve(import.meta.dirname, ".."), + encoding: "utf8", + }).trim(); + if (checkoutSha !== value) { + throw new Error(`source SHA ${value} does not match checkout HEAD ${checkoutSha}`); + } + return value; +} + +function median(values: readonly number[]): number { + const sorted = values.toSorted((left, right) => left - right); + const upperIndex = Math.floor(sorted.length / 2); + const upper = sorted[upperIndex] ?? 0; + const lower = sorted.length % 2 === 0 ? (sorted[upperIndex - 1] ?? upper) : upper; + return Number(((lower + upper) / 2).toFixed(3)); +} + +function percentile(values: readonly number[], fraction: number): number { + const sorted = values.toSorted((left, right) => left - right); + const index = Math.min(sorted.length - 1, Math.ceil(sorted.length * fraction) - 1); + return Number((sorted[Math.max(0, index)] ?? 0).toFixed(3)); +} + +/** Seeds a fully indexed linear transcript without charging setup to measured owner calls. */ +function seedTranscript(params: { + database: ReturnType; + eventCount: number; + payloadText: string; + sessionId: string; +}): void { + const { database, eventCount, payloadText, sessionId } = params; + const insertEvent = database.db.prepare( + `INSERT INTO transcript_events (session_id, seq, event_json, created_at) + VALUES (?, ?, ?, ?)`, + ); + const insertIdentity = database.db.prepare( + `INSERT INTO transcript_event_identities ( + session_id, event_id, seq, event_type, parent_id, message_idempotency_key, created_at + ) VALUES (?, ?, ?, 'message', ?, ?, ?)`, + ); + const insertActive = database.db.prepare( + `INSERT INTO session_transcript_active_events ( + session_id, active_position, event_seq, message_position + ) VALUES (?, ?, ?, ?)`, + ); + const now = Date.now(); + database.db.exec("BEGIN IMMEDIATE"); + try { + for (let seq = 0; seq < eventCount; seq += 1) { + const eventId = `benchmark-event-${seq}`; + const parentId = seq === 0 ? null : `benchmark-event-${seq - 1}`; + const idempotencyKey = `seed:${sessionId}:${seq}`; + const role = seq % 2 === 0 ? "user" : "assistant"; + const event = { + id: eventId, + message: { + content: role === "user" ? payloadText : [{ type: "text", text: payloadText }], + idempotencyKey, + role, + timestamp: now + seq, + }, + parentId, + timestamp: now + seq, + type: "message", + }; + insertEvent.run(sessionId, seq, JSON.stringify(event), now + seq); + const identityValues = [ + sessionId, + eventId, + seq, + parentId, + idempotencyKey, + now + seq, + ] satisfies SQLInputValue[]; + insertIdentity.run(...identityValues); + insertActive.run(sessionId, seq, seq, seq); + } + database.db + .prepare( + `INSERT INTO session_transcript_index_state ( + session_id, indexed_seq, leaf_event_id, needs_rebuild, + active_event_count, active_message_count, updated_at + ) VALUES (?, ?, ?, 0, ?, ?, ?)`, + ) + .run( + sessionId, + eventCount - 1, + `benchmark-event-${eventCount - 1}`, + eventCount, + eventCount, + now + eventCount, + ); + database.db.exec("COMMIT"); + } catch (error) { + database.db.exec("ROLLBACK"); + throw error; + } +} + +function instrumentWork(database: ReturnType): { + counters: WorkCounters; + reset: () => void; + restore: () => void; +} { + const counters: WorkCounters = { + fullTranscriptQueries: 0, + seededEventJsonParses: 0, + selectQueries: 0, + }; + const originalPrepare = database.db.prepare.bind(database.db); + const originalParse = JSON.parse; + Object.defineProperty(database.db, "prepare", { + configurable: true, + value: (sql: string) => { + const normalized = sql.replaceAll(/\s+/gu, " ").trim().toLowerCase(); + if (normalized.startsWith("select ")) { + counters.selectQueries += 1; + } + if ( + /from "?transcript_events"?/u.test(normalized) && + normalized.includes("event_json") && + /order by "?seq"? asc/u.test(normalized) + ) { + counters.fullTranscriptQueries += 1; + } + return originalPrepare(sql); + }, + }); + JSON.parse = ((text: string, reviver?: Parameters[1]) => { + if (text.includes('"id":"benchmark-event-')) { + counters.seededEventJsonParses += 1; + } + return originalParse(text, reviver); + }) as typeof JSON.parse; + return { + counters, + reset: () => { + counters.fullTranscriptQueries = 0; + counters.seededEventJsonParses = 0; + counters.selectQueries = 0; + }, + restore: () => { + Object.defineProperty(database.db, "prepare", { + configurable: true, + value: originalPrepare, + }); + JSON.parse = originalParse; + }, + }; +} + +function buildPromptFinalBatch(ordinal: number): AgentMessage[] { + return [ + attachCodexMirrorIdentity( + { + role: "user", + content: `benchmark prompt ${ordinal}`, + timestamp: 2_000_000_000_000 + ordinal, + } as AgentMessage, + `turn-${ordinal}:prompt`, + ), + attachCodexMirrorIdentity( + { + role: "assistant", + content: [{ type: "text", text: `benchmark final ${ordinal}` }], + timestamp: 2_000_000_100_000 + ordinal, + } as AgentMessage, + `turn-${ordinal}:assistant`, + ), + ]; +} + +async function runMirror(target: MirrorTarget, ordinal: number): Promise { + await codexTranscriptMirrorRuntime.mirror({ + ...target, + idempotencyScope: "codex-app-server:benchmark", + messages: buildPromptFinalBatch(ordinal), + }); +} + +async function main(): Promise { + const sourceSha = readSourceSha(); + const eventCount = readIntegerArg("events", DEFAULT_EVENT_COUNT); + const payloadBytes = readIntegerArg("payload-bytes", DEFAULT_PAYLOAD_BYTES); + const runs = readIntegerArg("runs", DEFAULT_RUNS); + const warmups = readIntegerArg("warmups", DEFAULT_WARMUPS); + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-codex-mirror-bench-")); + const agentId = "benchmark"; + const sessionId = "codex-mirror-benchmark"; + const sessionKey = `agent:${agentId}:${sessionId}`; + try { + const database = openOpenClawAgentDatabase({ + agentId, + path: path.join(stateDir, "openclaw-agent.sqlite"), + }); + await upsertSessionEntry( + { agentId, sessionKey, storePath: database.path }, + { sessionId, updatedAt: 1 }, + ); + seedTranscript({ + database, + eventCount, + payloadText: "x".repeat(payloadBytes), + sessionId, + }); + const target = { agentId, sessionId, sessionKey, storePath: database.path }; + const instrumentation = instrumentWork(database); + try { + for (let ordinal = 0; ordinal < warmups; ordinal += 1) { + await runMirror(target, ordinal); + } + instrumentation.reset(); + const beforeMaxRssKb = process.resourceUsage().maxRSS; + const durations: number[] = []; + for (let run = 0; run < runs; run += 1) { + const startedAt = performance.now(); + await runMirror(target, warmups + run); + durations.push(performance.now() - startedAt); + } + const afterMaxRssKb = process.resourceUsage().maxRSS; + const measuredWork = { ...instrumentation.counters }; + const lastOrdinal = warmups + runs - 1; + await runMirror(target, lastOrdinal); + const row = database.db + .prepare( + `SELECT COUNT(*) AS count, + SUM(LENGTH(CAST(event_json AS BLOB))) AS bytes + FROM transcript_events + WHERE session_id = ?`, + ) + .get(sessionId) as { bytes: number; count: number }; + const expectedEvents = eventCount + NEW_MESSAGES_PER_OPERATION * (warmups + runs); + if (row.count !== expectedEvents) { + throw new Error(`mirror wrote ${row.count} events; expected ${expectedEvents}`); + } + console.log( + JSON.stringify( + { + sourceSha, + fixture: { + initialMessageEvents: eventCount, + payloadBytes, + sqliteTranscriptBytesAfterOperations: row.bytes, + }, + operation: "real Codex mirror owner with one new prompt and one new final", + runtime: { + arch: process.arch, + node: process.version, + platform: `${os.platform()} ${os.release()}`, + }, + warmups, + runs, + latencyMs: { + median: median(durations), + p95: percentile(durations, 0.95), + raw: durations.map((value) => Number(value.toFixed(3))), + }, + memoryProxy: { + maxRssKbBeforeOperations: beforeMaxRssKb, + maxRssKbAfterOperations: afterMaxRssKb, + maxRssGrowthKb: Math.max(0, afterMaxRssKb - beforeMaxRssKb), + }, + measuredWork: { + ...measuredWork, + perOperation: { + fullTranscriptQueries: Number( + (measuredWork.fullTranscriptQueries / runs).toFixed(3), + ), + seededEventJsonParses: Number( + (measuredWork.seededEventJsonParses / runs).toFixed(3), + ), + selectQueries: Number((measuredWork.selectQueries / runs).toFixed(3)), + }, + }, + correctness: { + idempotentReplayAddedRows: 0, + storedEventCount: row.count, + }, + }, + null, + 2, + ), + ); + } finally { + instrumentation.restore(); + } + } finally { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + fs.rmSync(stateDir, { force: true, recursive: true }); + } +} + +await main(); diff --git a/scripts/lib/plugin-sdk-entrypoints.json b/scripts/lib/plugin-sdk-entrypoints.json index 6dd5f53274d9..8982ac6523c0 100644 --- a/scripts/lib/plugin-sdk-entrypoints.json +++ b/scripts/lib/plugin-sdk-entrypoints.json @@ -101,6 +101,7 @@ "cli-runtime", "cli-backend", "codex-mcp-projection", + "codex-session-transcript-runtime", "agent-harness-task-runtime", "agent-harness", "agent-harness-exec-review-runtime", diff --git a/scripts/lib/plugin-sdk-private-local-only-subpaths.json b/scripts/lib/plugin-sdk-private-local-only-subpaths.json index aa18b0ee0724..890aefc7f65b 100644 --- a/scripts/lib/plugin-sdk-private-local-only-subpaths.json +++ b/scripts/lib/plugin-sdk-private-local-only-subpaths.json @@ -29,6 +29,7 @@ "cli-backend", "cli-runtime", "codex-mcp-projection", + "codex-session-transcript-runtime", "command-status-runtime", "command-surface", "concurrency-runtime", diff --git a/scripts/stage-bundled-plugin-runtime.mjs b/scripts/stage-bundled-plugin-runtime.mjs index 29e486ce2241..15172ddcdffd 100644 --- a/scripts/stage-bundled-plugin-runtime.mjs +++ b/scripts/stage-bundled-plugin-runtime.mjs @@ -77,6 +77,7 @@ function writeJsonFile(targetPath, value) { const PRIVATE_LOCAL_ONLY_PLUGIN_SDK_DIST_FILE_NAME_FALLBACK = [ "codex-mcp-projection.js", + "codex-session-transcript-runtime.js", "qa-channel.js", "qa-channel-protocol.js", "qa-lab.js", diff --git a/src/config/sessions/session-accessor.sqlite-import.ts b/src/config/sessions/session-accessor.sqlite-import.ts new file mode 100644 index 000000000000..f4e89d85ef29 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-import.ts @@ -0,0 +1,110 @@ +import { runOpenClawAgentWriteTransaction } from "../../state/openclaw-agent-db.js"; +import type { TranscriptEvent } from "./session-accessor.sqlite-contract.js"; +import { readSessionEntryRow, writeSessionEntry } from "./session-accessor.sqlite-entry-store.js"; +import { readTranscriptEventJsonSetInTransaction } from "./session-accessor.sqlite-read.js"; +import { + formatSqliteSessionReferenceForScope, + resolveSqliteScope, + runExclusiveSqliteSessionWrite, + toDatabaseOptions, +} from "./session-accessor.sqlite-scope.js"; +import { + advanceTranscriptMutationAtInTransaction, + touchTranscriptMutationInTransaction, +} from "./session-accessor.sqlite-transcript-state.js"; +import { appendTranscriptEventInTransaction } from "./session-accessor.sqlite-transcript-store.js"; +import { reconcileSessionTranscriptIndexInTransaction } from "./session-transcript-index.js"; +import type { SessionEntry } from "./types.js"; + +/** Internal doctor/migration import target for one legacy session row. */ +type SqliteSessionImportRowsParams = { + agentId?: string; + env?: NodeJS.ProcessEnv; + storePath?: string; + sessionKey: string; + entry: SessionEntry; + readTranscriptEvents?: (append: (event: TranscriptEvent) => void) => void; + transcriptMtimeMs?: number; +}; + +/** Summary of rows written by an internal doctor/migration import. */ +type SqliteSessionImportRowsResult = { + sessionId: string; + sessionKey: string; + transcriptEvents: number; +}; + +/** Imports one legacy session entry and its transcript rows for doctor migration. */ +export async function importSqliteSessionRows( + params: SqliteSessionImportRowsParams, +): Promise { + const resolved = resolveSqliteScope({ + ...(params.agentId ? { agentId: params.agentId } : {}), + ...(params.env ? { env: params.env } : {}), + sessionKey: params.sessionKey, + ...(params.storePath ? { storePath: params.storePath } : {}), + }); + return await runExclusiveSqliteSessionWrite(resolved, async () => { + let transcriptEvents = 0; + runOpenClawAgentWriteTransaction((database) => { + const currentEntry = readSessionEntryRow(database, resolved.sessionKey)?.entry; + const preservedHarnessId = + params.entry.agentHarnessId === undefined && + currentEntry?.sessionId === params.entry.sessionId && + currentEntry.lifecycleRevision === params.entry.lifecycleRevision + ? currentEntry.agentHarnessId?.trim() + : undefined; + // Plugin doctor migrations can claim a legacy session before the full + // session import runs. Preserve that same-generation canonical owner. + const importedEntry = { + ...params.entry, + ...(preservedHarnessId ? { agentHarnessId: preservedHarnessId } : {}), + sessionFile: formatSqliteSessionReferenceForScope({ + ...resolved, + sessionId: params.entry.sessionId, + }), + }; + writeSessionEntry(database, resolved.sessionKey, importedEntry); + if (params.readTranscriptEvents) { + const transcriptScope = { + ...resolved, + sessionId: params.entry.sessionId, + }; + const existingEventJson = readTranscriptEventJsonSetInTransaction( + database, + params.entry.sessionId, + ); + params.readTranscriptEvents((event) => { + const eventJson = JSON.stringify(event); + if (existingEventJson.has(eventJson)) { + return; + } + if ( + appendTranscriptEventInTransaction(database, transcriptScope, event, { + scheduleProjectionReconcile: false, + touchMutation: false, + }) + ) { + existingEventJson.add(eventJson); + transcriptEvents += 1; + } + }); + reconcileSessionTranscriptIndexInTransaction(database.db, params.entry.sessionId); + } + if (params.transcriptMtimeMs !== undefined) { + advanceTranscriptMutationAtInTransaction( + database, + params.entry.sessionId, + params.transcriptMtimeMs, + ); + } else if (transcriptEvents > 0) { + touchTranscriptMutationInTransaction(database, params.entry.sessionId); + } + }, toDatabaseOptions(resolved)); + return { + sessionId: params.entry.sessionId, + sessionKey: resolved.sessionKey, + transcriptEvents, + }; + }); +} diff --git a/src/config/sessions/session-accessor.sqlite-read.ts b/src/config/sessions/session-accessor.sqlite-read.ts index 30f39e3ff67e..67571c430be3 100644 --- a/src/config/sessions/session-accessor.sqlite-read.ts +++ b/src/config/sessions/session-accessor.sqlite-read.ts @@ -159,25 +159,14 @@ export function readSqliteTranscriptSnapshot( database: OpenClawAgentDatabase, sessionId: string, ): { events: TranscriptEvent[]; rows: SqliteTranscriptSnapshotRow[] } { - const db = getSessionKysely(database.db); - const rows = executeSqliteQuerySync( - database.db, - db - .selectFrom("transcript_events") - .select(["event_json", "seq"]) - .where("session_id", "=", sessionId) - .orderBy("seq", "asc"), - ).rows; + const rows = readSqliteTranscriptEventRows(database, sessionId); return { - events: rows.map((row) => JSON.parse(row.event_json) as TranscriptEvent), - rows: rows.map((row) => ({ - eventJson: row.event_json, - seq: normalizeSqliteNumber(row.seq), - })), + events: rows.map((row) => JSON.parse(row.eventJson) as TranscriptEvent), + rows, }; } -/** Reads transcript event rows without parsing JSON (tolerant of malformed rows). Used by migrations. */ +/** Reads transcript rows without decoding payloads for snapshot comparison. */ export function readSqliteTranscriptEventRows( database: OpenClawAgentDatabase, sessionId: string, diff --git a/src/config/sessions/session-accessor.sqlite-scope.ts b/src/config/sessions/session-accessor.sqlite-scope.ts index 5549b0650d15..ee3a52961a42 100644 --- a/src/config/sessions/session-accessor.sqlite-scope.ts +++ b/src/config/sessions/session-accessor.sqlite-scope.ts @@ -35,6 +35,7 @@ type SessionSqliteDatabase = Pick< | "session_members" | "session_nodes" | "session_suggestions" + | "session_transcript_index_state" | "session_windows" | "transcript_rewrite_watermarks" | "trajectory_runtime_events" diff --git a/src/config/sessions/session-accessor.sqlite-transcript-mirror.ts b/src/config/sessions/session-accessor.sqlite-transcript-mirror.ts new file mode 100644 index 000000000000..e4e1c5926a86 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-transcript-mirror.ts @@ -0,0 +1,148 @@ +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, +} from "../../infra/kysely-sync.js"; +import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js"; +import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; +import type { TranscriptEvent } from "./session-accessor.sqlite-contract.js"; +import { + loadSqliteTranscriptEventsFromDatabase, + readTranscriptEventMessage, +} from "./session-accessor.sqlite-read.js"; +import { getSessionKysely } from "./session-accessor.sqlite-scope.js"; +import { readMessageIdempotencyKey } from "./session-accessor.sqlite-transcript-store.js"; + +// Keep supplied-key probes below SQLite's conservative variable ceiling. +const TRANSCRIPT_MIRROR_KEY_QUERY_BATCH_SIZE = 900; + +type TranscriptMirrorFacts = { + existingIdempotencyKeys: Set; + messagesByIdempotencyKey: Map; +}; + +/** Returns raw events only when the transcript identity projection is not current. */ +function loadTranscriptEventsForMirrorFallback( + database: OpenClawAgentDatabase, + sessionId: string, +): TranscriptEvent[] | undefined { + const db = getSessionKysely(database.db); + const latest = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("transcript_events") + .select("seq") + .where("session_id", "=", sessionId) + .orderBy("seq", "desc") + .limit(1), + ); + if (!latest) { + return []; + } + const state = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("session_transcript_index_state") + .select(["indexed_seq", "needs_rebuild"]) + .where("session_id", "=", sessionId), + ); + if (state && state.needs_rebuild === 0 && state.indexed_seq === latest.seq) { + return undefined; + } + // Raw rows stay authoritative if projection maintenance has not caught up. + return loadSqliteTranscriptEventsFromDatabase(database, sessionId); +} + +/** Reads the bounded identity facts needed by transcript mirrors. */ +export function readTranscriptMirrorFacts( + database: OpenClawAgentDatabase, + sessionId: string, + params: { + idempotencyKeys: readonly string[]; + }, +): TranscriptMirrorFacts { + return runSqliteDeferredTransactionSync( + database.db, + () => readTranscriptMirrorFactsInSnapshot(database, sessionId, params), + { + databaseLabel: database.path, + operationLabel: "session.transcript.mirror-facts", + }, + ); +} + +/** Reads mirror facts after the caller has established one SQLite snapshot. */ +function readTranscriptMirrorFactsInSnapshot( + database: OpenClawAgentDatabase, + sessionId: string, + params: { + idempotencyKeys: readonly string[]; + }, +): TranscriptMirrorFacts { + const idempotencyKeys = [...new Set(params.idempotencyKeys)]; + const fallbackEvents = loadTranscriptEventsForMirrorFallback(database, sessionId); + if (fallbackEvents !== undefined) { + return readMirrorFactsFromEvents(fallbackEvents, new Set(idempotencyKeys)); + } + + const db = getSessionKysely(database.db); + const facts: TranscriptMirrorFacts = { + existingIdempotencyKeys: new Set(), + messagesByIdempotencyKey: new Map(), + }; + for ( + let offset = 0; + offset < idempotencyKeys.length; + offset += TRANSCRIPT_MIRROR_KEY_QUERY_BATCH_SIZE + ) { + const batch = idempotencyKeys.slice(offset, offset + TRANSCRIPT_MIRROR_KEY_QUERY_BATCH_SIZE); + const rows = executeSqliteQuerySync( + database.db, + db + .selectFrom("transcript_event_identities as identity") + .innerJoin("transcript_events as event", (join) => + join + .onRef("event.session_id", "=", "identity.session_id") + .onRef("event.seq", "=", "identity.seq"), + ) + .select(["identity.message_idempotency_key", "event.event_json"]) + .where("identity.session_id", "=", sessionId) + .where("identity.message_idempotency_key", "in", batch) + .orderBy("identity.seq", "asc"), + ).rows; + for (const row of rows) { + const idempotencyKey = row.message_idempotency_key; + if (!idempotencyKey) { + continue; + } + facts.existingIdempotencyKeys.add(idempotencyKey); + const message = readTranscriptEventMessage(JSON.parse(row.event_json) as TranscriptEvent); + if (message !== undefined) { + facts.messagesByIdempotencyKey.set(idempotencyKey, message); + } + } + } + return facts; +} + +/** Extracts supplied mirror identities from authoritative transcript events. */ +function readMirrorFactsFromEvents( + events: readonly TranscriptEvent[], + candidateKeys: ReadonlySet, +): TranscriptMirrorFacts { + const facts: TranscriptMirrorFacts = { + existingIdempotencyKeys: new Set(), + messagesByIdempotencyKey: new Map(), + }; + for (const event of events) { + const message = readTranscriptEventMessage(event); + const idempotencyKey = readMessageIdempotencyKey(message); + if (!idempotencyKey || !candidateKeys.has(idempotencyKey)) { + continue; + } + facts.existingIdempotencyKeys.add(idempotencyKey); + if (message !== undefined) { + facts.messagesByIdempotencyKey.set(idempotencyKey, message); + } + } + return facts; +} diff --git a/src/config/sessions/session-accessor.sqlite-transcript-write.ts b/src/config/sessions/session-accessor.sqlite-transcript-write.ts index 1539317a82f6..b023c0193f39 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-write.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-write.ts @@ -28,27 +28,25 @@ import { } from "./session-accessor.sqlite-entry-store.js"; import { emitCommittedSessionIdentityDiff } from "./session-accessor.sqlite-identity.js"; import { + readSqliteTranscriptEventRows, readSqliteTranscriptSnapshot, - readTranscriptEventJsonSetInTransaction, type SqliteTranscriptSnapshotRow, } from "./session-accessor.sqlite-read.js"; import { cloneSessionEntry, - formatSqliteSessionReferenceForScope, - resolveSqliteScope, resolveSqliteTranscriptArchiveDirectory, resolveSqliteTranscriptScope, runExclusiveSqliteSessionWrite, toDatabaseOptions, type ResolvedTranscriptScope, } from "./session-accessor.sqlite-scope.js"; +import { readTranscriptMirrorFacts } from "./session-accessor.sqlite-transcript-mirror.js"; import { resolveTranscriptMessageAppendParent } from "./session-accessor.sqlite-transcript-parent.js"; -import { rememberCommittedSqliteTranscriptMessageSequencesInTransaction } from "./session-accessor.sqlite-transcript-sequences.js"; import { - advanceTranscriptMutationAtInTransaction, - readTranscriptGenerationInTransaction, - touchTranscriptMutationInTransaction, -} from "./session-accessor.sqlite-transcript-state.js"; + readCommittedSqliteTranscriptMessageSequence, + rememberCommittedSqliteTranscriptMessageSequencesInTransaction, +} from "./session-accessor.sqlite-transcript-sequences.js"; +import { readTranscriptGenerationInTransaction } from "./session-accessor.sqlite-transcript-state.js"; import { appendTranscriptEventInTransaction, ensureTranscriptHeader, @@ -61,7 +59,6 @@ import { rewriteSqliteTranscriptEventRowsInTransaction, } from "./session-accessor.sqlite-transcript-store.js"; import type { SessionTranscriptWriteTransactionContext } from "./session-accessor.types.js"; -import { reconcileSessionTranscriptIndexInTransaction } from "./session-transcript-index.js"; import type { SessionTranscriptTurnExpectedState, SessionTranscriptTurnLifecyclePatch, @@ -83,24 +80,6 @@ class SqliteTranscriptMutationConflictError extends Error { } } -/** Internal doctor/migration import target for one legacy session row. */ -type SqliteSessionImportRowsParams = { - agentId?: string; - env?: NodeJS.ProcessEnv; - storePath?: string; - sessionKey: string; - entry: SessionEntry; - readTranscriptEvents?: (append: (event: TranscriptEvent) => void) => void; - transcriptMtimeMs?: number; -}; - -/** Summary of rows written by an internal doctor/migration import. */ -type SqliteSessionImportRowsResult = { - sessionId: string; - sessionKey: string; - transcriptEvents: number; -}; - type SqliteExpectedSessionTranscriptTurnResult = { appendedMessages: TranscriptMessageAppendResult[]; rejectedReason?: "session-rebound"; @@ -112,6 +91,16 @@ type SqliteTranscriptWriteLockContext = { appendMessage: ( options: TranscriptMessageAppendOptions, ) => Promise | undefined>; + appendMessageWithMessageSequence: ( + options: TranscriptMessageAppendOptions, + ) => Promise<{ + messageSeq?: number; + result: TranscriptMessageAppendResult | undefined; + }>; + readMessageFacts: (params: { idempotencyKeys: readonly string[] }) => Promise<{ + existingIdempotencyKeys: Set; + messagesByIdempotencyKey: Map; + }>; readEvents: () => Promise; replaceEvents: (events: readonly TranscriptEvent[]) => Promise; }; @@ -191,13 +180,13 @@ export async function trimSqliteTranscriptForManualCompact( const resolved = resolveSqliteTranscriptScope(scope); return await runExclusiveSqliteSessionWrite(resolved, async () => { const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - const snapshot = readSqliteTranscriptSnapshot(database, resolved.sessionId); + const snapshotRows = readSqliteTranscriptEventRows(database, resolved.sessionId); const sessionSnapshot = readSqliteSessionEntrySelectionSnapshot( database, resolved.sessionKey, true, ); - const lines = snapshot.rows.map((row) => row.eventJson); + const lines = snapshotRows.map((row) => row.eventJson); const retainedLines = selectRetainedLines(lines); if (!retainedLines) { return { trimmed: false }; @@ -219,7 +208,7 @@ export async function trimSqliteTranscriptForManualCompact( let previousIdentity = new Map(); let currentIdentity = new Map(); runOpenClawAgentWriteTransaction((writeDatabase) => { - assertSqliteTranscriptSnapshotUnchanged(writeDatabase, resolved.sessionId, snapshot.rows); + assertSqliteTranscriptSnapshotUnchanged(writeDatabase, resolved.sessionId, snapshotRows); const freshSessionSnapshot = readSqliteSessionEntrySelectionSnapshot( writeDatabase, resolved.sessionKey, @@ -256,81 +245,6 @@ export async function trimSqliteTranscriptForManualCompact( }); } -/** Imports one legacy session entry and its transcript rows for doctor migration. */ -export async function importSqliteSessionRows( - params: SqliteSessionImportRowsParams, -): Promise { - const resolved = resolveSqliteScope({ - ...(params.agentId ? { agentId: params.agentId } : {}), - ...(params.env ? { env: params.env } : {}), - sessionKey: params.sessionKey, - ...(params.storePath ? { storePath: params.storePath } : {}), - }); - return await runExclusiveSqliteSessionWrite(resolved, async () => { - let transcriptEvents = 0; - runOpenClawAgentWriteTransaction((database) => { - const currentEntry = readSessionEntryRow(database, resolved.sessionKey)?.entry; - const preservedHarnessId = - params.entry.agentHarnessId === undefined && - currentEntry?.sessionId === params.entry.sessionId && - currentEntry.lifecycleRevision === params.entry.lifecycleRevision - ? currentEntry.agentHarnessId?.trim() - : undefined; - // Plugin doctor migrations can claim a legacy session before the full - // session import runs. Preserve that same-generation canonical owner. - const importedEntry = { - ...params.entry, - ...(preservedHarnessId ? { agentHarnessId: preservedHarnessId } : {}), - sessionFile: formatSqliteSessionReferenceForScope({ - ...resolved, - sessionId: params.entry.sessionId, - }), - }; - writeSessionEntry(database, resolved.sessionKey, importedEntry); - if (params.readTranscriptEvents) { - const transcriptScope = { - ...resolved, - sessionId: params.entry.sessionId, - }; - const existingEventJson = readTranscriptEventJsonSetInTransaction( - database, - params.entry.sessionId, - ); - params.readTranscriptEvents((event) => { - const eventJson = JSON.stringify(event); - if (existingEventJson.has(eventJson)) { - return; - } - if ( - appendTranscriptEventInTransaction(database, transcriptScope, event, { - scheduleProjectionReconcile: false, - touchMutation: false, - }) - ) { - existingEventJson.add(eventJson); - transcriptEvents += 1; - } - }); - reconcileSessionTranscriptIndexInTransaction(database.db, params.entry.sessionId); - } - if (params.transcriptMtimeMs !== undefined) { - advanceTranscriptMutationAtInTransaction( - database, - params.entry.sessionId, - params.transcriptMtimeMs, - ); - } else if (transcriptEvents > 0) { - touchTranscriptMutationInTransaction(database, params.entry.sessionId); - } - }, toDatabaseOptions(resolved)); - return { - sessionId: params.entry.sessionId, - sessionKey: resolved.sessionKey, - transcriptEvents, - }; - }); -} - /** Appends one raw transcript event to the additive SQLite transcript store. */ export async function appendSqliteTranscriptEvent( scope: SessionTranscriptAccessScope, @@ -553,6 +467,8 @@ export async function withSqliteTranscriptWriteLock( transcriptSnapshot = { kind: "current", rows: snapshot.rows }; return snapshot.events; }, + readMessageFacts: async (params) => + readTranscriptMirrorFacts(database, resolved.sessionId, params), replaceEvents: async (events) => { if (transcriptSnapshot?.kind === "stale") { throw new SqliteTranscriptMutationConflictError(resolved.sessionId); @@ -569,7 +485,7 @@ export async function withSqliteTranscriptWriteLock( ); } replaceSqliteTranscriptEventsInTransaction(writeDatabase, resolved, events); - return readSqliteTranscriptSnapshot(writeDatabase, resolved.sessionId).rows; + return readSqliteTranscriptEventRows(writeDatabase, resolved.sessionId); }, toDatabaseOptions(resolved)); transcriptSnapshot = { kind: "current", rows: nextSnapshot }; }, @@ -591,7 +507,7 @@ export async function withSqliteTranscriptWriteLock( nextSnapshotState = snapshotStillCurrent ? { kind: "current", - rows: readSqliteTranscriptSnapshot(writeDatabase, resolved.sessionId).rows, + rows: readSqliteTranscriptEventRows(writeDatabase, resolved.sessionId), } : { kind: "stale" }; } @@ -599,6 +515,25 @@ export async function withSqliteTranscriptWriteLock( transcriptSnapshot = nextSnapshotState; return result as TranscriptMessageAppendResult | undefined; }, + appendMessageWithMessageSequence: async (options) => { + let result: TranscriptMessageAppendResult | undefined; + let messageSeq: number | undefined; + runOpenClawAgentWriteTransaction((writeDatabase) => { + result = appendSqliteTranscriptMessageInTransaction(writeDatabase, resolved, options); + if (result) { + rememberCommittedSqliteTranscriptMessageSequencesInTransaction( + writeDatabase, + resolved.sessionId, + [result], + ); + messageSeq = readCommittedSqliteTranscriptMessageSequence(result); + } + }, toDatabaseOptions(resolved)); + return { + ...(messageSeq !== undefined ? { messageSeq } : {}), + result: result as TranscriptMessageAppendResult | undefined, + }; + }, }); }); } @@ -632,7 +567,7 @@ function isSqliteTranscriptSnapshotUnchanged( sessionId: string, expected: readonly SqliteTranscriptSnapshotRow[], ): boolean { - const current = readSqliteTranscriptSnapshot(database, sessionId).rows; + const current = readSqliteTranscriptEventRows(database, sessionId); return ( current.length === expected.length && current.every( diff --git a/src/config/sessions/session-accessor.sqlite.ts b/src/config/sessions/session-accessor.sqlite.ts index a1d931622569..fef0c84f6bb1 100644 --- a/src/config/sessions/session-accessor.sqlite.ts +++ b/src/config/sessions/session-accessor.sqlite.ts @@ -56,7 +56,6 @@ export { appendSqliteTranscriptEventSync, appendSqliteTranscriptMessage, appendSqliteTranscriptMessageSync, - importSqliteSessionRows, replaceSqliteTranscriptEvents, replaceSqliteTranscriptEventsSync, rewriteSqliteTranscriptEventRowsExact, @@ -64,6 +63,7 @@ export { withSqliteTranscriptWriteLock, withSqliteTranscriptWriteTransaction, } from "./session-accessor.sqlite-transcript-write.js"; +export { importSqliteSessionRows } from "./session-accessor.sqlite-import.js"; export { publishSqliteTranscriptUpdate } from "./session-accessor.sqlite-events.js"; export { readSqliteTranscriptRawDelta } from "./session-accessor.sqlite-delta.js"; export { diff --git a/src/config/sessions/session-accessor.types.ts b/src/config/sessions/session-accessor.types.ts index 0b7a38afdda2..dfffe5e0b52a 100644 --- a/src/config/sessions/session-accessor.types.ts +++ b/src/config/sessions/session-accessor.types.ts @@ -332,6 +332,18 @@ export type SessionTranscriptWriteLockAccessorContext = { appendMessage: ( options: TranscriptMessageAppendOptions, ) => Promise | undefined>; + /** Appends with commit-time idempotency and returns the committed visible sequence. */ + appendMessageWithMessageSequence: ( + options: TranscriptMessageAppendOptions, + ) => Promise<{ + messageSeq?: number; + result: TranscriptMessageAppendResult | undefined; + }>; + /** Reads bounded indexed facts for supplied transcript mirror identities. */ + readMessageFacts: (params: { idempotencyKeys: readonly string[] }) => Promise<{ + existingIdempotencyKeys: Set; + messagesByIdempotencyKey: Map; + }>; readEvents: () => Promise; replaceEvents: (events: readonly TranscriptEvent[]) => Promise; }; diff --git a/src/plugin-sdk/codex-session-transcript-runtime.ts b/src/plugin-sdk/codex-session-transcript-runtime.ts new file mode 100644 index 000000000000..0ed0d0d313a6 --- /dev/null +++ b/src/plugin-sdk/codex-session-transcript-runtime.ts @@ -0,0 +1,60 @@ +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import type { + TranscriptMessageAppendOptions, + TranscriptMessageAppendResult, +} from "../config/sessions/session-accessor.js"; +import type { AgentMessage } from "./agent-core.js"; +import { + withProjectedSessionTranscriptWriteLock, + type InternalSessionTranscriptWriteLockContext, + type InternalSessionTranscriptWriteLockParams, +} from "./session-transcript-lock-runtime.js"; +import { publishSessionTranscriptUpdateByIdentity } from "./session-transcript-runtime.js"; + +export type CodexSessionTranscriptMirrorWriteLockContext = + InternalSessionTranscriptWriteLockContext & { + appendMessageWithMessageSequence: ( + options: Omit, "config">, + ) => Promise<{ + messageSeq?: number; + result: TranscriptMessageAppendResult | undefined; + }>; + readMessageFacts: (params: { idempotencyKeys: readonly string[] }) => Promise<{ + existingIdempotencyKeys: Set; + messagesByIdempotencyKey: Map; + }>; + }; + +/** Runs the bundled Codex mirror under the transcript writer lock. */ +export async function withCodexSessionTranscriptMirrorWriteLock( + params: InternalSessionTranscriptWriteLockParams, + run: (context: CodexSessionTranscriptMirrorWriteLockContext) => Promise | T, +): Promise { + return await withProjectedSessionTranscriptWriteLock( + params, + run, + (context, locked) => ({ + ...context, + appendMessageWithMessageSequence: (options) => + locked.appendMessageWithMessageSequence({ + ...options, + ...(params.config !== undefined ? { config: params.config } : {}), + }), + readMessageFacts: async (factParams) => { + const facts = await locked.readMessageFacts(factParams); + const messagesByIdempotencyKey = new Map(); + for (const [idempotencyKey, message] of facts.messagesByIdempotencyKey) { + if (isAgentMessageRecord(message)) { + messagesByIdempotencyKey.set(idempotencyKey, message); + } + } + return { ...facts, messagesByIdempotencyKey }; + }, + }), + publishSessionTranscriptUpdateByIdentity, + ); +} + +function isAgentMessageRecord(value: unknown): value is AgentMessage & Record { + return isRecord(value) && typeof value.role === "string" && value.role.trim().length > 0; +} diff --git a/src/plugin-sdk/session-transcript-lock-runtime.ts b/src/plugin-sdk/session-transcript-lock-runtime.ts new file mode 100644 index 000000000000..60b7fa8dc71c --- /dev/null +++ b/src/plugin-sdk/session-transcript-lock-runtime.ts @@ -0,0 +1,114 @@ +import { + publishTranscriptUpdate, + resolveSessionTranscriptRuntimeTarget, + withTranscriptWriteLock, + type SessionTranscriptWriteLockAccessorContext, + type TranscriptMessageAppendOptions, + type TranscriptMessageAppendResult, + type TranscriptUpdatePayload, +} from "../config/sessions/session-accessor.js"; +import { normalizeAgentId } from "../routing/session-key.js"; +import { + formatSessionTranscriptMemoryHitKey, + type SessionTranscriptMemoryHitKey, + type SessionTranscriptReadParams, +} from "./session-transcript-memory-hit.js"; + +export type InternalSessionTranscriptTarget = { + agentId: string; + memoryKey: SessionTranscriptMemoryHitKey; + sessionId: string; + sessionKey: string; + targetKind: "runtime-session"; +}; + +export type InternalSessionTranscriptWriteLockParams = SessionTranscriptReadParams & { + config?: TranscriptMessageAppendOptions["config"]; +}; + +export type InternalSessionTranscriptWriteLockContext = { + appendMessage: ( + options: Omit, "config">, + ) => Promise | undefined>; + publishUpdate: (update?: TranscriptUpdatePayload) => Promise; + readEvents: () => Promise; + target: InternalSessionTranscriptTarget; +}; + +/** Resolves, locks, and publishes one projected transcript write context. */ +export async function withProjectedSessionTranscriptWriteLock< + T, + TContext extends InternalSessionTranscriptWriteLockContext, +>( + params: InternalSessionTranscriptWriteLockParams, + run: (context: TContext) => Promise | T, + projectContext: ( + context: InternalSessionTranscriptWriteLockContext, + locked: SessionTranscriptWriteLockAccessorContext, + ) => TContext, + publishQueuedUpdate?: ( + params: InternalSessionTranscriptWriteLockParams & { update?: TranscriptUpdatePayload }, + ) => Promise, +): Promise { + const storageTarget = await resolveSessionTranscriptRuntimeTarget(params); + const agentId = normalizeAgentId(storageTarget.agentId); + const target: InternalSessionTranscriptTarget = { + agentId, + memoryKey: formatSessionTranscriptMemoryHitKey({ + agentId, + sessionId: storageTarget.sessionId, + }), + sessionId: storageTarget.sessionId, + sessionKey: storageTarget.sessionKey, + targetKind: "runtime-session", + }; + const boundScope = { + ...params, + sessionId: storageTarget.sessionId, + sessionKey: storageTarget.sessionKey, + }; + // Publish only after the write callback commits, so failed transactions cannot + // expose transcript updates to gateway subscribers. + const queuedUpdates: Array = []; + const result = await withTranscriptWriteLock( + boundScope, + async (locked) => + await run( + projectContext( + { + target, + readEvents: locked.readEvents, + appendMessage: (options) => + locked.appendMessage({ + ...options, + ...(params.config !== undefined ? { config: params.config } : {}), + }), + publishUpdate: async (update) => { + queuedUpdates.push(update ? { ...update } : undefined); + }, + }, + locked, + ), + ), + ); + for (const update of queuedUpdates) { + if (publishQueuedUpdate) { + await publishQueuedUpdate({ + ...boundScope, + ...(update !== undefined ? { update } : {}), + }); + continue; + } + await publishTranscriptUpdate(boundScope, { + ...update, + agentId: storageTarget.agentId, + sessionKey: storageTarget.sessionKey, + target: { + agentId: storageTarget.agentId, + sessionId: storageTarget.sessionId, + sessionKey: storageTarget.sessionKey, + }, + }); + } + return result; +} diff --git a/src/plugin-sdk/session-transcript-mirror-runtime.test.ts b/src/plugin-sdk/session-transcript-mirror-runtime.test.ts new file mode 100644 index 000000000000..172df0ae959c --- /dev/null +++ b/src/plugin-sdk/session-transcript-mirror-runtime.test.ts @@ -0,0 +1,154 @@ +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { upsertSessionEntry } from "../config/sessions/session-accessor.js"; +import { + resolveSqliteTranscriptScope, + toDatabaseOptions, +} from "../config/sessions/session-accessor.sqlite-scope.js"; +import { waitForSessionTranscriptProjection } from "../config/sessions/session-transcript-reconcile.js"; +import { openOpenClawAgentDatabase } from "../state/openclaw-agent-db.js"; +import { withCodexSessionTranscriptMirrorWriteLock } from "./codex-session-transcript-runtime.js"; +import { readSessionTranscriptVisibleMessageDelta } from "./session-transcript-runtime.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +describe("private session transcript mirror runtime", () => { + let storePath: string; + + beforeEach(() => { + const tempDir = tempDirs.make("openclaw-sdk-transcript-mirror-"); + storePath = path.join(tempDir, "sessions.json"); + }); + + it("rechecks idempotency and publishes only committed visible sequences", async () => { + const scope = { + agentId: "main", + sessionId: "indexed-mirror-session", + sessionKey: "agent:main:indexed-mirror", + storePath, + }; + await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 1 }); + + await withCodexSessionTranscriptMirrorWriteLock(scope, async (locked) => { + expect(await locked.readMessageFacts({ idempotencyKeys: ["mirror-user"] })).toEqual({ + existingIdempotencyKeys: new Set(), + messagesByIdempotencyKey: new Map(), + }); + const first = await locked.appendMessageWithMessageSequence({ + idempotencyLookup: "scan", + message: { + role: "user", + content: [{ type: "text", text: "persist once" }], + idempotencyKey: "mirror-user", + timestamp: 1, + }, + }); + expect(first).toMatchObject({ + messageSeq: 1, + result: { appended: true, message: { role: "user" } }, + }); + expect(await locked.readMessageFacts({ idempotencyKeys: ["mirror-user"] })).toMatchObject({ + existingIdempotencyKeys: new Set(["mirror-user"]), + messagesByIdempotencyKey: new Map([ + ["mirror-user", expect.objectContaining({ role: "user" })], + ]), + }); + const replay = await locked.appendMessageWithMessageSequence({ + idempotencyLookup: "scan", + message: { + role: "user", + content: [{ type: "text", text: "must not replace persisted payload" }], + idempotencyKey: "mirror-user", + timestamp: 2, + }, + }); + expect(replay).toMatchObject({ + result: { + appended: false, + message: { content: [{ text: "persist once", type: "text" }], role: "user" }, + }, + }); + expect(replay.messageSeq).toBeUndefined(); + + if (!first.result) { + throw new Error("expected initial mirror append"); + } + await locked.appendMessage({ + message: { + role: "assistant", + content: [{ type: "text", text: "abandoned branch" }], + idempotencyKey: "mirror-abandoned", + timestamp: 3, + }, + parentId: first.result.messageId, + }); + const activeBranch = await locked.appendMessageWithMessageSequence({ + message: { + role: "assistant", + content: [{ type: "text", text: "final active branch" }], + idempotencyKey: "mirror-active", + timestamp: 4, + }, + parentId: first.result.messageId, + }); + expect(activeBranch).toMatchObject({ + result: { appended: true, message: { role: "assistant" } }, + }); + expect(activeBranch.messageSeq).toBeUndefined(); + }); + + const resolvedScope = resolveSqliteTranscriptScope(scope); + await expect( + readSessionTranscriptVisibleMessageDelta({ ...scope, maxMessages: 10 }), + ).resolves.toEqual({ + kind: "unavailable", + reason: "projection_rebuilding", + }); + await waitForSessionTranscriptProjection(scope); + await withCodexSessionTranscriptMirrorWriteLock(scope, async (locked) => { + const afterReconcile = await locked.appendMessageWithMessageSequence({ + message: { + role: "assistant", + content: [{ type: "text", text: "after active branch" }], + idempotencyKey: "mirror-after-reconcile", + timestamp: 5, + }, + }); + expect(afterReconcile).toMatchObject({ + messageSeq: 3, + result: { appended: true, message: { role: "assistant" } }, + }); + }); + + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolvedScope)); + const external = new DatabaseSync(database.path); + external + .prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?") + .run(scope.sessionId); + external.close(); + + await withCodexSessionTranscriptMirrorWriteLock(scope, async (locked) => { + expect(await locked.readMessageFacts({ idempotencyKeys: ["mirror-user"] })).toMatchObject({ + existingIdempotencyKeys: new Set(["mirror-user"]), + messagesByIdempotencyKey: new Map([ + ["mirror-user", expect.objectContaining({ role: "user" })], + ]), + }); + const dirtyProjection = await locked.appendMessageWithMessageSequence({ + idempotencyLookup: "scan", + message: { + role: "assistant", + content: [{ type: "text", text: "projection fallback" }], + idempotencyKey: "mirror-dirty", + timestamp: 6, + }, + }); + expect(dirtyProjection).toMatchObject({ + result: { appended: true, message: { role: "assistant" } }, + }); + expect(dirtyProjection.messageSeq).toBeUndefined(); + }); + }); +}); diff --git a/src/plugin-sdk/session-transcript-runtime.test.ts b/src/plugin-sdk/session-transcript-runtime.test.ts index 267f213a889e..02af33b96ea4 100644 --- a/src/plugin-sdk/session-transcript-runtime.test.ts +++ b/src/plugin-sdk/session-transcript-runtime.test.ts @@ -739,6 +739,8 @@ describe("session transcript runtime SDK", () => { const target = await withSessionTranscriptWriteLock(scope, async (locked) => { expect(await locked.readEvents()).toEqual([]); + expect(locked).not.toHaveProperty("appendMessageWithMessageSequence"); + expect(locked).not.toHaveProperty("readMessageFacts"); await locked.appendMessage({ message: { role: "assistant", diff --git a/src/plugin-sdk/session-transcript-runtime.ts b/src/plugin-sdk/session-transcript-runtime.ts index 8c1f89ad0f5f..3019a91c739a 100644 --- a/src/plugin-sdk/session-transcript-runtime.ts +++ b/src/plugin-sdk/session-transcript-runtime.ts @@ -37,6 +37,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { normalizeAgentId } from "../routing/session-key.js"; import { extractAssistantVisibleText } from "../shared/chat-message-content.js"; import type { AgentMessage } from "./agent-core.js"; +import { withProjectedSessionTranscriptWriteLock } from "./session-transcript-lock-runtime.js"; import { formatSessionTranscriptMemoryHitKey, parseSessionTranscriptMemoryHitKey, @@ -445,42 +446,7 @@ export async function withSessionTranscriptWriteLock( params: SessionTranscriptWriteLockParams, run: (context: SessionTranscriptWriteLockContext) => Promise | T, ): Promise { - const storageTarget = await resolveSessionTranscriptRuntimeTarget(params); - const target = projectPublicTarget({ - ...storageTarget, - targetKind: "runtime-session", - }); - const boundScope = { - ...params, - sessionId: storageTarget.sessionId, - sessionKey: storageTarget.sessionKey, - }; - // Treat publishUpdate as a post-commit callback: future transactional stores - // must not expose updates when the scoped write callback fails. - const queuedUpdates: Array = []; - const result = await withTranscriptWriteLock( - boundScope, - async (locked) => - await run({ - target, - readEvents: locked.readEvents, - appendMessage: (options) => - locked.appendMessage({ - ...options, - ...(params.config !== undefined ? { config: params.config } : {}), - }), - publishUpdate: async (update) => { - queuedUpdates.push(update ? { ...update } : undefined); - }, - }), - ); - for (const update of queuedUpdates) { - await publishSessionTranscriptUpdateByIdentity({ - ...boundScope, - update, - }); - } - return result; + return await withProjectedSessionTranscriptWriteLock(params, run, (context) => context); } function createAssistantMirrorMessage(params: { diff --git a/src/plugins/sdk-alias.ts b/src/plugins/sdk-alias.ts index efd9dbdfcb0f..c97cedb148dd 100644 --- a/src/plugins/sdk-alias.ts +++ b/src/plugins/sdk-alias.ts @@ -417,6 +417,7 @@ const cachedWorkspacePackageAliasMaps = new PluginLruCache { ); expect(latestMessagePlan).not.toContain("USE TEMP B-TREE FOR ORDER BY"); + const mirrorIdentityPlan = explainQueryPlan( + database.db, + ` + SELECT identity.message_idempotency_key, event.event_json + FROM transcript_event_identities AS identity + JOIN transcript_events AS event + ON event.session_id = identity.session_id AND event.seq = identity.seq + WHERE identity.session_id = ? + AND identity.message_idempotency_key IN (?, ?) + ORDER BY identity.seq ASC + `, + ["session-1", "prompt-key", "assistant-key"], + ); + expect(mirrorIdentityPlan).toContain("idx_agent_transcript_message_idempotency"); + expect(mirrorIdentityPlan).toContain("sqlite_autoindex_transcript_events_1"); + expect(mirrorIdentityPlan).not.toContain("SCAN transcript_events"); + + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_agent_transcript_event_sequence", + params: ["session-1", "message"], + sql: ` + SELECT COUNT(seq) + FROM transcript_event_identities + WHERE session_id = ? AND event_type = ? + `, + }); + expectPlanIncludes({ db: database.db, expected: "sqlite_autoindex_transcript_rewrite_watermarks_1", diff --git a/tsdown.config.ts b/tsdown.config.ts index 863fc4ceef77..838b698b1268 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -507,6 +507,9 @@ function buildUnifiedDistEntries(): Record { ), // Private bundled Codex helper for app-server user MCP config projection. "plugin-sdk/codex-mcp-projection": "src/plugin-sdk/codex-mcp-projection.ts", + // Private bundled Codex helper for app-server transcript mirroring. + "plugin-sdk/codex-session-transcript-runtime": + "src/plugin-sdk/codex-session-transcript-runtime.ts", ...Object.fromEntries( Object.entries(buildPluginSdkEntrySources(selectedPluginSdkEntrypoints)).map( ([entry, source]) => [`plugin-sdk/${entry}`, source],