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

fix(sessions): large histories no longer load entire transcripts (#108851)

Browse Source 
* perf(sessions): bound SQLite history reads

* fix(sessions): keep history pagination gap-free

* test(sessions): use shared temp cleanup

* fix(sessions): reconcile mixed transcript projections

* fix(sessions): preserve strict schema migration

* refactor(sessions): trim internal export surface

* refactor(sessions): keep reader helpers private

* fix(sessions): reconcile transcript projections off requests

* test(sessions): await transcript projection repair

* fix(sessions): normalize projection worker failures

* fix(sessions): satisfy projection release gates
This commit is contained in:
Peter Steinberger
2026-07-16 14:11:33 -07:00
committed by GitHub
parent a327e43023
commit 7d71d7cf6b
35 changed files with 2879 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/max-lines-baseline.txt
-1
View File
@@ -893,7 +893,6 @@ src/gateway/session-compaction-checkpoints.ts
src/gateway/session-create-service.ts
src/gateway/session-message-events.test.ts
src/gateway/session-reset-service.ts
src/gateway/session-transcript-readers.ts
src/gateway/session-utils.fs.test.ts
src/gateway/session-utils.fs.ts
src/gateway/session-utils.subagent.test.ts
docs/.generated/sqlite-session-transcript-schema-baseline.sha256
+1 -1
View File
@@ -1 +1 @@
1754707609cb8b5248307fc7082f3a213e0069883b6be5f392180f80bc0364c5 sqlite-session-transcript-schema-baseline.sql
286d1d7ffb924f2a2fb8e12e5c50ed05daddbc6451549c052540985062272761 sqlite-session-transcript-schema-baseline.sql
scripts/lib/sqlite-session-schema-baseline.ts
+1
View File
@@ -34,6 +34,7 @@ const TARGET_TABLES = new Set([
"session_entries",
"transcript_events",
"transcript_event_identities",
"session_transcript_active_events",
]);
function sha256(value: string): string {
scripts/release-check.ts
+1
View File
@@ -111,6 +111,7 @@ const requiredPathGroups = [
"dist/agents/compaction-planning.worker.js",
"dist/agents/model-provider-auth.worker.js",
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
"dist/build-info.json",
src/config/sessions/session-accessor.sqlite-active-events.test.ts
+605
View File
@@ -0,0 +1,605 @@
// Active transcript projection tests cover branch rebuilds and bounded large-history reads.
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
import { requireNodeSqlite } from "../../infra/node-sqlite.js";
import {
closeOpenClawAgentDatabasesForTest,
openOpenClawAgentDatabase,
runOpenClawAgentWriteTransaction,
} from "../../state/openclaw-agent-db.js";
import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js";
import { appendTranscriptEvent, persistSessionTranscriptTurn } from "./session-accessor.js";
import {
readRecentSessionTranscriptMessageEvents,
readSessionTranscriptMessageAnchorPage,
readSessionTranscriptMessageEventById,
readSessionTranscriptMessageEventCount,
readSessionTranscriptMessageEventPage,
SessionTranscriptProjectionUnavailableError,
} from "./session-accessor.sqlite-active-events.js";
import { runExclusiveSqliteSessionWrite } from "./session-accessor.sqlite-scope.js";
import { appendTranscriptEventsInTransaction } from "./session-accessor.sqlite-transcript-store.js";
import {
reconcileSessionTranscriptIndexes,
startSessionTranscriptIndexReconcile,
waitForSessionTranscriptIndexReconcile,
} from "./session-transcript-reconcile.js";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
describe("SQLite active transcript event projection", () => {
let stateDir: string;
let scope: {
agentId: string;
env: NodeJS.ProcessEnv;
sessionId: string;
sessionKey: string;
};
beforeEach(() => {
stateDir = tempDirs.make("openclaw-active-transcript-");
scope = {
agentId: "main",
env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
sessionId: "active-transcript-test",
sessionKey: "agent:main:active-transcript-test",
};
});
afterEach(() => {
closeOpenClawAgentDatabasesForTest();
closeOpenClawStateDatabaseForTest();
});
it("defers branch rewind rebuilds off history and writer stacks", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "root",
parentId: null,
message: { role: "user", content: "root" },
},
{
eventId: "inactive",
parentId: "root",
message: { role: "assistant", content: "inactive" },
},
{
eventId: "active",
parentId: "root",
message: { role: "assistant", content: "active" },
},
],
touchSessionEntry: false,
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
expect(
database.db
.prepare(
"SELECT needs_rebuild, active_message_count FROM session_transcript_index_state WHERE session_id = ?",
)
.get(scope.sessionId),
).toEqual({ active_message_count: 2, needs_rebuild: 1 });
expect(() => readSessionTranscriptMessageEventCount(scope)).toThrow(
SessionTranscriptProjectionUnavailableError,
);
await waitForSessionTranscriptIndexReconcile({ agentId: scope.agentId, env: scope.env });
const page = readSessionTranscriptMessageEventPage(scope, { maxMessages: 10, offset: 0 });
expect(page.events.map((entry) => (entry.event as { id?: unknown }).id)).toEqual([
"root",
"active",
]);
expect(page.events.map((entry) => entry.seq)).toEqual([1, 2]);
expect(page.totalMessages).toBe(2);
expect(
database.db
.prepare(
"SELECT needs_rebuild, active_event_count, active_message_count FROM session_transcript_index_state WHERE session_id = ?",
)
.get(scope.sessionId),
).toEqual({ active_event_count: 2, active_message_count: 2, needs_rebuild: 0 });
expect(
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),
).toEqual([
{ active_position: 0, event_seq: 1, message_position: 0 },
{ active_position: 1, event_seq: 3, message_position: 1 },
]);
});
it("defers mixed legacy and canonical rebuilds off request stacks", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "canonical-root",
parentId: null,
message: { role: "user", content: "canonical" },
},
],
touchSessionEntry: false,
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
await appendTranscriptEvent(scope, {
id: "legacy-child",
parentId: "canonical-root",
message: { role: "assistant", content: "legacy" },
});
expect(
database.db
.prepare(
"SELECT needs_rebuild, active_message_count FROM session_transcript_index_state WHERE session_id = ?",
)
.get(scope.sessionId),
).toEqual({ active_message_count: 1, needs_rebuild: 1 });
expect(() => readSessionTranscriptMessageEventCount(scope)).toThrow(
SessionTranscriptProjectionUnavailableError,
);
await waitForSessionTranscriptIndexReconcile({ agentId: scope.agentId, env: scope.env });
const page = readSessionTranscriptMessageEventPage(scope, { maxMessages: 10, offset: 0 });
expect(page.totalMessages).toBe(1);
expect(page.events.map((entry) => (entry.event as { id?: unknown }).id)).toEqual([
"canonical-root",
]);
expect(readSessionTranscriptMessageEventById(scope, "legacy-child")).toBeUndefined();
expect(
database.db
.prepare(
"SELECT needs_rebuild, active_message_count FROM session_transcript_index_state WHERE session_id = ?",
)
.get(scope.sessionId),
).toEqual({ active_message_count: 1, needs_rebuild: 0 });
});
it("fails fast and schedules maintenance when out-of-band state is dirty", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "seed",
parentId: null,
message: { role: "user", content: "seed" },
},
],
touchSessionEntry: false,
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
database.db
.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?")
.run(scope.sessionId);
expect(() => readSessionTranscriptMessageEventCount(scope)).toThrow(
SessionTranscriptProjectionUnavailableError,
);
expect(
database.db
.prepare("SELECT needs_rebuild FROM session_transcript_index_state WHERE session_id = ?")
.get(scope.sessionId),
).toEqual({ needs_rebuild: 1 });
await waitForSessionTranscriptIndexReconcile({ agentId: scope.agentId, env: scope.env });
expect(readSessionTranscriptMessageEventCount(scope)).toBe(1);
expect(
database.db
.prepare("SELECT needs_rebuild FROM session_transcript_index_state WHERE session_id = ?")
.get(scope.sessionId),
).toEqual({ needs_rebuild: 0 });
});
it("reconciles work scheduled while an earlier pass is yielding", async () => {
const secondScope = { ...scope, sessionId: "session-2", sessionKey: "agent:main:second" };
for (const target of [scope, secondScope]) {
await persistSessionTranscriptTurn(target, {
messages: [
{
eventId: `${target.sessionId}-seed`,
parentId: null,
message: { role: "user", content: target.sessionId },
},
],
touchSessionEntry: false,
});
}
const databaseOptions = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(databaseOptions);
const markDirty = (sessionId: string) =>
database.db
.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?")
.run(sessionId);
markDirty(scope.sessionId);
startSessionTranscriptIndexReconcile({
...databaseOptions,
preferredSessionId: scope.sessionId,
});
await new Promise<void>((resolve) => {
setImmediate(() => {
markDirty(secondScope.sessionId);
startSessionTranscriptIndexReconcile({
...databaseOptions,
preferredSessionId: secondScope.sessionId,
});
resolve();
});
});
await waitForSessionTranscriptIndexReconcile(databaseOptions);
expect(
database.db
.prepare(
"SELECT session_id FROM session_transcript_index_state WHERE needs_rebuild != 0 ORDER BY session_id",
)
.all(),
).toEqual([]);
});
it("keeps projection state and rows on one snapshot during a concurrent append", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "seed",
parentId: null,
message: { role: "toolResult", content: "seed" },
},
],
touchSessionEntry: false,
});
expect(readSessionTranscriptMessageEventCount(scope)).toBe(1);
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
const state = database.db
.prepare(
`
SELECT indexed_seq, active_event_count, active_message_count
FROM session_transcript_index_state
WHERE session_id = ?
`,
)
.get(scope.sessionId) as {
active_event_count: number;
active_message_count: number;
indexed_seq: number;
};
const nextSeq = state.indexed_seq + 1;
const appendedEvent = {
type: "message",
id: "concurrent",
parentId: "seed",
message: { role: "toolResult", content: "concurrent" },
};
const { DatabaseSync } = requireNodeSqlite();
const writer = new DatabaseSync(database.path);
writer.exec("PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 1000; PRAGMA foreign_keys = ON;");
let appended = false;
const options = {
maxBytes: 1024 * 1024,
maxLines: 10,
get maxMessages() {
if (!appended) {
appended = true;
writer.exec("BEGIN IMMEDIATE;");
try {
writer
.prepare(
`
INSERT INTO transcript_events (session_id, seq, event_json, created_at)
VALUES (?, ?, ?, ?)
`,
)
.run(scope.sessionId, nextSeq, JSON.stringify(appendedEvent), Date.now());
writer
.prepare(
`
INSERT INTO transcript_event_identities
(session_id, event_id, seq, event_type, parent_id,
message_idempotency_key, created_at)
VALUES (?, 'concurrent', ?, 'message', 'seed', NULL, ?)
`,
)
.run(scope.sessionId, nextSeq, Date.now());
writer
.prepare(
`
INSERT INTO session_transcript_active_events
(session_id, active_position, event_seq, message_position)
VALUES (?, ?, ?, ?)
`,
)
.run(scope.sessionId, state.active_event_count, nextSeq, state.active_message_count);
writer
.prepare(
`
UPDATE session_transcript_index_state
SET indexed_seq = ?, leaf_event_id = 'concurrent', needs_rebuild = 0,
active_event_count = active_event_count + 1,
active_message_count = active_message_count + 1,
updated_at = ?
WHERE session_id = ?
`,
)
.run(nextSeq, Date.now(), scope.sessionId);
writer.exec("COMMIT;");
} catch (error) {
writer.exec("ROLLBACK;");
throw error;
}
}
return 10;
},
};
try {
const concurrentRead = readRecentSessionTranscriptMessageEvents(scope, options);
expect(concurrentRead.totalMessages).toBe(1);
expect(concurrentRead.events.map((entry) => (entry.event as { id?: string }).id)).toEqual([
"seed",
]);
const afterCommit = readRecentSessionTranscriptMessageEvents(scope, {
maxBytes: 1024 * 1024,
maxLines: 10,
maxMessages: 10,
});
expect(afterCommit.totalMessages).toBe(2);
expect(afterCommit.events.map((entry) => (entry.event as { id?: string }).id)).toEqual([
"seed",
"concurrent",
]);
} finally {
writer.close();
}
});
it("awaits queued completion work after the preparation worker exits", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [{ eventId: "seed", message: { role: "user", content: "seed" } }],
touchSessionEntry: false,
});
let releaseWriter!: () => void;
let writerEntered!: () => void;
const entered = new Promise<void>((resolve) => {
writerEntered = resolve;
});
const release = new Promise<void>((resolve) => {
releaseWriter = resolve;
});
const heldWriter = runExclusiveSqliteSessionWrite(
{ agentId: scope.agentId, env: scope.env },
async () => {
writerEntered();
await release;
},
);
await entered;
const outcome = reconcileSessionTranscriptIndexes({
agentId: scope.agentId,
env: scope.env,
}).then(
(value) => ({ value }),
(error: unknown) => ({ error }),
);
await new Promise((resolve) => {
setTimeout(resolve, 1_000);
});
releaseWriter();
await heldWriter;
expect(await outcome).toEqual({ value: { reconciledSessions: 0 } });
}, 10_000);
it("keeps dirty batch appends off the synchronous writer stack", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [{ eventId: "root", message: { role: "user", content: "root" } }],
touchSessionEntry: false,
});
const databaseOptions = { agentId: scope.agentId, env: scope.env };
const database = openOpenClawAgentDatabase(databaseOptions);
const original = database.db
.prepare("SELECT event_json FROM transcript_events WHERE session_id = ? AND seq = 1")
.get(scope.sessionId) as { event_json: string };
database.db
.prepare("UPDATE transcript_events SET event_json = '{' WHERE session_id = ? AND seq = 1")
.run(scope.sessionId);
runOpenClawAgentWriteTransaction((writeDatabase) => {
expect(
appendTranscriptEventsInTransaction(writeDatabase, scope, [
{ type: "leaf", id: "batch-leaf", parentId: "root", targetId: "root" },
]),
).toBe(1);
}, databaseOptions);
database.db
.prepare("UPDATE transcript_events SET event_json = ? WHERE session_id = ? AND seq = 1")
.run(original.event_json, scope.sessionId);
expect(
database.db
.prepare("SELECT needs_rebuild FROM session_transcript_index_state WHERE session_id = ?")
.get(scope.sessionId),
).toEqual({ needs_rebuild: 1 });
await waitForSessionTranscriptIndexReconcile(databaseOptions);
expect(readSessionTranscriptMessageEventCount(scope)).toBe(1);
});
it("keeps 100k-message reads bounded while rebuilds yield to live writes", async () => {
await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "seed",
message: { role: "toolResult", content: "seed" },
},
],
touchSessionEntry: false,
});
const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env });
const insertEvent = database.db.prepare(`
INSERT INTO transcript_events (session_id, seq, event_json, created_at)
VALUES (?, ?, ?, ?)
`);
const insertIdentity = database.db.prepare(`
INSERT INTO transcript_event_identities
(session_id, event_id, seq, event_type, parent_id, message_idempotency_key, created_at)
VALUES (?, ?, ?, 'message', ?, NULL, ?)
`);
const insertActive = database.db.prepare(`
INSERT INTO session_transcript_active_events
(session_id, active_position, event_seq, message_position)
VALUES (?, ?, ?, ?)
`);
database.db.exec("BEGIN IMMEDIATE;");
try {
database.db
.prepare("DELETE FROM session_transcript_fts WHERE session_id = ?")
.run(scope.sessionId);
database.db
.prepare("DELETE FROM session_transcript_index_state WHERE session_id = ?")
.run(scope.sessionId);
database.db
.prepare("DELETE FROM transcript_event_identities WHERE session_id = ?")
.run(scope.sessionId);
database.db
.prepare("DELETE FROM transcript_events WHERE session_id = ?")
.run(scope.sessionId);
insertEvent.run(
scope.sessionId,
0,
JSON.stringify({ id: scope.sessionId, type: "session", version: 3 }),
0,
);
for (let index = 1; index <= 100_000; index += 1) {
const eventId = `message-${index}`;
const parentId = index === 1 ? null : `message-${index - 1}`;
insertEvent.run(
scope.sessionId,
index,
JSON.stringify({
type: "message",
id: eventId,
parentId,
message: { role: "toolResult", content: `payload-${index}` },
}),
index,
);
insertIdentity.run(scope.sessionId, eventId, index, parentId, index);
insertActive.run(scope.sessionId, index - 1, index, index - 1);
}
database.db
.prepare(
`
INSERT INTO session_transcript_index_state
(session_id, indexed_seq, leaf_event_id, needs_rebuild,
active_event_count, active_message_count, updated_at)
VALUES (?, 100000, 'message-100000', 0, 100000, 100000, 100000)
`,
)
.run(scope.sessionId);
database.db.exec("COMMIT;");
} catch (error) {
database.db.exec("ROLLBACK;");
throw error;
}
// Parse sentinel: any accidental full materialization fails before reaching the bounded tail.
database.db
.prepare("UPDATE transcript_events SET event_json = '{' WHERE session_id = ? AND seq = 1")
.run(scope.sessionId);
const page = readSessionTranscriptMessageEventPage(scope, { maxMessages: 25, offset: 0 });
const recent = readRecentSessionTranscriptMessageEvents(scope, {
maxBytes: 1024 * 1024,
maxLines: 10,
maxMessages: 10,
});
const lineCappedRecent = readRecentSessionTranscriptMessageEvents(scope, {
maxBytes: 1024 * 1024,
maxLines: 3,
maxMessages: 10,
});
const byId = readSessionTranscriptMessageEventById(scope, "message-100000");
const anchor = readSessionTranscriptMessageAnchorPage(scope, {
maxMessages: 5,
messageId: "message-100000",
});
expect(page.totalMessages).toBe(100_000);
expect(page.events).toHaveLength(25);
expect(page.events.map((entry) => entry.seq)).toEqual(
Array.from({ length: 25 }, (_, index) => 99_976 + index),
);
expect(recent.totalMessages).toBe(100_000);
expect(recent.events).toHaveLength(10);
expect(recent.events.at(-1)?.seq).toBe(100_000);
expect(lineCappedRecent.events).toHaveLength(3);
expect(lineCappedRecent.events.at(-1)?.seq).toBe(100_000);
expect(readSessionTranscriptMessageEventCount(scope)).toBe(100_000);
expect(byId?.seq).toBe(100_000);
expect(anchor).toMatchObject({
found: true,
hasOverreadContext: true,
offset: 0,
totalMessages: 100_000,
});
expect(anchor.events).toHaveLength(6);
expect(anchor.events.at(-1)?.seq).toBe(100_000);
database.db
.prepare("UPDATE transcript_events SET event_json = ? WHERE session_id = ? AND seq = 1")
.run(
JSON.stringify({
type: "message",
id: "message-1",
parentId: null,
message: { role: "toolResult", content: "payload-1" },
}),
scope.sessionId,
);
database.db
.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?")
.run(scope.sessionId);
expect(() => readSessionTranscriptMessageEventCount(scope)).toThrow(
SessionTranscriptProjectionUnavailableError,
);
const order: string[] = [];
const reconciliation = waitForSessionTranscriptIndexReconcile({
agentId: scope.agentId,
env: scope.env,
}).then(() => order.push("reconciled"));
await new Promise<void>((resolve) => {
setImmediate(() => {
setTimeout(() => {
order.push("event-loop-responsive");
resolve();
}, 0);
});
});
expect(order).toEqual(["event-loop-responsive"]);
const liveWrite = await persistSessionTranscriptTurn(scope, {
messages: [
{
eventId: "message-100001",
parentId: "message-100000",
message: { role: "toolResult", content: "live-write" },
},
],
touchSessionEntry: false,
});
expect(liveWrite.appendedCount).toBe(1);
order.push("live-write");
await reconciliation;
expect(order).toEqual(["event-loop-responsive", "live-write", "reconciled"]);
expect(readSessionTranscriptMessageEventCount(scope)).toBe(100_001);
}, 30_000);
});
src/config/sessions/session-accessor.sqlite-active-events.ts
+382
View File
@@ -0,0 +1,382 @@
// Bounded reads over the materialized active transcript path. Dirty paths
// schedule maintenance and fail fast; clean reads deserialize selected rows.
import {
executeSqliteQuerySync,
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
openOpenClawAgentDatabase,
type OpenClawAgentDatabase,
} from "../../state/openclaw-agent-db.js";
import type {
SessionTranscriptReadScope,
TranscriptEvent,
} from "./session-accessor.sqlite-contract.js";
import {
resolveSqliteTranscriptReadScope,
toDatabaseOptions,
} from "./session-accessor.sqlite-scope.js";
import type { SessionTranscriptProjectionState } from "./session-transcript-index.js";
import { startSessionTranscriptIndexReconcile } from "./session-transcript-reconcile.js";
type ActiveTranscriptDatabase = Pick<
OpenClawAgentKyselyDatabase,
| "session_transcript_active_events"
| "session_transcript_index_state"
| "transcript_event_identities"
| "transcript_events"
>;
export type SessionTranscriptMessageEvent = {
event: TranscriptEvent;
seq: number;
};
export type SessionTranscriptMessageEventPage = {
events: SessionTranscriptMessageEvent[];
totalMessages: number;
};
export type SessionTranscriptMessageAnchorPage = SessionTranscriptMessageEventPage & {
found: boolean;
hasOverreadContext: boolean;
offset: number;
};
export class SessionTranscriptProjectionUnavailableError extends Error {
constructor(readonly sessionId: string) {
super(`Session transcript projection is rebuilding: ${sessionId}`);
this.name = "SessionTranscriptProjectionUnavailableError";
}
}
export function isSessionTranscriptProjectionUnavailableError(
error: unknown,
): error is SessionTranscriptProjectionUnavailableError {
return error instanceof SessionTranscriptProjectionUnavailableError;
}
type CurrentProjection = {
database: OpenClawAgentDatabase;
resolved: ReturnType<typeof resolveSqliteTranscriptReadScope>;
state: SessionTranscriptProjectionState;
};
const EMPTY_PROJECTION_STATE: SessionTranscriptProjectionState = {
activeEventCount: 0,
activeMessageCount: 0,
indexedSeq: -1,
leafEventId: null,
needsRebuild: false,
};
function getActiveTranscriptKysely(database: OpenClawAgentDatabase) {
return getNodeSqliteKysely<ActiveTranscriptDatabase>(database.db);
}
function readProjectionSnapshot(
database: OpenClawAgentDatabase,
sessionId: string,
): { latestSeq: number; state?: SessionTranscriptProjectionState } | undefined {
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")
.select([
"latest.seq as latest_seq",
"state.active_event_count",
"state.active_message_count",
"state.indexed_seq",
"state.leaf_event_id",
"state.needs_rebuild",
])
.where("latest.session_id", "=", sessionId)
.orderBy("latest.seq", "desc")
.limit(1),
);
if (!row) {
return undefined;
}
return {
latestSeq: row.latest_seq,
...(typeof row.indexed_seq === "number"
? {
state: {
activeEventCount: row.active_event_count ?? 0,
activeMessageCount: row.active_message_count ?? 0,
indexedSeq: row.indexed_seq,
leafEventId: row.leaf_event_id,
needsRebuild: row.needs_rebuild !== 0,
},
}
: {}),
};
}
function withCurrentProjectionSnapshot<T>(
scope: SessionTranscriptReadScope,
read: (projection: CurrentProjection) => T,
): T {
const resolved = resolveSqliteTranscriptReadScope(scope);
const databaseOptions = toDatabaseOptions(resolved);
const database = openOpenClawAgentDatabase(databaseOptions);
const result = runSqliteDeferredTransactionSync(
database.db,
() => {
const snapshot = readProjectionSnapshot(database, resolved.sessionId);
if (!snapshot) {
return {
kind: "value" as const,
value: read({ database, resolved, state: EMPTY_PROJECTION_STATE }),
};
}
if (
snapshot.state &&
!snapshot.state.needsRebuild &&
snapshot.state.indexedSeq === snapshot.latestSeq
) {
return {
kind: "value" as const,
value: read({ database, resolved, state: snapshot.state }),
};
}
return { kind: "unavailable" as const };
},
{
databaseLabel: database.path,
operationLabel: "sessions.history.read",
},
);
if (result.kind === "value") {
return result.value;
}
// Request latency never scales with transcript size. The maintenance owner
// rebuilds after this stack unwinds; callers return a retryable response.
startSessionTranscriptIndexReconcile({
...databaseOptions,
preferredSessionId: resolved.sessionId,
});
throw new SessionTranscriptProjectionUnavailableError(resolved.sessionId);
}
function parseMessageEventRow(row: {
event_json: string;
message_position: number | null;
}): SessionTranscriptMessageEvent {
if (row.message_position === null) {
throw new Error("Active transcript message row is missing its message position");
}
return {
event: JSON.parse(row.event_json) as TranscriptEvent,
// Gateway cursors use the visible-message ordinal, matching the JSONL index.
// Raw event seq includes headers/control rows and would make pages overlap.
seq: row.message_position + 1,
};
}
function readMessageRange(
projection: CurrentProjection,
start: number,
endExclusive: number,
): SessionTranscriptMessageEvent[] {
if (endExclusive <= start) {
return [];
}
const db = getActiveTranscriptKysely(projection.database);
return executeSqliteQuerySync(
projection.database.db,
db
.selectFrom("session_transcript_active_events as active")
.innerJoin("transcript_events as event", (join) =>
join
.onRef("event.session_id", "=", "active.session_id")
.onRef("event.seq", "=", "active.event_seq"),
)
.select(["active.message_position", "event.event_json"])
.where("active.session_id", "=", projection.resolved.sessionId)
.where("active.message_position", "is not", null)
.where("active.message_position", ">=", start)
.where("active.message_position", "<", endExclusive)
.orderBy("active.message_position", "asc"),
).rows.map(parseMessageEventRow);
}
/** Reads every message event on the active path. Full callers remain intentionally O(output). */
export function readSessionTranscriptMessageEvents(
scope: SessionTranscriptReadScope,
): SessionTranscriptMessageEvent[] {
return withCurrentProjectionSnapshot(scope, (projection) =>
readMessageRange(projection, 0, projection.state.activeMessageCount),
);
}
/** Reads a bounded active-path tail while preserving transcript line and byte caps. */
export function readRecentSessionTranscriptMessageEvents(
scope: SessionTranscriptReadScope,
options: { maxBytes: number; maxLines: number; maxMessages: number },
): SessionTranscriptMessageEventPage {
return withCurrentProjectionSnapshot(scope, (projection) => {
const maxMessages = Math.max(
0,
Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 0),
);
const maxLines = Math.max(
0,
Math.floor(Number.isFinite(options.maxLines) ? options.maxLines : 0),
);
if (maxMessages === 0 || maxLines === 0) {
return { events: [], totalMessages: projection.state.activeMessageCount };
}
const maxBytes = Math.max(
1024,
Math.floor(Number.isFinite(options.maxBytes) ? options.maxBytes : 8 * 1024 * 1024),
);
const db = getActiveTranscriptKysely(projection.database);
const rows = executeSqliteQuerySync(
projection.database.db,
db
.selectFrom("session_transcript_active_events as active")
.innerJoin("transcript_events as event", (join) =>
join
.onRef("event.session_id", "=", "active.session_id")
.onRef("event.seq", "=", "active.event_seq"),
)
.select(["active.event_seq", "active.message_position", "event.event_json"])
.where("active.session_id", "=", projection.resolved.sessionId)
.orderBy("active.active_position", "desc")
.limit(maxLines),
).rows;
const selected: typeof rows = [];
let bytes = 0;
for (const row of rows) {
const rowBytes = Buffer.byteLength(row.event_json) + 1;
if (selected.length > 0 && bytes + rowBytes > maxBytes) {
break;
}
selected.push(row);
bytes += rowBytes;
}
const events = selected
.toReversed()
.filter((row) => row.message_position !== null)
.map(parseMessageEventRow);
return {
events: events.length > maxMessages ? events.slice(-maxMessages) : events,
totalMessages: projection.state.activeMessageCount,
};
});
}
/** Reads one tail-relative message page with index range predicates, never OFFSET scanning. */
export function readSessionTranscriptMessageEventPage(
scope: SessionTranscriptReadScope,
options: { maxMessages: number; offset: number },
): SessionTranscriptMessageEventPage {
return withCurrentProjectionSnapshot(scope, (projection) => {
const totalMessages = projection.state.activeMessageCount;
const offset = Math.min(
Math.max(0, Math.floor(Number.isFinite(options.offset) ? options.offset : 0)),
totalMessages,
);
const maxMessages = Math.max(
0,
Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 0),
);
const endExclusive = Math.max(0, totalMessages - offset);
const start = Math.max(0, endExclusive - maxMessages);
return {
events: readMessageRange(projection, start, endExclusive),
totalMessages,
};
});
}
/** Counts active-path messages from the transactionally maintained watermark. */
export function readSessionTranscriptMessageEventCount(scope: SessionTranscriptReadScope): number {
return withCurrentProjectionSnapshot(scope, (projection) => projection.state.activeMessageCount);
}
/** Reads one active message by event id without materializing sibling rows. */
export function readSessionTranscriptMessageEventById(
scope: SessionTranscriptReadScope,
messageId: string,
): SessionTranscriptMessageEvent | undefined {
return withCurrentProjectionSnapshot(scope, (projection) => {
const db = getActiveTranscriptKysely(projection.database);
const row = executeSqliteQueryTakeFirstSync(
projection.database.db,
db
.selectFrom("transcript_event_identities as identity")
.innerJoin("session_transcript_active_events as active", (join) =>
join
.onRef("active.session_id", "=", "identity.session_id")
.onRef("active.event_seq", "=", "identity.seq"),
)
.innerJoin("transcript_events as event", (join) =>
join
.onRef("event.session_id", "=", "active.session_id")
.onRef("event.seq", "=", "active.event_seq"),
)
.select(["active.message_position", "event.event_json"])
.where("identity.session_id", "=", projection.resolved.sessionId)
.where("identity.event_id", "=", messageId)
.where("active.message_position", "is not", null),
);
return row ? parseMessageEventRow(row) : undefined;
});
}
/** Reads a centered active-message page plus one older context row for split rendering. */
export function readSessionTranscriptMessageAnchorPage(
scope: SessionTranscriptReadScope,
options: { maxMessages: number; messageId: string },
): SessionTranscriptMessageAnchorPage {
return withCurrentProjectionSnapshot(scope, (projection) => {
const db = getActiveTranscriptKysely(projection.database);
const anchor = executeSqliteQueryTakeFirstSync(
projection.database.db,
db
.selectFrom("transcript_event_identities as identity")
.innerJoin("session_transcript_active_events as active", (join) =>
join
.onRef("active.session_id", "=", "identity.session_id")
.onRef("active.event_seq", "=", "identity.seq"),
)
.select("active.message_position")
.where("identity.session_id", "=", projection.resolved.sessionId)
.where("identity.event_id", "=", options.messageId)
.where("active.message_position", "is not", null),
);
const totalMessages = projection.state.activeMessageCount;
if (anchor?.message_position === null || anchor?.message_position === undefined) {
return {
events: [],
found: false,
hasOverreadContext: false,
offset: 0,
totalMessages,
};
}
const pageSize = Math.max(
1,
Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 1),
);
const newerMessages = Math.floor(pageSize / 2);
const olderMessages = pageSize - newerMessages - 1;
const latestStart = Math.max(0, totalMessages - pageSize);
const start = Math.min(Math.max(0, anchor.message_position - olderMessages), latestStart);
const endExclusive = Math.min(totalMessages, start + pageSize);
const readStart = Math.max(0, start - 1);
return {
events: readMessageRange(projection, readStart, endExclusive),
found: true,
hasOverreadContext: readStart < start,
offset: totalMessages - endExclusive,
totalMessages,
};
});
}
src/config/sessions/session-accessor.sqlite-transcript-store.ts
+47 -4
View File
@@ -23,7 +23,11 @@ import {
readNextTranscriptSeq,
touchTranscriptMutationInTransaction,
} from "./session-accessor.sqlite-transcript-state.js";
import { indexAppendedTranscriptEventInTransaction } from "./session-transcript-index.js";
import {
indexAppendedTranscriptEventInTransaction,
reconcileSessionTranscriptIndexInTransaction,
} from "./session-transcript-index.js";
import { startSessionTranscriptIndexReconcile } from "./session-transcript-reconcile.js";
import { createSessionTranscriptHeader } from "./transcript-header.js";
import {
isSessionTranscriptLeafControl,
@@ -35,7 +39,12 @@ export function appendTranscriptEventInTransaction(
database: OpenClawAgentDatabase,
scope: ResolvedTranscriptScope,
event: TranscriptEvent,
options: { dedupeByMessageIdempotency?: boolean; touchMutation?: boolean } = {},
options: {
dedupeByMessageIdempotency?: boolean;
onProjectionReconcileNeeded?: () => void;
scheduleProjectionReconcile?: boolean;
touchMutation?: boolean;
} = {},
): boolean {
const db = getSessionKysely(database.db);
const createdAt = readEventTimestamp(event) ?? Date.now();
@@ -68,14 +77,18 @@ export function appendTranscriptEventInTransaction(
if (options.touchMutation !== false) {
touchTranscriptMutationInTransaction(database, scope.sessionId);
}
indexAppendedTranscriptEventInTransaction(database.db, {
const projectionNeedsRebuild = indexAppendedTranscriptEventInTransaction(database.db, {
sessionId: scope.sessionId,
seq,
event,
eventId: identity?.eventId ?? null,
createdAt,
});
if (projectionNeedsRebuild) {
options.onProjectionReconcileNeeded?.();
}
if (!identity) {
scheduleTranscriptProjectionReconcile(database, scope, projectionNeedsRebuild, options);
return true;
}
// Caller-checked appends may retain a duplicate key in the payload, but the
@@ -105,22 +118,51 @@ export function appendTranscriptEventInTransaction(
})
.onConflict((conflict) => conflict.columns(["session_id", "event_id"]).doNothing()),
);
scheduleTranscriptProjectionReconcile(database, scope, projectionNeedsRebuild, options);
return true;
}
function scheduleTranscriptProjectionReconcile(
database: OpenClawAgentDatabase,
scope: ResolvedTranscriptScope,
projectionNeedsRebuild: boolean,
options: { scheduleProjectionReconcile?: boolean },
): void {
if (!projectionNeedsRebuild || options.scheduleProjectionReconcile === false) {
return;
}
// setImmediate in the reconcile owner runs only after this synchronous
// SQLite transaction commits, keeping full-tree work off the writer stack.
startSessionTranscriptIndexReconcile({
agentId: scope.agentId,
path: database.path,
preferredSessionId: scope.sessionId,
});
}
export function appendTranscriptEventsInTransaction(
database: OpenClawAgentDatabase,
scope: ResolvedTranscriptScope,
events: readonly TranscriptEvent[],
): number {
let appended = 0;
let projectionNeedsRebuild = false;
for (const event of events) {
if (appendTranscriptEventInTransaction(database, scope, event, { touchMutation: false })) {
if (
appendTranscriptEventInTransaction(database, scope, event, {
onProjectionReconcileNeeded: () => {
projectionNeedsRebuild = true;
},
scheduleProjectionReconcile: false,
touchMutation: false,
})
) {
appended += 1;
}
}
if (appended > 0) {
touchTranscriptMutationInTransaction(database, scope.sessionId);
scheduleTranscriptProjectionReconcile(database, scope, projectionNeedsRebuild, {});
}
return appended;
}
@@ -291,6 +333,7 @@ export function replaceSqliteTranscriptEventsInTransaction(
}
if (deleted || seq > 0) {
touchTranscriptMutationInTransaction(database, resolved.sessionId);
reconcileSessionTranscriptIndexInTransaction(database.db, resolved.sessionId);
}
}
src/config/sessions/session-accessor.sqlite-transcript-write.ts
+3
View File
@@ -51,6 +51,7 @@ import {
redactTranscriptMessageForStorage,
replaceSqliteTranscriptEventsInTransaction,
} from "./session-accessor.sqlite-transcript-store.js";
import { reconcileSessionTranscriptIndexInTransaction } from "./session-transcript-index.js";
import type {
SessionTranscriptTurnExpectedState,
SessionTranscriptTurnLifecyclePatch,
@@ -185,6 +186,7 @@ export async function importSqliteSessionRows(
}
if (
appendTranscriptEventInTransaction(database, transcriptScope, event, {
scheduleProjectionReconcile: false,
touchMutation: false,
})
) {
@@ -192,6 +194,7 @@ export async function importSqliteSessionRows(
transcriptEvents += 1;
}
});
reconcileSessionTranscriptIndexInTransaction(database.db, params.entry.sessionId);
}
if (params.transcriptMtimeMs !== undefined) {
advanceTranscriptMutationAtInTransaction(
src/config/sessions/session-accessor.ts
+15
View File
@@ -180,6 +180,21 @@ export {
withTranscriptWriteTransaction,
} from "./session-accessor.transcript.js";
export { persistSessionTranscriptTurn } from "./session-accessor.transcript-turn.js";
export {
isSessionTranscriptProjectionUnavailableError,
readRecentSessionTranscriptMessageEvents,
readSessionTranscriptMessageAnchorPage,
readSessionTranscriptMessageEventById,
readSessionTranscriptMessageEventCount,
readSessionTranscriptMessageEventPage,
readSessionTranscriptMessageEvents,
SessionTranscriptProjectionUnavailableError,
} from "./session-accessor.sqlite-active-events.js";
export type {
SessionTranscriptMessageAnchorPage,
SessionTranscriptMessageEvent,
SessionTranscriptMessageEventPage,
} from "./session-accessor.sqlite-active-events.js";
export {
resolveSessionTranscriptReadTarget,
resolveSessionTranscriptRuntimeReadTarget,
src/config/sessions/session-transcript-index.ts
+196 -101
View File
@@ -1,11 +1,11 @@
// Transcript FTS index maintenance shared by the SQLite session accessor
// (in-transaction hooks) and session-transcript-search (reconcile + query).
// The index mirrors the ACTIVE transcript branch only. Invariant: the
// Active transcript projection maintenance shared by the SQLite session
// accessor, bounded history readers, and full-text search. Both projections
// mirror the ACTIVE transcript branch only. Invariant: the
// watermark's leaf_event_id always equals the append parent the accessor
// would resolve next; an append that chains onto it forward-indexes in the
// same transaction, anything ambiguous (leaf controls, branch switches)
// marks the session dirty and the next search rebuilds it from the same
// visible-path resolution sessions_history uses.
// marks the session dirty for its write or maintenance owner to rebuild from
// the canonical visible-path resolver.
import type { DatabaseSync } from "node:sqlite";
import {
executeSqliteQuerySync,
@@ -13,6 +13,12 @@ import {
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
extractTranscriptIndexEntry,
hasTranscriptMessage,
shouldProjectActiveEvent,
type TranscriptIndexEntry,
} from "./session-transcript-projection-rebuild.js";
import {
isCanonicalSessionTranscriptEntry,
isSessionTranscriptLeafControl,
@@ -26,107 +32,53 @@ import {
type TranscriptIndexDatabase = Pick<
OpenClawAgentKyselyDatabase,
"sessions" | "session_transcript_fts" | "session_transcript_index_state" | "transcript_events"
| "sessions"
| "session_transcript_active_events"
| "session_transcript_fts"
| "session_transcript_index_state"
| "transcript_events"
>;
type TranscriptIndexEntry = {
messageId: string;
role: "assistant" | "user";
text: string;
timestamp: number;
};
type TranscriptIndexWatermark = {
export type SessionTranscriptProjectionState = {
activeEventCount: number;
activeMessageCount: number;
indexedSeq: number;
leafEventId: string | null;
needsRebuild: boolean;
};
type SessionTranscriptProjectionSourceRow = {
event: unknown;
seq: number;
};
function getIndexKysely(db: DatabaseSync) {
return getNodeSqliteKysely<TranscriptIndexDatabase>(db);
}
function readMessageText(message: unknown): string | undefined {
if (!message || typeof message !== "object" || Array.isArray(message)) {
return undefined;
}
const record = message as { content?: unknown; role?: unknown; text?: unknown };
if (record.role !== "user" && record.role !== "assistant") {
return undefined;
}
if (typeof record.content === "string") {
return record.content.trim() || undefined;
}
if (typeof record.text === "string") {
return record.text.trim() || undefined;
}
if (!Array.isArray(record.content)) {
return undefined;
}
const parts = record.content.flatMap((block) => {
if (!block || typeof block !== "object" || Array.isArray(block)) {
return [];
}
const part = block as { text?: unknown; type?: unknown };
if (part.type !== "text" && part.type !== "input_text" && part.type !== "output_text") {
return [];
}
return typeof part.text === "string" && part.text.trim() ? [part.text] : [];
});
return parts.length > 0 ? parts.join("\n") : undefined;
}
/**
* Extracts the searchable payload from one transcript event. Only user and
* assistant message text is indexed; tool results, reasoning blocks, and
* images stay out of the index by construction.
*/
function extractTranscriptIndexEntry(
event: unknown,
fallbackTimestamp: number,
): TranscriptIndexEntry | undefined {
if (!event || typeof event !== "object" || Array.isArray(event)) {
return undefined;
}
const record = event as { id?: unknown; message?: unknown; timestamp?: unknown; type?: unknown };
if (record.type !== "message" || typeof record.id !== "string" || !record.id.trim()) {
return undefined;
}
const message = record.message as { role?: unknown } | undefined;
const role = message?.role;
if (role !== "user" && role !== "assistant") {
return undefined;
}
const text = readMessageText(message);
if (!text) {
return undefined;
}
const timestamp =
typeof record.timestamp === "number"
? record.timestamp
: typeof record.timestamp === "string"
? Date.parse(record.timestamp)
: Number.NaN;
return {
messageId: record.id.trim(),
role,
text,
timestamp: Number.isFinite(timestamp) ? timestamp : fallbackTimestamp,
};
}
function readWatermark(db: DatabaseSync, sessionId: string): TranscriptIndexWatermark | undefined {
function readSessionTranscriptProjectionState(
db: DatabaseSync,
sessionId: string,
): SessionTranscriptProjectionState | undefined {
const row = executeSqliteQueryTakeFirstSync(
db,
getIndexKysely(db)
.selectFrom("session_transcript_index_state")
.select(["indexed_seq", "leaf_event_id", "needs_rebuild"])
.select([
"active_event_count",
"active_message_count",
"indexed_seq",
"leaf_event_id",
"needs_rebuild",
])
.where("session_id", "=", sessionId),
);
if (!row) {
return undefined;
}
return {
activeEventCount: row.active_event_count,
activeMessageCount: row.active_message_count,
indexedSeq: row.indexed_seq,
leafEventId: row.leaf_event_id,
needsRebuild: row.needs_rebuild !== 0,
@@ -136,7 +88,7 @@ function readWatermark(db: DatabaseSync, sessionId: string): TranscriptIndexWate
function writeWatermark(
db: DatabaseSync,
sessionId: string,
watermark: TranscriptIndexWatermark,
watermark: SessionTranscriptProjectionState,
now: number,
): void {
executeSqliteQuerySync(
@@ -145,6 +97,8 @@ function writeWatermark(
.insertInto("session_transcript_index_state")
.values({
session_id: sessionId,
active_event_count: watermark.activeEventCount,
active_message_count: watermark.activeMessageCount,
indexed_seq: watermark.indexedSeq,
leaf_event_id: watermark.leafEventId,
needs_rebuild: watermark.needsRebuild ? 1 : 0,
@@ -152,6 +106,8 @@ function writeWatermark(
})
.onConflict((conflict) =>
conflict.column("session_id").doUpdateSet({
active_event_count: watermark.activeEventCount,
active_message_count: watermark.activeMessageCount,
indexed_seq: watermark.indexedSeq,
leaf_event_id: watermark.leafEventId,
needs_rebuild: watermark.needsRebuild ? 1 : 0,
@@ -161,6 +117,35 @@ function writeWatermark(
);
}
function insertActiveEventRow(
db: DatabaseSync,
params: {
activePosition: number;
eventSeq: number;
messagePosition: number | null;
sessionId: string;
},
): void {
executeSqliteQuerySync(
db,
getIndexKysely(db).insertInto("session_transcript_active_events").values({
session_id: params.sessionId,
active_position: params.activePosition,
event_seq: params.eventSeq,
message_position: params.messagePosition,
}),
);
}
function deleteActiveEventRows(db: DatabaseSync, sessionId: string): void {
executeSqliteQuerySync(
db,
getIndexKysely(db)
.deleteFrom("session_transcript_active_events")
.where("session_id", "=", sessionId),
);
}
function insertFtsRow(db: DatabaseSync, sessionId: string, entry: TranscriptIndexEntry): void {
executeSqliteQuerySync(
db,
@@ -202,24 +187,30 @@ export function indexAppendedTranscriptEventInTransaction(
eventId: string | null;
createdAt: number;
},
): void {
const watermark = readWatermark(db, params.sessionId);
): boolean {
const watermark = readSessionTranscriptProjectionState(db, params.sessionId);
if (!watermark) {
if (params.seq !== 0) {
// Pre-existing rows without index state (e.g. doctor-migrated
// transcripts): stay unindexed until reconcile rebuilds the session.
return;
return true;
}
applyForwardIndex(db, params, { indexedSeq: -1, leafEventId: null, needsRebuild: false });
return;
applyForwardIndex(db, params, {
activeEventCount: 0,
activeMessageCount: 0,
indexedSeq: -1,
leafEventId: null,
needsRebuild: false,
});
return false;
}
if (watermark.needsRebuild) {
return;
return true;
}
if (params.seq !== watermark.indexedSeq + 1) {
// Out-of-band writes bypassed the hook; reconcile recomputes the truth.
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
return;
return true;
}
if (
isSessionTranscriptLeafControl(params.event) ||
@@ -229,14 +220,32 @@ export function indexAppendedTranscriptEventInTransaction(
// the main chain; the visible path must be re-resolved rather than
// guessed at append time.
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
return;
return true;
}
const isCanonicalEvent = isCanonicalSessionTranscriptEntry(params.event);
if (isCanonicalEvent && watermark.leafEventId === null && watermark.activeEventCount > 0) {
// A canonical tree supersedes legacy flat message rows. Re-resolve once
// instead of retaining rows that are no longer on the selected path.
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
return true;
}
const treeEntry = parseSessionTranscriptTreeEntry(params.event);
if (
!isCanonicalEvent &&
watermark.leafEventId !== null &&
shouldProjectActiveEvent(params.event)
) {
// A noncanonical row after a tracked tree cursor may be a flat fallback or
// an opaque append ancestor. Only the full resolver can decide visibility.
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
return true;
}
if (treeEntry && treeEntry.parentId !== watermark.leafEventId) {
markSessionTranscriptIndexDirtyInTransaction(db, params.sessionId);
return;
return true;
}
applyForwardIndex(db, params, watermark);
return false;
}
function applyForwardIndex(
@@ -248,12 +257,22 @@ function applyForwardIndex(
eventId: string | null;
createdAt: number;
},
watermark: TranscriptIndexWatermark,
watermark: SessionTranscriptProjectionState,
): void {
const entry = extractTranscriptIndexEntry(params.event, params.createdAt);
if (entry) {
insertFtsRow(db, params.sessionId, entry);
}
const projectsActiveEvent = shouldProjectActiveEvent(params.event);
const projectsMessage = projectsActiveEvent && hasTranscriptMessage(params.event);
if (projectsActiveEvent) {
insertActiveEventRow(db, {
activePosition: watermark.activeEventCount,
eventSeq: params.seq,
messagePosition: projectsMessage ? watermark.activeMessageCount : null,
sessionId: params.sessionId,
});
}
// Mirror scanSessionTranscriptTree's leaf advancement: canonical entries
// (parent-linked or parentless) become the tip the next append chains to;
// headers and unknown control rows leave the tip untouched.
@@ -262,6 +281,8 @@ function applyForwardIndex(
db,
params.sessionId,
{
activeEventCount: watermark.activeEventCount + (projectsActiveEvent ? 1 : 0),
activeMessageCount: watermark.activeMessageCount + (projectsMessage ? 1 : 0),
indexedSeq: params.seq,
leafEventId: advancesLeaf ? params.eventId : watermark.leafEventId,
needsRebuild: false,
@@ -273,11 +294,13 @@ function applyForwardIndex(
/** Marks one session for lazy rebuild without touching its FTS rows. */
function markSessionTranscriptIndexDirtyInTransaction(db: DatabaseSync, sessionId: string): void {
const now = Date.now();
const watermark = readWatermark(db, sessionId);
const watermark = readSessionTranscriptProjectionState(db, sessionId);
writeWatermark(
db,
sessionId,
{
activeEventCount: watermark?.activeEventCount ?? 0,
activeMessageCount: watermark?.activeMessageCount ?? 0,
indexedSeq: watermark?.indexedSeq ?? -1,
leafEventId: watermark?.leafEventId ?? null,
needsRebuild: true,
@@ -292,6 +315,7 @@ export function deleteSessionTranscriptIndexInTransaction(
sessionId: string,
): void {
deleteFtsRows(db, sessionId);
deleteActiveEventRows(db, sessionId);
executeSqliteQuerySync(
db,
getIndexKysely(db)
@@ -305,25 +329,45 @@ export function deleteSessionTranscriptIndexInTransaction(
* rows, indexes the resolved active branch, and resets the watermark to the
* same append parent the accessor's next append will resolve.
*/
export function rebuildSessionTranscriptIndexInTransaction(
function rebuildSessionTranscriptIndexInTransaction(
db: DatabaseSync,
sessionId: string,
events: readonly unknown[],
maxSeq: number,
rows: readonly SessionTranscriptProjectionSourceRow[],
): void {
deleteFtsRows(db, sessionId);
deleteActiveEventRows(db, sessionId);
const now = Date.now();
const events = rows.map((row) => row.event);
let activeEventCount = 0;
let activeMessageCount = 0;
for (const entry of selectVisibleTranscriptEventEntries(events)) {
const indexed = extractTranscriptIndexEntry(entry.event, now);
if (indexed) {
insertFtsRow(db, sessionId, indexed);
}
const source = rows[entry.seq - 1];
if (!source || !shouldProjectActiveEvent(entry.event)) {
continue;
}
const projectsMessage = hasTranscriptMessage(entry.event);
insertActiveEventRow(db, {
activePosition: activeEventCount,
eventSeq: source.seq,
messagePosition: projectsMessage ? activeMessageCount : null,
sessionId,
});
activeEventCount += 1;
if (projectsMessage) {
activeMessageCount += 1;
}
}
writeWatermark(
db,
sessionId,
{
indexedSeq: maxSeq,
activeEventCount,
activeMessageCount,
indexedSeq: rows.at(-1)?.seq ?? -1,
leafEventId: resolveVisibleTranscriptAppendParentId(events),
needsRebuild: false,
},
@@ -331,6 +375,47 @@ export function rebuildSessionTranscriptIndexInTransaction(
);
}
/** Rebuilds one lagging projection under its current write transaction. */
export function reconcileSessionTranscriptIndexInTransaction(
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) {
deleteSessionTranscriptIndexInTransaction(db, sessionId);
return false;
}
const state = readSessionTranscriptProjectionState(db, sessionId);
if (state && !state.needsRebuild && state.indexedSeq === latest.seq) {
return false;
}
const rows = executeSqliteQuerySync(
db,
getIndexKysely(db)
.selectFrom("transcript_events")
.select(["event_json", "seq"])
.where("session_id", "=", sessionId)
.orderBy("seq", "asc"),
).rows;
rebuildSessionTranscriptIndexInTransaction(
db,
sessionId,
rows.map((row) => ({
event: JSON.parse(row.event_json) as unknown,
seq: row.seq,
})),
);
return true;
}
/**
* Sessions whose index needs reconcile work: flagged rebuilds, transcripts
* that gained rows without index state (doctor imports), and watermarks
@@ -376,6 +461,16 @@ export function listSessionsNeedingTranscriptIndexReconcile(db: DatabaseSync): s
/** Drops index rows for sessions whose transcript rows are gone. */
export function deleteOrphanedTranscriptIndexRowsInTransaction(db: DatabaseSync): void {
const kysely = getIndexKysely(db);
executeSqliteQuerySync(
db,
kysely
.deleteFrom("session_transcript_active_events")
.where(
"session_id",
"not in",
kysely.selectFrom("transcript_events").select("session_id").distinct(),
),
);
executeSqliteQuerySync(
db,
kysely
src/config/sessions/session-transcript-projection-rebuild.ts
+429
View File
@@ -0,0 +1,429 @@
import type { DatabaseSync } from "node:sqlite";
import type { Generated } from "kysely";
import {
executeSqliteQuerySync,
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
} from "../../infra/kysely-sync.js";
import { runSqliteDeferredTransactionSync } from "../../infra/sqlite-transaction.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
isCanonicalSessionTranscriptEntry,
parseSessionTranscriptTreeEntry,
} from "./transcript-tree.js";
import {
resolveVisibleTranscriptAppendParentId,
selectVisibleTranscriptEventEntries,
} from "./transcript-visible-events.js";
type TranscriptProjectionDatabase = Pick<
OpenClawAgentKyselyDatabase,
"sessions" | "session_transcript_index_state" | "transcript_events"
> & {
session_transcript_active_events: OpenClawAgentKyselyDatabase["session_transcript_active_events"] & {
rowid: Generated<number>;
};
session_transcript_fts: OpenClawAgentKyselyDatabase["session_transcript_fts"] & {
rowid: Generated<number>;
};
};
export type TranscriptIndexEntry = {
messageId: string;
role: "assistant" | "user";
text: string;
timestamp: number;
};
export type PreparedSessionTranscriptProjectionMetadata = {
activeEventCount: number;
activeMessageCount: number;
leafEventId: string | null;
sessionId: string;
sourceIndexedSeq: number;
sourceTranscriptUpdatedAt: number | null;
};
export type PreparedSessionTranscriptProjection = PreparedSessionTranscriptProjectionMetadata & {
activeRows: Array<{
activePosition: number;
eventSeq: number;
messagePosition: number | null;
}>;
ftsRows: TranscriptIndexEntry[];
};
type ProjectionDeleteChunkResult = {
hasMore: boolean;
owned: boolean;
};
function getProjectionKysely(db: DatabaseSync) {
return getNodeSqliteKysely<TranscriptProjectionDatabase>(db);
}
function readMessageText(message: unknown): string | undefined {
if (!message || typeof message !== "object" || Array.isArray(message)) {
return undefined;
}
const record = message as { content?: unknown; role?: unknown; text?: unknown };
if (record.role !== "user" && record.role !== "assistant") {
return undefined;
}
if (typeof record.content === "string") {
return record.content.trim() || undefined;
}
if (typeof record.text === "string") {
return record.text.trim() || undefined;
}
if (!Array.isArray(record.content)) {
return undefined;
}
const parts = record.content.flatMap((block) => {
if (!block || typeof block !== "object" || Array.isArray(block)) {
return [];
}
const part = block as { text?: unknown; type?: unknown };
if (part.type !== "text" && part.type !== "input_text" && part.type !== "output_text") {
return [];
}
return typeof part.text === "string" && part.text.trim() ? [part.text] : [];
});
return parts.length > 0 ? parts.join("\n") : undefined;
}
/** Extracts the searchable user/assistant text from one transcript event. */
export function extractTranscriptIndexEntry(
event: unknown,
fallbackTimestamp: number,
): TranscriptIndexEntry | undefined {
if (!event || typeof event !== "object" || Array.isArray(event)) {
return undefined;
}
const record = event as { id?: unknown; message?: unknown; timestamp?: unknown; type?: unknown };
if (record.type !== "message" || typeof record.id !== "string" || !record.id.trim()) {
return undefined;
}
const message = record.message as { role?: unknown } | undefined;
const role = message?.role;
if (role !== "user" && role !== "assistant") {
return undefined;
}
const text = readMessageText(message);
if (!text) {
return undefined;
}
const timestamp =
typeof record.timestamp === "number"
? record.timestamp
: typeof record.timestamp === "string"
? Date.parse(record.timestamp)
: Number.NaN;
return {
messageId: record.id.trim(),
role,
text,
timestamp: Number.isFinite(timestamp) ? timestamp : fallbackTimestamp,
};
}
export function hasTranscriptMessage(event: unknown): boolean {
return (
typeof event === "object" &&
event !== null &&
!Array.isArray(event) &&
Object.hasOwn(event, "message") &&
(event as { message?: unknown }).message !== undefined
);
}
export function shouldProjectActiveEvent(event: unknown): boolean {
if (!event || typeof event !== "object" || Array.isArray(event)) {
return false;
}
const record = event as { type?: unknown };
if (record.type === "session") {
return false;
}
return (
isCanonicalSessionTranscriptEntry(event) ||
parseSessionTranscriptTreeEntry(event) !== undefined ||
hasTranscriptMessage(event)
);
}
/** Reads and resolves one projection on a worker-owned SQLite snapshot. */
export function prepareSessionTranscriptProjection(
db: DatabaseSync,
sessionId: string,
): PreparedSessionTranscriptProjection | undefined {
return runSqliteDeferredTransactionSync(
db,
() => {
const kysely = getProjectionKysely(db);
const session = executeSqliteQueryTakeFirstSync(
db,
kysely
.selectFrom("sessions")
.select("transcript_updated_at")
.where("session_id", "=", sessionId),
);
const rows = executeSqliteQuerySync(
db,
kysely
.selectFrom("transcript_events")
.select(["event_json", "seq"])
.where("session_id", "=", sessionId)
.orderBy("seq", "asc"),
).rows;
if (!session || rows.length === 0) {
return undefined;
}
const now = Date.now();
const events = rows.map((row) => JSON.parse(row.event_json) as unknown);
const activeRows: PreparedSessionTranscriptProjection["activeRows"] = [];
const ftsRows: TranscriptIndexEntry[] = [];
let activeMessageCount = 0;
for (const entry of selectVisibleTranscriptEventEntries(events)) {
const indexed = extractTranscriptIndexEntry(entry.event, now);
if (indexed) {
ftsRows.push(indexed);
}
const source = rows[entry.seq - 1];
if (!source || !shouldProjectActiveEvent(entry.event)) {
continue;
}
const projectsMessage = hasTranscriptMessage(entry.event);
activeRows.push({
activePosition: activeRows.length,
eventSeq: source.seq,
messagePosition: projectsMessage ? activeMessageCount : null,
});
if (projectsMessage) {
activeMessageCount += 1;
}
}
return {
activeEventCount: activeRows.length,
activeMessageCount,
activeRows,
ftsRows,
leafEventId: resolveVisibleTranscriptAppendParentId(events),
sessionId,
sourceIndexedSeq: rows.at(-1)?.seq ?? -1,
sourceTranscriptUpdatedAt: session.transcript_updated_at,
};
},
{
databaseLabel: "agent transcript projection",
operationLabel: "sessions.transcript-index.prepare",
},
);
}
function sourceSnapshotMatches(
db: DatabaseSync,
plan: PreparedSessionTranscriptProjectionMetadata,
): boolean {
const kysely = getProjectionKysely(db);
const session = executeSqliteQueryTakeFirstSync(
db,
kysely
.selectFrom("sessions")
.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),
);
return (
session?.transcript_updated_at === plan.sourceTranscriptUpdatedAt &&
latest?.seq === plan.sourceIndexedSeq
);
}
function projectionClaimIsOwned(db: DatabaseSync, sessionId: string, claimId: number): boolean {
const row = executeSqliteQueryTakeFirstSync(
db,
getProjectionKysely(db)
.selectFrom("session_transcript_index_state")
.select(["needs_rebuild", "updated_at"])
.where("session_id", "=", sessionId),
);
return row?.needs_rebuild !== 0 && row?.updated_at === claimId;
}
/** Claims a prepared snapshot. Later chunks publish only while this claim remains current. */
export function claimPreparedSessionTranscriptProjectionInTransaction(
db: DatabaseSync,
plan: PreparedSessionTranscriptProjectionMetadata,
claimId: number,
): boolean {
if (!sourceSnapshotMatches(db, plan)) {
return false;
}
const kysely = getProjectionKysely(db);
const current = executeSqliteQueryTakeFirstSync(
db,
kysely
.selectFrom("session_transcript_index_state")
.select(["indexed_seq", "needs_rebuild"])
.where("session_id", "=", plan.sessionId),
);
if (current?.needs_rebuild === 0 && current.indexed_seq === plan.sourceIndexedSeq) {
return false;
}
executeSqliteQuerySync(
db,
kysely
.insertInto("session_transcript_index_state")
.values({
active_event_count: 0,
active_message_count: 0,
indexed_seq: -1,
leaf_event_id: null,
needs_rebuild: 1,
session_id: plan.sessionId,
updated_at: claimId,
})
.onConflict((conflict) =>
conflict.column("session_id").doUpdateSet({
active_event_count: 0,
active_message_count: 0,
indexed_seq: -1,
leaf_event_id: null,
needs_rebuild: 1,
updated_at: claimId,
}),
),
);
return true;
}
/** 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 },
): ProjectionDeleteChunkResult {
if (!projectionClaimIsOwned(db, params.sessionId, params.claimId)) {
return { hasMore: false, owned: false };
}
// Hidden rowid batching is the narrow SQLite primitive that keeps each
// writer transaction bounded for both ordinary and FTS5 projection rows.
const kysely = getProjectionKysely(db);
const active = Number(
executeSqliteQuerySync(
db,
kysely
.deleteFrom("session_transcript_active_events")
.where(
"rowid",
"in",
kysely
.selectFrom("session_transcript_active_events")
.select("rowid")
.where("session_id", "=", params.sessionId)
.limit(params.maxRowsPerTable),
),
).numAffectedRows ?? 0n,
);
const fts = Number(
executeSqliteQuerySync(
db,
kysely
.deleteFrom("session_transcript_fts")
.where(
"rowid",
"in",
kysely
.selectFrom("session_transcript_fts")
.select("rowid")
.where("session_id", "=", params.sessionId)
.limit(params.maxRowsPerTable),
),
).numAffectedRows ?? 0n,
);
return {
hasMore: active === params.maxRowsPerTable || fts === params.maxRowsPerTable,
owned: true,
};
}
/** Appends one bounded projection chunk while its claim remains current. */
export function appendPreparedSessionTranscriptProjectionChunkInTransaction(
db: DatabaseSync,
params: {
activeRows?: PreparedSessionTranscriptProjection["activeRows"];
claimId: number;
ftsRows?: PreparedSessionTranscriptProjection["ftsRows"];
sessionId: string;
},
): boolean {
if (!projectionClaimIsOwned(db, params.sessionId, params.claimId)) {
return false;
}
const kysely = getProjectionKysely(db);
if (params.activeRows && params.activeRows.length > 0) {
executeSqliteQuerySync(
db,
kysely.insertInto("session_transcript_active_events").values(
params.activeRows.map((row) => ({
active_position: row.activePosition,
event_seq: row.eventSeq,
message_position: row.messagePosition,
session_id: params.sessionId,
})),
),
);
}
if (params.ftsRows && params.ftsRows.length > 0) {
executeSqliteQuerySync(
db,
kysely.insertInto("session_transcript_fts").values(
params.ftsRows.map((row) => ({
message_id: row.messageId,
role: row.role,
session_id: params.sessionId,
text: row.text,
timestamp: row.timestamp as unknown as string,
})),
),
);
}
return true;
}
/** Publishes counts and the append cursor only if the transcript snapshot stayed current. */
export function finalizePreparedSessionTranscriptProjectionInTransaction(
db: DatabaseSync,
plan: PreparedSessionTranscriptProjectionMetadata,
claimId: number,
): boolean {
if (!projectionClaimIsOwned(db, plan.sessionId, claimId) || !sourceSnapshotMatches(db, plan)) {
return false;
}
executeSqliteQuerySync(
db,
getProjectionKysely(db)
.updateTable("session_transcript_index_state")
.set({
active_event_count: plan.activeEventCount,
active_message_count: plan.activeMessageCount,
indexed_seq: plan.sourceIndexedSeq,
leaf_event_id: plan.leafEventId,
needs_rebuild: 0,
updated_at: Date.now(),
})
.where("session_id", "=", plan.sessionId)
.where("needs_rebuild", "!=", 0)
.where("updated_at", "=", claimId),
);
return true;
}
src/config/sessions/session-transcript-reconcile.ts
+389
View File
@@ -0,0 +1,389 @@
// Transcript projection reconciliation owner. Gateway startup awaits it;
// request paths may only schedule it and return a bounded retryable response.
import { randomInt } from "node:crypto";
import path from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
import { Worker } from "node:worker_threads";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import {
resolveOpenClawAgentSqlitePath,
runOpenClawAgentWriteTransaction,
type OpenClawAgentDatabase,
type OpenClawAgentDatabaseOptions,
} from "../../state/openclaw-agent-db.js";
import { runExclusiveSqliteSessionWrite } from "./session-accessor.sqlite-scope.js";
import { deleteOrphanedTranscriptIndexRowsInTransaction } from "./session-transcript-index.js";
import {
appendPreparedSessionTranscriptProjectionChunkInTransaction,
claimPreparedSessionTranscriptProjectionInTransaction,
deletePreparedSessionTranscriptProjectionChunkInTransaction,
finalizePreparedSessionTranscriptProjectionInTransaction,
type PreparedSessionTranscriptProjectionMetadata,
} from "./session-transcript-projection-rebuild.js";
import type {
EncodedTranscriptFtsChunk,
SessionTranscriptReconcileWorkerInput,
SessionTranscriptReconcileWorkerMessage,
} from "./session-transcript-reconcile.worker.js";
const log = createSubsystemLogger("sessions/transcript-index");
const PROJECTION_WRITE_CHUNK_ROWS = 512;
type RunningReconcile = {
pending: boolean;
preferredSessionId?: string;
promise?: Promise<SessionTranscriptReconcileResult>;
};
const runningReconciles = new Map<string, RunningReconcile>();
export type SessionTranscriptReconcileResult = {
reconciledSessions: number;
};
type SessionTranscriptReconcileParams = OpenClawAgentDatabaseOptions & {
preferredSessionId?: string;
};
type ActivePreparedProjection = {
claimId: number;
plan: PreparedSessionTranscriptProjectionMetadata;
};
function reconcileKey(params: OpenClawAgentDatabaseOptions): string {
return resolveOpenClawAgentSqlitePath(params);
}
function resolveSessionTranscriptReconcileWorkerUrl(currentModuleUrl = import.meta.url): URL {
const currentPath = fileURLToPath(currentModuleUrl);
const normalized = currentPath.replaceAll(path.sep, "/");
const distMarker = "/dist/";
const distIndex = normalized.lastIndexOf(distMarker);
if (distIndex >= 0) {
const distRoot = currentPath.slice(0, distIndex + distMarker.length);
return pathToFileURL(
path.join(distRoot, "config", "sessions", "session-transcript-reconcile.worker.js"),
);
}
const extension = path.extname(currentPath) || ".js";
return new URL(`./session-transcript-reconcile.worker${extension}`, currentModuleUrl);
}
function yieldToGateway(): Promise<void> {
return new Promise((resolve) => {
setImmediate(resolve);
});
}
function nextProjectionClaimId(): number {
return -randomInt(1, 2 ** 47);
}
function normalizeReconcileError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}
// Node Worker messages take a transfer list, unlike Window.postMessage.
// Keep the empty list explicit so the platform contract stays unambiguous.
function continueProjectionWorker(worker: Worker, accepted: boolean): void {
worker.postMessage({ accepted, type: "continue" }, []);
}
async function runProjectionWrite<T>(
databaseOptions: OpenClawAgentDatabaseOptions,
operationLabel: string,
operation: (database: OpenClawAgentDatabase) => T,
): Promise<T> {
return await runExclusiveSqliteSessionWrite(databaseOptions, async () =>
runOpenClawAgentWriteTransaction(operation, databaseOptions, { operationLabel }),
);
}
async function claimPreparedSessionTranscriptProjection(
databaseOptions: OpenClawAgentDatabaseOptions,
plan: PreparedSessionTranscriptProjectionMetadata,
): Promise<ActivePreparedProjection | undefined> {
const claimId = nextProjectionClaimId();
const claimed = await runProjectionWrite(
databaseOptions,
"sessions.transcript-index.claim",
(database) => claimPreparedSessionTranscriptProjectionInTransaction(database.db, plan, claimId),
);
if (!claimed) {
return undefined;
}
let deleteResult = { hasMore: true, owned: true };
while (deleteResult.hasMore && deleteResult.owned) {
deleteResult = await runProjectionWrite(
databaseOptions,
"sessions.transcript-index.delete-chunk",
(database) =>
deletePreparedSessionTranscriptProjectionChunkInTransaction(database.db, {
maxRowsPerTable: PROJECTION_WRITE_CHUNK_ROWS,
sessionId: plan.sessionId,
claimId,
}),
);
await yieldToGateway();
}
if (!deleteResult.owned) {
return undefined;
}
return { claimId, plan };
}
function decodeFtsChunk(chunk: EncodedTranscriptFtsChunk) {
const decoder = new TextDecoder();
return chunk.rows.map((row) => ({
messageId: row.messageId,
role: row.role,
text: decoder.decode(
chunk.textBytes.subarray(row.textByteOffset, row.textByteOffset + row.textByteLength),
),
timestamp: row.timestamp,
}));
}
async function appendPreparedProjectionChunk(
databaseOptions: OpenClawAgentDatabaseOptions,
active: ActivePreparedProjection,
rows:
| {
activeRows: Parameters<
typeof appendPreparedSessionTranscriptProjectionChunkInTransaction
>[1]["activeRows"];
}
| {
ftsRows: Parameters<
typeof appendPreparedSessionTranscriptProjectionChunkInTransaction
>[1]["ftsRows"];
},
): Promise<boolean> {
const owned = await runProjectionWrite(
databaseOptions,
"activeRows" in rows
? "sessions.transcript-index.active-chunk"
: "sessions.transcript-index.fts-chunk",
(database) =>
appendPreparedSessionTranscriptProjectionChunkInTransaction(database.db, {
...rows,
claimId: active.claimId,
sessionId: active.plan.sessionId,
}),
);
await yieldToGateway();
return owned;
}
async function finalizePreparedProjection(
databaseOptions: OpenClawAgentDatabaseOptions,
active: ActivePreparedProjection,
): Promise<boolean> {
return await runProjectionWrite(
databaseOptions,
"sessions.transcript-index.finalize",
(database) =>
finalizePreparedSessionTranscriptProjectionInTransaction(
database.db,
active.plan,
active.claimId,
),
);
}
/** Prepares full trees off-thread, then commits bounded chunks through the runtime writer owner. */
export function reconcileSessionTranscriptIndexes(
params: SessionTranscriptReconcileParams,
): Promise<SessionTranscriptReconcileResult> {
const databasePath = resolveOpenClawAgentSqlitePath(params);
const databaseOptions: OpenClawAgentDatabaseOptions = {
agentId: params.agentId,
...(params.env ? { env: params.env } : {}),
path: databasePath,
};
const workerUrl = resolveSessionTranscriptReconcileWorkerUrl();
const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined;
const input: SessionTranscriptReconcileWorkerInput = {
agentId: params.agentId,
path: databasePath,
...(params.preferredSessionId ? { preferredSessionId: params.preferredSessionId } : {}),
};
let worker: Worker;
try {
worker = new Worker(workerUrl, { workerData: input, execArgv: sourceWorkerExecArgv });
} catch (error) {
return Promise.reject(normalizeReconcileError(error));
}
return new Promise<SessionTranscriptReconcileResult>((resolve, reject) => {
let active: ActivePreparedProjection | undefined;
let doneReceived = false;
let reconciledSessions = 0;
let settled = false;
const settle = (finish: () => void, terminate: boolean) => {
if (settled) {
return;
}
settled = true;
worker.removeAllListeners();
if (terminate) {
void worker.terminate();
}
finish();
};
const handleMessage = async (message: SessionTranscriptReconcileWorkerMessage) => {
if (message.type === "failed") {
settle(() => reject(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,
);
return;
}
try {
await runProjectionWrite(
databaseOptions,
"sessions.transcript-index.orphan-sweep",
(database) => deleteOrphanedTranscriptIndexRowsInTransaction(database.db),
);
} catch (error) {
settle(() => reject(normalizeReconcileError(error)), true);
return;
}
settle(() => resolve({ reconciledSessions }), false);
return;
}
try {
if (message.type === "plan-start") {
if (active) {
throw new Error("session transcript reconcile worker started overlapping plans");
}
active = await claimPreparedSessionTranscriptProjection(databaseOptions, message.plan);
continueProjectionWorker(worker, active !== undefined);
return;
}
if (!active || active.plan.sessionId !== message.sessionId) {
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);
active = undefined;
if (finalized) {
reconciledSessions += 1;
}
continueProjectionWorker(worker, finalized);
return;
}
const owned = await appendPreparedProjectionChunk(
databaseOptions,
active,
message.type === "active-chunk"
? { activeRows: message.rows }
: { ftsRows: decodeFtsChunk(message.chunk) },
);
if (!owned) {
active = undefined;
}
continueProjectionWorker(worker, owned);
} catch (error) {
settle(() => reject(normalizeReconcileError(error)), true);
}
};
worker.on("message", (message: SessionTranscriptReconcileWorkerMessage) => {
void handleMessage(message);
});
worker.once("error", (error) => {
settle(() => reject(normalizeReconcileError(error)), true);
});
worker.once("exit", (code) => {
if (doneReceived && code === 0) {
return;
}
settle(
() => reject(new Error(`session transcript reconcile worker exited with code ${code}`)),
false,
);
});
});
}
/** Starts one deferred reconcile. No transcript rows are read on the caller's stack. */
export function startSessionTranscriptIndexReconcile(
params: SessionTranscriptReconcileParams,
): void {
const key = reconcileKey(params);
const running = runningReconciles.get(key);
if (running) {
// The active pass snapshots dirty sessions. Latch later writes so it
// rescans before ownership is released instead of losing their work.
running.pending = true;
running.preferredSessionId ??= params.preferredSessionId;
return;
}
const state: RunningReconcile = {
pending: false,
...(params.preferredSessionId ? { preferredSessionId: params.preferredSessionId } : {}),
};
const pending = yieldToGateway()
.then(async () => {
let reconciledSessions = 0;
while (true) {
state.pending = false;
const preferredSessionId = state.preferredSessionId;
delete state.preferredSessionId;
const result = await reconcileSessionTranscriptIndexes({
...params,
...(preferredSessionId ? { preferredSessionId } : {}),
});
reconciledSessions += result.reconciledSessions;
if (state.pending) {
continue;
}
// Check and relinquish ownership without an async boundary. A later
// request either latches above or creates a fresh owner below.
if (runningReconciles.get(key) === state) {
runningReconciles.delete(key);
}
return { reconciledSessions };
}
})
.catch(async (error: unknown) => {
log.warn(
`session transcript reconcile failed agent=${params.agentId} error=${error instanceof Error ? error.message : String(error)}`,
);
const shouldHandoff = state.pending;
const preferredSessionId = state.preferredSessionId;
if (runningReconciles.get(key) === state) {
runningReconciles.delete(key);
}
if (shouldHandoff) {
startSessionTranscriptIndexReconcile({
...params,
...(preferredSessionId ? { preferredSessionId } : {}),
});
await waitForSessionTranscriptIndexReconcile(params);
}
return { reconciledSessions: 0 };
});
state.promise = pending;
runningReconciles.set(key, state);
}
export function isSessionTranscriptIndexReconcileRunning(
params: OpenClawAgentDatabaseOptions,
): boolean {
return runningReconciles.has(reconcileKey(params));
}
/** Test and maintenance wait hook for an already-scheduled reconcile. */
export async function waitForSessionTranscriptIndexReconcile(
params: OpenClawAgentDatabaseOptions,
): Promise<void> {
await runningReconciles.get(reconcileKey(params))?.promise;
}
src/config/sessions/session-transcript-reconcile.worker.ts
+199
View File
@@ -0,0 +1,199 @@
/** Worker entrypoint for transcript parsing and active-branch resolution only. */
import { parentPort, workerData } from "node:worker_threads";
import {
closeOpenClawAgentDatabaseByPath,
openOpenClawAgentDatabase,
} from "../../state/openclaw-agent-db.js";
import { listSessionsNeedingTranscriptIndexReconcile } from "./session-transcript-index.js";
import {
prepareSessionTranscriptProjection,
type PreparedSessionTranscriptProjection,
type PreparedSessionTranscriptProjectionMetadata,
type TranscriptIndexEntry,
} from "./session-transcript-projection-rebuild.js";
const ACTIVE_ROWS_PER_CHUNK = 512;
const FTS_ROWS_PER_CHUNK = 128;
const FTS_TEXT_BYTES_PER_CHUNK = 256 * 1024;
export type SessionTranscriptReconcileWorkerInput = {
agentId: string;
path: string;
preferredSessionId?: string;
};
export type EncodedTranscriptFtsChunk = {
rows: Array<{
messageId: string;
role: "assistant" | "user";
textByteLength: number;
textByteOffset: number;
timestamp: number;
}>;
textBytes: Uint8Array<ArrayBuffer>;
};
export type SessionTranscriptReconcileWorkerMessage =
| {
type: "active-chunk";
rows: PreparedSessionTranscriptProjection["activeRows"];
sessionId: string;
}
| { type: "done" }
| { type: "failed"; error: string }
| { type: "fts-chunk"; chunk: EncodedTranscriptFtsChunk; sessionId: string }
| { type: "plan-finish"; sessionId: string }
| { type: "plan-start"; plan: PreparedSessionTranscriptProjectionMetadata };
type SessionTranscriptReconcileWorkerCommand = { accepted: boolean; type: "continue" };
function parseWorkerInput(value: unknown): SessionTranscriptReconcileWorkerInput | undefined {
if (!value || typeof value !== "object" || Array.isArray(value)) {
return undefined;
}
const input = value as Record<string, unknown>;
if (typeof input.agentId !== "string" || typeof input.path !== "string") {
return undefined;
}
if (input.preferredSessionId !== undefined && typeof input.preferredSessionId !== "string") {
return undefined;
}
return {
agentId: input.agentId,
path: input.path,
...(typeof input.preferredSessionId === "string"
? { preferredSessionId: input.preferredSessionId }
: {}),
};
}
function orderSessionIds(sessionIds: string[], preferredSessionId: string | undefined): string[] {
if (!preferredSessionId || !sessionIds.includes(preferredSessionId)) {
return sessionIds;
}
return [
preferredSessionId,
...sessionIds.filter((sessionId) => sessionId !== preferredSessionId),
];
}
const input = parseWorkerInput(workerData);
if (!parentPort || !input) {
throw new Error("session transcript reconcile worker requires valid worker data");
}
const port = parentPort;
const reconcileInput: SessionTranscriptReconcileWorkerInput = input;
function waitForContinue(): Promise<boolean> {
return new Promise((resolve, reject) => {
port.once("message", (message: SessionTranscriptReconcileWorkerCommand) => {
if (message?.type !== "continue" || typeof message.accepted !== "boolean") {
reject(new Error("session transcript reconcile worker received an invalid command"));
return;
}
resolve(message.accepted);
});
});
}
async function postAndWait(
message: SessionTranscriptReconcileWorkerMessage,
transferList: ArrayBuffer[] = [],
): Promise<boolean> {
port.postMessage(message, transferList);
return await waitForContinue();
}
function encodeFtsChunk(rows: readonly TranscriptIndexEntry[]): EncodedTranscriptFtsChunk {
const encoder = new TextEncoder();
const encoded = rows.map((row) => ({ bytes: encoder.encode(row.text), row }));
const textBytes = new Uint8Array(encoded.reduce((total, entry) => total + entry.bytes.length, 0));
let textByteOffset = 0;
const metadata = encoded.map(({ bytes, row }) => {
textBytes.set(bytes, textByteOffset);
const result = {
messageId: row.messageId,
role: row.role,
textByteLength: bytes.length,
textByteOffset,
timestamp: row.timestamp,
};
textByteOffset += bytes.length;
return result;
});
return { rows: metadata, textBytes };
}
function takeFtsChunkEnd(rows: readonly TranscriptIndexEntry[], start: number): number {
let bytes = 0;
let end = start;
while (end < rows.length && end - start < FTS_ROWS_PER_CHUNK) {
const rowBytes = Buffer.byteLength(rows[end]?.text ?? "", "utf8");
if (end > start && bytes + rowBytes > FTS_TEXT_BYTES_PER_CHUNK) {
break;
}
bytes += rowBytes;
end += 1;
}
return end;
}
async function streamPreparedProjection(plan: PreparedSessionTranscriptProjection): Promise<void> {
const { activeRows, ftsRows, ...metadata } = plan;
if (!(await postAndWait({ type: "plan-start", plan: metadata }))) {
return;
}
for (let offset = 0; offset < activeRows.length; offset += ACTIVE_ROWS_PER_CHUNK) {
if (
!(await postAndWait({
type: "active-chunk",
rows: activeRows.slice(offset, offset + ACTIVE_ROWS_PER_CHUNK),
sessionId: plan.sessionId,
}))
) {
return;
}
}
for (let offset = 0; offset < ftsRows.length;) {
const end = takeFtsChunkEnd(ftsRows, offset);
const chunk = encodeFtsChunk(ftsRows.slice(offset, end));
const accepted = await postAndWait({ type: "fts-chunk", chunk, sessionId: plan.sessionId }, [
chunk.textBytes.buffer,
]);
if (!accepted) {
return;
}
offset = end;
}
await postAndWait({ type: "plan-finish", sessionId: plan.sessionId });
}
async function run(): Promise<void> {
try {
const database = openOpenClawAgentDatabase({
agentId: reconcileInput.agentId,
path: reconcileInput.path,
});
const sessionIds = orderSessionIds(
listSessionsNeedingTranscriptIndexReconcile(database.db),
reconcileInput.preferredSessionId,
);
for (const sessionId of sessionIds) {
const plan = prepareSessionTranscriptProjection(database.db, sessionId);
if (plan) {
await streamPreparedProjection(plan);
}
}
port.postMessage({ type: "done" } satisfies SessionTranscriptReconcileWorkerMessage);
} catch (error) {
port.postMessage({
type: "failed",
error: error instanceof Error ? error.message : String(error),
} satisfies SessionTranscriptReconcileWorkerMessage);
} finally {
closeOpenClawAgentDatabaseByPath(reconcileInput.path);
port.close();
}
}
void run();
src/config/sessions/session-transcript-search.test.ts
+28 -1
View File
@@ -175,7 +175,7 @@ describe("searchSessionTranscripts", () => {
expect(result.hits[0]?.messageId).toBe("m-new");
});
it("only surfaces the active branch after a leaf-control rewind", async () => {
it("only surfaces the active branch after a deferred leaf-control rebuild", async () => {
const scope = transcriptScope("session-1", "agent:main:main");
await replaceSqliteTranscriptEvents(scope, [
{
@@ -207,6 +207,33 @@ describe("searchSessionTranscripts", () => {
expect(search("alpha").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);
await replaceSqliteTranscriptEvents(
scope,
["alpha-stream", "beta-stream", "gamma-stream"].map((marker, index) => ({
type: "message",
id: `m${index + 1}`,
parentId: index === 0 ? null : `m${index}`,
message: { role: "user", content: [{ type: "text", text: `${marker} ${largeText}` }] },
})) as unknown as TranscriptEvent[],
);
await appendSqliteTranscriptEvent(scope, {
type: "leaf",
id: "leaf-large",
parentId: "m3",
targetId: "m3",
} as unknown as TranscriptEvent);
expect(search("gamma-stream").indexing).toBe(true);
await waitForSearchReconcile("gamma-stream");
expect(search("alpha-stream").hits).toHaveLength(1);
expect(search("beta-stream").hits).toHaveLength(1);
expect(search("gamma-stream").hits).toHaveLength(1);
});
it("backfills transcripts that predate the index via reconcile", async () => {
await appendUserMessage("session-1", "agent:main:main", "historic knowledge");
const { db, kysely } = agentKysely();
src/config/sessions/session-transcript-search.ts
+13 -90
View File
@@ -1,22 +1,15 @@
// Full-text search over per-agent transcript rows. Appends index themselves
// inside the accessor's write transactions (session-transcript-index.ts);
// this module owns the query path and the lazy reconcile that backfills
// doctor-migrated transcripts and rebuilds branch-rewound sessions.
import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js";
import {
openOpenClawAgentDatabase,
runOpenClawAgentWriteTransaction,
} from "../../state/openclaw-agent-db.js";
// 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 { truncateUtf16Safe } from "../../utils.js";
import { listSessionsNeedingTranscriptIndexReconcile } from "./session-transcript-index.js";
import {
deleteOrphanedTranscriptIndexRowsInTransaction,
listSessionsNeedingTranscriptIndexReconcile,
rebuildSessionTranscriptIndexInTransaction,
} from "./session-transcript-index.js";
isSessionTranscriptIndexReconcileRunning,
startSessionTranscriptIndexReconcile,
} from "./session-transcript-reconcile.js";
const log = createSubsystemLogger("sessions/search-index");
const SEARCH_SNIPPET_MAX_CHARS = 500;
const SEARCH_LIMIT_MAX = 25;
const SEARCH_QUERY_MAX_CHARS = 4096;
@@ -37,78 +30,6 @@ type SessionTranscriptSearchResult = {
truncated: boolean;
};
const runningReconciles = new Map<string, Promise<void>>();
/**
* Rebuilds every session whose index state lags its transcript rows, then
* sweeps orphaned index rows. One write transaction per session keeps the
* agent DB responsive to live appends between rebuilds.
*/
async function reconcileSessionTranscriptIndex(params: {
agentId: string;
env?: NodeJS.ProcessEnv;
}): Promise<void> {
const database = openOpenClawAgentDatabase({
agentId: params.agentId,
...(params.env ? { env: params.env } : {}),
});
const sessionIds = listSessionsNeedingTranscriptIndexReconcile(database.db);
for (const sessionId of sessionIds) {
runOpenClawAgentWriteTransaction(
(agentDatabase) => {
// Rows are reread inside the transaction: a live append that landed
// after the dirty scan is either included here or re-flagged by its
// own in-transaction hook, so the rebuild can never go stale.
const rows = executeSqliteQuerySync(
agentDatabase.db,
getNodeSqliteKysely<Pick<OpenClawAgentKyselyDatabase, "transcript_events">>(
agentDatabase.db,
)
.selectFrom("transcript_events")
.select(["seq", "event_json"])
.where("session_id", "=", sessionId)
.orderBy("seq", "asc"),
).rows;
if (rows.length === 0) {
return;
}
const events = rows.map((row) => JSON.parse(row.event_json) as unknown);
const maxSeq = rows[rows.length - 1]?.seq ?? -1;
rebuildSessionTranscriptIndexInTransaction(agentDatabase.db, sessionId, events, maxSeq);
},
{ agentId: params.agentId, ...(params.env ? { env: params.env } : {}) },
{ operationLabel: "sessions.search.reconcile" },
);
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
}
runOpenClawAgentWriteTransaction(
(agentDatabase) => {
deleteOrphanedTranscriptIndexRowsInTransaction(agentDatabase.db);
},
{ agentId: params.agentId, ...(params.env ? { env: params.env } : {}) },
{ operationLabel: "sessions.search.orphan-sweep" },
);
}
function startReconcile(params: { agentId: string; env?: NodeJS.ProcessEnv }): void {
if (runningReconciles.has(params.agentId)) {
return;
}
const pending = reconcileSessionTranscriptIndex(params)
.catch((error: unknown) => {
// The next search re-detects dirty sessions and retries.
log.warn(
`session transcript reconcile failed agent=${params.agentId} error=${error instanceof Error ? error.message : String(error)}`,
);
})
.finally(() => {
runningReconciles.delete(params.agentId);
});
runningReconciles.set(params.agentId, pending);
}
function toFtsQuery(query: string): string {
return query
.trim()
@@ -132,15 +53,17 @@ export function searchSessionTranscripts(params: {
if (query.length > SEARCH_QUERY_MAX_CHARS) {
throw new Error(`query must not exceed ${SEARCH_QUERY_MAX_CHARS} characters`);
}
const database = openOpenClawAgentDatabase({
const databaseOptions = {
agentId: params.agentId,
...(params.env ? { env: params.env } : {}),
});
};
const database = openOpenClawAgentDatabase(databaseOptions);
const dirtySessions = listSessionsNeedingTranscriptIndexReconcile(database.db);
if (dirtySessions.length > 0) {
startReconcile(params);
startSessionTranscriptIndexReconcile(params);
}
const indexing = dirtySessions.length > 0 || runningReconciles.has(params.agentId);
const indexing =
dirtySessions.length > 0 || isSessionTranscriptIndexReconcileRunning(databaseOptions);
const limit = Math.min(Math.max(1, params.limit ?? 10), SEARCH_LIMIT_MAX);
const sessionKeys = params.sessionKeys ?? [];
const whereSession =
src/config/sessions/transcript-visible-events.ts
+1 -1
View File
@@ -3,7 +3,7 @@ import {
selectSessionTranscriptTreePathNodes,
} from "./transcript-tree.js";
export type VisibleTranscriptEventEntry<T> = {
type VisibleTranscriptEventEntry<T> = {
event: T;
/** Parent id after active-branch normalization; null when no visible parent exists. */
parentId: string | null;
src/gateway/server-methods/chat.ts
+34 -14
View File
@@ -27,7 +27,10 @@ import { resolveProviderIdForAuth } from "../../agents/provider-auth-aliases.js"
import { createAgentRunRestartAbortError } from "../../agents/run-termination.js";
import { dispatchInboundMessage } from "../../auto-reply/dispatch.js";
import { resolveSessionWorkStartError } from "../../config/sessions.js";
import { resolveTranscriptSessionKeyBySessionId } from "../../config/sessions/session-accessor.js";
import {
isSessionTranscriptProjectionUnavailableError,
resolveTranscriptSessionKeyBySessionId,
} from "../../config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
clearAgentRunContext,
@@ -567,19 +570,36 @@ async function handleChatHistoryRequest({
const max = Math.min(1000, requested);
const maxHistoryBytes = getMaxChatHistoryMessagesBytes();
const effectiveMaxChars = resolveEffectiveChatHistoryMaxChars(cfg, maxChars);
const historyPage = await readChatHistoryPage({
entry: historyEntry,
provider: resolvedSessionModel.provider,
sessionId,
storePath,
sessionAgentId,
canonicalKey,
max,
maxHistoryBytes,
effectiveMaxChars,
offset,
messageId,
});
let historyPage: Awaited<ReturnType<typeof readChatHistoryPage>>;
try {
historyPage = await readChatHistoryPage({
entry: historyEntry,
provider: resolvedSessionModel.provider,
sessionId,
storePath,
sessionAgentId,
canonicalKey,
max,
maxHistoryBytes,
effectiveMaxChars,
offset,
messageId,
});
} catch (error) {
if (!isSessionTranscriptProjectionUnavailableError(error)) {
throw error;
}
respond(
false,
undefined,
errorShape(ErrorCodes.UNAVAILABLE, "session history is rebuilding; retry shortly", {
details: { method },
retryable: true,
retryAfterMs: 250,
}),
);
return;
}
const normalized = enrichChatHistoryCompactionMarkers(historyPage.messages, historyEntry);
const perMessageHardCap = Math.min(CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES, maxHistoryBytes);
const replaced = replaceOversizedChatHistoryMessages({
src/gateway/server-startup-session-migration.test.ts
+32
View File
@@ -11,6 +11,9 @@ type ResolveStoreTargets = NonNullable<
>;
type SweepStoreTemps = NonNullable<StartupMigrationDeps["sweepOrphanSessionStoreTemps"]>;
type RunDoctorSessionSqlite = NonNullable<StartupMigrationDeps["runDoctorSessionSqlite"]>;
type ReconcileSessionTranscriptIndexes = NonNullable<
StartupMigrationDeps["reconcileSessionTranscriptIndexes"]
>;
function makeLog() {
return {
@@ -31,6 +34,9 @@ function makeDeps(
migrate: MigrateSessionKeys,
removedFiles = 0,
runDoctorSessionSqlite: RunDoctorSessionSqlite = makeSessionSqliteImport(),
reconcileSessionTranscriptIndexes: ReconcileSessionTranscriptIndexes = vi
.fn<ReconcileSessionTranscriptIndexes>()
.mockResolvedValue({ reconciledSessions: 0 }),
) {
return {
migrateOrphanedSessionKeys: migrate,
@@ -43,6 +49,7 @@ function makeDeps(
.mockResolvedValueOnce(removedFiles)
.mockResolvedValue(0),
runDoctorSessionSqlite,
reconcileSessionTranscriptIndexes,
};
}
@@ -199,6 +206,31 @@ describe("runStartupSessionMigration", () => {
expect(log.warn).not.toHaveBeenCalled();
});
it("reconciles configured agent transcript projections before startup completes", async () => {
const log = makeLog();
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
const reconcile = vi
.fn<ReconcileSessionTranscriptIndexes>()
.mockResolvedValueOnce({ reconciledSessions: 2 })
.mockResolvedValueOnce({ reconciledSessions: 1 });
const cfg = {
agents: { defaults: {}, list: [{ id: "main" }, { id: "ops" }] },
session: {},
} as Parameters<typeof runStartupSessionMigration>[0]["cfg"];
await runStartupSessionMigration({
cfg,
log,
deps: makeDeps(migrate, 0, makeSessionSqliteImport(), reconcile),
});
expect(reconcile).toHaveBeenNthCalledWith(1, { agentId: "main" });
expect(reconcile).toHaveBeenNthCalledWith(2, { agentId: "ops" });
expect(log.info).toHaveBeenCalledWith(
"session: rebuilt 3 transcript projection(s) before serving history",
);
});
it("warns without blocking when hot legacy session SQLite import reports legacy file issues", async () => {
const log = makeLog();
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
src/gateway/server-startup-session-migration.ts
+28
View File
@@ -1,3 +1,4 @@
import { listAgentIds } from "../agents/agent-scope.js";
import type {
DoctorSessionSqliteIssue,
DoctorSessionSqliteReport,
@@ -27,6 +28,7 @@ type SessionSqliteStartupFailureReportWriter = (
) => { jsonPath: string; markdownPath: string };
type SessionMigrationDeps = Parameters<typeof runSessionStartupMigration>[0]["deps"] & {
reconcileSessionTranscriptIndexes?: typeof import("../config/sessions/session-transcript-reconcile.js").reconcileSessionTranscriptIndexes;
restoreSessionSqliteMigrationRun?: SessionSqliteStartupRestoreRunner;
runDoctorSessionSqlite?: SessionSqliteStartupImportRunner;
writeSessionSqliteMigrationFailureReports?: SessionSqliteStartupFailureReportWriter;
@@ -54,6 +56,32 @@ export async function runStartupSessionMigration(params: {
}): Promise<void> {
await runSessionStartupMigration(params);
await runStartupSessionSqliteImport(params);
await reconcileStartupSessionTranscriptIndexes(params);
}
async function reconcileStartupSessionTranscriptIndexes(params: {
cfg: OpenClawConfig;
env?: NodeJS.ProcessEnv;
log: SessionStartupMigrationLogger;
deps?: SessionMigrationDeps;
}): Promise<void> {
const reconcile =
params.deps?.reconcileSessionTranscriptIndexes ??
(await import("../config/sessions/session-transcript-reconcile.js"))
.reconcileSessionTranscriptIndexes;
let reconciledSessions = 0;
for (const agentId of listAgentIds(params.cfg)) {
const result = await reconcile({
agentId,
...(params.env ? { env: params.env } : {}),
});
reconciledSessions += result.reconciledSessions;
}
if (reconciledSessions > 0) {
params.log.info(
`session: rebuilt ${reconciledSessions} transcript projection(s) before serving history`,
);
}
}
async function runStartupSessionSqliteImport(params: {
src/gateway/server.chat.gateway-server-chat-b.test.ts
+103 -1
View File
@@ -21,11 +21,13 @@ import {
replaceSessionEntry,
withTranscriptWriteLock,
} from "../config/sessions/session-accessor.js";
import { waitForSessionTranscriptIndexReconcile } from "../config/sessions/session-transcript-reconcile.js";
import { invalidateSessionStoreCache } from "../config/sessions/store-cache.js";
import type { AgentModelConfig } from "../config/types.agents-shared.js";
import { rotateAgentEventLifecycleGeneration } from "../infra/agent-events.js";
import { onDiagnosticEvent, type DiagnosticPayloadLargeEvent } from "../infra/diagnostic-events.js";
import { runExclusiveSessionLifecycleMutation } from "../sessions/session-lifecycle-admission.js";
import { openOpenClawAgentDatabase } from "../state/openclaw-agent-db.js";
import { createDeferred } from "../test-utils/deferred.js";
import { captureEnv, setTestEnvValue } from "../test-utils/env.js";
import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js";
@@ -5911,6 +5913,10 @@ describe("gateway server chat", () => {
targetId: "msg-active",
}),
]);
await waitForSessionTranscriptIndexReconcile({
agentId: "main",
path: path.join(sessionDir, "openclaw-agent.sqlite"),
});
const stale = await fetchChatMessage(ws, {
sessionKey: "main",
@@ -6045,6 +6051,35 @@ describe("gateway server chat", () => {
});
});
test("chat.history returns retryable unavailable while a dirty projection rebuilds", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
const sessionDir = await prepareMainHistoryHarness({ ws, createSessionDir });
await writeMainSessionTranscript(sessionDir, [
JSON.stringify({ message: { role: "user", content: "ready after rebuild" } }),
]);
const databaseOptions = {
agentId: "main",
path: path.join(sessionDir, "openclaw-agent.sqlite"),
};
const database = openOpenClawAgentDatabase(databaseOptions);
database.db
.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1 WHERE session_id = ?")
.run("sess-main");
const rebuilding = await rpcReq(ws, "chat.history", { sessionKey: "main", limit: 1 });
expect(rebuilding.ok).toBe(false);
expect(rebuilding.error).toMatchObject({ code: "UNAVAILABLE", retryable: true });
await waitForSessionTranscriptIndexReconcile(databaseOptions);
const ready = await rpcReq<{ messages?: unknown[] }>(ws, "chat.history", {
sessionKey: "main",
limit: 1,
});
expect(ready.ok).toBe(true);
expect(JSON.stringify(ready.payload?.messages)).toContain("ready after rebuild");
});
});
test("chat.history offset pagination advances from the projected first-page boundary", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
const sessionDir = await prepareMainHistoryHarness({ ws, createSessionDir });
@@ -6174,6 +6209,69 @@ describe("gateway server chat", () => {
});
});
test("chat.history pagination ignores non-message event sequence gaps", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
await prepareMainHistoryHarness({ ws, createSessionDir });
const storePath = testState.sessionStorePath;
if (!storePath) {
throw new Error("session store path was not initialized");
}
const scope = {
agentId: "main",
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath,
};
let parentId: string | null = null;
for (let index = 1; index <= 5; index += 1) {
const messageId = `message-${index}`;
await appendTranscriptMessage(scope, {
eventId: messageId,
parentId,
message: {
role: index % 2 === 0 ? "assistant" : "user",
content: [{ type: "text", text: `message ${index}` }],
timestamp: Date.now() + index,
},
});
parentId = messageId;
if (index < 5) {
const controlId = `control-${index}`;
await appendTranscriptEvent(scope, { type: "custom", id: controlId, parentId });
parentId = controlId;
}
}
type HistoryPage = {
messages?: Array<{ __openclaw?: { seq?: number } }>;
nextOffset?: number;
hasMore?: boolean;
totalMessages?: number;
};
const pages: HistoryPage[] = [];
let offset: number | undefined;
do {
const page = await rpcReq<HistoryPage>(ws, "chat.history", {
sessionKey: "main",
limit: 2,
...(offset !== undefined ? { offset } : {}),
});
expect(page.ok).toBe(true);
pages.push(page.payload ?? {});
offset = page.payload?.nextOffset;
} while (pages.at(-1)?.hasMore);
expect(pages.map((page) => page.messages?.map(readOpenClawSeq))).toEqual([
[4, 5],
[2, 3],
[1],
]);
expect(pages.map((page) => page.nextOffset)).toEqual([2, 4, undefined]);
expect(pages.map((page) => page.hasMore)).toEqual([true, true, false]);
expect(pages.map((page) => page.totalMessages)).toEqual([5, 5, 5]);
});
});
test("chat.history centers a bounded page around a message id", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
const sessionDir = await prepareMainHistoryHarness({ ws, createSessionDir });
@@ -6203,6 +6301,10 @@ describe("gateway server chat", () => {
},
);
}
await waitForSessionTranscriptIndexReconcile({
agentId: "main",
path: path.join(sessionDir, "openclaw-agent.sqlite"),
});
const history = await rpcReq<{
messages?: Array<{ __openclaw?: { seq?: number } }>;
@@ -6219,7 +6321,7 @@ describe("gateway server chat", () => {
});
expect(history.ok).toBe(true);
expect(history.payload?.messages?.map(readOpenClawSeq)).toEqual([4, 5, 6]);
expect(history.payload?.messages?.map(readOpenClawSeq)).toEqual([2, 3, 4]);
expect(history.payload?.offset).toBeUndefined();
expect(history.payload?.nextOffset).toBeUndefined();
expect(history.payload?.hasMore).toBeUndefined();
src/gateway/session-transcript-anchor-reader.ts
+17 -16
View File
@@ -1,15 +1,15 @@
import type { SessionTranscriptReadScope } from "../config/sessions/session-accessor.js";
import {
readSessionTranscriptMessageAnchorPage,
type SessionTranscriptReadScope,
} from "../config/sessions/session-accessor.js";
import {
isSqliteReadTarget,
readSqliteMessageRecords,
resolveTranscriptReadTarget,
sqliteRecordMessageWithSeq,
sqliteMessageEventWithSeq,
toTranscriptReadScope,
type ReadRecentSessionMessagesResult,
} from "./session-transcript-readers.js";
import {
readSessionMessagesAroundIdWithStatsAsync as readSessionMessagesAroundIdWithStatsAsyncFile,
resolveSessionMessageAnchorBounds,
} from "./session-utils.fs-anchor.js";
import { readSessionMessagesAroundIdWithStatsAsync as readSessionMessagesAroundIdWithStatsAsyncFile } from "./session-utils.fs-anchor.js";
type ReadSessionMessagesAroundIdResult = ReadRecentSessionMessagesResult & {
found: boolean;
@@ -30,9 +30,8 @@ export async function readSessionMessagesAroundIdWithStatsAsync(
? undefined
: target.sessionFile;
if (isSqliteReadTarget(target)) {
const records = await readSqliteMessageRecords(target);
const bounds = resolveSessionMessageAnchorBounds(records, opts.messageId, opts.maxMessages);
if (!bounds) {
const page = readSessionTranscriptMessageAnchorPage(toTranscriptReadScope(target), opts);
if (!page.found) {
if (opts.allowResetArchiveFallback === true) {
return await readSessionMessagesAroundIdWithStatsAsyncFile(
target.sessionId,
@@ -47,17 +46,19 @@ export async function readSessionMessagesAroundIdWithStatsAsync(
hasOverreadContext: false,
messages: [],
offset: 0,
totalMessages: records.length,
totalMessages: page.totalMessages,
transcriptPath: target.sessionFile,
};
}
const readStart = Math.max(0, bounds.start - 1);
return {
found: true,
hasOverreadContext: readStart < bounds.start,
messages: records.slice(readStart, bounds.endExclusive).map(sqliteRecordMessageWithSeq),
offset: bounds.offset,
totalMessages: records.length,
hasOverreadContext: page.hasOverreadContext,
messages: page.events.flatMap((entry) => {
const message = sqliteMessageEventWithSeq(entry);
return message === undefined ? [] : [message];
}),
offset: page.offset,
totalMessages: page.totalMessages,
transcriptPath: target.sessionFile,
};
}
src/gateway/session-transcript-readers.test.ts
+10 -5
View File
@@ -6,6 +6,7 @@ import {
persistSessionTranscriptTurn,
upsertSessionEntry,
} from "../config/sessions/session-accessor.js";
import { waitForSessionTranscriptIndexReconcile } from "../config/sessions/session-transcript-reconcile.js";
import { formatSqliteSessionFileMarker } from "../config/sessions/sqlite-marker.js";
import { captureEnv, setTestEnvValue } from "../test-utils/env.js";
import { readSessionMessagesAroundIdWithStatsAsync } from "./session-transcript-anchor-reader.js";
@@ -273,7 +274,7 @@ describe("session transcript reader facade", () => {
]);
await expect(
readSessionMessagesAsync(scope, { mode: "recent", maxMessages: 1 }),
).resolves.toMatchObject([{ content: "sqlite follow-up", __openclaw: { seq: 4 } }]);
).resolves.toMatchObject([{ content: "sqlite follow-up", __openclaw: { seq: 3 } }]);
await expect(readSessionMessageCountAsync(scope)).resolves.toBe(3);
});
@@ -315,7 +316,7 @@ describe("session transcript reader facade", () => {
).resolves.toMatchObject([{ content: "marker scoped prompt" }]);
await expect(
readSessionMessageByIdAsync({ sessionFile: marker, sessionId }, "marker-message"),
).resolves.toMatchObject({ found: true, seq: 2 });
).resolves.toMatchObject({ found: true, seq: 1 });
});
test("projects SQLite transcript reads to the active branch", async () => {
@@ -346,6 +347,10 @@ describe("session transcript reader facade", () => {
],
touchSessionEntry: false,
});
await waitForSessionTranscriptIndexReconcile({
agentId: "main",
path: path.join(tempDir, "openclaw-agent.sqlite"),
});
const messages = await readSessionMessagesAsync(scope, {
mode: "full",
@@ -358,7 +363,7 @@ describe("session transcript reader facade", () => {
).toEqual(["root", "active"]);
expect(
messages.map((message) => (message as { __openclaw?: { seq?: number } })["__openclaw"]?.seq),
).toEqual([2, 4]);
).toEqual([1, 2]);
await expect(readSessionMessageCountAsync(scope)).resolves.toBe(2);
});
@@ -394,7 +399,7 @@ describe("session transcript reader facade", () => {
page.messages.map(
(message) => (message as { __openclaw?: { seq?: number } })["__openclaw"]?.seq,
),
).toEqual([3, 4]);
).toEqual([2, 3]);
});
test("honors agent ids when no store path or session file is provided", async () => {
@@ -416,7 +421,7 @@ describe("session transcript reader facade", () => {
await expect(readSessionMessageCountAsync(scope)).resolves.toBe(1);
await expect(readSessionMessageByIdAsync(scope, "agent-message")).resolves.toMatchObject({
found: true,
seq: 2,
seq: 1,
});
await expect(
readSessionMessagesAsync(scope, { mode: "full", reason: "facade agent scope test" }),
src/gateway/session-transcript-readers.ts
+57 -95
View File
@@ -4,17 +4,17 @@ import {
normalizeUsage,
type UsageLike,
} from "../agents/usage.js";
import type { SessionTranscriptReadScope } from "../config/sessions/session-accessor.js";
import {
loadTranscriptEvents,
loadTranscriptEventsSync,
readRecentSessionTranscriptMessageEvents,
readSessionTranscriptMessageEventById,
readSessionTranscriptMessageEventCount,
readSessionTranscriptMessageEventPage,
readSessionTranscriptMessageEvents,
resolveSessionTranscriptReadTarget,
type SessionTranscriptMessageEvent,
type SessionTranscriptReadScope,
} from "../config/sessions/session-accessor.js";
import { parseSqliteSessionFileMarker } from "../config/sessions/sqlite-marker.js";
import {
selectVisibleTranscriptEventEntries,
type VisibleTranscriptEventEntry,
} from "../config/sessions/transcript-visible-events.js";
import { hasInterSessionUserProvenance } from "../sessions/input-provenance.js";
import type {
ReadRecentSessionMessagesOptions,
@@ -103,7 +103,9 @@ export function isSqliteReadTarget(target: ResolvedTranscriptReadTarget): boolea
return parseSqliteSessionFileMarker(target.sessionFile) !== undefined;
}
function toTranscriptReadScope(target: ResolvedTranscriptReadTarget): SessionTranscriptReadScope {
export function toTranscriptReadScope(
target: ResolvedTranscriptReadTarget,
): SessionTranscriptReadScope {
return {
...(target.agentId ? { agentId: target.agentId } : {}),
sessionId: target.sessionId,
@@ -137,39 +139,33 @@ function extractMessageRecord(
};
}
function extractMessageRecordsFromEventEntries(entries: VisibleTranscriptEventEntry<unknown>[]): {
type SqliteMessageRecord = {
id?: string;
message: unknown;
recordTimestampMs?: number;
seq: number;
}[] {
};
function extractMessageRecordsFromEventEntries(
entries: readonly SessionTranscriptMessageEvent[],
): SqliteMessageRecord[] {
return entries.flatMap((entry) => {
const record = extractMessageRecord(entry.event);
return record ? [{ ...record, seq: entry.seq }] : [];
});
}
function readSqliteMessageRecordsSync(target: ResolvedTranscriptReadTarget): {
id?: string;
message: unknown;
recordTimestampMs?: number;
seq: number;
}[] {
function readSqliteMessageRecordsSync(target: ResolvedTranscriptReadTarget): SqliteMessageRecord[] {
return extractMessageRecordsFromEventEntries(
selectVisibleTranscriptEventEntries(loadTranscriptEventsSync(toTranscriptReadScope(target))),
readSessionTranscriptMessageEvents(toTranscriptReadScope(target)),
);
}
export async function readSqliteMessageRecords(target: ResolvedTranscriptReadTarget): Promise<
{
id?: string;
message: unknown;
recordTimestampMs?: number;
seq: number;
}[]
> {
async function readSqliteMessageRecords(
target: ResolvedTranscriptReadTarget,
): Promise<SqliteMessageRecord[]> {
return extractMessageRecordsFromEventEntries(
selectVisibleTranscriptEventEntries(await loadTranscriptEvents(toTranscriptReadScope(target))),
readSessionTranscriptMessageEvents(toTranscriptReadScope(target)),
);
}
@@ -191,57 +187,31 @@ function normalizeRecentSqliteReadOptions(opts?: Partial<ReadRecentSessionMessag
return { maxMessages, maxBytes, maxLines };
}
function selectRecentSqliteEventEntries(
entries: VisibleTranscriptEventEntry<unknown>[],
opts: { maxBytes: number; maxLines: number },
) {
const selected: VisibleTranscriptEventEntry<unknown>[] = [];
let bytes = 0;
for (const entry of entries.toReversed()) {
const line = JSON.stringify(entry.event);
const lineBytes = Buffer.byteLength(line) + 1;
if (selected.length > 0 && bytes + lineBytes > opts.maxBytes) {
break;
}
selected.push(entry);
bytes += lineBytes;
if (selected.length >= opts.maxLines) {
break;
}
}
return selected.toReversed();
}
async function readRecentSqliteMessageRecords(
target: ResolvedTranscriptReadTarget,
opts?: Partial<ReadRecentSessionMessagesOptions>,
): Promise<{ id?: string; message: unknown; recordTimestampMs?: number; seq: number }[]> {
): Promise<{ records: SqliteMessageRecord[]; totalMessages: number }> {
const normalized = normalizeRecentSqliteReadOptions(opts);
const entries = selectVisibleTranscriptEventEntries(
await loadTranscriptEvents(toTranscriptReadScope(target)),
);
const records = extractMessageRecordsFromEventEntries(
selectRecentSqliteEventEntries(entries, normalized),
);
return normalized.maxMessages > 0 ? records.slice(-normalized.maxMessages) : [];
const page = readRecentSessionTranscriptMessageEvents(toTranscriptReadScope(target), normalized);
return {
records: extractMessageRecordsFromEventEntries(page.events),
totalMessages: page.totalMessages,
};
}
function readRecentSqliteUsageMessages(
target: ResolvedTranscriptReadTarget,
maxBytes: number,
): unknown[] {
const entries = selectVisibleTranscriptEventEntries(
loadTranscriptEventsSync(toTranscriptReadScope(target)),
);
return extractMessageRecordsFromEventEntries(
selectRecentSqliteEventEntries(entries, {
maxBytes: Math.max(1024, Math.floor(Number.isFinite(maxBytes) ? maxBytes : 8 * 1024 * 1024)),
maxLines: 1000,
}),
).map((record) => record.message);
const page = readRecentSessionTranscriptMessageEvents(toTranscriptReadScope(target), {
maxBytes: Math.max(1024, Math.floor(Number.isFinite(maxBytes) ? maxBytes : 8 * 1024 * 1024)),
maxLines: 1000,
maxMessages: 1000,
});
return extractMessageRecordsFromEventEntries(page.events).map((record) => record.message);
}
export function sqliteRecordMessageWithSeq(record: {
function sqliteRecordMessageWithSeq(record: {
id?: string;
message: unknown;
recordTimestampMs?: number;
@@ -256,6 +226,11 @@ export function sqliteRecordMessageWithSeq(record: {
});
}
export function sqliteMessageEventWithSeq(entry: SessionTranscriptMessageEvent): unknown {
const record = extractMessageRecord(entry.event);
return record ? sqliteRecordMessageWithSeq({ ...record, seq: entry.seq }) : undefined;
}
function extractMessageRole(message: unknown): string | undefined {
return message && typeof message === "object" && !Array.isArray(message)
? ((message as { role?: unknown }).role as string | undefined)
@@ -451,7 +426,7 @@ export async function readSessionMessagesAsync(
const target = resolveTranscriptReadTarget(scope);
if (isSqliteReadTarget(target)) {
if (opts.mode === "recent") {
const records = await readRecentSqliteMessageRecords(target, opts);
const { records } = await readRecentSqliteMessageRecords(target, opts);
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
return await readRecentSessionMessagesAsyncFile(
target.sessionId,
@@ -493,7 +468,7 @@ export async function readSessionMessagesWithSourceAsync(
if (isSqliteReadTarget(target)) {
const records =
opts.mode === "recent"
? await readRecentSqliteMessageRecords(target, opts)
? (await readRecentSqliteMessageRecords(target, opts)).records
: await readSqliteMessageRecords(target);
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
return await readSessionMessagesWithSourceAsyncFile(
@@ -527,9 +502,13 @@ export async function readSessionMessageByIdAsync(
): Promise<ReadSessionMessageByIdResult> {
const target = resolveTranscriptReadTarget(scope);
if (isSqliteReadTarget(target)) {
const found = (await readSqliteMessageRecords(target)).find(
(record) => record.id === messageId,
const foundEvent = readSessionTranscriptMessageEventById(
toTranscriptReadScope(target),
messageId,
);
const found = foundEvent
? extractMessageRecordsFromEventEntries([foundEvent]).at(0)
: undefined;
if (found) {
return { found: true, message: found.message, oversized: false, seq: found.seq };
}
@@ -584,7 +563,7 @@ export async function readSessionMessageCountAsync(
): Promise<number> {
const target = resolveTranscriptReadTarget(scope);
if (isSqliteReadTarget(target)) {
return (await readSqliteMessageRecords(target)).length;
return readSessionTranscriptMessageEventCount(toTranscriptReadScope(target));
}
return await readSessionMessageCountAsyncFile(
target.sessionId,
@@ -601,13 +580,8 @@ export async function readRecentSessionMessagesWithStatsAsync(
): Promise<ReadRecentSessionMessagesResult> {
const target = resolveTranscriptReadTarget(scope);
if (isSqliteReadTarget(target)) {
const records = await readSqliteMessageRecords(target);
const recentRecords = await readRecentSqliteMessageRecords(target, opts);
if (
records.length === 0 &&
recentRecords.length === 0 &&
opts.allowResetArchiveFallback === true
) {
const { records, totalMessages } = await readRecentSqliteMessageRecords(target, opts);
if (totalMessages === 0 && records.length === 0 && opts.allowResetArchiveFallback === true) {
return await readRecentSessionMessagesWithStatsAsyncFile(
target.sessionId,
target.storePath,
@@ -617,8 +591,8 @@ export async function readRecentSessionMessagesWithStatsAsync(
);
}
return {
messages: recentRecords.map(sqliteRecordMessageWithSeq),
totalMessages: records.length,
messages: records.map(sqliteRecordMessageWithSeq),
totalMessages,
transcriptPath: target.sessionFile,
};
}
@@ -638,8 +612,8 @@ export async function readSessionMessagesPageWithStatsAsync(
): Promise<ReadRecentSessionMessagesResult> {
const target = resolveTranscriptReadTarget(scope);
if (isSqliteReadTarget(target)) {
const records = await readSqliteMessageRecords(target);
if (records.length === 0 && opts.allowResetArchiveFallback === true) {
const page = readSessionTranscriptMessageEventPage(toTranscriptReadScope(target), opts);
if (page.totalMessages === 0 && opts.allowResetArchiveFallback === true) {
return await readSessionMessagesPageWithStatsAsyncFile(
target.sessionId,
target.storePath,
@@ -648,20 +622,9 @@ export async function readSessionMessagesPageWithStatsAsync(
target.agentId,
);
}
const totalMessages = records.length;
const offset = Math.min(
Math.max(0, Math.floor(Number.isFinite(opts.offset) ? opts.offset : 0)),
totalMessages,
);
const maxMessages = Math.max(
0,
Math.floor(Number.isFinite(opts.maxMessages) ? opts.maxMessages : 0),
);
const endExclusive = Math.max(0, totalMessages - offset);
const start = Math.max(0, endExclusive - maxMessages);
return {
messages: records.slice(start, endExclusive).map(sqliteRecordMessageWithSeq),
totalMessages,
messages: extractMessageRecordsFromEventEntries(page.events).map(sqliteRecordMessageWithSeq),
totalMessages: page.totalMessages,
transcriptPath: target.sessionFile,
};
}
@@ -763,4 +726,3 @@ export function readSessionPreviewItemsFromTranscript(
maxChars,
);
}
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
src/gateway/session-utils.fs-anchor.ts
+1 -1
View File
@@ -19,7 +19,7 @@ type ReadSessionMessagesAroundIdResult = ReadRecentSessionMessagesResult & {
offset: number;
};
export function resolveSessionMessageAnchorBounds(
function resolveSessionMessageAnchorBounds(
records: readonly { id?: string }[],
messageId: string,
maxMessages: number,
src/gateway/sessions-history-http.revocation.test.ts
+28 -2
View File
@@ -17,6 +17,7 @@ let gatewayConfig: {
allowRealIpFallback: false,
};
let authCheckCalls = 0;
let transcriptReadError: Error | undefined;
vi.mock("../config/config.js", () => ({
getRuntimeConfig: () => ({
@@ -94,9 +95,19 @@ vi.mock("./session-utils.js", () => ({
}));
vi.mock("./session-transcript-readers.js", () => ({
readRecentSessionMessagesWithStatsAsync: async () => ({ messages: [], totalMessages: 0 }),
readRecentSessionMessagesWithStatsAsync: async () => {
if (transcriptReadError) {
throw transcriptReadError;
}
return { messages: [], totalMessages: 0 };
},
readSessionMessagesAsync: async () => [],
readSessionMessagesWithSourceAsync: async () => ({ messages: [] }),
readSessionMessagesWithSourceAsync: async () => {
if (transcriptReadError) {
throw transcriptReadError;
}
return { messages: [] };
},
}));
vi.mock("./session-history-state.js", () => ({
@@ -117,6 +128,7 @@ vi.mock("./session-history-state.js", () => ({
},
}));
import { SessionTranscriptProjectionUnavailableError } from "../config/sessions/session-accessor.js";
import { handleSessionHistoryHttpRequest } from "./sessions-history-http.js";
const SESSION_HISTORY_URL = "/sessions/agent%3Amain/history";
@@ -331,6 +343,7 @@ afterEach(() => {
transcriptUpdateHandler = undefined;
authRevoked = false;
authCheckCalls = 0;
transcriptReadError = undefined;
gatewayConfig = {
trustedProxies: ["10.0.0.1"],
allowRealIpFallback: false,
@@ -338,6 +351,19 @@ afterEach(() => {
});
describe("session history SSE auth revocation", () => {
it("returns retryable HTTP unavailable while a dirty projection rebuilds", async () => {
transcriptReadError = new SessionTranscriptProjectionUnavailableError("session-1");
const { req, res } = await openSessionHistoryStreamPair(TRUSTED_PROXY_STARTUP_OPTIONS, {
expectSubscribed: false,
});
expect(res.statusCode).toBe(503);
expect(res.headers.get("retry-after")).toBe("1");
expect(res.writes.join("")).toContain('"retryable":true');
expect(req.listenerCount("error")).toBe(0);
});
it("closes the stream before delivering transcript updates after auth is revoked", async () => {
const res = await openSessionHistoryStream({ auth: { mode: "trusted-proxy" } as never });
src/gateway/sessions-history-http.ts
+56 -35
View File
@@ -6,6 +6,7 @@ import {
normalizeOptionalString,
} from "@openclaw/normalization-core/string-coerce";
import { getRuntimeConfig } from "../config/io.js";
import { isSessionTranscriptProjectionUnavailableError } from "../config/sessions/session-accessor.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
import { normalizeAgentId } from "../routing/session-key.js";
import { onInternalSessionTranscriptUpdate } from "../sessions/transcript-events.js";
@@ -143,41 +144,61 @@ export async function handleSessionHistoryHttpRequest(
const limit = resolveLimit(req);
const cursor = normalizeOptionalString(getRequestUrl(req).searchParams.get("cursor"));
const effectiveMaxChars = DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS;
const boundedSnapshot =
cursor === undefined && typeof limit === "number"
? await readRecentSessionMessagesWithStatsAsync(
{
agentId: target.agentId,
sessionEntry: entry,
sessionId: entry.sessionId,
sessionKey: target.canonicalKey,
storePath: target.storePath,
},
{
...resolveSessionHistoryTailReadOptions(limit),
allowResetArchiveFallback: true,
},
)
: undefined;
// Cursor reads still need an arbitrary historical window. The common first
// page path is bounded above so `limit=1` cannot materialize huge transcripts.
const fullSnapshot =
boundedSnapshot === undefined && entry?.sessionId
? await readSessionMessagesWithSourceAsync(
{
agentId: target.agentId,
sessionEntry: entry,
sessionId: entry.sessionId,
sessionKey: target.canonicalKey,
storePath: target.storePath,
},
{
mode: "full",
reason: "session history cursor pagination",
allowResetArchiveFallback: true,
},
)
: undefined;
let boundedSnapshot:
| Awaited<ReturnType<typeof readRecentSessionMessagesWithStatsAsync>>
| undefined;
let fullSnapshot: Awaited<ReturnType<typeof readSessionMessagesWithSourceAsync>> | undefined;
try {
boundedSnapshot =
cursor === undefined && typeof limit === "number"
? await readRecentSessionMessagesWithStatsAsync(
{
agentId: target.agentId,
sessionEntry: entry,
sessionId: entry.sessionId,
sessionKey: target.canonicalKey,
storePath: target.storePath,
},
{
...resolveSessionHistoryTailReadOptions(limit),
allowResetArchiveFallback: true,
},
)
: undefined;
// Cursor reads still need an arbitrary historical window. The common first
// page path is bounded above so `limit=1` cannot materialize huge transcripts.
fullSnapshot =
boundedSnapshot === undefined && entry?.sessionId
? await readSessionMessagesWithSourceAsync(
{
agentId: target.agentId,
sessionEntry: entry,
sessionId: entry.sessionId,
sessionKey: target.canonicalKey,
storePath: target.storePath,
},
{
mode: "full",
reason: "session history cursor pagination",
allowResetArchiveFallback: true,
},
)
: undefined;
} catch (error) {
if (!isSessionTranscriptProjectionUnavailableError(error)) {
throw error;
}
res.setHeader("Retry-After", "1");
sendJson(res, 503, {
ok: false,
error: {
type: "unavailable",
message: "session history is rebuilding; retry shortly",
retryable: true,
},
});
return true;
}
const rawSnapshot = boundedSnapshot?.messages ?? fullSnapshot?.messages ?? [];
const historySnapshot = buildSessionHistorySnapshot({
rawMessages: rawSnapshot,
src/state/openclaw-agent-db.generated.d.ts
Vendored
+10
View File
@@ -121,6 +121,13 @@ export interface SessionRoutes {
updated_at: number;
}
export interface SessionTranscriptActiveEvents {
active_position: number;
event_seq: number;
message_position: number | null;
session_id: string;
}
export interface SessionTranscriptFts {
message_id: string | null;
role: string | null;
@@ -160,6 +167,8 @@ export interface SessionTranscriptFtsIdx {
}
export interface SessionTranscriptIndexState {
active_event_count: Generated<number>;
active_message_count: Generated<number>;
indexed_seq: number;
leaf_event_id: string | null;
needs_rebuild: Generated<number>;
@@ -233,6 +242,7 @@ export interface DB {
session_conversations: SessionConversations;
session_entries: SessionEntries;
session_routes: SessionRoutes;
session_transcript_active_events: SessionTranscriptActiveEvents;
session_transcript_fts: SessionTranscriptFts;
session_transcript_fts_config: SessionTranscriptFtsConfig;
session_transcript_fts_content: SessionTranscriptFtsContent;
src/state/openclaw-agent-db.test.ts
+88
View File
@@ -532,6 +532,9 @@ describe("openclaw agent database", () => {
DROP TABLE memory_index_sources_strict;
INSERT INTO memory_index_sources (path, source, hash, mtime, size)
VALUES ('MEMORY.md', 'memory', 'legacy-hash', 10.75, 20);
DROP TABLE session_transcript_active_events;
ALTER TABLE session_transcript_index_state DROP COLUMN active_event_count;
ALTER TABLE session_transcript_index_state DROP COLUMN active_message_count;
PRAGMA user_version = 8;
UPDATE schema_meta SET schema_version = 8 WHERE meta_key = 'primary';
`);
@@ -553,6 +556,19 @@ describe("openclaw agent database", () => {
.prepare("SELECT mtime, typeof(mtime) AS storage_type FROM memory_index_sources")
.get(),
).toEqual({ mtime: 10.75, storage_type: "real" });
expect(
migrated.db
.prepare(
"SELECT strict FROM pragma_table_list WHERE name = 'session_transcript_active_events'",
)
.get(),
).toEqual({ strict: 1 });
expect(
migrated.db
.prepare("PRAGMA table_info(session_transcript_index_state)")
.all()
.map((column) => (column as { name: string }).name),
).toEqual(expect.arrayContaining(["active_event_count", "active_message_count"]));
expect(readSqliteNumberPragma(migrated.db, "user_version")).toBe(OPENCLAW_AGENT_SCHEMA_VERSION);
expect(
migrated.db
@@ -1560,6 +1576,78 @@ describe("openclaw agent database", () => {
expect(readSqliteNumberPragma(database.db, "user_version")).toBe(OPENCLAW_AGENT_SCHEMA_VERSION);
});
it("adds the active transcript projection when upgrading v9 databases", () => {
const stateDir = createTempStateDir();
const databasePath = path.join(
stateDir,
"agents",
"worker-1",
"agent",
"openclaw-agent.sqlite",
);
fs.mkdirSync(path.dirname(databasePath), { recursive: true });
const currentSchema = fs.readFileSync(
new URL("./openclaw-agent-schema.sql", import.meta.url),
"utf8",
);
const { DatabaseSync } = requireNodeSqlite();
const db = new DatabaseSync(databasePath);
db.exec(currentSchema);
db.exec(`
DROP TABLE session_transcript_active_events;
ALTER TABLE session_transcript_index_state DROP COLUMN active_event_count;
ALTER TABLE session_transcript_index_state DROP COLUMN active_message_count;
INSERT INTO schema_meta
(meta_key, role, schema_version, agent_id, app_version, created_at, updated_at)
VALUES ('primary', 'agent', 9, 'worker-1', NULL, 1, 1);
INSERT INTO sessions
(session_id, session_key, created_at, updated_at)
VALUES ('session-1', 'agent:worker-1:main', 10, 20);
INSERT INTO transcript_events
(session_id, seq, event_json, created_at)
VALUES
('session-1', 0, '{"type":"session","id":"session-1"}', 10),
('session-1', 1, '{"type":"message","id":"m1","parentId":null,"message":{"role":"user","content":"hello"}}', 20);
INSERT INTO session_transcript_index_state
(session_id, indexed_seq, leaf_event_id, needs_rebuild, updated_at)
VALUES ('session-1', 1, 'm1', 0, 20);
PRAGMA user_version = 9;
`);
db.close();
const database = openOpenClawAgentDatabase({
agentId: "worker-1",
env: { OPENCLAW_STATE_DIR: stateDir },
});
const columns = database.db
.prepare("PRAGMA table_info(session_transcript_index_state)")
.all() as Array<{ name?: unknown }>;
expect(columns.map((column) => column.name)).toEqual(
expect.arrayContaining(["active_event_count", "active_message_count"]),
);
expect(
database.db
.prepare(
"SELECT indexed_seq, needs_rebuild, active_event_count, active_message_count FROM session_transcript_index_state WHERE session_id = ?",
)
.get("session-1"),
).toEqual({
active_event_count: 0,
active_message_count: 0,
indexed_seq: 1,
needs_rebuild: 1,
});
expect(
database.db
.prepare(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_transcript_active_events'",
)
.get(),
).toEqual({ name: "session_transcript_active_events" });
expect(readSqliteNumberPragma(database.db, "user_version")).toBe(OPENCLAW_AGENT_SCHEMA_VERSION);
});
it("inspects registered database ownership without mutating the database", () => {
const stateDir = createTempStateDir();
const database = openOpenClawAgentDatabase({
src/state/openclaw-agent-db.ts
+31 -3
View File
@@ -70,14 +70,14 @@ export { resolveOpenClawAgentSqlitePath } from "./openclaw-agent-db.paths.js";
* per pathname, protected with private file modes, and registered in the shared
* OpenClaw state database for discovery and maintenance.
*/
// v9 = SQLite STRICT tables. v8 added per-transcript session provenance.
// v7 added per-entry lifecycle status projection.
// v10 = materialized active transcript paths. v9 added SQLite STRICT tables.
// v8 added per-transcript session provenance. v7 added per-entry lifecycle status projection.
// v6 added session/transcript hot-path indexes.
// v5 added transcript mutation watermarks.
// The v4 session/transcript flip and main's v2 memory-identity
// change is folded in structure-gated (migrateMemoryIndexSourcesIdentity), so
// v2 main DBs and pre-merge v4 flip DBs both converge on this schema.
export const OPENCLAW_AGENT_SCHEMA_VERSION = 9;
export const OPENCLAW_AGENT_SCHEMA_VERSION = 10;
const OPENCLAW_AGENT_DB_DIR_MODE = 0o700;
const OPENCLAW_AGENT_DB_FILE_MODE = 0o600;
const OPENCLAW_AGENT_DB_SLOW_OPEN_MS = 1_000;
@@ -358,6 +358,33 @@ function migrateOpenClawAgentSchema(db: DatabaseSync): void {
backfillTranscriptMutationWatermarks(db);
}
function migrateSessionTranscriptActiveProjection(db: DatabaseSync, previousVersion: number): void {
if (previousVersion >= 10) {
return;
}
const columns = readSqliteTableColumns(db, "session_transcript_index_state");
if (columns && !columns.has("active_event_count")) {
db.exec(
"ALTER TABLE session_transcript_index_state ADD COLUMN active_event_count INTEGER NOT NULL DEFAULT 0;",
);
}
if (columns && !columns.has("active_message_count")) {
db.exec(
"ALTER TABLE session_transcript_index_state ADD COLUMN active_message_count INTEGER NOT NULL DEFAULT 0;",
);
}
// This table is derived state. Gateway startup rebuilds it after all legacy
// imports finish, keeping schema-open work cheap and history reads bounded.
db.exec(`
DELETE FROM session_transcript_active_events;
UPDATE session_transcript_index_state
SET needs_rebuild = 1,
active_event_count = 0,
active_message_count = 0,
updated_at = ${Date.now()};
`);
}
function parseMigratedSessionEntry(value: unknown): MigratedSessionEntry | null {
if (typeof value !== "string") {
return null;
@@ -695,6 +722,7 @@ function ensureAgentSchema(db: DatabaseSync, agentId: string, pathname: string):
migrateMemoryIndexSourcesIdentity(db);
migrateOpenClawAgentSchema(db);
db.exec(OPENCLAW_AGENT_SCHEMA_SQL);
migrateSessionTranscriptActiveProjection(db, previousVersion);
if (previousVersion < OPENCLAW_AGENT_SCHEMA_VERSION) {
migrateSqliteSchemaToStrictInTransaction(db, OPENCLAW_AGENT_SCHEMA_SQL, {
databaseLabel: pathname,
src/state/openclaw-agent-schema.generated.ts
+18
View File
@@ -254,9 +254,27 @@ CREATE TABLE IF NOT EXISTS session_transcript_index_state (
indexed_seq INTEGER NOT NULL,
leaf_event_id TEXT,
needs_rebuild INTEGER NOT NULL DEFAULT 0,
active_event_count INTEGER NOT NULL DEFAULT 0,
active_message_count INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL
) STRICT;
CREATE TABLE IF NOT EXISTS session_transcript_active_events (
session_id TEXT NOT NULL,
active_position INTEGER NOT NULL CHECK (active_position >= 0),
event_seq INTEGER NOT NULL,
message_position INTEGER CHECK (message_position IS NULL OR message_position >= 0),
PRIMARY KEY (session_id, active_position),
FOREIGN KEY (session_id, event_seq) REFERENCES transcript_events(session_id, seq) ON DELETE CASCADE
) STRICT;
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_transcript_active_event_seq
ON session_transcript_active_events(session_id, event_seq);
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_transcript_active_messages
ON session_transcript_active_events(session_id, message_position)
WHERE message_position IS NOT NULL;
CREATE VIRTUAL TABLE IF NOT EXISTS session_transcript_fts USING fts5(
text,
session_id UNINDEXED,
src/state/openclaw-agent-schema.sql
+18
View File
@@ -249,9 +249,27 @@ CREATE TABLE IF NOT EXISTS session_transcript_index_state (
indexed_seq INTEGER NOT NULL,
leaf_event_id TEXT,
needs_rebuild INTEGER NOT NULL DEFAULT 0,
active_event_count INTEGER NOT NULL DEFAULT 0,
active_message_count INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL
) STRICT;
CREATE TABLE IF NOT EXISTS session_transcript_active_events (
session_id TEXT NOT NULL,
active_position INTEGER NOT NULL CHECK (active_position >= 0),
event_seq INTEGER NOT NULL,
message_position INTEGER CHECK (message_position IS NULL OR message_position >= 0),
PRIMARY KEY (session_id, active_position),
FOREIGN KEY (session_id, event_seq) REFERENCES transcript_events(session_id, seq) ON DELETE CASCADE
) STRICT;
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_transcript_active_event_seq
ON session_transcript_active_events(session_id, event_seq);
CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_transcript_active_messages
ON session_transcript_active_events(session_id, message_position)
WHERE message_position IS NOT NULL;
CREATE VIRTUAL TABLE IF NOT EXISTS session_transcript_fts USING fts5(
text,
session_id UNINDEXED,
src/state/sqlite-query-plan.test.ts
+33
View File
@@ -208,5 +208,38 @@ describe("sqlite hot query plans", () => {
"USING COVERING INDEX idx_agent_transcript_event_sequence (session_id=? AND event_type=?)",
);
expect(latestMessagePlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
const historyPagePlan = explainQueryPlan(
database.db,
`
SELECT active.event_seq, event.event_json
FROM session_transcript_active_events AS active
JOIN transcript_events AS event
ON event.session_id = active.session_id AND event.seq = active.event_seq
WHERE active.session_id = ?
AND active.message_position IS NOT NULL
AND active.message_position >= ?
AND active.message_position < ?
ORDER BY active.message_position ASC
`,
["session-1", 100, 125],
);
expect(historyPagePlan).toContain("idx_agent_transcript_active_messages");
expect(historyPagePlan).toContain("sqlite_autoindex_transcript_events_1");
expect(historyPagePlan).not.toContain("USE TEMP B-TREE FOR ORDER BY");
const historyAnchorPlan = explainQueryPlan(
database.db,
`
SELECT active.message_position
FROM transcript_event_identities AS identity
JOIN session_transcript_active_events AS active
ON active.session_id = identity.session_id AND active.event_seq = identity.seq
WHERE identity.session_id = ? AND identity.event_id = ?
`,
["session-1", "message-1"],
);
expect(historyAnchorPlan).toContain("sqlite_autoindex_transcript_event_identities_1");
expect(historyAnchorPlan).toContain("idx_agent_transcript_active_event_seq");
});
});
test/release-check.test.ts
+2
View File
@@ -709,6 +709,7 @@ describe("collectMissingPackPaths", () => {
"dist/agents/compaction-planning.worker.js",
"dist/agents/model-provider-auth.worker.js",
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
bundledDistPluginFile("telegram", "runtime-api.js"),
@@ -744,6 +745,7 @@ describe("collectMissingPackPaths", () => {
"dist/agents/compaction-planning.worker.js",
"dist/agents/model-provider-auth.worker.js",
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
"dist/build-info.json",
test/scripts/sqlite-session-schema-baseline.test.ts
+1
View File
@@ -18,6 +18,7 @@ describe("SQLite sessions/transcripts schema baseline", () => {
expect(rendered.sql).toContain("CREATE TABLE IF NOT EXISTS sessions");
expect(rendered.sql).toContain("CREATE TABLE IF NOT EXISTS transcript_events");
expect(rendered.sql).toContain("CREATE TABLE IF NOT EXISTS transcript_event_identities");
expect(rendered.sql).toContain("CREATE TABLE IF NOT EXISTS session_transcript_active_events");
expect(rendered.sql).not.toContain("idx_agent_transcript_events_session");
expect(rendered.sql).not.toContain("CREATE TABLE IF NOT EXISTS cache_entries");
expect(rendered.sql).not.toContain("CREATE TABLE IF NOT EXISTS auth_profile_store");
tsdown.config.ts
+2
View File
@@ -275,6 +275,8 @@ function buildCoreDistEntries(): Record<string, string> {
"agents/compaction-planning.worker": "src/agents/compaction-planning.worker.ts",
"agents/model-provider-auth.worker": "src/agents/model-provider-auth.worker.ts",
"audit/audit-event-writer.worker": "src/audit/audit-event-writer.worker.ts",
"config/sessions/session-transcript-reconcile.worker":
"src/config/sessions/session-transcript-reconcile.worker.ts",
"acp/control-plane/manager": "src/acp/control-plane/manager.ts",
"cli/gateway-lifecycle.runtime": "src/cli/gateway-cli/lifecycle.runtime.ts",
"provider-dispatcher.runtime": "src/auto-reply/reply/provider-dispatcher.runtime.ts",