ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 21:07:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: prevent stale transcript projections from publishing (#126947)

Browse Source 
Fence transcript projections by source generation so stale rebuilds cannot publish or satisfy readers.

Refs #126914.
This commit is contained in:
Shakker
2026-08-24 17:38:34 +01:00
committed by GitHub
parent 59abf3de62
commit 0fc18f7447
49 changed files with 2172 additions and 409 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/.generated/sqlite-session-transcript-schema-baseline.sha256
+1 -1
View File
@@ -1 +1 @@
1ba742b789c40e2b6006b6c24abd371319ecdaf8a8a939d0ff61d60d2f4eec7c sqlite-session-transcript-schema-baseline.sql
9c5037d4a08541485a0fb4b6901872ba258832b5e120bcd2cb230df7348c18d8 sqlite-session-transcript-schema-baseline.sql
src/agents/harness/context-engine-turn-attempt.test.ts
+3 -3
View File
@@ -355,7 +355,7 @@ describe("accepted context-engine turn finalization", () => {
if (siblingIdentity?.seq === undefined) {
throw new Error("expected sibling transcript identity");
}
// Model a stale/concurrent projection that assigns a later active position
// Model a published malformed projection that assigns a later active position
// to a sibling. Position order alone must not make it an accepted descendant.
database.db
.prepare(
@@ -369,9 +369,9 @@ describe("accepted context-engine turn finalization", () => {
);
database.db
.prepare(
"UPDATE session_transcript_index_state SET indexed_seq = ?, needs_rebuild = 0 WHERE session_id = ?",
"UPDATE session_transcript_index_state SET indexed_seq = ?, needs_rebuild = 0, source_generation = ? WHERE session_id = ?",
)
.run(siblingIdentity.seq, target.sessionId);
.run(siblingIdentity.seq, terminal.anchor.generation, target.sessionId);
const siblingAnchor = readActiveTranscriptEntryAnchor({
...target,
entryId: sibling.messageId,
src/config/sessions/session-accessor.sqlite-active-events.test.ts
+7 -3
View File
@@ -959,11 +959,15 @@ describe("SQLite active transcript event projection", () => {
`
INSERT INTO session_transcript_index_state
(session_id, indexed_seq, leaf_event_id, needs_rebuild,
active_event_count, active_message_count, updated_at)
VALUES (?, 100000, 'm100000', 0, 100000, 100000, 100000)
active_event_count, active_message_count, source_generation, updated_at)
VALUES (
?, 100000, 'm100000', 0, 100000, 100000,
(SELECT generation FROM transcript_rewrite_watermarks WHERE session_id = ?),
100000
)
`,
)
.run(scope.sessionId);
.run(scope.sessionId, scope.sessionId);
database.db.exec("COMMIT;");
} catch (error) {
database.db.exec("ROLLBACK;");
src/config/sessions/session-accessor.sqlite-active-projection.ts
+12 -2
View File
@@ -5,6 +5,7 @@ import {
openOpenClawAgentDatabase,
type OpenClawAgentDatabase,
} from "../../state/openclaw-agent-db.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "../../state/openclaw-agent-transcript-projection-source-schema.js";
import type { SessionTranscriptReadScope } from "./session-accessor.sqlite-contract.js";
import {
resolveSqliteTranscriptReadScope,
@@ -13,6 +14,7 @@ import {
import type { SessionTranscriptProjectionState } from "./session-transcript-index.js";
import { SessionTranscriptProjectionUnavailableError } from "./session-transcript-projection-error.js";
import { startSessionTranscriptIndexReconcile } from "./session-transcript-reconcile.js";
import { EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ } from "./session-transcript-source-generation.js";
type ActiveTranscriptDatabase = Pick<
OpenClawAgentKyselyDatabase,
@@ -32,9 +34,10 @@ export type CurrentTranscriptProjection = {
const EMPTY_PROJECTION_STATE: SessionTranscriptProjectionState = {
activeEventCount: 0,
activeMessageCount: 0,
indexedSeq: -1,
indexedSeq: EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
leafEventId: null,
needsRebuild: false,
sourceGeneration: null,
};
export function getActiveTranscriptKysely(database: OpenClawAgentDatabase) {
@@ -45,11 +48,13 @@ function readProjectionSnapshot(
database: OpenClawAgentDatabase,
sessionId: string,
): { latestSeq: number; state?: SessionTranscriptProjectionState } | undefined {
ensureOpenClawAgentTranscriptProjectionSourceColumns(database.db);
const row = executeSqliteQueryTakeFirstSync(
database.db,
getActiveTranscriptKysely(database)
.selectFrom("transcript_events as latest")
.leftJoin("session_transcript_index_state as state", "state.session_id", "latest.session_id")
.leftJoin("transcript_rewrite_watermarks as source", "source.session_id", "latest.session_id")
.select([
"latest.seq as latest_seq",
"state.active_event_count",
@@ -57,6 +62,8 @@ function readProjectionSnapshot(
"state.indexed_seq",
"state.leaf_event_id",
"state.needs_rebuild",
"state.source_generation",
"source.generation",
])
.where("latest.session_id", "=", sessionId)
.orderBy("latest.seq", "desc")
@@ -75,6 +82,8 @@ function readProjectionSnapshot(
indexedSeq: row.indexed_seq,
leafEventId: row.leaf_event_id,
needsRebuild: row.needs_rebuild !== 0,
sourceGeneration:
row.source_generation === row.generation ? row.source_generation : null,
},
}
: {}),
@@ -101,7 +110,8 @@ export function withCurrentProjectionSnapshot<T>(
if (
snapshot.state &&
!snapshot.state.needsRebuild &&
snapshot.state.indexedSeq === snapshot.latestSeq
snapshot.state.indexedSeq === snapshot.latestSeq &&
snapshot.state.sourceGeneration !== null
) {
return {
kind: "value" as const,
src/config/sessions/session-accessor.sqlite-canonical-repair.ts
+18 -21
View File
@@ -23,7 +23,6 @@ import {
} from "./session-accessor.sqlite-scope.js";
import { bindSessionWindowEntryProjection } from "./session-accessor.sqlite-session-row.js";
import { parseSessionEntryJson } from "./session-accessor.sqlite-status.js";
import { ensureTranscriptGenerationInTransaction } from "./session-accessor.sqlite-transcript-state.js";
import { canonicalSessionKeyMigrationRequiredError } from "./session-canonical-key.js";
import { invalidateExistingSessionTranscriptDisplayInTransaction } from "./session-transcript-display.js";
import {
@@ -34,6 +33,11 @@ import {
startSessionTranscriptDisplayReconcile,
startSessionTranscriptIndexReconcile,
} from "./session-transcript-reconcile.js";
import {
ensureSessionTranscriptSourceGenerationInTransaction,
readSessionTranscriptSourceGenerationInTransaction,
replaceSessionTranscriptSourceGenerationInTransaction,
} from "./session-transcript-source-generation.js";
import { normalizeStoreSessionKey } from "./store-entry.js";
import type { SessionEntry } from "./types.js";
@@ -167,16 +171,15 @@ export async function ensureSqliteTranscriptGenerationsForCanonicalRepair(
),
]);
const db = getSessionKysely(database.db);
const eventSessionIds = executeSqliteQuerySync(
const windowSessionIds = executeSqliteQuerySync(
database.db,
db
.selectFrom("transcript_events")
.selectFrom("session_windows")
.select("session_id")
.where("session_id", "in", sessionIds)
.groupBy("session_id"),
.where("session_id", "in", sessionIds),
).rows;
for (const row of eventSessionIds) {
ensureTranscriptGenerationInTransaction(database, row.session_id);
for (const row of windowSessionIds) {
ensureSessionTranscriptSourceGenerationInTransaction(database, row.session_id);
}
}, toDatabaseOptions(group.resolved));
});
@@ -529,13 +532,10 @@ function copySqliteSessionGenerationRows(params: {
.selectAll()
.where("session_id", "=", params.sessionId),
).rows;
const rewriteWatermarks = executeSqliteQuerySync(
const sourceGeneration = readSessionTranscriptSourceGenerationInTransaction(
params.source.db,
sourceDb
.selectFrom("transcript_rewrite_watermarks")
.selectAll()
.where("session_id", "=", params.sessionId),
).rows;
params.sessionId,
);
const trajectoryEvents = executeSqliteQuerySync(
params.source.db,
sourceDb
@@ -554,16 +554,18 @@ function copySqliteSessionGenerationRows(params: {
!params.sourceWindowPresent &&
transcriptEvents.length === 0 &&
transcriptIdentities.length === 0 &&
rewriteWatermarks.length === 0 &&
sourceGeneration === undefined &&
trajectoryEvents.length === 0 &&
parentStreamEvents.length === 0
) {
return false;
}
if (params.sourceWindowPresent && !sourceGeneration) {
throw new Error(`Canonical transcript source ${params.sessionId} has no generation`);
}
for (const table of [
"transcript_event_identities",
"transcript_events",
"transcript_rewrite_watermarks",
"trajectory_runtime_events",
"acp_parent_stream_events",
] as const) {
@@ -584,12 +586,7 @@ function copySqliteSessionGenerationRows(params: {
destinationDb.insertInto("transcript_event_identities").values(row),
);
}
for (const row of rewriteWatermarks) {
executeSqliteQuerySync(
params.destination.db,
destinationDb.insertInto("transcript_rewrite_watermarks").values(row),
);
}
replaceSessionTranscriptSourceGenerationInTransaction(params.destination, params.sessionId);
for (const row of trajectoryEvents) {
executeSqliteQuerySync(
params.destination.db,
src/config/sessions/session-accessor.sqlite-data-version.test.ts
+2 -1
View File
@@ -272,7 +272,8 @@ describe("SQLite session entry cache", () => {
sessionId: "same-connection-non-entry-2",
updatedAt: 1,
});
ensureOpenClawAgentDisplayRowSchema(openOpenClawAgentDatabase(scope).db);
const database = openOpenClawAgentDatabase(scope).db;
ensureOpenClawAgentDisplayRowSchema(database);
const first = listSessionEntriesCore({ ...scope, clone: false, projection: "list" });
await appendTranscriptMessage(
src/config/sessions/session-accessor.sqlite-entry-store.ts
+4 -4
View File
@@ -49,7 +49,7 @@ import {
hasValidSessionEntryIdentity,
parseSessionEntryJson as parseSessionEntryRow,
} from "./session-accessor.sqlite-status.js";
import { readTranscriptMutationStateInTransaction } from "./session-accessor.sqlite-transcript-state.js";
import * as transcript from "./session-accessor.sqlite-transcript-state.js";
import {
assertCanonicalSessionEntryLineageWrite,
assertCanonicalSqliteSessionKeysCurrent,
@@ -629,8 +629,8 @@ export function writeSessionEntry(
// Registry writes snapshot the current transcript watermark so recovery can
// distinguish same-millisecond transcript writes before and after this row.
const transcriptObservedAt =
readTranscriptMutationStateInTransaction(database, normalizedEntry.sessionId).updatedAt ??
updatedAt;
transcript.readTranscriptMutationStateInTransaction(database, normalizedEntry.sessionId)
.updatedAt ?? updatedAt;
const boundSessionRoot = bindSessionRoot({ entry: normalizedEntry, sessionKey, updatedAt });
const conversation = prepareSessionConversationForWrite({
database,
@@ -728,6 +728,7 @@ export function writeSessionEntry(
}),
),
);
transcript.ensureSessionTranscriptSourceGenerationInTransaction(database, sessionRow.session_id);
if (conversation) {
linkSessionConversation({
database,
@@ -739,5 +740,4 @@ export function writeSessionEntry(
}
publishSessionEntryCacheInvalidation(database, sessionNode, writeGeneration);
}
/** Resolves the parent fork decision using SQLite transcript rows when totals are stale. */
src/config/sessions/session-accessor.sqlite-import.ts
-2
View File
@@ -21,7 +21,6 @@ import {
} from "./session-accessor.sqlite-scope.js";
import {
advanceTranscriptMutationAtInTransaction,
ensureTranscriptGenerationInTransaction,
ensureTranscriptSessionRoot,
touchTranscriptMutationInTransaction,
} from "./session-accessor.sqlite-transcript-state.js";
@@ -147,7 +146,6 @@ function importSqliteSessionRowsInTransaction(
ensureTranscriptSessionRoot(database, transcriptScope, exactTranscriptRows[0]!.createdAt, {
allowStoredAlias: true,
});
ensureTranscriptGenerationInTransaction(database, params.entry.sessionId);
for (const [seq, row] of exactTranscriptRows.entries()) {
executeSqliteQuerySync(
database.db,
src/config/sessions/session-accessor.sqlite-title-probes.ts
+9 -1
View File
@@ -6,6 +6,7 @@ import {
openOpenClawAgentDatabase,
type OpenClawAgentDatabase,
} from "../../state/openclaw-agent-db.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "../../state/openclaw-agent-transcript-projection-source-schema.js";
import type {
SessionTranscriptReadScope,
TranscriptEvent,
@@ -60,6 +61,7 @@ function readTitleProbeChunk(
database: OpenClawAgentDatabase,
sessionIds: readonly string[],
): Map<string, SessionTranscriptTitleProbe> {
ensureOpenClawAgentTranscriptProjectionSourceColumns(database.db);
const db = getTitleProbeKysely(database);
const rows = runSqliteDeferredTransactionSync(
database.db,
@@ -104,6 +106,7 @@ function readTitleProbeChunk(
"state.indexed_seq",
"state.needs_rebuild",
"rewrite.generation",
"state.source_generation",
"active.message_position",
"event.event_json",
eb
@@ -138,7 +141,12 @@ function readTitleProbeChunk(
const probes = new Map<string, SessionTranscriptTitleProbe>();
for (const row of rows) {
const emptyTranscript = row.latest_seq === null;
const projectionCurrent = row.needs_rebuild === 0 && row.indexed_seq === row.latest_seq;
const projectionCurrent =
typeof row.generation === "string" &&
typeof row.source_generation === "string" &&
row.needs_rebuild === 0 &&
row.indexed_seq === row.latest_seq &&
row.source_generation === row.generation;
if (
(!emptyTranscript && !projectionCurrent) ||
parseEventType(row.latest_boundary_json) === "reset"
src/config/sessions/session-accessor.sqlite-transcript-anchor.ts
+11
View File
@@ -10,6 +10,7 @@ import {
type ResolvedTranscriptScope,
} from "./session-accessor.sqlite-scope.js";
import { readMessageIdempotencyKey } from "./session-accessor.sqlite-transcript-store.js";
import { readCurrentSessionTranscriptActiveSourceInTransaction } from "./session-transcript-source-generation.js";
import type { TranscriptEntryAnchor } from "./transcript-entry-anchor.js";
/** Reads one active message identity from the caller's current SQLite transaction. */
@@ -18,7 +19,17 @@ export function readActiveTranscriptEntryAnchorInTransaction(params: {
resolved: ResolvedTranscriptScope;
entryId: string;
message?: unknown;
projectionCurrent?: boolean;
}): TranscriptEntryAnchor | undefined {
if (
params.projectionCurrent !== true &&
!readCurrentSessionTranscriptActiveSourceInTransaction(
params.database.db,
params.resolved.sessionId,
)
) {
return undefined;
}
const db = getSessionKysely(params.database.db);
const row = executeSqliteQueryTakeFirstSync(
params.database.db,
src/config/sessions/session-accessor.sqlite-transcript-message-append.ts
+11 -1
View File
@@ -49,12 +49,14 @@ export function appendTranscriptMessageInTransaction<TMessage>(
const readAnchor = (params: {
message: unknown;
messageId: string;
projectionCurrent?: boolean;
}): TranscriptMessageAppendResult<TMessage>["anchor"] =>
readActiveTranscriptEntryAnchorInTransaction({
database,
resolved,
entryId: params.messageId,
message: params.message,
projectionCurrent: params.projectionCurrent,
});
const existingAppendResult = (found: { message: unknown; messageId: string }) => {
const anchor = readAnchor(found);
@@ -108,11 +110,15 @@ export function appendTranscriptMessageInTransaction<TMessage>(
timestamp: resolveTimestampMsToIsoString(now),
message: finalMessage,
};
let projectionNeedsRebuild = false;
const appended = appendTranscriptEventInTransaction(database, resolved, event, {
dedupeByMessageIdempotency:
options.idempotencyLookup !== "caller-checked" &&
options.idempotencyLookup !== "scan-assistant",
maintainDisplayProjection: options.maintainDisplayProjection === true ? true : undefined,
onProjectionReconcileNeeded: () => {
projectionNeedsRebuild = true;
},
});
if (!appended && idempotencyKey && options.idempotencyLookup !== "caller-checked") {
const existing = readTranscriptMessageByScopedIdempotencyKey(
@@ -146,7 +152,11 @@ export function appendTranscriptMessageInTransaction<TMessage>(
if (!appended) {
throw new Error(`SQLite transcript append did not insert message ${messageId}.`);
}
const anchor = readAnchor({ message: finalMessage, messageId });
const anchor = readAnchor({
message: finalMessage,
messageId,
projectionCurrent: !projectionNeedsRebuild,
});
return {
appended: true,
...(anchor ? { anchor } : {}),
src/config/sessions/session-accessor.sqlite-transcript-message-rewrite.ts
+5 -2
View File
@@ -7,8 +7,8 @@ import {
runExclusiveSqliteSessionWrite,
toDatabaseOptions,
} from "./session-accessor.sqlite-scope.js";
import { readTranscriptGenerationInTransaction } from "./session-accessor.sqlite-transcript-state.js";
import { rewriteSqliteTranscriptEventRowsInTransaction } from "./session-accessor.sqlite-transcript-store.js";
import { readSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
import type { TranscriptEntryAnchor } from "./transcript-entry-anchor.js";
type TranscriptMessageAnchorRewriteResult<TMessage> = {
@@ -52,7 +52,10 @@ export async function rewriteTranscriptMessageAtAnchor<TMessage>(
seq: anchor.rawSeq,
},
]);
const generation = readTranscriptGenerationInTransaction(database, resolved.sessionId);
const generation = readSessionTranscriptSourceGenerationInTransaction(
database.db,
resolved.sessionId,
)?.generation;
if (generation) {
result = { generation, message };
}
src/config/sessions/session-accessor.sqlite-transcript-mirror.ts
+5 -10
View File
@@ -12,6 +12,7 @@ import {
import { getSessionKysely, type ResolvedTranscriptScope } from "./session-accessor.sqlite-scope.js";
import { readActiveTranscriptEntryAnchorInTransaction } from "./session-accessor.sqlite-transcript-anchor.js";
import { readMessageIdempotencyKey } from "./session-accessor.sqlite-transcript-store.js";
import { readCurrentSessionTranscriptActiveSourceInTransaction } from "./session-transcript-source-generation.js";
import type { TranscriptEntryAnchor } from "./transcript-entry-anchor.js";
// Keep supplied-key probes below SQLite's conservative variable ceiling.
@@ -29,6 +30,10 @@ function loadTranscriptEventsForMirrorFallback(
sessionId: string,
): TranscriptEvent[] | undefined {
const db = getSessionKysely(database.db);
const source = readCurrentSessionTranscriptActiveSourceInTransaction(database.db, sessionId);
if (source) {
return undefined;
}
const latest = executeSqliteQueryTakeFirstSync(
database.db,
db
@@ -41,16 +46,6 @@ function loadTranscriptEventsForMirrorFallback(
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 loadTranscriptEventsFromDatabase(database, sessionId);
}
src/config/sessions/session-accessor.sqlite-transcript-sequences.test.ts
+64
View File
@@ -0,0 +1,64 @@
import { afterEach, describe, expect, it } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
import {
closeOpenClawAgentDatabasesForTest,
openOpenClawAgentDatabase,
runOpenClawAgentWriteTransaction,
} from "../../state/openclaw-agent-db.js";
import { persistSessionTranscriptTurn } from "./session-accessor.js";
import {
readCommittedTranscriptMessageSequence,
rememberCommittedTranscriptMessageSequencesInTransaction,
} from "./session-accessor.sqlite-transcript-sequences.js";
import { replaceSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
describe("committed transcript message sequences", () => {
afterEach(() => {
closeOpenClawAgentDatabasesForTest();
});
it("omits a sequence when the active projection source is stale", async () => {
const scope = {
agentId: "main",
env: { OPENCLAW_STATE_DIR: tempDirs.make("openclaw-committed-sequence-") },
sessionId: "committed-sequence",
sessionKey: "agent:main:committed-sequence",
};
await persistSessionTranscriptTurn(scope, {
messages: [{ eventId: "seed", message: { role: "user", content: "seed" } }],
touchSessionEntry: false,
});
const message = {
appended: true,
message: { role: "user", content: "seed" },
messageId: "seed",
};
const options = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(options);
runOpenClawAgentWriteTransaction((writeDatabase) => {
rememberCommittedTranscriptMessageSequencesInTransaction(writeDatabase, scope.sessionId, [
message,
]);
}, options);
expect(readCommittedTranscriptMessageSequence(message)).toBe(1);
runOpenClawAgentWriteTransaction((writeDatabase) => {
replaceSessionTranscriptSourceGenerationInTransaction(writeDatabase, scope.sessionId);
rememberCommittedTranscriptMessageSequencesInTransaction(writeDatabase, scope.sessionId, [
message,
]);
}, options);
expect(readCommittedTranscriptMessageSequence(message)).toBeUndefined();
expect(
database.db
.prepare(
"SELECT source_generation FROM session_transcript_index_state WHERE session_id = ?",
)
.get(scope.sessionId),
).toEqual({ source_generation: null });
});
});
src/config/sessions/session-accessor.sqlite-transcript-sequences.ts
+11 -12
View File
@@ -13,6 +13,7 @@ import {
toDatabaseOptions,
} from "./session-accessor.sqlite-scope.js";
import { readTranscriptIdentityByEventId } from "./session-accessor.sqlite-transcript-store.js";
import { readCurrentSessionTranscriptActiveSourceInTransaction } from "./session-transcript-source-generation.js";
// Append results are public SDK contracts. Keep commit-only cursor metadata
// attached to their object lifetime without changing the returned message shape.
@@ -38,23 +39,21 @@ export function rememberCommittedTranscriptMessageSequencesInTransaction(
if (appendedMessages.length === 0) {
return;
}
const appendedProjectionCurrent = appendedMessages.every((message) => message.anchor);
const db = getNodeSqliteKysely<
Pick<
OpenClawAgentKyselyDatabase,
"session_transcript_active_events" | "session_transcript_index_state"
>
Pick<OpenClawAgentKyselyDatabase, "session_transcript_active_events">
>(database.db);
const projection = executeSqliteQueryTakeFirstSync(
database.db,
db
.selectFrom("session_transcript_index_state")
.select("needs_rebuild")
.where("session_id", "=", sessionId),
);
if (projection?.needs_rebuild !== 0) {
if (
!appendedProjectionCurrent &&
!readCurrentSessionTranscriptActiveSourceInTransaction(database.db, sessionId)
) {
return;
}
for (const message of appendedMessages) {
if (appendedProjectionCurrent && message.anchor) {
committedTranscriptMessageSequences.set(message, message.anchor.activeMessagePosition + 1);
continue;
}
const identity = readTranscriptIdentityByEventId(database, sessionId, message.messageId);
if (!identity) {
continue;
src/config/sessions/session-accessor.sqlite-transcript-state.ts
+1 -55
View File
@@ -1,4 +1,3 @@
import { randomUUID } from "node:crypto";
import { uniqueStrings } from "@openclaw/normalization-core/string-normalization";
import {
executeSqliteQuerySync,
@@ -21,60 +20,7 @@ import {
resolveDeliveryProvenCanonicalSessionKey,
} from "./store-entry.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("transcript_rewrite_watermarks")
.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("transcript_rewrite_watermarks")
.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("transcript_rewrite_watermarks")
.values({ session_id: sessionId, generation, updated_at: Date.now() })
.onConflict((conflict) =>
conflict.column("session_id").doUpdateSet({ generation, updated_at: Date.now() }),
),
);
return generation;
}
export { ensureSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
export function ensureTranscriptSessionRoot(
database: OpenClawAgentDatabase,
src/config/sessions/session-accessor.sqlite-transcript-store.ts
+30 -30
View File
@@ -21,12 +21,9 @@ import { getSessionKysely, type ResolvedTranscriptScope } from "./session-access
import {
advanceTranscriptMutationAtInTransaction,
deleteTranscriptEventsInTransaction,
ensureTranscriptGenerationInTransaction,
ensureTranscriptSessionRoot,
readTranscriptGenerationInTransaction,
readTranscriptMutationStateInTransaction,
readNextTranscriptSeq,
rotateTranscriptGenerationInTransaction,
touchTranscriptMutationInTransaction,
} from "./session-accessor.sqlite-transcript-state.js";
import { invalidateExistingSessionTranscriptDisplayInTransaction } from "./session-transcript-display.js";
@@ -39,6 +36,10 @@ import {
startSessionTranscriptDisplayReconcile,
startSessionTranscriptIndexReconcile,
} from "./session-transcript-reconcile.js";
import {
ensureSessionTranscriptSourceGenerationInTransaction,
replaceSessionTranscriptSourceGenerationInTransaction,
} from "./session-transcript-source-generation.js";
import { createSessionTranscriptHeader } from "./transcript-header.js";
import {
isSessionTranscriptLeafControl,
@@ -69,7 +70,10 @@ export function appendTranscriptEventInTransaction(
ensureTranscriptSessionRoot(database, scope, createdAt, {
allowStoredAlias: options.allowStoredAlias === true,
});
ensureTranscriptGenerationInTransaction(database, scope.sessionId);
const sourceGeneration = ensureSessionTranscriptSourceGenerationInTransaction(
database,
scope.sessionId,
);
const identity = readTranscriptEventIdentity(persistedEvent);
if (identity && readTranscriptIdentityByEventId(database, scope.sessionId, identity.eventId)) {
return false;
@@ -105,6 +109,7 @@ export function appendTranscriptEventInTransaction(
eventId: identity?.eventId ?? null,
createdAt,
maintainDisplayProjection: options.maintainDisplayProjection,
sourceGeneration,
});
if (projectionNeedsRebuild) {
options.onProjectionReconcileNeeded?.();
@@ -348,38 +353,33 @@ export function replaceSqliteTranscriptEventsInTransaction(
options.preserveSessionWindowRecency === true
? readTranscriptMutationStateInTransaction(database, resolved.sessionId).updatedAt
: undefined;
const previousGeneration = readTranscriptGenerationInTransaction(database, resolved.sessionId);
const deleted = deleteTranscriptEventsInTransaction(database, resolved.sessionId);
if (events.length === 0) {
if (deleted || previousGeneration) {
rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
const displayProjectionInvalidated = invalidateExistingSessionTranscriptDisplayInTransaction(
database.db,
resolved.sessionId,
);
recordTranscriptReplacementMutation(
database,
resolved.sessionId,
preservedTranscriptUpdatedAt,
);
scheduleTranscriptProjectionReconcile(database, resolved, true, {
maintainDisplayProjection: displayProjectionInvalidated,
});
}
const db = getSessionKysely(database.db);
const sourceWindow = executeSqliteQueryTakeFirstSync(
database.db,
db
.selectFrom("session_windows")
.select("updated_at")
.where("session_id", "=", resolved.sessionId),
);
if (!sourceWindow && events.length === 0) {
return;
}
if (!deleted || options.preserveSessionWindowRecency !== true) {
if (!sourceWindow || (events.length > 0 && options.preserveSessionWindowRecency !== true)) {
ensureTranscriptSessionRoot(database, resolved, readEventTimestamp(events[0]) ?? Date.now());
}
if (deleted || previousGeneration) {
rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
} else {
ensureTranscriptGenerationInTransaction(database, resolved.sessionId);
}
const deleted = deleteTranscriptEventsInTransaction(database, resolved.sessionId);
replaceSessionTranscriptSourceGenerationInTransaction(database, resolved.sessionId);
const displayProjectionInvalidated = invalidateExistingSessionTranscriptDisplayInTransaction(
database.db,
resolved.sessionId,
);
if (events.length === 0) {
recordTranscriptReplacementMutation(database, resolved.sessionId, preservedTranscriptUpdatedAt);
scheduleTranscriptProjectionReconcile(database, resolved, true, {
maintainDisplayProjection: displayProjectionInvalidated,
});
return;
}
let seq = 0;
const seenEventIds = new Set<string>();
const seenMessageIdempotencyKeys = new Set<string>();
@@ -456,7 +456,7 @@ export function rewriteSqliteTranscriptEventRowsInTransaction(
);
}
}
rotateTranscriptGenerationInTransaction(database, resolved.sessionId);
replaceSessionTranscriptSourceGenerationInTransaction(database, resolved.sessionId);
const displayProjectionInvalidated = invalidateExistingSessionTranscriptDisplayInTransaction(
database.db,
resolved.sessionId,
@@ -490,7 +490,7 @@ export function updateSqliteTranscriptEventJsonInTransaction(
.where("seq", "=", seq),
);
}
rotateTranscriptGenerationInTransaction(database, sessionId);
replaceSessionTranscriptSourceGenerationInTransaction(database, sessionId);
const displayProjectionInvalidated = invalidateExistingSessionTranscriptDisplayInTransaction(
database.db,
sessionId,
src/config/sessions/session-accessor.sqlite-transcript-write.ts
+7 -3
View File
@@ -46,13 +46,13 @@ import {
readCommittedTranscriptMessageSequence,
rememberCommittedTranscriptMessageSequencesInTransaction,
} from "./session-accessor.sqlite-transcript-sequences.js";
import { readTranscriptGenerationInTransaction } from "./session-accessor.sqlite-transcript-state.js";
import {
appendTranscriptEventInTransaction,
replaceSqliteTranscriptEventsInTransaction,
rewriteSqliteTranscriptEventRowsInTransaction,
} from "./session-accessor.sqlite-transcript-store.js";
import type { SessionTranscriptWriteTransactionContext } from "./session-accessor.types.js";
import { readSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
import type {
SessionTranscriptTurnExpectedState,
SessionTranscriptTurnLifecyclePatch,
@@ -137,14 +137,18 @@ export async function rewriteTranscriptEventRowsExact(
let result: { generation: string } | null = null;
runOpenClawAgentWriteTransaction((database) => {
const currentGeneration =
readTranscriptGenerationInTransaction(database, resolved.sessionId) ?? null;
readSessionTranscriptSourceGenerationInTransaction(database.db, resolved.sessionId)
?.generation ?? null;
const initialGenerationMaterialized =
params.allowInitialGenerationMaterialization === true && params.expectedGeneration === null;
if (currentGeneration !== params.expectedGeneration && !initialGenerationMaterialized) {
return;
}
rewriteSqliteTranscriptEventRowsInTransaction(database, resolved, params.rows);
const generation = readTranscriptGenerationInTransaction(database, resolved.sessionId);
const generation = readSessionTranscriptSourceGenerationInTransaction(
database.db,
resolved.sessionId,
)?.generation;
if (generation) {
result = { generation };
}
src/config/sessions/session-accessor.transcript-range.ts
+2 -22
View File
@@ -11,6 +11,7 @@ import {
resolveSqliteTranscriptScope,
toDatabaseOptions,
} from "./session-accessor.sqlite-scope.js";
import { readCurrentSessionTranscriptActiveSourceInTransaction } from "./session-transcript-source-generation.js";
import type { TranscriptEntryAnchor, TranscriptTurnBoundary } from "./transcript-entry-anchor.js";
export type ClosedTranscriptTurnReadResult =
@@ -126,28 +127,7 @@ export function readClosedTranscriptTurn(params: {
if (!binding) {
return { kind: "session-rebound" } as const;
}
const frontier = executeSqliteQueryTakeFirstSync(
database.db,
db
.selectFrom("transcript_events")
.select("seq")
.where("session_id", "=", target.sessionId)
.orderBy("seq", "desc")
.limit(1),
)?.seq;
const projection = executeSqliteQueryTakeFirstSync(
database.db,
db
.selectFrom("session_transcript_index_state")
.select(["indexed_seq", "needs_rebuild"])
.where("session_id", "=", target.sessionId),
);
if (
frontier === undefined ||
!projection ||
projection.needs_rebuild !== 0 ||
projection.indexed_seq !== frontier
) {
if (!readCurrentSessionTranscriptActiveSourceInTransaction(database.db, target.sessionId)) {
return { kind: "projection-unavailable" } as const;
}
const readAnchor = (anchor: TranscriptEntryAnchor) =>
src/config/sessions/session-transcript-display-rebuild-store.test.ts
+8
View File
@@ -5,6 +5,7 @@ import {
appendSessionTranscriptDisplayChunkInTransaction,
claimSessionTranscriptDisplayInTransaction,
} from "./session-transcript-display-rebuild-store.js";
import { ensureSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
vi.mock("../../infra/tmp-openclaw-dir.js", () => ({
DEFAULT_POSIX_TMP_ROOT: "/tmp/openclaw",
@@ -77,12 +78,17 @@ describe("transcript display rebuild persistence", () => {
);
}
const generation = `generation-${entry.sessionId}`;
const sourceGeneration = ensureSessionTranscriptSourceGenerationInTransaction(
{ db },
entry.sessionId,
);
const claimId = Date.now();
db.exec("BEGIN IMMEDIATE");
expect(
claimSessionTranscriptDisplayInTransaction(db, {
claimId,
generation,
previousGeneration: null,
sessionId: entry.sessionId,
}),
).toBe(true);
@@ -100,6 +106,8 @@ describe("transcript display rebuild persistence", () => {
sourceEventSeq,
})),
sessionId: entry.sessionId,
sourceGeneration,
sourceIndexedSeq: ROW_COUNT - 1,
}),
).toBe(true);
db.exec("COMMIT");
src/config/sessions/session-transcript-display-rebuild-store.ts
+59 -21
View File
@@ -1,10 +1,6 @@
import type { DatabaseSync } from "node:sqlite";
import type { Generated } from "kysely";
import {
executeSqliteQuerySync,
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js";
import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js";
import type {
PreparedSessionTranscriptDisplayCarry,
@@ -30,6 +26,11 @@ import {
writeDisplayReducerCarry,
writeDisplayState,
} from "./session-transcript-display-store.js";
import {
EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
readSessionTranscriptSourceGenerationInTransaction,
sessionTranscriptSourceGenerationMatchesInTransaction,
} from "./session-transcript-source-generation.js";
const SESSION_TRANSCRIPT_DISPLAY_PAGE_MAX_ROWS = 200;
// Node's SQLite builds default to 32,766 variables per statement. Leave room for
@@ -82,26 +83,33 @@ export function claimSessionTranscriptDisplayInTransaction(
params: {
claimId: number;
generation: string;
previousGeneration: string | null;
sessionId: string;
},
): boolean {
const state = readSessionTranscriptDisplayState(db, params.sessionId);
if (!state) {
if (params.previousGeneration !== null) {
return false;
}
writeDisplayState(db, params.sessionId, {
generation: params.generation,
indexedSeq: -1,
indexedSeq: EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
needsRebuild: true,
rowCount: 0,
sourceGeneration: null,
updatedAt: params.claimId,
});
return true;
}
if (state.generation !== params.generation) {
if (state.generation !== params.previousGeneration) {
return false;
}
writeDisplayState(db, params.sessionId, {
...state,
generation: params.generation,
needsRebuild: true,
sourceGeneration: null,
updatedAt: params.claimId,
});
return true;
@@ -109,13 +117,39 @@ export function claimSessionTranscriptDisplayInTransaction(
function displayClaimIsOwned(
db: DatabaseSync,
params: { claimId: number; generation: string; sessionId: string },
params: {
claimId: number;
generation: string;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): boolean {
const state = readSessionTranscriptDisplayState(db, params.sessionId);
return Boolean(
state?.needsRebuild &&
state.generation === params.generation &&
state.updatedAt === params.claimId,
state.updatedAt === params.claimId &&
sessionTranscriptSourceGenerationMatchesInTransaction(db, params.sessionId, {
generation: params.sourceGeneration,
indexedSeq: params.sourceIndexedSeq,
}),
);
}
export function abandonSessionTranscriptDisplayClaimInTransaction(
db: DatabaseSync,
params: { claimId: number; generation: string; sessionId: string },
): void {
executeSqliteQuerySync(
db,
getDisplayKysely(db)
.updateTable(SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE)
.set({ source_generation: null, updated_at: Date.now() })
.where("session_id", "=", params.sessionId)
.where("generation", "=", params.generation)
.where("needs_rebuild", "!=", 0)
.where("updated_at", "=", params.claimId),
);
}
@@ -126,6 +160,8 @@ export function deleteSessionTranscriptDisplayChunkInTransaction(
generation: string;
maxRows: number;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): DisplayDeleteChunkResult {
if (!displayClaimIsOwned(db, params)) {
@@ -164,6 +200,8 @@ export function appendSessionTranscriptDisplayChunkInTransaction(
generation: string;
rows: readonly PreparedSessionTranscriptDisplayRow[];
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): boolean {
if (!displayClaimIsOwned(db, params)) {
@@ -235,6 +273,7 @@ export function finalizeSessionTranscriptDisplayInTransaction(
carry: readonly PreparedSessionTranscriptDisplayCarry[];
rowCount: number;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): boolean {
@@ -250,6 +289,7 @@ export function finalizeSessionTranscriptDisplayInTransaction(
indexed_seq: params.sourceIndexedSeq,
needs_rebuild: 0,
row_count: params.rowCount,
source_generation: params.sourceGeneration,
updated_at: Date.now(),
})
.where("session_id", "=", params.sessionId)
@@ -257,7 +297,10 @@ export function finalizeSessionTranscriptDisplayInTransaction(
.where("needs_rebuild", "!=", 0)
.where("updated_at", "=", params.claimId),
);
return result.numAffectedRows === 1n;
if (result.numAffectedRows !== 1n) {
return false;
}
return true;
}
function normalizeDisplayPageLimit(limit: number): number {
@@ -273,21 +316,16 @@ function readSessionTranscriptDisplayRowsSnapshot(
params: SessionTranscriptDisplayReadParams,
): SessionTranscriptDisplayReadResult {
const state = readSessionTranscriptDisplayState(db, sessionId);
const latest = executeSqliteQueryTakeFirstSync(
db,
getDisplayKysely(db)
.selectFrom("transcript_events")
.select("seq")
.where("session_id", "=", sessionId)
.orderBy("seq", "desc")
.limit(1),
);
const latestSeq = latest?.seq ?? -1;
const source = state
? readSessionTranscriptSourceGenerationInTransaction(db, sessionId)
: undefined;
if (
!source ||
!state ||
state.generation !== params.expectedGeneration ||
state.needsRebuild ||
state.indexedSeq !== latestSeq
state.indexedSeq !== source.indexedSeq ||
state.sourceGeneration !== source.generation
) {
return { generation: state?.generation ?? null, kind: "reset" };
}
src/config/sessions/session-transcript-display-semantics.test.ts
+9
View File
@@ -15,6 +15,7 @@ import {
projectionRow,
readDisplaySnapshot,
} from "./session-transcript-display.test-support.js";
import { readSessionTranscriptSourceGenerationTokenInTransaction } from "./session-transcript-source-generation.js";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
@@ -48,10 +49,18 @@ describe("session transcript display semantics", () => {
"INSERT INTO transcript_events (session_id, seq, event_json, created_at) VALUES (?, ?, ?, ?)",
)
.run(sessionId, seq, JSON.stringify(event), seq + 1);
const sourceGeneration = readSessionTranscriptSourceGenerationTokenInTransaction(
database.db,
sessionId,
);
if (!sourceGeneration) {
throw new Error("expected transcript source generation");
}
appendEligibleSessionTranscriptDisplayRowInTransaction(database.db, {
event,
seq,
sessionId,
sourceGeneration,
});
},
{ agentId, env },
src/config/sessions/session-transcript-display-store.ts
+22 -3
View File
@@ -31,12 +31,14 @@ import {
SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE,
validateOpenClawAgentDisplayRowSchema,
} from "../../state/openclaw-agent-display-row-schema.js";
import { EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ } from "./session-transcript-source-generation.js";
type SessionTranscriptDisplayState = {
generation: string;
indexedSeq: number;
needsRebuild: boolean;
rowCount: number;
sourceGeneration: string | null;
updatedAt: number;
};
type DisplayRowDatabase = Omit<
@@ -72,7 +74,14 @@ export function readSessionTranscriptDisplayState(
db,
getDisplayKysely(db)
.selectFrom(SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE)
.select(["generation", "indexed_seq", "needs_rebuild", "row_count", "updated_at"])
.select([
"generation",
"indexed_seq",
"needs_rebuild",
"row_count",
"source_generation",
"updated_at",
])
.where("session_id", "=", sessionId),
);
return row
@@ -81,6 +90,7 @@ export function readSessionTranscriptDisplayState(
indexedSeq: row.indexed_seq,
needsRebuild: row.needs_rebuild !== 0,
rowCount: row.row_count,
sourceGeneration: row.source_generation,
updatedAt: row.updated_at,
}
: undefined;
@@ -101,6 +111,7 @@ export function writeDisplayState(
needs_rebuild: state.needsRebuild ? 1 : 0,
row_count: state.rowCount,
session_id: sessionId,
source_generation: state.sourceGeneration,
updated_at: state.updatedAt,
})
.onConflict((conflict) =>
@@ -109,6 +120,7 @@ export function writeDisplayState(
indexed_seq: state.indexedSeq,
needs_rebuild: state.needsRebuild ? 1 : 0,
row_count: state.rowCount,
source_generation: state.sourceGeneration,
updated_at: state.updatedAt,
}),
),
@@ -125,9 +137,10 @@ export function invalidateSessionTranscriptDisplayInTransaction(
const generation = createDisplayGeneration();
writeDisplayState(db, sessionId, {
generation,
indexedSeq: state?.indexedSeq ?? -1,
indexedSeq: state?.indexedSeq ?? EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
needsRebuild: true,
rowCount: state?.rowCount ?? 0,
sourceGeneration: null,
updatedAt: Date.now(),
});
return generation;
@@ -546,13 +559,17 @@ function createDatabaseDisplayEffects(
/** Extends one ready display generation after active-path eligibility is already proven. */
export function appendEligibleSessionTranscriptDisplayRowInTransaction(
db: DatabaseSync,
params: { event: unknown; seq: number; sessionId: string },
params: { event: unknown; seq: number; sessionId: string; sourceGeneration: string },
): boolean {
ensureOpenClawAgentDisplayRowSchema(db);
const state = readSessionTranscriptDisplayState(db, params.sessionId);
if (state?.needsRebuild) {
return true;
}
if (state && state.sourceGeneration !== params.sourceGeneration) {
invalidateSessionTranscriptDisplayInTransaction(db, params.sessionId);
return true;
}
if (state && params.seq !== state.indexedSeq + 1) {
invalidateSessionTranscriptDisplayInTransaction(db, params.sessionId);
return true;
@@ -569,6 +586,7 @@ export function appendEligibleSessionTranscriptDisplayRowInTransaction(
indexedSeq: params.seq - 1,
needsRebuild: false,
rowCount: 0,
sourceGeneration: params.sourceGeneration,
updatedAt: Date.now(),
});
}
@@ -606,6 +624,7 @@ export function appendEligibleSessionTranscriptDisplayRowInTransaction(
indexedSeq: params.seq,
needsRebuild: false,
rowCount: effects.rowCount(),
sourceGeneration: params.sourceGeneration,
updatedAt: Date.now(),
});
return false;
src/config/sessions/session-transcript-display.test-support.ts
+15 -4
View File
@@ -1,9 +1,8 @@
import type { DatabaseSync } from "node:sqlite";
import { openOpenClawAgentDatabase } from "../../state/openclaw-agent-db.js";
import { prepareSessionTranscriptDisplayProjection } from "./session-transcript-display.js";
import {
buildSessionTranscriptProjection,
type SessionTranscriptProjectionSourceRow,
} from "./session-transcript-projection-rebuild.js";
import { buildSessionTranscriptProjection } from "./session-transcript-projection-rebuild.js";
import { readSessionTranscriptSourceGenerationTokenInTransaction } from "./session-transcript-source-generation.js";
type DatabaseScope = {
agentId: string;
@@ -12,6 +11,9 @@ type DatabaseScope = {
type PreparedSessionTranscriptDisplayProjection = ReturnType<
typeof prepareSessionTranscriptDisplayProjection
>;
type SessionTranscriptProjectionSourceRow = Parameters<
typeof buildSessionTranscriptProjection
>[0]["rows"][number];
export function projectionRow(
seq: number,
@@ -25,10 +27,19 @@ export function projectionFixture(rows: SessionTranscriptProjectionSourceRow[])
return buildSessionTranscriptProjection({
rows,
sessionId: "projection-session",
sourceGeneration: "test-source-generation",
sourceTranscriptUpdatedAt: 42,
});
}
export function readRequiredSourceGeneration(db: DatabaseSync, sessionId: string) {
const generation = readSessionTranscriptSourceGenerationTokenInTransaction(db, sessionId);
if (!generation) {
throw new Error("expected transcript source generation");
}
return generation;
}
export function canvasUrlWithLength(length: number): string {
const prefix = "/__openclaw__/canvas/documents/cv/";
const segmentCount = 16;
src/config/sessions/session-transcript-display.test.ts
+165 -2
View File
@@ -13,6 +13,8 @@ import {
} from "./session-accessor.js";
import { copySqliteSessionOwnedStateForCanonicalRepair } from "./session-accessor.sqlite-canonical-repair.js";
import { importSqliteSessionRows } from "./session-accessor.sqlite-import.js";
import { resolveSqliteTranscriptScope } from "./session-accessor.sqlite-scope.js";
import { appendTranscriptEventInTransaction } from "./session-accessor.sqlite-transcript-store.js";
import {
replaceTranscriptEvents,
rewriteTranscriptEventRowsExact,
@@ -29,6 +31,15 @@ import {
readSessionTranscriptDisplayRowsInTransaction,
readSessionTranscriptDisplayState,
} from "./session-transcript-display.js";
import {
plannedDisplaySnapshot,
readDisplaySnapshot,
} from "./session-transcript-display.test-support.js";
import {
claimPreparedSessionTranscriptProjectionInTransaction,
finalizePreparedSessionTranscriptProjectionInTransaction,
prepareSessionTranscriptProjection,
} from "./session-transcript-projection-rebuild.js";
import {
reconcileSessionTranscriptDisplayProjection,
waitForSessionTranscriptIndexReconcile,
@@ -109,6 +120,20 @@ describe("SQLite transcript display rows", () => {
.all(sessionId) as DisplayRow[];
}
function readSourceIdentity(sessionId = scope.sessionId) {
const db = database().db;
const generation = db
.prepare("SELECT generation FROM transcript_rewrite_watermarks WHERE session_id = ?")
.get(sessionId) as { generation: string };
const frontier = db
.prepare("SELECT MAX(seq) AS indexed_seq FROM transcript_events WHERE session_id = ?")
.get(sessionId) as { indexed_seq: number | null };
return {
sourceGeneration: generation.generation,
sourceIndexedSeq: frontier.indexed_seq ?? -1,
};
}
function trackDisplayGenerationWrites(sessionId: string): void {
const db = database().db;
readSessionTranscriptDisplayState(db, sessionId);
@@ -272,6 +297,26 @@ describe("SQLite transcript display rows", () => {
expect(readRows()).toEqual(secondRows);
});
it("resets a ready display when its source generation is stale", async () => {
await appendPlainPair();
const ready = readState();
database()
.db.prepare(
`UPDATE session_transcript_display_state
SET source_generation = 'stale-source'
WHERE session_id = ?`,
)
.run(scope.sessionId);
expect(
readPage({
expectedGeneration: ready.generation,
fromOrdinal: 0,
limit: 10,
}),
).toEqual({ generation: ready.generation, kind: "reset" });
});
it("rotates on a boundary and publishes one dense rebuilt generation", async () => {
await appendPlainPair();
const before = readState();
@@ -418,6 +463,7 @@ describe("SQLite transcript display rows", () => {
return { event: JSON.parse(source.event_json), seq: source.seq };
});
const plan = prepareSessionTranscriptDisplayProjection(sourceRows);
const source = readSourceIdentity();
expect(plan.rows[0]?.semanticSources).toMatchObject([
{ relation: "tts_supplement", sourceEventSeq: 2 },
]);
@@ -434,6 +480,7 @@ describe("SQLite transcript display rows", () => {
claimSessionTranscriptDisplayInTransaction(agentDatabase.db, {
claimId: firstClaim,
generation: firstGeneration,
previousGeneration: firstGeneration,
sessionId: scope.sessionId,
}),
).toBe(true);
@@ -444,6 +491,7 @@ describe("SQLite transcript display rows", () => {
generation: firstGeneration,
maxRows: 1,
sessionId: scope.sessionId,
...source,
});
expect(deleted.owned).toBe(true);
} while (deleted.hasMore);
@@ -453,6 +501,7 @@ describe("SQLite transcript display rows", () => {
generation: firstGeneration,
rows: plan.rows.slice(0, 1),
sessionId: scope.sessionId,
...source,
}),
).toBe(true);
},
@@ -472,6 +521,7 @@ describe("SQLite transcript display rows", () => {
generation: firstGeneration,
rows: plan.rows.slice(1),
sessionId: scope.sessionId,
...source,
}),
).toBe(false);
expect(
@@ -481,7 +531,7 @@ describe("SQLite transcript display rows", () => {
generation: firstGeneration,
rowCount: plan.rows.length,
sessionId: scope.sessionId,
sourceIndexedSeq: sourceRows.at(-1)?.seq ?? -1,
...source,
}),
).toBe(false);
},
@@ -502,6 +552,7 @@ describe("SQLite transcript display rows", () => {
claimSessionTranscriptDisplayInTransaction(agentDatabase.db, {
claimId: retryClaim,
generation: retryGeneration,
previousGeneration: retryGeneration,
sessionId: scope.sessionId,
}),
).toBe(true);
@@ -512,6 +563,7 @@ describe("SQLite transcript display rows", () => {
generation: retryGeneration,
maxRows: 1,
sessionId: scope.sessionId,
...source,
});
} while (deleted.hasMore);
expect(
@@ -520,6 +572,7 @@ describe("SQLite transcript display rows", () => {
generation: retryGeneration,
rows: plan.rows,
sessionId: scope.sessionId,
...source,
}),
).toBe(true);
expect(
@@ -529,7 +582,7 @@ describe("SQLite transcript display rows", () => {
generation: retryGeneration,
rowCount: plan.rows.length,
sessionId: scope.sessionId,
sourceIndexedSeq: sourceRows.at(-1)?.seq ?? -1,
...source,
}),
).toBe(true);
},
@@ -570,6 +623,116 @@ describe("SQLite transcript display rows", () => {
]);
});
it("rotates a lag rebuild identity in its claim before replacing rows", async () => {
await appendPlainPair();
const before = readState();
const beforeRows = readRows();
const resolved = resolveSqliteTranscriptScope(scope);
runOpenClawAgentWriteTransaction(
(agentDatabase) => {
expect(
appendTranscriptEventInTransaction(
agentDatabase,
resolved,
{
type: "message",
id: "lagged-assistant",
parentId: "assistant-1",
message: { role: "assistant", content: "lagged" },
},
{ maintainDisplayProjection: false, scheduleProjectionReconcile: false },
),
).toBe(true);
},
{ agentId: scope.agentId, env: scope.env },
);
const plan = prepareSessionTranscriptProjection(database().db, scope.sessionId, {
includeDisplayProjection: true,
});
expect(plan).toMatchObject({
activeNeedsRebuild: false,
displayNeedsRebuild: true,
displayPreviousGeneration: before.generation,
});
expect(plan?.displayGeneration).not.toBe(before.generation);
const claimId = -404;
runOpenClawAgentWriteTransaction(
(agentDatabase) => {
expect(
claimPreparedSessionTranscriptProjectionInTransaction(agentDatabase.db, plan!, claimId),
).toBe(true);
expect(readSessionTranscriptDisplayState(agentDatabase.db, scope.sessionId)).toMatchObject({
generation: plan!.displayGeneration,
needsRebuild: true,
updatedAt: claimId,
});
expect(readRows()).toEqual(beforeRows);
},
{ agentId: scope.agentId, env: scope.env },
);
expect(readPage({ expectedGeneration: before.generation, fromOrdinal: 0, limit: 10 })).toEqual({
generation: plan!.displayGeneration,
kind: "reset",
});
const source = {
sourceGeneration: plan!.sourceGeneration,
sourceIndexedSeq: plan!.sourceIndexedSeq,
};
runOpenClawAgentWriteTransaction(
(agentDatabase) => {
let deleted;
do {
deleted = deleteSessionTranscriptDisplayChunkInTransaction(agentDatabase.db, {
claimId,
generation: plan!.displayGeneration,
maxRows: 1,
sessionId: scope.sessionId,
...source,
});
expect(deleted.owned).toBe(true);
} while (deleted.hasMore);
expect(
appendSessionTranscriptDisplayChunkInTransaction(agentDatabase.db, {
claimId,
generation: plan!.displayGeneration,
rows: plan!.displayRows,
sessionId: scope.sessionId,
...source,
}),
).toBe(true);
expect(
finalizePreparedSessionTranscriptProjectionInTransaction(
agentDatabase.db,
plan!,
claimId,
),
).toBe(true);
},
{ agentId: scope.agentId, env: scope.env },
);
expect(
readDisplaySnapshot({ agentId: scope.agentId, env: scope.env }, scope.sessionId),
).toEqual(
plannedDisplaySnapshot(
database()
.db.prepare(
"SELECT seq, event_json, created_at FROM transcript_events WHERE session_id = ? ORDER BY seq",
)
.all(scope.sessionId)
.map((entry) => {
const row = entry as { created_at: number; event_json: string; seq: number };
return { createdAt: row.created_at, event: JSON.parse(row.event_json), seq: row.seq };
}),
),
);
expect(readRows().map((row) => row.display_ordinal)).toEqual(
readRows().map((_, index) => index),
);
});
it("publishes an empty ready generation after clearing a transcript", async () => {
await appendPlainPair();
const before = readState();
src/config/sessions/session-transcript-index.ts
+202 -95
View File
@@ -6,6 +6,7 @@
// same transaction, anything ambiguous (leaf controls, branch switches)
// marks the session dirty for its write or maintenance owner to rebuild from
// the canonical visible-path resolver.
import { randomInt } from "node:crypto";
import type { DatabaseSync } from "node:sqlite";
import type { ColumnType } from "kysely";
import {
@@ -15,6 +16,8 @@ import {
} from "../../infra/kysely-sync.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import { ensureOpenClawAgentDisplayRowSchema } from "../../state/openclaw-agent-display-row-schema.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "../../state/openclaw-agent-transcript-projection-source-schema.js";
import { chunkItems } from "../../utils/chunk-items.js";
import {
appendEligibleSessionTranscriptDisplayRowInTransaction,
hasTranscriptMessage,
@@ -24,11 +27,19 @@ import {
shouldProjectActiveEvent,
} from "./session-transcript-display.js";
import {
appendPreparedSessionTranscriptProjectionChunkInTransaction,
buildSessionTranscriptProjection,
claimPreparedSessionTranscriptProjectionInTransaction,
deletePreparedSessionTranscriptProjectionChunkInTransaction,
extractTranscriptIndexEntry,
type SessionTranscriptProjectionSourceRow,
finalizePreparedSessionTranscriptProjectionInTransaction,
type TranscriptIndexEntry,
} from "./session-transcript-projection-rebuild.js";
import {
EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
readSessionTranscriptSourceGenerationInTransaction,
readSessionTranscriptSourceGenerationTokenInTransaction,
} from "./session-transcript-source-generation.js";
import {
isCanonicalSessionTranscriptEntry,
isSessionTranscriptLeafControl,
@@ -46,6 +57,7 @@ type TranscriptIndexDatabase = Omit<
| "session_transcript_display_state"
| "session_transcript_fts"
| "session_transcript_index_state"
| "transcript_rewrite_watermarks"
| "transcript_events"
>,
"session_transcript_fts"
@@ -58,12 +70,16 @@ type TranscriptIndexDatabase = Omit<
};
};
const SYNCHRONOUS_PROJECTION_CHUNK_ROWS = 512;
const SYNCHRONOUS_PROJECTION_FTS_CHUNK_ROWS = 128;
export type SessionTranscriptProjectionState = {
activeEventCount: number;
activeMessageCount: number;
indexedSeq: number;
leafEventId: string | null;
needsRebuild: boolean;
sourceGeneration: string | null;
};
function getIndexKysely(db: DatabaseSync) {
@@ -74,6 +90,7 @@ function readSessionTranscriptProjectionState(
db: DatabaseSync,
sessionId: string,
): SessionTranscriptProjectionState | undefined {
ensureOpenClawAgentTranscriptProjectionSourceColumns(db);
const row = executeSqliteQueryTakeFirstSync(
db,
getIndexKysely(db)
@@ -84,6 +101,7 @@ function readSessionTranscriptProjectionState(
"indexed_seq",
"leaf_event_id",
"needs_rebuild",
"source_generation",
])
.where("session_id", "=", sessionId),
);
@@ -96,24 +114,22 @@ function readSessionTranscriptProjectionState(
indexedSeq: row.indexed_seq,
leafEventId: row.leaf_event_id,
needsRebuild: row.needs_rebuild !== 0,
sourceGeneration: row.source_generation,
};
}
export function sessionTranscriptIndexNeedsReconcile(db: DatabaseSync, sessionId: string): boolean {
const latest = executeSqliteQueryTakeFirstSync(
db,
getIndexKysely(db)
.selectFrom("transcript_events")
.select("seq")
.where("session_id", "=", sessionId)
.orderBy("seq", "desc")
.limit(1),
);
if (!latest) {
const source = readSessionTranscriptSourceGenerationInTransaction(db, sessionId);
if (!source || source.indexedSeq < 0) {
return false;
}
const state = readSessionTranscriptProjectionState(db, sessionId);
return !state || state.needsRebuild || state.indexedSeq !== latest.seq;
return (
!state ||
state.needsRebuild ||
state.indexedSeq !== source.indexedSeq ||
state.sourceGeneration !== source.generation
);
}
function writeWatermark(
@@ -121,7 +137,11 @@ function writeWatermark(
sessionId: string,
watermark: SessionTranscriptProjectionState,
now: number,
sourceGeneration?: string,
): void {
if (!watermark.needsRebuild && !sourceGeneration) {
throw new Error(`Transcript source generation is missing for ${sessionId}`);
}
executeSqliteQuerySync(
db,
getIndexKysely(db)
@@ -133,6 +153,7 @@ function writeWatermark(
indexed_seq: watermark.indexedSeq,
leaf_event_id: watermark.leafEventId,
needs_rebuild: watermark.needsRebuild ? 1 : 0,
source_generation: watermark.needsRebuild ? null : sourceGeneration,
updated_at: now,
})
.onConflict((conflict) =>
@@ -142,6 +163,7 @@ function writeWatermark(
indexed_seq: watermark.indexedSeq,
leaf_event_id: watermark.leafEventId,
needs_rebuild: watermark.needsRebuild ? 1 : 0,
source_generation: watermark.needsRebuild ? null : sourceGeneration,
updated_at: now,
}),
),
@@ -226,11 +248,18 @@ export function indexAppendedTranscriptEventInTransaction(
createdAt: number;
/** True maintains, false skips for a batch owner, omission invalidates adopted state. */
maintainDisplayProjection?: boolean;
sourceGeneration?: string;
},
): boolean {
const existingDisplayInvalidated =
params.maintainDisplayProjection === undefined &&
invalidateExistingSessionTranscriptDisplayInTransaction(db, params.sessionId);
const sourceGeneration =
params.sourceGeneration ??
readSessionTranscriptSourceGenerationTokenInTransaction(db, params.sessionId);
if (!sourceGeneration) {
throw new Error(`Transcript source generation is missing for ${params.sessionId}`);
}
const watermark = readSessionTranscriptProjectionState(db, params.sessionId);
if (!watermark) {
if (params.seq !== 0) {
@@ -239,15 +268,24 @@ export function indexAppendedTranscriptEventInTransaction(
invalidateDisplayProjectionForAppend(db, params);
return true;
}
applyForwardIndex(db, params, {
activeEventCount: 0,
activeMessageCount: 0,
indexedSeq: -1,
leafEventId: null,
needsRebuild: false,
});
applyForwardIndex(
db,
params,
{
activeEventCount: 0,
activeMessageCount: 0,
indexedSeq: EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
leafEventId: null,
needsRebuild: false,
sourceGeneration,
},
sourceGeneration,
);
return params.maintainDisplayProjection === true
? appendEligibleSessionTranscriptDisplayRowInTransaction(db, params)
? appendEligibleSessionTranscriptDisplayRowInTransaction(db, {
...params,
sourceGeneration,
})
: existingDisplayInvalidated;
}
if (watermark.needsRebuild) {
@@ -260,6 +298,11 @@ export function indexAppendedTranscriptEventInTransaction(
}
return true;
}
if (watermark.sourceGeneration !== sourceGeneration) {
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
invalidateDisplayProjectionForAppend(db, params);
return true;
}
if (params.seq !== watermark.indexedSeq + 1) {
// Out-of-band writes bypassed the hook; reconcile recomputes the truth.
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
@@ -278,7 +321,7 @@ export function indexAppendedTranscriptEventInTransaction(
return true;
}
if (isSessionTranscriptDisplayBoundary(params.event)) {
applyForwardIndex(db, params, watermark);
applyForwardIndex(db, params, watermark, sourceGeneration);
invalidateDisplayProjectionForAppend(db, params);
return true;
}
@@ -307,9 +350,12 @@ export function indexAppendedTranscriptEventInTransaction(
invalidateDisplayProjectionForAppend(db, params);
return true;
}
applyForwardIndex(db, params, watermark);
applyForwardIndex(db, params, watermark, sourceGeneration);
return params.maintainDisplayProjection === true
? appendEligibleSessionTranscriptDisplayRowInTransaction(db, params)
? appendEligibleSessionTranscriptDisplayRowInTransaction(db, {
...params,
sourceGeneration,
})
: existingDisplayInvalidated;
}
@@ -323,6 +369,7 @@ function applyForwardIndex(
createdAt: number;
},
watermark: SessionTranscriptProjectionState,
sourceGeneration: string,
): void {
const entry = extractTranscriptIndexEntry(params.event, params.createdAt);
if (entry) {
@@ -351,8 +398,10 @@ function applyForwardIndex(
indexedSeq: params.seq,
leafEventId: advancesLeaf ? params.eventId : watermark.leafEventId,
needsRebuild: false,
sourceGeneration,
},
params.createdAt,
sourceGeneration,
);
}
@@ -366,9 +415,10 @@ function markSessionTranscriptIndexDirtyInTransaction(db: DatabaseSync, sessionI
{
activeEventCount: watermark?.activeEventCount ?? 0,
activeMessageCount: watermark?.activeMessageCount ?? 0,
indexedSeq: watermark?.indexedSeq ?? -1,
indexedSeq: watermark?.indexedSeq ?? EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
leafEventId: watermark?.leafEventId ?? null,
needsRebuild: true,
sourceGeneration: null,
},
now,
);
@@ -389,44 +439,6 @@ export function deleteSessionTranscriptIndexInTransaction(
);
}
/**
* Rebuilds one session's index from its full event set: drops existing FTS
* rows, indexes the resolved active branch, and resets the watermark to the
* same append parent the accessor's next append will resolve.
*/
function rebuildSessionTranscriptIndexInTransaction(
db: DatabaseSync,
sessionId: string,
rows: readonly SessionTranscriptProjectionSourceRow[],
): void {
const projection = buildSessionTranscriptProjection({
includeDisplayRows: false,
rows,
sessionId,
sourceTranscriptUpdatedAt: null,
});
deleteFtsRows(db, sessionId);
deleteActiveEventRows(db, sessionId);
for (const entry of projection.ftsRows) {
insertFtsRow(db, sessionId, entry);
}
for (const row of projection.activeRows) {
insertActiveEventRow(db, { ...row, sessionId });
}
writeWatermark(
db,
sessionId,
{
activeEventCount: projection.activeEventCount,
activeMessageCount: projection.activeMessageCount,
indexedSeq: projection.sourceIndexedSeq,
leafEventId: projection.leafEventId,
needsRebuild: false,
},
Date.now(),
);
}
/** Rebuilds one lagging projection under its current write transaction. */
export function reconcileSessionTranscriptIndexInTransaction(
db: DatabaseSync,
@@ -448,6 +460,17 @@ export function reconcileSessionTranscriptIndexInTransaction(
if (!sessionTranscriptIndexNeedsReconcile(db, sessionId)) {
return false;
}
const source = readSessionTranscriptSourceGenerationInTransaction(db, sessionId);
const session = executeSqliteQueryTakeFirstSync(
db,
getIndexKysely(db)
.selectFrom("session_windows")
.select("transcript_updated_at")
.where("session_id", "=", sessionId),
);
if (!source || !session) {
throw new Error(`Transcript source generation is missing for ${sessionId}`);
}
const rows = executeSqliteQuerySync(
db,
getIndexKysely(db)
@@ -456,15 +479,65 @@ export function reconcileSessionTranscriptIndexInTransaction(
.where("session_id", "=", sessionId)
.orderBy("seq", "asc"),
).rows;
rebuildSessionTranscriptIndexInTransaction(
db,
sessionId,
rows.map((row) => ({
const projection = buildSessionTranscriptProjection({
activeNeedsRebuild: true,
displayNeedsRebuild: false,
includeDisplayRows: false,
rows: rows.map((row) => ({
createdAt: row.created_at,
event: JSON.parse(row.event_json) as unknown,
seq: row.seq,
createdAt: row.created_at,
})),
);
sessionId,
sourceGeneration: source.generation,
sourceTranscriptUpdatedAt: session.transcript_updated_at,
});
const claimId = -randomInt(1, 2 ** 47);
if (!claimPreparedSessionTranscriptProjectionInTransaction(db, projection, claimId)) {
return false;
}
let deleted;
do {
deleted = deletePreparedSessionTranscriptProjectionChunkInTransaction(db, {
claimId,
maxRowsPerTable: SYNCHRONOUS_PROJECTION_CHUNK_ROWS,
sessionId,
sourceGeneration: projection.sourceGeneration,
sourceIndexedSeq: projection.sourceIndexedSeq,
});
if (!deleted.owned) {
throw new Error(`Transcript projection claim changed while rebuilding ${sessionId}`);
}
} while (deleted.hasMore);
for (const activeRows of chunkItems(projection.activeRows, SYNCHRONOUS_PROJECTION_CHUNK_ROWS)) {
if (
!appendPreparedSessionTranscriptProjectionChunkInTransaction(db, {
activeRows,
claimId,
sessionId,
sourceGeneration: projection.sourceGeneration,
sourceIndexedSeq: projection.sourceIndexedSeq,
})
) {
throw new Error(`Transcript projection claim changed while rebuilding ${sessionId}`);
}
}
for (const ftsRows of chunkItems(projection.ftsRows, SYNCHRONOUS_PROJECTION_FTS_CHUNK_ROWS)) {
if (
!appendPreparedSessionTranscriptProjectionChunkInTransaction(db, {
claimId,
ftsRows,
sessionId,
sourceGeneration: projection.sourceGeneration,
sourceIndexedSeq: projection.sourceIndexedSeq,
})
) {
throw new Error(`Transcript projection claim changed while rebuilding ${sessionId}`);
}
}
if (!finalizePreparedSessionTranscriptProjectionInTransaction(db, projection, claimId)) {
throw new Error(`Transcript projection claim changed while finalizing ${sessionId}`);
}
return true;
}
@@ -474,16 +547,19 @@ export function reconcileSessionTranscriptIndexInTransaction(
* behind the newest row. Ordered for deterministic reconcile passes.
*/
function hasTranscriptRows(db: DatabaseSync): boolean {
return Boolean(
db.prepare("SELECT 1 FROM transcript_events LIMIT 1").get(), // sqlite-allow-raw -- Avoid creating the lazy display group for an unused agent database.
);
return Boolean(db.prepare("SELECT 1 FROM transcript_events LIMIT 1").get()); // sqlite-allow-raw -- Avoid creating the lazy display group for an unused agent database.
}
/** Lists sessions whose active and FTS projections lag canonical transcript rows. */
function sessionIds(rows: readonly { session_id: unknown }[]): string[] {
return rows.flatMap(({ session_id }) => (typeof session_id === "string" ? [session_id] : []));
}
/** Lists sessions whose active and FTS projection is not bound to the current source. */
export function listSessionsNeedingTranscriptIndexReconcile(db: DatabaseSync): string[] {
if (!hasTranscriptRows(db)) {
return [];
}
ensureOpenClawAgentTranscriptProjectionSourceColumns(db);
const rows = executeSqliteQuerySync(
db,
getIndexKysely(db)
@@ -509,31 +585,42 @@ export function listSessionsNeedingTranscriptIndexReconcile(db: DatabaseSync): s
"st.session_id",
"session_windows.session_id",
)
.leftJoin(
"transcript_rewrite_watermarks as source",
"source.session_id",
"session_windows.session_id",
)
.select("session_windows.session_id")
.where((eb) =>
eb.or([
eb(eb.fn.coalesce("st.needs_rebuild", eb.val(1)), "!=", 0),
eb("latest.seq", ">", eb.fn.coalesce("st.indexed_seq", eb.val(-1))),
eb(
"latest.seq",
">",
eb.fn.coalesce("st.indexed_seq", eb.val(EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ)),
),
eb("st.source_generation", "is", null),
eb("source.generation", "is", null),
eb("st.source_generation", "!=", eb.ref("source.generation")),
]),
)
.orderBy("session_windows.session_id"),
).rows;
return rows.flatMap((row) => (typeof row.session_id === "string" ? [row.session_id] : []));
return sessionIds(rows);
}
/** Lists sessions whose active, FTS, or display projection requires repair. */
export function listSessionsNeedingTranscriptProjectionReconcile(db: DatabaseSync): string[] {
const transcriptRowsPresent = hasTranscriptRows(db);
const hasDisplayStateTable = Boolean(
// sqlite-allow-raw -- Avoid installing the lazy display group solely to decide whether reconcile has work.
db
.prepare(/* sqlite-allow-raw */ SQLITE_TABLE_EXISTS_SQL)
.get("session_transcript_display_state"),
// sqlite-allow-raw -- Probe the lazy schema without installing it.
db.prepare(SQLITE_TABLE_EXISTS_SQL).get("session_transcript_display_state"),
);
if (!transcriptRowsPresent && !hasDisplayStateTable) {
return [];
}
ensureOpenClawAgentDisplayRowSchema(db);
ensureOpenClawAgentTranscriptProjectionSourceColumns(db);
const kysely = getIndexKysely(db);
const rows = transcriptRowsPresent
? executeSqliteQuerySync(
@@ -561,36 +648,56 @@ export function listSessionsNeedingTranscriptProjectionReconcile(db: DatabaseSyn
"display.session_id",
"session_windows.session_id",
)
.leftJoin(
"transcript_rewrite_watermarks as source",
"source.session_id",
"session_windows.session_id",
)
.select("session_windows.session_id")
.where((eb) =>
eb.or([
eb(eb.fn.coalesce("display.needs_rebuild", eb.val(1)), "!=", 0),
eb("latest.seq", ">", eb.fn.coalesce("display.indexed_seq", eb.val(-1))),
eb(
"latest.seq",
">",
eb.fn.coalesce(
"display.indexed_seq",
eb.val(EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ),
),
),
eb("display.source_generation", "is", null),
eb("source.generation", "is", null),
eb("display.source_generation", "!=", eb.ref("source.generation")),
]),
)
// The transcript PK makes the correlated latest-row lookup one index seek per session.
// Grouping transcript_events here made every healthy search rescan the entire history.
.orderBy("session_windows.session_id"),
).rows
: [];
const emptyDirtyRows = executeSqliteQuerySync(
const displayDirtyRows = executeSqliteQuerySync(
db,
kysely
.selectFrom("session_transcript_display_state")
.select("session_id")
.where("needs_rebuild", "!=", 0),
.selectFrom("session_transcript_display_state as display")
.leftJoin(
"transcript_rewrite_watermarks as source",
"source.session_id",
"display.session_id",
)
.select("display.session_id")
.where((eb) =>
eb.or([
eb("display.needs_rebuild", "!=", 0),
eb("display.source_generation", "is", null),
eb("source.generation", "is", null),
eb("display.source_generation", "!=", eb.ref("source.generation")),
]),
),
).rows;
return [
...new Set(
[...listSessionsNeedingTranscriptIndexReconcile(db), ...rows, ...emptyDirtyRows].flatMap(
(row) =>
typeof row === "string"
? [row]
: typeof row.session_id === "string"
? [row.session_id]
: [],
),
),
...new Set([
...listSessionsNeedingTranscriptIndexReconcile(db),
...sessionIds(rows),
...sessionIds(displayDirtyRows),
]),
].toSorted();
}
src/config/sessions/session-transcript-projection-rebuild.test.ts
+3 -1
View File
@@ -24,11 +24,12 @@ import {
projectionRow as row,
readDisplayRowIdentities,
readDisplaySnapshot,
readRequiredSourceGeneration,
serializeDisplayTables,
} from "./session-transcript-display.test-support.js";
import type { SessionTranscriptProjectionSourceRow } from "./session-transcript-projection-rebuild.js";
import { reconcileSessionTranscriptDisplayProjection } from "./session-transcript-reconcile.js";
type SessionTranscriptProjectionSourceRow = ReturnType<typeof row>;
const SESSION_ID = "projection-session";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
@@ -97,6 +98,7 @@ describe("canonical session transcript projection", () => {
event,
seq,
sessionId,
sourceGeneration: readRequiredSourceGeneration(database.db, sessionId),
});
},
{ agentId: scope.agentId, env },
src/config/sessions/session-transcript-projection-rebuild.ts
+118 -35
View File
@@ -9,6 +9,7 @@ import {
import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
abandonSessionTranscriptDisplayClaimInTransaction,
claimSessionTranscriptDisplayInTransaction,
finalizeSessionTranscriptDisplayInTransaction,
hasTranscriptMessage,
@@ -19,6 +20,11 @@ import {
type PreparedSessionTranscriptDisplayCarry,
type PreparedSessionTranscriptDisplayRow,
} from "./session-transcript-display.js";
import {
EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
readSessionTranscriptSourceGenerationInTransaction,
sessionTranscriptSourceGenerationMatchesInTransaction,
} from "./session-transcript-source-generation.js";
import {
resolveVisibleTranscriptAppendParentId,
selectVisibleTranscriptEventEntries,
@@ -54,9 +60,11 @@ export type PreparedSessionTranscriptProjectionMetadata = {
displayCarry: PreparedSessionTranscriptDisplayCarry[];
displayGeneration: string;
displayNeedsRebuild: boolean;
displayPreviousGeneration: string | null;
displayRowCount: number;
leafEventId: string | null;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
sourceTranscriptUpdatedAt: number | null;
};
@@ -71,7 +79,7 @@ export type PreparedSessionTranscriptProjection = PreparedSessionTranscriptProje
ftsRows: TranscriptIndexEntry[];
};
export type SessionTranscriptProjectionSourceRow = {
type SessionTranscriptProjectionSourceRow = {
createdAt: number;
event: unknown;
seq: number;
@@ -157,9 +165,11 @@ export function buildSessionTranscriptProjection(params: {
activeNeedsRebuild?: boolean;
displayGeneration?: string;
displayNeedsRebuild?: boolean;
displayPreviousGeneration?: string | null;
includeDisplayRows?: boolean;
rows: readonly SessionTranscriptProjectionSourceRow[];
sessionId: string;
sourceGeneration: string;
sourceTranscriptUpdatedAt: number | null;
}): PreparedSessionTranscriptProjection {
const now = Date.now();
@@ -203,12 +213,14 @@ export function buildSessionTranscriptProjection(params: {
displayCarry: displayProjection.carry,
displayGeneration: params.displayGeneration ?? randomUUID().replaceAll("-", ""),
displayNeedsRebuild: params.displayNeedsRebuild ?? true,
displayPreviousGeneration: params.displayPreviousGeneration ?? null,
displayRowCount: displayRows.length,
displayRows,
ftsRows,
leafEventId: resolveVisibleTranscriptAppendParentId(events),
sessionId: params.sessionId,
sourceIndexedSeq: params.rows.at(-1)?.seq ?? -1,
sourceGeneration: params.sourceGeneration,
sourceIndexedSeq: params.rows.at(-1)?.seq ?? EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
sourceTranscriptUpdatedAt: params.sourceTranscriptUpdatedAt,
};
}
@@ -241,12 +253,16 @@ export function prepareSessionTranscriptProjection(
if (!session) {
return undefined;
}
const latestSeq = rows.at(-1)?.seq ?? -1;
const source = readSessionTranscriptSourceGenerationInTransaction(db, sessionId);
if (!source) {
return undefined;
}
const latestSeq = source.indexedSeq;
const activeState = executeSqliteQueryTakeFirstSync(
db,
kysely
.selectFrom("session_transcript_index_state")
.select(["indexed_seq", "needs_rebuild"])
.select(["indexed_seq", "needs_rebuild", "source_generation"])
.where("session_id", "=", sessionId),
);
const includeDisplayProjection = options.includeDisplayProjection === true;
@@ -255,23 +271,33 @@ export function prepareSessionTranscriptProjection(
: undefined;
const displayNeedsRebuild =
includeDisplayProjection &&
(!displayState || displayState.needsRebuild || displayState.indexedSeq !== latestSeq);
(!displayState ||
displayState.needsRebuild ||
displayState.indexedSeq !== latestSeq ||
displayState.sourceGeneration !== source.generation);
const displayGeneration =
displayState && (!displayNeedsRebuild || displayState.needsRebuild)
? displayState.generation
: randomUUID().replaceAll("-", "");
return buildSessionTranscriptProjection({
activeNeedsRebuild:
latestSeq >= 0 &&
(!activeState ||
activeState.needs_rebuild !== 0 ||
activeState.indexed_seq !== latestSeq),
displayGeneration: displayState?.generation ?? randomUUID().replaceAll("-", ""),
activeState.indexed_seq !== latestSeq ||
activeState.source_generation !== source.generation),
displayGeneration,
displayNeedsRebuild,
includeDisplayRows: includeDisplayProjection,
displayPreviousGeneration: displayState?.generation ?? null,
rows: rows.map((row) => ({
createdAt: row.created_at,
event: JSON.parse(row.event_json) as Record<string, unknown>,
seq: row.seq,
})),
sessionId,
sourceGeneration: source.generation,
sourceTranscriptUpdatedAt: session.transcript_updated_at,
});
},
@@ -294,30 +320,40 @@ function sourceSnapshotMatches(
.select("transcript_updated_at")
.where("session_id", "=", plan.sessionId),
);
const latest = executeSqliteQueryTakeFirstSync(
db,
kysely
.selectFrom("transcript_events")
.select("seq")
.where("session_id", "=", plan.sessionId)
.orderBy("seq", "desc")
.limit(1),
);
const source = readSessionTranscriptSourceGenerationInTransaction(db, plan.sessionId);
return (
session?.transcript_updated_at === plan.sourceTranscriptUpdatedAt &&
(latest?.seq ?? -1) === plan.sourceIndexedSeq
source?.generation === plan.sourceGeneration &&
source.indexedSeq === plan.sourceIndexedSeq
);
}
function projectionClaimIsOwned(db: DatabaseSync, sessionId: string, claimId: number): boolean {
function projectionClaimIsOwned(
db: DatabaseSync,
params: {
claimId: number;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): boolean {
const row = executeSqliteQueryTakeFirstSync(
db,
getProjectionKysely(db)
.selectFrom("session_transcript_index_state")
.select(["needs_rebuild", "updated_at"])
.where("session_id", "=", sessionId),
.select(["needs_rebuild", "source_generation", "updated_at"])
.where("session_id", "=", params.sessionId),
);
return Boolean(
row &&
row.needs_rebuild !== 0 &&
row.source_generation === null &&
row.updated_at === params.claimId &&
sessionTranscriptSourceGenerationMatchesInTransaction(db, params.sessionId, {
generation: params.sourceGeneration,
indexedSeq: params.sourceIndexedSeq,
}),
);
return row?.needs_rebuild !== 0 && row?.updated_at === claimId;
}
/** Claims a prepared snapshot. Later chunks publish only while this claim remains current. */
@@ -334,12 +370,15 @@ export function claimPreparedSessionTranscriptProjectionInTransaction(
db,
kysely
.selectFrom("session_transcript_index_state")
.select(["indexed_seq", "needs_rebuild"])
.select(["indexed_seq", "needs_rebuild", "source_generation"])
.where("session_id", "=", plan.sessionId),
);
const activeNeedsRebuild =
plan.sourceIndexedSeq >= 0 &&
(!current || current.needs_rebuild !== 0 || current.indexed_seq !== plan.sourceIndexedSeq);
(!current ||
current.needs_rebuild !== 0 ||
current.indexed_seq !== plan.sourceIndexedSeq ||
current.source_generation !== plan.sourceGeneration);
if (
activeNeedsRebuild !== plan.activeNeedsRebuild ||
(!plan.activeNeedsRebuild && !plan.displayNeedsRebuild)
@@ -348,14 +387,11 @@ export function claimPreparedSessionTranscriptProjectionInTransaction(
}
if (plan.displayNeedsRebuild) {
const readiness = readSessionTranscriptDisplayRowsInTransaction(db, plan.sessionId, {
expectedGeneration: plan.displayGeneration,
expectedGeneration: plan.displayPreviousGeneration ?? plan.displayGeneration,
fromOrdinal: 0,
limit: 1,
});
if (
readiness.kind === "ready" ||
(readiness.generation !== null && readiness.generation !== plan.displayGeneration)
) {
if (readiness.kind === "ready" || readiness.generation !== plan.displayPreviousGeneration) {
return false;
}
}
@@ -364,6 +400,7 @@ export function claimPreparedSessionTranscriptProjectionInTransaction(
!claimSessionTranscriptDisplayInTransaction(db, {
claimId,
generation: plan.displayGeneration,
previousGeneration: plan.displayPreviousGeneration,
sessionId: plan.sessionId,
})
) {
@@ -377,19 +414,21 @@ export function claimPreparedSessionTranscriptProjectionInTransaction(
.values({
active_event_count: 0,
active_message_count: 0,
indexed_seq: -1,
indexed_seq: EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
leaf_event_id: null,
needs_rebuild: 1,
session_id: plan.sessionId,
source_generation: null,
updated_at: claimId,
})
.onConflict((conflict) =>
conflict.column("session_id").doUpdateSet({
active_event_count: 0,
active_message_count: 0,
indexed_seq: -1,
indexed_seq: EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
leaf_event_id: null,
needs_rebuild: 1,
source_generation: null,
updated_at: claimId,
}),
),
@@ -398,12 +437,43 @@ export function claimPreparedSessionTranscriptProjectionInTransaction(
return true;
}
export function abandonPreparedSessionTranscriptProjectionInTransaction(
db: DatabaseSync,
plan: PreparedSessionTranscriptProjectionMetadata,
claimId: number,
): void {
if (plan.activeNeedsRebuild) {
executeSqliteQuerySync(
db,
getProjectionKysely(db)
.updateTable("session_transcript_index_state")
.set({ source_generation: null, updated_at: Date.now() })
.where("session_id", "=", plan.sessionId)
.where("needs_rebuild", "!=", 0)
.where("updated_at", "=", claimId),
);
}
if (plan.displayNeedsRebuild) {
abandonSessionTranscriptDisplayClaimInTransaction(db, {
claimId,
generation: plan.displayGeneration,
sessionId: plan.sessionId,
});
}
}
/** Deletes old rows in bounded rowid batches while the prepared claim is current. */
export function deletePreparedSessionTranscriptProjectionChunkInTransaction(
db: DatabaseSync,
params: { claimId: number; maxRowsPerTable: number; sessionId: string },
params: {
claimId: number;
maxRowsPerTable: number;
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): ProjectionDeleteChunkResult {
if (!projectionClaimIsOwned(db, params.sessionId, params.claimId)) {
if (!projectionClaimIsOwned(db, params)) {
return { hasMore: false, owned: false };
}
// Hidden rowid batching is the narrow SQLite primitive that keeps each
@@ -455,9 +525,11 @@ export function appendPreparedSessionTranscriptProjectionChunkInTransaction(
claimId: number;
ftsRows?: PreparedSessionTranscriptProjection["ftsRows"];
sessionId: string;
sourceGeneration: string;
sourceIndexedSeq: number;
},
): boolean {
if (!projectionClaimIsOwned(db, params.sessionId, params.claimId)) {
if (!projectionClaimIsOwned(db, params)) {
return false;
}
const kysely = getProjectionKysely(db);
@@ -498,7 +570,13 @@ export function finalizePreparedSessionTranscriptProjectionInTransaction(
claimId: number,
): boolean {
if (
(plan.activeNeedsRebuild && !projectionClaimIsOwned(db, plan.sessionId, claimId)) ||
(plan.activeNeedsRebuild &&
!projectionClaimIsOwned(db, {
claimId,
sessionId: plan.sessionId,
sourceGeneration: plan.sourceGeneration,
sourceIndexedSeq: plan.sourceIndexedSeq,
})) ||
!sourceSnapshotMatches(db, plan)
) {
return false;
@@ -511,6 +589,7 @@ export function finalizePreparedSessionTranscriptProjectionInTransaction(
generation: plan.displayGeneration,
rowCount: plan.displayRowCount,
sessionId: plan.sessionId,
sourceGeneration: plan.sourceGeneration,
sourceIndexedSeq: plan.sourceIndexedSeq,
})
) {
@@ -529,11 +608,15 @@ export function finalizePreparedSessionTranscriptProjectionInTransaction(
indexed_seq: plan.sourceIndexedSeq,
leaf_event_id: plan.leafEventId,
needs_rebuild: 0,
source_generation: plan.sourceGeneration,
updated_at: Date.now(),
})
.where("session_id", "=", plan.sessionId)
.where("needs_rebuild", "!=", 0)
.where("updated_at", "=", claimId),
);
return result.numAffectedRows === 1n;
if (result.numAffectedRows !== 1n) {
return false;
}
return true;
}
src/config/sessions/session-transcript-read-fence.ts
+8
View File
@@ -3,6 +3,7 @@ import { executeSqliteQueryTakeFirstSync } from "../../infra/kysely-sync.js";
import type { UserTurnTranscriptAdmissionReceipt } from "../../sessions/user-turn-transcript.types.js";
import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js";
import { getSessionKysely } from "./session-accessor.sqlite-scope.js";
import { readCurrentSessionTranscriptActiveSourceInTransaction } from "./session-transcript-source-generation.js";
const transcriptReadFenceStorage = new AsyncLocalStorage<UserTurnTranscriptAdmissionReceipt>();
@@ -61,6 +62,13 @@ export function resolveSqliteSessionTranscriptReadFence(params: {
"Current-turn transcript admission belongs to a different session key",
);
}
if (
!readCurrentSessionTranscriptActiveSourceInTransaction(params.database.db, params.sessionId)
) {
throw new SessionTranscriptReadFenceError(
`Current-turn transcript admission projection changed: ${receipt.entryId}`,
);
}
const db = getSessionKysely(params.database.db);
const boundary = executeSqliteQueryTakeFirstSync(
params.database.db,
src/config/sessions/session-transcript-reconcile.ts
+61 -14
View File
@@ -31,6 +31,7 @@ import {
sessionTranscriptIndexNeedsReconcile,
} from "./session-transcript-index.js";
import {
abandonPreparedSessionTranscriptProjectionInTransaction,
appendPreparedSessionTranscriptProjectionChunkInTransaction,
claimPreparedSessionTranscriptProjectionInTransaction,
deletePreparedSessionTranscriptProjectionChunkInTransaction,
@@ -42,6 +43,7 @@ import type {
SessionTranscriptReconcileWorkerInput,
SessionTranscriptReconcileWorkerMessage,
} from "./session-transcript-reconcile.worker.js";
import { ensureAllSessionTranscriptSourceGenerationsInTransaction } from "./session-transcript-source-generation.js";
const log = createSubsystemLogger("sessions/transcript-index");
const PROJECTION_WRITE_CHUNK_ROWS = 512;
@@ -142,9 +144,11 @@ async function claimPreparedSessionTranscriptProjection(
"sessions.transcript-index.delete-chunk",
(database) =>
deletePreparedSessionTranscriptProjectionChunkInTransaction(database.db, {
claimId,
maxRowsPerTable: PROJECTION_WRITE_CHUNK_ROWS,
sessionId: plan.sessionId,
claimId,
sourceGeneration: plan.sourceGeneration,
sourceIndexedSeq: plan.sourceIndexedSeq,
}),
);
}
@@ -158,12 +162,15 @@ async function claimPreparedSessionTranscriptProjection(
generation: plan.displayGeneration,
maxRows: PROJECTION_WRITE_CHUNK_ROWS,
sessionId: plan.sessionId,
sourceGeneration: plan.sourceGeneration,
sourceIndexedSeq: plan.sourceIndexedSeq,
}),
);
}
await yieldToGateway();
}
if (!deleteResult.owned || !displayDeleteResult.owned) {
await abandonPreparedProjection(databaseOptions, { claimId, plan });
return undefined;
}
return { claimId, plan };
@@ -213,12 +220,16 @@ async function appendPreparedProjectionChunk(
generation: active.plan.displayGeneration,
rows: rows.displayRows,
sessionId: active.plan.sessionId,
sourceGeneration: active.plan.sourceGeneration,
sourceIndexedSeq: active.plan.sourceIndexedSeq,
});
}
return appendPreparedSessionTranscriptProjectionChunkInTransaction(database.db, {
...rows,
claimId: active.claimId,
sessionId: active.plan.sessionId,
sourceGeneration: active.plan.sourceGeneration,
sourceIndexedSeq: active.plan.sourceIndexedSeq,
});
},
);
@@ -242,6 +253,19 @@ async function finalizePreparedProjection(
);
}
async function abandonPreparedProjection(
databaseOptions: OpenClawAgentDatabaseOptions,
active: ActivePreparedProjection,
): Promise<void> {
await runProjectionWrite(databaseOptions, "sessions.transcript-index.abandon", (database) =>
abandonPreparedSessionTranscriptProjectionInTransaction(
database.db,
active.plan,
active.claimId,
),
);
}
/** Prepares full trees off-thread, then commits bounded chunks through the runtime writer owner. */
export async function reconcileSessionTranscriptIndexes(
params: SessionTranscriptReconcileParams,
@@ -272,6 +296,7 @@ async function reconcileSessionTranscriptProjections(
databaseOptions,
"sessions.transcript-index.preflight",
(database) => {
ensureAllSessionTranscriptSourceGenerationsInTransaction(database);
deleteOrphanedTranscriptIndexRowsInTransaction(database.db);
return (
(includeDisplayProjection
@@ -309,6 +334,7 @@ async function reconcileSessionTranscriptProjections(
let active: ActivePreparedProjection | undefined;
let doneReceived = false;
let reconciledSessions = 0;
let settling = false;
let settled = false;
const settle = (finish: () => void, terminate: boolean) => {
if (settled) {
@@ -321,18 +347,37 @@ async function reconcileSessionTranscriptProjections(
}
finish();
};
const fail = async (error: unknown, terminate: boolean) => {
if (settled || settling) {
return;
}
settling = true;
let failure = toStringifiedError(error);
const claimed = active;
active = undefined;
if (claimed) {
try {
await abandonPreparedProjection(databaseOptions, claimed);
} catch (abandonError) {
failure = new Error(
`${failure.message}; transcript projection claim abandonment failed: ${toStringifiedError(abandonError).message}`,
);
}
}
settle(() => reject(failure), terminate);
};
const handleMessage = async (message: SessionTranscriptReconcileWorkerMessage) => {
if (settled || settling) {
return;
}
if (message.type === "failed") {
settle(() => reject(new Error(message.error)), false);
await fail(new Error(message.error), false);
return;
}
if (message.type === "done") {
doneReceived = true;
if (active) {
settle(
() => reject(new Error("session transcript reconcile worker ended mid-plan")),
true,
);
await fail(new Error("session transcript reconcile worker ended mid-plan"), true);
return;
}
try {
@@ -342,7 +387,7 @@ async function reconcileSessionTranscriptProjections(
(database) => deleteOrphanedTranscriptIndexRowsInTransaction(database.db),
);
} catch (error) {
settle(() => reject(toStringifiedError(error)), true);
await fail(error, true);
return;
}
await workerExit;
@@ -362,7 +407,11 @@ async function reconcileSessionTranscriptProjections(
throw new Error("session transcript reconcile worker sent a chunk for no active plan");
}
if (message.type === "plan-finish") {
const finalized = await finalizePreparedProjection(databaseOptions, active);
const claimed = active;
const finalized = await finalizePreparedProjection(databaseOptions, claimed);
if (!finalized) {
await abandonPreparedProjection(databaseOptions, claimed);
}
active = undefined;
if (finalized) {
reconciledSessions += 1;
@@ -380,27 +429,25 @@ async function reconcileSessionTranscriptProjections(
: { ftsRows: decodeFtsChunk(message.chunk) },
);
if (!owned) {
await abandonPreparedProjection(databaseOptions, active);
active = undefined;
}
continueProjectionWorker(worker, owned);
} catch (error) {
settle(() => reject(toStringifiedError(error)), true);
await fail(error, true);
}
};
worker.on("message", (message: SessionTranscriptReconcileWorkerMessage) => {
void handleMessage(message);
});
worker.once("error", (error) => {
settle(() => reject(toStringifiedError(error)), true);
void fail(error, true);
});
worker.once("exit", (code) => {
if (doneReceived && code === 0) {
return;
}
settle(
() => reject(new Error(`session transcript reconcile worker exited with code ${code}`)),
false,
);
void fail(new Error(`session transcript reconcile worker exited with code ${code}`), false);
});
});
}
src/config/sessions/session-transcript-search.test.ts
+45 -3
View File
@@ -3,7 +3,11 @@ import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js";
import {
executeSqliteQuerySync,
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
closeOpenClawAgentDatabasesForTest,
@@ -93,7 +97,10 @@ function agentKysely() {
kysely: getNodeSqliteKysely<
Pick<
OpenClawAgentKyselyDatabase,
"session_transcript_fts" | "session_transcript_index_state" | "transcript_events"
| "session_transcript_fts"
| "session_transcript_index_state"
| "transcript_events"
| "transcript_rewrite_watermarks"
>
>(database.db),
};
@@ -208,6 +215,24 @@ describe("searchSessionTranscripts", () => {
expect(search("alpha").hits).toHaveLength(1);
});
it("hides same-generation FTS rows until their frontier is current", async () => {
await appendUserMessage("session-1", "agent:main:main", "frontier guarded");
const { db, kysely } = agentKysely();
executeSqliteQuerySync(
db,
kysely
.updateTable("session_transcript_index_state")
.set({ indexed_seq: -1 })
.where("session_id", "=", "session-1"),
);
const lagging = search("frontier");
expect(lagging.indexing).toBe(true);
expect(lagging.hits).toEqual([]);
await waitForSearchReconcile("frontier");
expect(search("frontier").hits).toHaveLength(1);
});
it("streams large searchable projections to the writer in bounded chunks", async () => {
const scope = transcriptScope("session-1", "agent:main:main");
const largeText = "x".repeat(140 * 1024);
@@ -248,13 +273,30 @@ describe("searchSessionTranscripts", () => {
expect(result.hits).toHaveLength(1);
});
it("detects missing, dirty, and lagging transcript index watermarks", async () => {
it("detects missing source, dirty, and lagging transcript index watermarks", async () => {
await appendUserMessage("session-1", "agent:main:main", "indexed message");
const { db, kysely } = agentKysely();
const pending = () => listSessionsNeedingTranscriptIndexReconcile(db);
expect(pending()).toEqual([]);
const source = executeSqliteQueryTakeFirstSync(
db,
kysely
.deleteFrom("transcript_rewrite_watermarks")
.where("session_id", "=", "session-1")
.returning(["generation", "updated_at"]),
);
expect(pending()).toEqual(["session-1"]);
executeSqliteQuerySync(
db,
kysely.insertInto("transcript_rewrite_watermarks").values({
generation: source!.generation,
session_id: "session-1",
updated_at: source!.updated_at,
}),
);
executeSqliteQuerySync(
db,
kysely
src/config/sessions/session-transcript-search.ts
+15 -3
View File
@@ -3,6 +3,7 @@
// this module owns the query path and schedules the shared reconcile owner
// when doctor imports or out-of-band writes leave derived rows behind.
import { openOpenClawAgentDatabase } from "../../state/openclaw-agent-db.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "../../state/openclaw-agent-transcript-projection-source-schema.js";
import { truncateUtf16Safe } from "../../utils.js";
import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js";
import { listSessionsNeedingTranscriptIndexReconcile } from "./session-transcript-index.js";
@@ -63,6 +64,7 @@ export function searchSessionTranscripts(params: {
...(databasePath ? { path: databasePath } : {}),
};
const database = openOpenClawAgentDatabase(databaseOptions);
ensureOpenClawAgentTranscriptProjectionSourceColumns(database.db);
const dirtySessions = listSessionsNeedingTranscriptIndexReconcile(database.db);
if (dirtySessions.length > 0) {
startSessionTranscriptIndexReconcile(databaseOptions);
@@ -87,10 +89,20 @@ export function searchSessionTranscripts(params: {
bm25(session_transcript_fts) AS rank
FROM session_transcript_fts
JOIN session_windows ON session_windows.session_id = session_transcript_fts.session_id
WHERE session_transcript_fts MATCH ?${whereSession}
AND session_transcript_fts.session_id NOT IN (
SELECT session_id FROM session_transcript_index_state WHERE needs_rebuild != 0
JOIN transcript_rewrite_watermarks AS source
ON source.session_id = session_transcript_fts.session_id
JOIN session_transcript_index_state AS state
ON state.session_id = session_transcript_fts.session_id
AND state.needs_rebuild = 0
AND state.source_generation = source.generation
AND state.indexed_seq = (
SELECT latest.seq
FROM transcript_events AS latest
WHERE latest.session_id = session_transcript_fts.session_id
ORDER BY latest.seq DESC
LIMIT 1
)
WHERE session_transcript_fts MATCH ?${whereSession}
ORDER BY rank ASC, timestamp DESC, message_id ASC
LIMIT ?
`);
src/config/sessions/session-transcript-source-generation.test.ts
+644
View File
@@ -0,0 +1,644 @@
import { EventEmitter } from "node:events";
import type { Worker } from "node:worker_threads";
import { afterEach, describe, expect, it, vi } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
import {
closeOpenClawAgentDatabasesForTest,
openOpenClawAgentDatabase,
runOpenClawAgentWriteTransaction,
} from "../../state/openclaw-agent-db.js";
import {
appendTranscriptMessage,
persistSessionTranscriptTurn,
upsertSessionEntryCore,
} from "./session-accessor.js";
import { readSessionTranscriptMessageEventCount } from "./session-accessor.sqlite-active-events.js";
import { ensureSqliteTranscriptGenerationsForCanonicalRepair } from "./session-accessor.sqlite-canonical-repair.js";
import { importSqliteSessionRows } from "./session-accessor.sqlite-import.js";
import { resolveSqliteTranscriptScope } from "./session-accessor.sqlite-scope.js";
import { appendTranscriptEventInTransaction } from "./session-accessor.sqlite-transcript-store.js";
import { replaceTranscriptEvents } from "./session-accessor.sqlite-transcript-write.js";
import {
appendSessionTranscriptDisplayChunkInTransaction,
deleteSessionTranscriptDisplayChunkInTransaction,
} from "./session-transcript-display.js";
import { reconcileSessionTranscriptIndexInTransaction } from "./session-transcript-index.js";
import {
abandonPreparedSessionTranscriptProjectionInTransaction,
appendPreparedSessionTranscriptProjectionChunkInTransaction,
claimPreparedSessionTranscriptProjectionInTransaction,
deletePreparedSessionTranscriptProjectionChunkInTransaction,
finalizePreparedSessionTranscriptProjectionInTransaction,
prepareSessionTranscriptProjection,
} from "./session-transcript-projection-rebuild.js";
import { reconcileSessionTranscriptIndexes } from "./session-transcript-reconcile.js";
import { replaceSessionTranscriptSourceGenerationInTransaction } from "./session-transcript-source-generation.js";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
afterEach(() => {
closeOpenClawAgentDatabasesForTest();
});
function createScope(name: string) {
return {
agentId: "main",
env: { OPENCLAW_STATE_DIR: tempDirs.make(`openclaw-source-generation-${name}-`) },
sessionId: `session-${name}`,
sessionKey: `agent:main:${name}`,
};
}
function readGeneration(scope: ReturnType<typeof createScope>): string | undefined {
return (
openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env })
.db.prepare("SELECT generation FROM transcript_rewrite_watermarks WHERE session_id = ?")
.get(scope.sessionId) as { generation: string } | undefined
)?.generation;
}
describe("session transcript source generation", () => {
it("keeps an empty replacement for an absent session as a no-op", async () => {
const scope = createScope("absent-empty-replacement");
await replaceTranscriptEvents(scope, []);
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
expect(
database.db
.prepare(
`SELECT
(SELECT COUNT(*) FROM session_windows WHERE session_id = ?) AS windows,
(SELECT COUNT(*) FROM transcript_rewrite_watermarks WHERE session_id = ?) AS generations,
(SELECT COUNT(*) FROM sqlite_schema
WHERE type = 'table' AND name = 'session_transcript_display_state') AS display_tables`,
)
.get(scope.sessionId, scope.sessionId),
).toEqual({ display_tables: 0, generations: 0, windows: 0 });
});
it("owns an empty generation and rotates it once for an empty replacement", async () => {
const scope = createScope("empty-replacement");
await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 10 });
const initial = readGeneration(scope);
expect(initial).toMatch(/^[0-9a-f]{32}$/u);
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db.exec(`
CREATE TEMP TABLE tracked_source_generation_writes (write_count INTEGER NOT NULL);
INSERT INTO tracked_source_generation_writes (write_count) VALUES (0);
CREATE TEMP TRIGGER track_source_generation_insert
AFTER INSERT ON transcript_rewrite_watermarks
WHEN NEW.session_id = '${scope.sessionId}'
BEGIN
UPDATE tracked_source_generation_writes SET write_count = write_count + 1;
END;
CREATE TEMP TRIGGER track_source_generation_update
AFTER UPDATE ON transcript_rewrite_watermarks
WHEN NEW.session_id = '${scope.sessionId}'
BEGIN
UPDATE tracked_source_generation_writes SET write_count = write_count + 1;
END;
`);
await replaceTranscriptEvents(scope, []);
expect(readGeneration(scope)).toMatch(/^[0-9a-f]{32}$/u);
expect(readGeneration(scope)).not.toBe(initial);
expect(
database.db.prepare("SELECT write_count FROM tracked_source_generation_writes").get(),
).toEqual({ write_count: 1 });
});
it("gives an empty legacy import an authoritative generation", async () => {
const scope = createScope("empty-import");
await importSqliteSessionRows({
agentId: scope.agentId,
env: scope.env,
entry: { sessionId: scope.sessionId, updatedAt: 10 },
sessionKey: scope.sessionKey,
});
expect(readGeneration(scope)).toMatch(/^[0-9a-f]{32}$/u);
});
it("repairs the generation of an empty legacy canonical source", async () => {
const scope = createScope("empty-canonical-repair");
const entry = { sessionId: scope.sessionId, updatedAt: 10 };
await upsertSessionEntryCore(scope, entry);
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db
.prepare("DELETE FROM transcript_rewrite_watermarks WHERE session_id = ?")
.run(scope.sessionId);
await ensureSqliteTranscriptGenerationsForCanonicalRepair([
{
agentId: scope.agentId,
entry,
sessionKey: scope.sessionKey,
storePath: database.path,
},
]);
expect(readGeneration(scope)).toMatch(/^[0-9a-f]{32}$/u);
});
it("publishes source ownership with active and display projection state", async () => {
const scope = createScope("projection-source-ownership");
await upsertSessionEntryCore(scope, { sessionId: scope.sessionId, updatedAt: 10 });
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "projection source" },
});
const sourceGeneration = readGeneration(scope);
expect(
database.db
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation,
display.generation AS display_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = ?`,
)
.get(scope.sessionId),
).toEqual({
active_source_generation: sourceGeneration,
display_generation: expect.stringMatching(/^[0-9a-f]{32}$/u),
display_source_generation: sourceGeneration,
});
});
it("rejects prepared work after a same-sequence source replacement", async () => {
const scope = createScope("stale-preparation");
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "prepared source" },
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db.exec(`
UPDATE session_transcript_index_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
UPDATE session_transcript_display_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
`);
const plan = prepareSessionTranscriptProjection(database.db, scope.sessionId);
expect(plan).toBeDefined();
runOpenClawAgentWriteTransaction(
(writeDatabase) => {
replaceSessionTranscriptSourceGenerationInTransaction(writeDatabase, scope.sessionId);
expect(
writeDatabase.db
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = ?`,
)
.get(scope.sessionId),
).toEqual({ active_source_generation: null, display_source_generation: null });
expect(
claimPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -101),
).toBe(false);
},
{ agentId: scope.agentId, env: scope.env },
);
});
it("rejects final publication after a same-sequence source replacement", async () => {
const scope = createScope("stale-finalization");
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "claimed source" },
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db.exec(`
UPDATE session_transcript_index_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
UPDATE session_transcript_display_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
`);
const plan = prepareSessionTranscriptProjection(database.db, scope.sessionId);
expect(plan).toBeDefined();
runOpenClawAgentWriteTransaction(
(writeDatabase) => {
expect(
claimPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -202),
).toBe(true);
},
{ agentId: scope.agentId, env: scope.env },
);
runOpenClawAgentWriteTransaction(
(writeDatabase) => {
replaceSessionTranscriptSourceGenerationInTransaction(writeDatabase, scope.sessionId);
expect(
finalizePreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -202),
).toBe(false);
},
{ agentId: scope.agentId, env: scope.env },
);
});
it("does not forward-publish rows from an unbound projection", async () => {
const scope = createScope("stale-forward-append");
const initial = await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "bound source" },
});
expect(initial).toBeDefined();
const options = { agentId: scope.agentId, env: scope.env };
const resolved = resolveSqliteTranscriptScope(scope);
runOpenClawAgentWriteTransaction((writeDatabase) => {
replaceSessionTranscriptSourceGenerationInTransaction(writeDatabase, scope.sessionId);
expect(
appendTranscriptEventInTransaction(
writeDatabase,
resolved,
{
id: "stale-forward",
message: { role: "assistant", content: "must rebuild" },
parentId: initial!.messageId,
type: "message",
},
{ scheduleProjectionReconcile: false },
),
).toBe(true);
}, options);
const database = openOpenClawAgentDatabase(options);
expect(
database.db
.prepare(
`SELECT
(SELECT needs_rebuild FROM session_transcript_index_state WHERE session_id = ?) AS active_needs_rebuild,
(SELECT source_generation FROM session_transcript_index_state WHERE session_id = ?) AS active_source_generation,
(SELECT needs_rebuild FROM session_transcript_display_state WHERE session_id = ?) AS display_needs_rebuild,
(SELECT source_generation FROM session_transcript_display_state WHERE session_id = ?) AS display_source_generation`,
)
.get(scope.sessionId, scope.sessionId, scope.sessionId, scope.sessionId),
).toEqual({
active_needs_rebuild: 1,
active_source_generation: null,
display_needs_rebuild: 1,
display_source_generation: null,
});
});
it.each(["generation", "frontier"] as const)(
"rejects every claimed chunk after the source %s changes",
async (change) => {
const scope = createScope(`stale-chunks-${change}`);
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "claimed source" },
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db.exec(`
UPDATE session_transcript_index_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
`);
const plan = prepareSessionTranscriptProjection(database.db, scope.sessionId);
expect(plan).toBeDefined();
const claimId = -303;
runOpenClawAgentWriteTransaction(
(writeDatabase) => {
expect(
claimPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, claimId),
).toBe(true);
},
{ agentId: scope.agentId, env: scope.env },
);
runOpenClawAgentWriteTransaction(
(writeDatabase) => {
if (change === "generation") {
replaceSessionTranscriptSourceGenerationInTransaction(writeDatabase, scope.sessionId);
} else {
writeDatabase.db
.prepare(
"INSERT INTO transcript_events (session_id, seq, event_json, created_at) VALUES (?, ?, ?, ?)",
)
.run(
scope.sessionId,
plan!.sourceIndexedSeq + 1,
JSON.stringify({ type: "message", id: "raced" }),
Date.now(),
);
}
const before = writeDatabase.db
.prepare(
`SELECT
(SELECT COUNT(*) FROM session_transcript_active_events WHERE session_id = ?) AS active_count,
(SELECT COUNT(*) FROM session_transcript_fts WHERE session_id = ?) AS fts_count,
(SELECT COUNT(*) FROM session_transcript_display_rows WHERE session_id = ?) AS display_count`,
)
.get(scope.sessionId, scope.sessionId, scope.sessionId);
const source = {
sourceGeneration: plan!.sourceGeneration,
sourceIndexedSeq: plan!.sourceIndexedSeq,
};
expect(
deletePreparedSessionTranscriptProjectionChunkInTransaction(writeDatabase.db, {
claimId,
maxRowsPerTable: 1,
sessionId: scope.sessionId,
...source,
}),
).toEqual({ hasMore: false, owned: false });
expect(
appendPreparedSessionTranscriptProjectionChunkInTransaction(writeDatabase.db, {
activeRows: plan!.activeRows,
claimId,
ftsRows: plan!.ftsRows,
sessionId: scope.sessionId,
...source,
}),
).toBe(false);
expect(
deleteSessionTranscriptDisplayChunkInTransaction(writeDatabase.db, {
claimId,
generation: plan!.displayGeneration,
maxRows: 1,
sessionId: scope.sessionId,
...source,
}),
).toEqual({ hasMore: false, owned: false });
expect(
appendSessionTranscriptDisplayChunkInTransaction(writeDatabase.db, {
claimId,
generation: plan!.displayGeneration,
rows: plan!.displayRows,
sessionId: scope.sessionId,
...source,
}),
).toBe(false);
expect(
finalizePreparedSessionTranscriptProjectionInTransaction(
writeDatabase.db,
plan!,
claimId,
),
).toBe(false);
expect(
writeDatabase.db
.prepare(
`SELECT
(SELECT COUNT(*) FROM session_transcript_active_events WHERE session_id = ?) AS active_count,
(SELECT COUNT(*) FROM session_transcript_fts WHERE session_id = ?) AS fts_count,
(SELECT COUNT(*) FROM session_transcript_display_rows WHERE session_id = ?) AS display_count`,
)
.get(scope.sessionId, scope.sessionId, scope.sessionId),
).toEqual(before);
},
{ agentId: scope.agentId, env: scope.env },
);
},
);
it("does not let delayed abandonment invalidate a newer publication", async () => {
const scope = createScope("delayed-abandonment");
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "fresh owner" },
});
const options = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(options);
database.db.exec(`
UPDATE session_transcript_index_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
UPDATE session_transcript_display_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
`);
const plan = prepareSessionTranscriptProjection(database.db, scope.sessionId);
expect(plan).toBeDefined();
runOpenClawAgentWriteTransaction((writeDatabase) => {
expect(
claimPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -501),
).toBe(true);
expect(
claimPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -502),
).toBe(true);
expect(
finalizePreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -502),
).toBe(true);
abandonPreparedSessionTranscriptProjectionInTransaction(writeDatabase.db, plan!, -501);
}, options);
expect(
database.db
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = ?`,
)
.get(scope.sessionId),
).toEqual({
active_source_generation: plan!.sourceGeneration,
display_source_generation: plan!.sourceGeneration,
});
});
it.each(["failed", "done", "error", "exit", "rejected-continuation"] as const)(
"abandons a claimed projection after worker %s and permits a fresh retry",
async (terminal) => {
const scope = createScope(`worker-${terminal}`);
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "seed",
maintainDisplayProjection: true,
message: { role: "user", content: "seed" },
},
],
touchSessionEntry: false,
});
const databaseOptions = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(databaseOptions);
database.db.exec(`
UPDATE session_transcript_index_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
UPDATE session_transcript_display_state
SET needs_rebuild = 1
WHERE session_id = '${scope.sessionId}';
`);
const prepared = prepareSessionTranscriptProjection(database.db, scope.sessionId);
expect(prepared).toBeDefined();
const {
activeRows: _activeRows,
displayRows: _displayRows,
ftsRows: _ftsRows,
...plan
} = prepared!;
const worker = Object.assign(new EventEmitter(), {
postMessage: vi.fn((message: { accepted: boolean; type: "continue" }) => {
if (!message.accepted) {
queueMicrotask(() => {
worker.emit("message", { type: "done" });
worker.emit("exit", 0);
});
return;
}
queueMicrotask(() => {
if (terminal === "failed") {
worker.emit("message", { error: "injected worker failure", type: "failed" });
} else if (terminal === "done") {
worker.emit("message", { type: "done" });
worker.emit("exit", 0);
} else if (terminal === "error") {
worker.emit("error", new Error("injected worker error"));
} else if (terminal === "exit") {
worker.emit("exit", 7);
} else {
runOpenClawAgentWriteTransaction((writeDatabase) => {
replaceSessionTranscriptSourceGenerationInTransaction(
writeDatabase,
scope.sessionId,
);
}, databaseOptions);
worker.emit("message", {
rows: prepared!.activeRows.slice(0, 1),
sessionId: scope.sessionId,
type: "active-chunk",
});
}
});
}),
terminate: vi.fn(async () => 0),
});
const createWorker = vi.fn(() => {
queueMicrotask(() => worker.emit("message", { plan, type: "plan-start" }));
return worker as unknown as Worker;
});
const outcome = reconcileSessionTranscriptIndexes({
...databaseOptions,
createWorker,
});
if (terminal === "rejected-continuation") {
await expect(outcome).resolves.toEqual({ reconciledSessions: 0 });
} else {
await expect(outcome).rejects.toThrow();
}
expect(createWorker).toHaveBeenCalledTimes(1);
const abandoned = database.db
.prepare(
`SELECT
(SELECT updated_at FROM session_transcript_index_state WHERE session_id = ?) AS active_claim,
(SELECT source_generation FROM session_transcript_index_state WHERE session_id = ?) AS active_source_generation`,
)
.get(scope.sessionId, scope.sessionId) as {
active_claim: number;
active_source_generation: string | null;
};
expect(abandoned.active_claim).toBeGreaterThanOrEqual(0);
expect(abandoned.active_source_generation).toBeNull();
await expect(reconcileSessionTranscriptIndexes(databaseOptions)).resolves.toEqual({
reconciledSessions: 1,
});
expect(readSessionTranscriptMessageEventCount(scope)).toBe(1);
},
20_000,
);
it("rolls back a synchronous rebuild when its claimed source changes", async () => {
const scope = createScope("synchronous-claim-race");
await appendTranscriptMessage(scope, {
maintainDisplayProjection: true,
message: { role: "user", content: "synchronous source" },
});
const options = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(options);
database.db
.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?")
.run(scope.sessionId);
const before = {
active: database.db
.prepare(
"SELECT active_position, event_seq, message_position FROM session_transcript_active_events WHERE session_id = ? ORDER BY active_position",
)
.all(scope.sessionId),
projectionSources: database.db
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = ?`,
)
.get(scope.sessionId),
fts: database.db
.prepare(
"SELECT message_id, role, text FROM session_transcript_fts WHERE session_id = ? ORDER BY message_id",
)
.all(scope.sessionId),
generation: readGeneration(scope),
};
database.db.exec(`
CREATE TEMP TRIGGER race_synchronous_projection_claim
AFTER UPDATE OF updated_at ON main.session_transcript_index_state
WHEN NEW.session_id = '${scope.sessionId}' AND NEW.updated_at < 0
BEGIN
UPDATE transcript_rewrite_watermarks
SET generation = 'raced-source-generation'
WHERE session_id = NEW.session_id;
END;
`);
expect(() =>
runOpenClawAgentWriteTransaction((writeDatabase) => {
reconcileSessionTranscriptIndexInTransaction(writeDatabase.db, scope.sessionId);
}, options),
).toThrow(`Transcript projection claim changed while rebuilding ${scope.sessionId}`);
expect({
active: database.db
.prepare(
"SELECT active_position, event_seq, message_position FROM session_transcript_active_events WHERE session_id = ? ORDER BY active_position",
)
.all(scope.sessionId),
projectionSources: database.db
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = ?`,
)
.get(scope.sessionId),
fts: database.db
.prepare(
"SELECT message_id, role, text FROM session_transcript_fts WHERE session_id = ? ORDER BY message_id",
)
.all(scope.sessionId),
generation: readGeneration(scope),
}).toEqual(before);
});
});
src/config/sessions/session-transcript-source-generation.ts
+212
View File
@@ -0,0 +1,212 @@
import { randomUUID } from "node:crypto";
import type { DatabaseSync } from "node:sqlite";
import {
executeSqliteQuerySync,
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "../../state/openclaw-agent-transcript-projection-source-schema.js";
import { tableExists } from "../../state/openclaw-state-db-schema-helpers.js";
type SourceGenerationDatabase = Pick<
OpenClawAgentKyselyDatabase,
| "session_transcript_display_state"
| "session_transcript_index_state"
| "session_windows"
| "transcript_events"
| "transcript_rewrite_watermarks"
>;
export const EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ = -1;
type SessionTranscriptSourceGeneration = {
generation: string;
indexedSeq: number;
};
function getSourceGenerationKysely(db: DatabaseSync) {
return getNodeSqliteKysely<SourceGenerationDatabase>(db);
}
function createTranscriptGeneration(): string {
return randomUUID().replaceAll("-", "");
}
/** Reads the authoritative source token when the caller already owns the append frontier. */
export function readSessionTranscriptSourceGenerationTokenInTransaction(
db: DatabaseSync,
sessionId: string,
): string | undefined {
return executeSqliteQueryTakeFirstSync(
db,
getSourceGenerationKysely(db)
.selectFrom("transcript_rewrite_watermarks")
.select("generation")
.where("session_id", "=", sessionId),
)?.generation;
}
/** Reads the authoritative source generation and frontier from one SQLite snapshot. */
export function readSessionTranscriptSourceGenerationInTransaction(
db: DatabaseSync,
sessionId: string,
): SessionTranscriptSourceGeneration | undefined {
const row = executeSqliteQueryTakeFirstSync(
db,
getSourceGenerationKysely(db)
.selectFrom("session_windows as window")
.innerJoin(
"transcript_rewrite_watermarks as rewrite",
"rewrite.session_id",
"window.session_id",
)
.select((eb) => [
"rewrite.generation",
eb
.selectFrom("transcript_events as event")
.select((inner) => inner.fn.max<number>("event.seq").as("indexed_seq"))
.whereRef("event.session_id", "=", "window.session_id")
.as("indexed_seq"),
])
.where("window.session_id", "=", sessionId),
);
return row
? {
generation: row.generation,
indexedSeq: row.indexed_seq ?? EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ,
}
: undefined;
}
/** Materializes one source generation; ordinary appends preserve an existing token. */
export function ensureSessionTranscriptSourceGenerationInTransaction(
database: Pick<OpenClawAgentDatabase, "db">,
sessionId: string,
): string {
ensureOpenClawAgentTranscriptProjectionSourceColumns(database.db);
const existing = readSessionTranscriptSourceGenerationTokenInTransaction(database.db, sessionId);
if (existing) {
return existing;
}
const generation = createTranscriptGeneration();
const db = getSourceGenerationKysely(database.db);
const inserted = executeSqliteQuerySync(
database.db,
db
.insertInto("transcript_rewrite_watermarks")
.values({ session_id: sessionId, generation, updated_at: Date.now() })
.onConflict((conflict) => conflict.column("session_id").doNothing()),
);
return inserted.numAffectedRows === 1n
? generation
: (readSessionTranscriptSourceGenerationTokenInTransaction(database.db, sessionId) ??
generation);
}
/** Backfills legacy windows through the same source-generation policy before reconciliation. */
export function ensureAllSessionTranscriptSourceGenerationsInTransaction(
database: Pick<OpenClawAgentDatabase, "db">,
): number {
const db = getSourceGenerationKysely(database.db);
const missing = executeSqliteQuerySync(
database.db,
db
.selectFrom("session_windows as window")
.leftJoin(
"transcript_rewrite_watermarks as rewrite",
"rewrite.session_id",
"window.session_id",
)
.select("window.session_id")
.where("rewrite.session_id", "is", null),
).rows;
for (const row of missing) {
ensureSessionTranscriptSourceGenerationInTransaction(database, row.session_id);
}
return missing.length;
}
export function sessionTranscriptSourceGenerationMatchesInTransaction(
db: DatabaseSync,
sessionId: string,
expected: SessionTranscriptSourceGeneration,
): boolean {
const source = readSessionTranscriptSourceGenerationInTransaction(db, sessionId);
return source?.generation === expected.generation && source.indexedSeq === expected.indexedSeq;
}
/** Returns source identity only while the active projection is fully current. */
export function readCurrentSessionTranscriptActiveSourceInTransaction(
db: DatabaseSync,
sessionId: string,
): SessionTranscriptSourceGeneration | undefined {
ensureOpenClawAgentTranscriptProjectionSourceColumns(db);
const row = executeSqliteQueryTakeFirstSync(
db,
getSourceGenerationKysely(db)
.selectFrom("session_transcript_index_state as state")
.innerJoin("transcript_rewrite_watermarks as source", "source.session_id", "state.session_id")
.select((eb) => [
"source.generation",
"state.indexed_seq",
"state.needs_rebuild",
"state.source_generation",
eb
.selectFrom("transcript_events as event")
.select((inner) => inner.fn.max<number>("event.seq").as("source_indexed_seq"))
.whereRef("event.session_id", "=", "state.session_id")
.as("source_indexed_seq"),
])
.where("state.session_id", "=", sessionId),
);
const sourceIndexedSeq = row?.source_indexed_seq ?? EMPTY_SESSION_TRANSCRIPT_SOURCE_INDEXED_SEQ;
return row &&
row.needs_rebuild === 0 &&
row.indexed_seq === sourceIndexedSeq &&
row.source_generation === row.generation
? { generation: row.generation, indexedSeq: sourceIndexedSeq }
: undefined;
}
/** Replaces source identity and invalidates every derived projection atomically. */
export function replaceSessionTranscriptSourceGenerationInTransaction(
database: Pick<OpenClawAgentDatabase, "db">,
sessionId: string,
source: { generation?: string; updatedAt?: number } = {},
): string {
const generation = source.generation ?? createTranscriptGeneration();
const updatedAt = source.updatedAt ?? Date.now();
ensureOpenClawAgentTranscriptProjectionSourceColumns(database.db);
const db = getSourceGenerationKysely(database.db);
executeSqliteQuerySync(
database.db,
db
.insertInto("transcript_rewrite_watermarks")
.values({ generation, session_id: sessionId, updated_at: updatedAt })
.onConflict((conflict) =>
conflict.column("session_id").doUpdateSet({
generation,
updated_at: updatedAt,
}),
),
);
executeSqliteQuerySync(
database.db,
db
.updateTable("session_transcript_index_state")
.set({ source_generation: null })
.where("session_id", "=", sessionId),
);
if (tableExists(database.db, "session_transcript_display_state")) {
executeSqliteQuerySync(
database.db,
db
.updateTable("session_transcript_display_state")
.set({ source_generation: null })
.where("session_id", "=", sessionId),
);
}
return generation;
}
src/gateway/session-transcript-readers.test.ts
+8
View File
@@ -820,6 +820,14 @@ describe("session transcript reader facade", () => {
.run("f".repeat(32), sessionId);
vi.clearAllMocks();
expect(readSessionTitleFieldsFromTranscript(scope)).toEqual({
firstUserMessage: null,
lastMessagePreview: null,
});
await waitForSessionTranscriptIndexReconcile({
agentId: "main",
path: path.join(tempDir, "openclaw-agent.sqlite"),
});
expect(readSessionTitleFieldsFromTranscript(scope)).toEqual({
firstUserMessage: "generation prompt",
lastMessagePreview: "generation reply",
src/infra/state-migrations.media-persistence.historical-schema.test-support.ts
+1
View File
@@ -41,6 +41,7 @@ function removeSchemaRange(sql: string, startMarker: string, endMarker: string):
/** Exact schema bytes from 509a5f0373764, derived from current SQL with later additions removed. */
export function historicalV15AgentSchemaSql(): string {
let sql = restoreHistoricalAgentLeaseSchema(OPENCLAW_AGENT_SCHEMA_SQL)
.replaceAll(" source_generation TEXT,\n", "")
.replace(" entry_valid INTEGER NOT NULL DEFAULT 0 CHECK (entry_valid IN (-1, 0, 1)),\n", "")
.replace(" project_id TEXT,\n", "")
.replace(" route_context_json TEXT,\n", "")
src/plugin-sdk/session-transcript-runtime.test.ts
+6 -1
View File
@@ -220,7 +220,12 @@ describe("session transcript runtime SDK", () => {
});
await expect(
readSessionTranscriptRawDelta({ ...missingScope, maxBytes: 10, maxEvents: 1 }),
).resolves.toEqual({ kind: "missing" });
).resolves.toMatchObject({
kind: "page",
events: [],
hasMore: false,
serializedBytes: 0,
});
const scope = {
...missingScope,
src/state/openclaw-agent-db-additive-columns.ts
+21
View File
@@ -6,6 +6,12 @@ type LazyAdditiveAgentColumnDefinition = {
tableName: "session_nodes";
};
type TranscriptProjectionSourceColumnDefinition = {
columnName: "source_generation";
dataType: "TEXT";
tableName: "session_transcript_display_state" | "session_transcript_index_state";
};
// Session responsibility is feature-local and remains absent until the first
// explicit assignment. Bare nullable declarations keep older readers safe.
export const FIRST_USE_ADDITIVE_AGENT_COLUMN_DEFINITIONS = [
@@ -15,3 +21,18 @@ export const FIRST_USE_ADDITIVE_AGENT_COLUMN_DEFINITIONS = [
{ columnName: "owner_assigned_by_id", dataType: "TEXT", tableName: "session_nodes" },
{ columnName: "owner_assigned_at", dataType: "INTEGER", tableName: "session_nodes" },
] as const satisfies readonly LazyAdditiveAgentColumnDefinition[];
// Projection ownership is derived state. A missing value makes old rows stale,
// while the bare nullable column remains safe for older same-version readers.
export const TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS = [
{
columnName: "source_generation",
dataType: "TEXT",
tableName: "session_transcript_index_state",
},
{
columnName: "source_generation",
dataType: "TEXT",
tableName: "session_transcript_display_state",
},
] as const satisfies readonly TranscriptProjectionSourceColumnDefinition[];
src/state/openclaw-agent-db-schema-helpers.ts
+7 -1
View File
@@ -23,7 +23,10 @@ import {
ensureOpenClawAgentBoardSchemaInTransaction,
} from "./openclaw-agent-board-schema.js";
import { CONTEXT_ENGINE_TURN_OUTBOX_TABLE } from "./openclaw-agent-context-engine-turn-outbox-schema.js";
import { FIRST_USE_ADDITIVE_AGENT_COLUMN_DEFINITIONS } from "./openclaw-agent-db-additive-columns.js";
import {
FIRST_USE_ADDITIVE_AGENT_COLUMN_DEFINITIONS,
TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS,
} from "./openclaw-agent-db-additive-columns.js";
import { OPENCLAW_AGENT_SCHEMA_VERSION } from "./openclaw-agent-db-contract.js";
import { OpenClawAgentDatabaseMediaMigrationRequiredError } from "./openclaw-agent-db-migration-required.js";
import {
@@ -91,6 +94,9 @@ const AGENT_SCHEMA_COMPATIBILITY = {
...FIRST_USE_ADDITIVE_AGENT_COLUMN_DEFINITIONS.map(
({ columnName, tableName }) => `${tableName}.${columnName}`,
),
...TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS.map(
({ columnName, tableName }) => `${tableName}.${columnName}`,
),
],
allowedColumnDefinitions: {
"conversations.delivery_target": ["delivery_target TEXT NOT NULL DEFAULT ''"],
src/state/openclaw-agent-db-schema.ts
+3 -10
View File
@@ -44,7 +44,7 @@ import {
backfillSessionConversations,
ensureSessionAdditiveColumns,
ensureSessionEntryValidityProjection,
hasPendingSessionConversationRouteContextColumn,
hasPendingSessionAdditiveSchemaMigration,
migrateConversationDeliveryTargetColumn,
migrateSessionEntryStatusProjection,
readSqliteTableColumns,
@@ -156,11 +156,6 @@ function hasPendingSessionKeyContractSchemaMigration(db: DatabaseSync): boolean
return !sessionNodeColumns.has("entry_valid") || !hasContractTable;
}
function hasPendingSessionProjectColumn(db: DatabaseSync): boolean {
const columns = readSqliteTableColumns(db, "session_nodes");
return Boolean(columns && !columns.has("project_id"));
}
function migrateMemoryChunkMetadataSchema(db: DatabaseSync): void {
ensureMemoryRecallMetadataSchema(db);
ensureMemoryChunkProvenance(db);
@@ -345,8 +340,7 @@ function migrateSessionTranscriptGenerations(db: DatabaseSync, previousVersion:
db.prepare(
`INSERT OR IGNORE INTO transcript_rewrite_watermarks (session_id, generation, updated_at)
SELECT session_id, lower(hex(randomblob(16))), ?
FROM transcript_events
GROUP BY session_id`,
FROM session_windows`,
).run(Date.now());
}
@@ -571,8 +565,7 @@ export function assertAgentDatabaseIntegrityBeforeMutation(
(hasPendingMemoryChunkMetadataMigration(database) ||
hasPendingSessionKeyContractSchemaMigration(database) ||
hasRetiredAgentStateLeaseSchema(database) ||
hasPendingSessionConversationRouteContextColumn(database) ||
hasPendingSessionProjectColumn(database));
hasPendingSessionAdditiveSchemaMigration(database));
if (userVersion === OPENCLAW_AGENT_SCHEMA_VERSION && !hasPendingCurrentVersionMigration) {
verifyAndRepairCanonicalSqliteIndexes(database, pathname, AGENT_BASE_SCHEMA_SQL, {
allowMissingColumns: true,
src/state/openclaw-agent-db-session-migrations.ts
+15 -1
View File
@@ -7,6 +7,10 @@ import { parseSqliteSessionEntryRecord } from "../config/sessions/session-entry-
import { normalizeAccountId } from "../routing/account-id.js";
import { buildConversationRef, normalizeConversationPeerId } from "../routing/conversation-ref.js";
import { deriveSessionChatTypeFromKey } from "../sessions/session-chat-type-shared.js";
import {
dropRetiredTranscriptProjectionBindingSchema,
hasRetiredTranscriptProjectionBindingSchema,
} from "./openclaw-agent-transcript-projection-source-schema.js";
type MigratedConversationEntry = Record<string, unknown>;
@@ -272,6 +276,7 @@ export function readSqliteTableColumns(db: DatabaseSync, tableName: string): Set
/** Installs same-version session projections on first updated-binary open. */
export function ensureSessionAdditiveColumns(db: DatabaseSync): void {
dropRetiredTranscriptProjectionBindingSchema(db);
const columns = readSqliteTableColumns(db, "session_nodes");
if (columns && !columns.has("project_id")) {
db.exec("ALTER TABLE session_nodes ADD COLUMN project_id TEXT;");
@@ -298,11 +303,20 @@ export function ensureSessionAdditiveColumns(db: DatabaseSync): void {
}
}
export function hasPendingSessionConversationRouteContextColumn(db: DatabaseSync): boolean {
function hasPendingSessionConversationRouteContextColumn(db: DatabaseSync): boolean {
const columns = readSqliteTableColumns(db, "session_conversations");
return Boolean(columns && !columns.has("route_context_json"));
}
export function hasPendingSessionAdditiveSchemaMigration(db: DatabaseSync): boolean {
const sessionNodeColumns = readSqliteTableColumns(db, "session_nodes");
return Boolean(
hasRetiredTranscriptProjectionBindingSchema(db) ||
hasPendingSessionConversationRouteContextColumn(db) ||
(sessionNodeColumns && !sessionNodeColumns.has("project_id")),
);
}
/** Adds the v11 exact delivery target before the conversation backfill writes canonical rows. */
export function migrateConversationDeliveryTargetColumn(db: DatabaseSync): void {
const columns = readSqliteTableColumns(db, "conversations");
src/state/openclaw-agent-db.generated.d.ts
Vendored
+2
View File
@@ -376,6 +376,7 @@ export interface SessionTranscriptDisplayState {
needs_rebuild: Generated<number>;
row_count: number;
session_id: string;
source_generation: string | null;
updated_at: number;
}
@@ -424,6 +425,7 @@ export interface SessionTranscriptIndexState {
leaf_event_id: string | null;
needs_rebuild: Generated<number>;
session_id: string;
source_generation: string | null;
updated_at: number;
}
src/state/openclaw-agent-db.test.ts
+40 -18
View File
@@ -61,6 +61,7 @@ import {
import { resolveOpenClawStateSqlitePath } from "./openclaw-state-db.paths.js";
import {
collectSqliteSchemaShape,
createSqliteSchemaShapeFromSql,
normalizeSqliteSchemaShapeSql,
replaceNamedIndexesWithNoncanonicalIndexes,
} from "./sqlite-schema-shape.test-support.js";
@@ -79,17 +80,6 @@ function createTempStateDir(): string {
return makeTempDir(agentDbTempDirs, "openclaw-agent-db-");
}
function createCurrentAgentRuntimeSchemaShape() {
const { DatabaseSync } = requireNodeSqlite();
const database = new DatabaseSync(":memory:");
try {
database.exec(AGENT_BASE_SCHEMA_SQL);
return collectSqliteSchemaShape(database);
} finally {
database.close();
}
}
function ensureSharedStateDatabaseTemplate(): string {
if (sharedStateDatabaseTemplatePath) {
return sharedStateDatabaseTemplatePath;
@@ -1284,7 +1274,9 @@ describe("openclaw agent database", () => {
env: { OPENCLAW_STATE_DIR: stateDir },
});
expect(collectSqliteSchemaShape(database.db)).toEqual(createCurrentAgentRuntimeSchemaShape());
expect(collectSqliteSchemaShape(database.db)).toEqual(
createSqliteSchemaShapeFromSql(AGENT_BASE_SCHEMA_SQL),
);
expect(
database.db
.prepare("SELECT name FROM sqlite_schema WHERE type = 'table' AND name = 'state_leases'")
@@ -1578,7 +1570,7 @@ describe("openclaw agent database", () => {
).toEqual({ schema_version: OPENCLAW_AGENT_SCHEMA_VERSION });
});
it("backfills one generation per existing transcript when upgrading v12", () => {
it("backfills one generation per existing transcript window when upgrading v12", () => {
const stateDir = createTempStateDir();
const env = { OPENCLAW_STATE_DIR: stateDir };
const databasePath = materializeV13WorkerAgentDatabase(stateDir);
@@ -1605,9 +1597,13 @@ describe("openclaw agent database", () => {
)
.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(generations.map((row) => row.session_id)).toEqual([
"with-transcript",
"without-transcript",
]);
for (const row of generations) {
expect(row.generation).toMatch(/^[0-9a-f]{32}$/);
}
expect(
migrated.db
.prepare("SELECT strict FROM pragma_table_list WHERE name = ?")
@@ -3433,7 +3429,9 @@ describe("openclaw agent database", () => {
const stateDir = createTempStateDir();
const env = { OPENCLAW_STATE_DIR: stateDir };
const databasePath = materializeCurrentWorkerAgentDatabase(stateDir);
const canonicalShape = normalizeSqliteSchemaShapeSql(createCurrentAgentRuntimeSchemaShape());
const canonicalShape = normalizeSqliteSchemaShapeSql(
createSqliteSchemaShapeFromSql(AGENT_BASE_SCHEMA_SQL),
);
const { DatabaseSync } = requireNodeSqlite();
const drifted = new DatabaseSync(databasePath);
@@ -3512,7 +3510,7 @@ describe("openclaw agent database", () => {
const repaired = migrateAndOpenLegacyAgentDatabaseForTest({ agentId: "worker-1", env });
expect(normalizeSqliteSchemaShapeSql(collectSqliteSchemaShape(repaired.db))).toEqual(
normalizeSqliteSchemaShapeSql(createCurrentAgentRuntimeSchemaShape()),
normalizeSqliteSchemaShapeSql(createSqliteSchemaShapeFromSql(AGENT_BASE_SCHEMA_SQL)),
);
expect(
repaired.db
@@ -4365,6 +4363,30 @@ describe("openclaw agent database", () => {
);
});
it("rejects a newer user version after a validated handle is physically reopened", () => {
const stateDir = createTempStateDir();
const env = { OPENCLAW_STATE_DIR: stateDir };
const databasePath = openOpenClawAgentDatabase({ agentId: "worker-1", env }).path;
expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true);
const { DatabaseSync } = requireNodeSqlite();
const future = new DatabaseSync(databasePath);
try {
future.exec(`PRAGMA user_version = ${OPENCLAW_AGENT_SCHEMA_VERSION + 1};`);
} finally {
future.close();
}
expect(() => openOpenClawAgentDatabase({ agentId: "worker-1", env })).toThrow(
expect.objectContaining({
name: "SqliteSchemaVersionError",
message: expect.stringContaining(
`newer schema version ${OPENCLAW_AGENT_SCHEMA_VERSION + 1}`,
),
}),
);
});
it.each([0, OPENCLAW_AGENT_SCHEMA_VERSION - 1])(
"rechecks the media version guard at v%d after a validated handle is physically reopened",
(version) => {
src/state/openclaw-agent-db.ts
+7 -2
View File
@@ -43,6 +43,7 @@ import {
import {
assertCanonicalAgentMediaPersistenceVersion,
assertExistingAgentSchemaOwner,
assertOpenClawAgentCurrentRuntimeSchema,
assertSupportedAgentSchemaVersion,
readExistingAgentSchemaMeta,
} from "./openclaw-agent-db-schema-helpers.js";
@@ -324,8 +325,8 @@ export function openOpenClawAgentDatabase(
let maintenance: OpenClawAgentDatabase["walMaintenance"] | undefined;
try {
db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
assertSupportedAgentSchemaVersion(db, pathname);
if (!isValidatedReopen) {
assertSupportedAgentSchemaVersion(db, pathname);
assertExistingAgentSchemaOwner(readExistingAgentSchemaMeta(db), agentId, pathname);
}
// Integrity is not process-stable: the file can be damaged while evicted.
@@ -343,7 +344,11 @@ export function openOpenClawAgentDatabase(
synchronous: "NORMAL",
});
openedWalMaintenance = maintenance;
if (!isValidatedReopen) {
if (isValidatedReopen) {
// The process cache skips write-capable convergence, not shape validation:
// same-version lazy groups can drift while a physical handle is closed.
assertOpenClawAgentCurrentRuntimeSchema(db, { agentId, pathname });
} else {
ensureOpenClawAgentSchema(db, agentId, pathname);
}
return maintenance;
src/state/openclaw-agent-display-row-schema.test.ts
+160 -4
View File
@@ -3,7 +3,9 @@ import { DatabaseSync } from "node:sqlite";
import { afterEach, describe, expect, it } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
import { assertSqliteSchemaContains } from "../infra/sqlite-schema-contract.js";
import { assertOpenClawAgentSchemaContains } from "./openclaw-agent-db-schema-helpers.js";
import {
closeOpenClawAgentDatabaseByPath,
closeOpenClawAgentDatabasesForTest,
openOpenClawAgentDatabase,
} from "./openclaw-agent-db.js";
@@ -15,8 +17,10 @@ import {
SESSION_TRANSCRIPT_DISPLAY_ROWS_TABLE,
SESSION_TRANSCRIPT_DISPLAY_ROW_SOURCES_TABLE,
SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE,
validateOpenClawAgentDisplayRowSchema,
} from "./openclaw-agent-display-row-schema.js";
import { OPENCLAW_AGENT_SCHEMA_SQL } from "./openclaw-agent-schema.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "./openclaw-agent-transcript-projection-source-schema.js";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
@@ -292,10 +296,17 @@ describe("agent display-row schema", () => {
ensureOpenClawAgentDisplayRowSchema(database.db);
insertDisplayOwnerRows(database.db);
insertDisplayStateAndRow(database.db);
database.db.exec(`
ALTER TABLE session_transcript_index_state ADD COLUMN source_generation TEXT;
ALTER TABLE session_transcript_display_state ADD COLUMN source_generation TEXT;
`);
for (const tableName of [
"session_transcript_index_state",
SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE,
]) {
const present = database.db
.prepare(`SELECT 1 FROM pragma_table_info('${tableName}') WHERE name = 'source_generation'`)
.get();
if (!present) {
database.db.exec(`ALTER TABLE ${tableName} ADD COLUMN source_generation TEXT;`);
}
}
closeOpenClawAgentDatabasesForTest();
expect(() => {
@@ -398,3 +409,148 @@ describe("agent display-row schema", () => {
}
});
});
describe("agent transcript projection source columns", () => {
const priorSchema = OPENCLAW_AGENT_SCHEMA_SQL.replaceAll(" source_generation TEXT,\n", "");
it("accepts and lazily upgrades the prior same-version shape", () => {
const database = new DatabaseSync(":memory:");
try {
database.exec(priorSchema);
database.exec(`
INSERT INTO session_nodes
(session_key, current_session_id, entry_json, entry_valid, updated_at)
VALUES ('agent:main:upgrade', 'session-upgrade', '{}', -1, 1);
INSERT INTO session_windows
(session_id, session_key, session_scope, created_at, updated_at)
VALUES ('session-upgrade', 'agent:main:upgrade', 'conversation', 1, 1);
INSERT INTO transcript_rewrite_watermarks (session_id, generation, updated_at)
VALUES ('session-upgrade', 'source-generation', 1);
INSERT INTO transcript_events (session_id, seq, event_json, created_at)
VALUES ('session-upgrade', 0, '{"type":"session","id":"session-upgrade"}', 1);
INSERT INTO session_transcript_index_state
(session_id, indexed_seq, needs_rebuild, active_event_count, active_message_count, updated_at)
VALUES ('session-upgrade', 0, 0, 1, 0, 1);
INSERT INTO session_transcript_display_state
(session_id, generation, indexed_seq, row_count, needs_rebuild, updated_at)
VALUES ('session-upgrade', 'display-generation', 0, 1, 0, 1);
`);
expect(() =>
assertOpenClawAgentSchemaContains(
database,
"previous agent schema",
OPENCLAW_AGENT_SCHEMA_SQL,
),
).not.toThrow();
expect(validateOpenClawAgentDisplayRowSchema(database)).toBe(true);
ensureOpenClawAgentTranscriptProjectionSourceColumns(database);
expect(
database
.prepare(
`SELECT name FROM pragma_table_info('session_transcript_index_state')
WHERE name = 'source_generation'`,
)
.get(),
).toEqual({ name: "source_generation" });
expect(
database
.prepare(
`SELECT name FROM pragma_table_info('session_transcript_display_state')
WHERE name = 'source_generation'`,
)
.get(),
).toEqual({ name: "source_generation" });
expect(
database
.prepare(
`SELECT
active.source_generation AS active_source_generation,
display.source_generation AS display_source_generation
FROM session_transcript_index_state AS active
JOIN session_transcript_display_state AS display
ON display.session_id = active.session_id
WHERE active.session_id = 'session-upgrade'`,
)
.get(),
).toEqual({
active_source_generation: "source-generation",
display_source_generation: "source-generation",
});
} finally {
database.close();
}
});
it("retries a lazy upgrade rolled back by its owner", () => {
const database = new DatabaseSync(":memory:");
try {
database.exec(priorSchema);
database.exec("BEGIN IMMEDIATE;");
ensureOpenClawAgentTranscriptProjectionSourceColumns(database);
database.exec("ROLLBACK;");
ensureOpenClawAgentTranscriptProjectionSourceColumns(database);
expect(
database
.prepare(
`SELECT COUNT(*) AS count
FROM pragma_table_info('session_transcript_index_state')
WHERE name = 'source_generation'`,
)
.get(),
).toEqual({ count: 1 });
} finally {
database.close();
}
});
it("drops the retired candidate binding table on the next process open", () => {
const stateDir = tempDirs.make("openclaw-retired-projection-binding-");
const options = { agentId: "main", env: { OPENCLAW_STATE_DIR: stateDir } };
const initial = openOpenClawAgentDatabase(options);
initial.db.exec(
"CREATE TABLE session_transcript_projection_bindings (session_id TEXT) STRICT;",
);
closeOpenClawAgentDatabasesForTest();
const reopened = openOpenClawAgentDatabase(options);
expect(tableExists(reopened.db, "session_transcript_projection_bindings")).toBe(false);
});
});
describe("agent display-row physical reopen", () => {
it.each([
{
damage: `DROP TABLE ${SESSION_TRANSCRIPT_DISPLAY_ROWS_TABLE};`,
name: "partial display foundation",
},
{
damage: `
DROP TABLE ${SESSION_TRANSCRIPT_DISPLAY_ROWS_TABLE};
DROP TABLE ${SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE};
`,
name: "orphaned display semantics",
},
{
damage: `DROP TABLE ${SESSION_TRANSCRIPT_DISPLAY_CANVAS_TABLE};`,
name: "partial display semantics",
},
])("rejects a $name after physical reopen", ({ damage }) => {
const stateDir = tempDirs.make("openclaw-display-row-reopen-");
const options = { agentId: "main", env: { OPENCLAW_STATE_DIR: stateDir } };
const initial = openOpenClawAgentDatabase(options);
const databasePath = initial.path;
ensureOpenClawAgentDisplayRowSchema(initial.db);
expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true);
const damaged = new DatabaseSync(databasePath);
damaged.exec(damage);
damaged.close();
expect(() => openOpenClawAgentDatabase(options)).toThrow(
/display-row (?:semantics )?schema is partially present/u,
);
});
});
src/state/openclaw-agent-display-row-schema.ts
+31 -16
View File
@@ -2,6 +2,7 @@ import type { DatabaseSync } from "node:sqlite";
import { assertSqliteSchemaContains } from "../infra/sqlite-schema-contract.js";
import { runSqliteImmediateTransactionSync } from "../infra/sqlite-transaction.js";
import { OPENCLAW_AGENT_SCHEMA_SQL } from "./openclaw-agent-schema.js";
import { ensureOpenClawAgentTranscriptProjectionSourceColumns } from "./openclaw-agent-transcript-projection-source-schema.js";
export const SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE = "session_transcript_display_state";
export const SESSION_TRANSCRIPT_DISPLAY_ROWS_TABLE = "session_transcript_display_rows";
@@ -12,15 +13,21 @@ export const SESSION_TRANSCRIPT_DISPLAY_CARRY_TABLE = "session_transcript_displa
const DISPLAY_ROW_SCHEMA_START = `CREATE TABLE IF NOT EXISTS ${SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE} (`;
const DISPLAY_SEMANTICS_SCHEMA_START = `CREATE TABLE IF NOT EXISTS ${SESSION_TRANSCRIPT_DISPLAY_ROW_SOURCES_TABLE} (`;
const DISPLAY_ROW_SCHEMA_END =
const TRANSCRIPT_FTS_SCHEMA_START =
"CREATE VIRTUAL TABLE IF NOT EXISTS session_transcript_fts USING fts5(";
const SQLITE_TABLE_EXISTS_SQL = "SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?";
const ENSURED_DATABASES = new WeakSet<DatabaseSync>();
const ABSENT_DATABASES = new WeakSet<DatabaseSync>();
const FOUNDATION_ONLY_DATABASES = new WeakSet<DatabaseSync>();
const DISPLAY_SCHEMA_COMPATIBILITY = { allowCompatibleAdditiveColumns: true } as const;
const DISPLAY_ROW_SCHEMA_COMPATIBILITY = {
allowCompatibleAdditiveColumns: true,
allowedMissingColumns: [`${SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE}.source_generation`],
};
function splitDisplayRowSchema(sql: string): {
function splitDisplayRowSchema(
sql: string,
endMarker = TRANSCRIPT_FTS_SCHEMA_START,
): {
displayFoundation: string;
displayRows: string;
displaySemantics: string;
@@ -28,7 +35,7 @@ function splitDisplayRowSchema(sql: string): {
} {
const start = sql.indexOf(DISPLAY_ROW_SCHEMA_START);
const semanticsStart = sql.indexOf(DISPLAY_SEMANTICS_SCHEMA_START, start);
const end = sql.indexOf(DISPLAY_ROW_SCHEMA_END, start);
const end = sql.indexOf(endMarker, start);
if (start === -1 || semanticsStart === -1 || end === -1) {
throw new Error("OpenClaw agent display-row schema markers are missing.");
}
@@ -45,7 +52,8 @@ const displayRowSchema = splitDisplayRowSchema(OPENCLAW_AGENT_SCHEMA_SQL);
const AGENT_DISPLAY_ROW_SCHEMA_SQL = displayRowSchema.displayRows;
const AGENT_DISPLAY_ROW_FOUNDATION_SCHEMA_SQL = displayRowSchema.displayFoundation;
const AGENT_DISPLAY_ROW_SEMANTICS_SCHEMA_SQL = displayRowSchema.displaySemantics;
export const AGENT_BASE_SCHEMA_SQL = displayRowSchema.withoutDisplayRows;
export const AGENT_BASE_SCHEMA_SQL =
splitDisplayRowSchema(OPENCLAW_AGENT_SCHEMA_SQL).withoutDisplayRows;
function hasDisplayRowTable(db: DatabaseSync, tableName: string): boolean {
return Boolean(
@@ -63,7 +71,15 @@ export function validateOpenClawAgentDisplayRowSchema(db: DatabaseSync): boolean
}
const statePresent = hasDisplayRowTable(db, SESSION_TRANSCRIPT_DISPLAY_STATE_TABLE);
const rowsPresent = hasDisplayRowTable(db, SESSION_TRANSCRIPT_DISPLAY_ROWS_TABLE);
if (!statePresent && !rowsPresent) {
const semanticTables = [
SESSION_TRANSCRIPT_DISPLAY_ROW_SOURCES_TABLE,
SESSION_TRANSCRIPT_DISPLAY_CANVAS_TABLE,
SESSION_TRANSCRIPT_DISPLAY_CARRY_TABLE,
];
const presentSemanticTables = semanticTables.filter((tableName) =>
hasDisplayRowTable(db, tableName),
);
if (!statePresent && !rowsPresent && presentSemanticTables.length === 0) {
ABSENT_DATABASES.add(db);
return false;
}
@@ -74,15 +90,7 @@ export function validateOpenClawAgentDisplayRowSchema(db: DatabaseSync): boolean
db,
"OpenClaw agent display-row foundation schema",
AGENT_DISPLAY_ROW_FOUNDATION_SCHEMA_SQL,
DISPLAY_SCHEMA_COMPATIBILITY,
);
const semanticTables = [
SESSION_TRANSCRIPT_DISPLAY_ROW_SOURCES_TABLE,
SESSION_TRANSCRIPT_DISPLAY_CANVAS_TABLE,
SESSION_TRANSCRIPT_DISPLAY_CARRY_TABLE,
];
const presentSemanticTables = semanticTables.filter((tableName) =>
hasDisplayRowTable(db, tableName),
DISPLAY_ROW_SCHEMA_COMPATIBILITY,
);
if (presentSemanticTables.length === 0) {
FOUNDATION_ONLY_DATABASES.add(db);
@@ -95,7 +103,7 @@ export function validateOpenClawAgentDisplayRowSchema(db: DatabaseSync): boolean
db,
"OpenClaw agent display-row schema",
AGENT_DISPLAY_ROW_SCHEMA_SQL,
DISPLAY_SCHEMA_COMPATIBILITY,
DISPLAY_ROW_SCHEMA_COMPATIBILITY,
);
ENSURED_DATABASES.add(db);
return true;
@@ -140,6 +148,13 @@ export function ensureOpenClawAgentDisplayRowSchema(db: DatabaseSync): void {
}
ABSENT_DATABASES.delete(db);
FOUNDATION_ONLY_DATABASES.delete(db);
assertSqliteSchemaContains(
db,
"OpenClaw agent display-row schema",
AGENT_DISPLAY_ROW_SCHEMA_SQL,
DISPLAY_ROW_SCHEMA_COMPATIBILITY,
);
ensureOpenClawAgentTranscriptProjectionSourceColumns(db);
assertSqliteSchemaContains(
db,
"OpenClaw agent display-row schema",
src/state/openclaw-agent-schema.sql
+2
View File
@@ -629,6 +629,7 @@ CREATE TABLE IF NOT EXISTS session_transcript_index_state (
needs_rebuild INTEGER NOT NULL DEFAULT 0,
active_event_count INTEGER NOT NULL DEFAULT 0,
active_message_count INTEGER NOT NULL DEFAULT 0,
source_generation TEXT,
updated_at INTEGER NOT NULL,
FOREIGN KEY (session_id) REFERENCES session_windows(session_id) ON DELETE CASCADE
) STRICT;
@@ -655,6 +656,7 @@ CREATE TABLE IF NOT EXISTS session_transcript_display_state (
indexed_seq INTEGER NOT NULL CHECK (indexed_seq >= -1),
row_count INTEGER NOT NULL CHECK (row_count >= 0),
needs_rebuild INTEGER NOT NULL DEFAULT 0 CHECK (needs_rebuild IN (0, 1)),
source_generation TEXT,
updated_at INTEGER NOT NULL,
FOREIGN KEY (session_id) REFERENCES session_windows(session_id) ON DELETE CASCADE
) STRICT;
src/state/openclaw-agent-transcript-projection-source-schema.ts
+78
View File
@@ -0,0 +1,78 @@
import type { DatabaseSync } from "node:sqlite";
import { TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS } from "./openclaw-agent-db-additive-columns.js";
import { ensureColumn, tableExists, tableHasColumn } from "./openclaw-state-db-schema-helpers.js";
const ENSURED_DATABASES = new WeakSet<DatabaseSync>();
type TranscriptProjectionStateTable =
(typeof TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS)[number]["tableName"];
function projectionSourceColumnsPresent(db: DatabaseSync): boolean {
return TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS.every(
({ columnName, tableName }) =>
!tableExists(db, tableName) || tableHasColumn(db, tableName, columnName),
);
}
function adoptReadyProjectionSourceGeneration(
db: DatabaseSync,
tableName: TranscriptProjectionStateTable,
): void {
// sqlite-allow-raw -- One-time same-version column migration. Later NULL values remain stale.
db.exec(`
UPDATE ${tableName}
SET source_generation = (
SELECT generation
FROM transcript_rewrite_watermarks
WHERE session_id = ${tableName}.session_id
)
WHERE source_generation IS NULL
AND needs_rebuild = 0
AND indexed_seq = COALESCE((
SELECT MAX(seq)
FROM transcript_events
WHERE session_id = ${tableName}.session_id
), -1);
`);
}
export function hasRetiredTranscriptProjectionBindingSchema(db: DatabaseSync): boolean {
return tableExists(db, "session_transcript_projection_bindings");
}
export function dropRetiredTranscriptProjectionBindingSchema(db: DatabaseSync): void {
if (!hasRetiredTranscriptProjectionBindingSchema(db)) {
return;
}
// Retired derived ownership rows can become stale across a downgrade.
// Removing them makes any older reader rebuild instead of trusting them.
db.exec("DROP TABLE session_transcript_projection_bindings;"); // sqlite-allow-raw -- Retired additive DDL cleanup.
}
/** Adds the nullable generation owner to each present projection-state table once. */
export function ensureOpenClawAgentTranscriptProjectionSourceColumns(db: DatabaseSync): void {
if (ENSURED_DATABASES.has(db)) {
return;
}
let addedColumn = false;
for (const {
columnName,
dataType,
tableName,
} of TRANSCRIPT_PROJECTION_SOURCE_COLUMN_DEFINITIONS) {
if (!ensureColumn(db, tableName, `${columnName} ${dataType}`)) {
continue;
}
addedColumn = true;
adoptReadyProjectionSourceGeneration(db, tableName);
}
if (!addedColumn || !db.isTransaction) {
ENSURED_DATABASES.add(db);
return;
}
setImmediate(() => {
if (db.isOpen && !db.isTransaction && projectionSourceColumnsPresent(db)) {
ENSURED_DATABASES.add(db);
}
});
}
src/state/sqlite-schema-shape.test-support.ts
+2 -2
View File
@@ -75,10 +75,10 @@ type IndexXInfoRow = {
};
/** Execute schema SQL in memory and return its comparable shape. */
export function createSqliteSchemaShapeFromSql(schemaUrl: URL): SqliteSchemaShape {
export function createSqliteSchemaShapeFromSql(schema: URL | string): SqliteSchemaShape {
const db = new DatabaseSync(":memory:");
try {
db.exec(readFileSync(schemaUrl, "utf8"));
db.exec(typeof schema === "string" ? schema : readFileSync(schema, "utf8"));
return collectSqliteSchemaShape(db);
} finally {
db.close();