From 7d71d7cf6b3c8bcbc0f05fb80d187808686a1f1d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 14:11:33 -0700 Subject: [PATCH] fix(sessions): large histories no longer load entire transcripts (#108851) * 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 --- config/max-lines-baseline.txt | 1 - ...-session-transcript-schema-baseline.sha256 | 2 +- scripts/lib/sqlite-session-schema-baseline.ts | 1 + scripts/release-check.ts | 1 + ...sion-accessor.sqlite-active-events.test.ts | 605 ++++++++++++++++++ .../session-accessor.sqlite-active-events.ts | 382 +++++++++++ ...ession-accessor.sqlite-transcript-store.ts | 51 +- ...ession-accessor.sqlite-transcript-write.ts | 3 + src/config/sessions/session-accessor.ts | 15 + .../sessions/session-transcript-index.ts | 297 ++++++--- .../session-transcript-projection-rebuild.ts | 429 +++++++++++++ .../sessions/session-transcript-reconcile.ts | 389 +++++++++++ .../session-transcript-reconcile.worker.ts | 199 ++++++ .../session-transcript-search.test.ts | 29 +- .../sessions/session-transcript-search.ts | 103 +-- .../sessions/transcript-visible-events.ts | 2 +- src/gateway/server-methods/chat.ts | 48 +- .../server-startup-session-migration.test.ts | 32 + .../server-startup-session-migration.ts | 28 + .../server.chat.gateway-server-chat-b.test.ts | 104 ++- .../session-transcript-anchor-reader.ts | 33 +- .../session-transcript-readers.test.ts | 15 +- src/gateway/session-transcript-readers.ts | 152 ++--- src/gateway/session-utils.fs-anchor.ts | 2 +- .../sessions-history-http.revocation.test.ts | 30 +- src/gateway/sessions-history-http.ts | 91 ++- src/state/openclaw-agent-db.generated.d.ts | 10 + src/state/openclaw-agent-db.test.ts | 88 +++ src/state/openclaw-agent-db.ts | 34 +- src/state/openclaw-agent-schema.generated.ts | 18 + src/state/openclaw-agent-schema.sql | 18 + src/state/sqlite-query-plan.test.ts | 33 + test/release-check.test.ts | 2 + .../sqlite-session-schema-baseline.test.ts | 1 + tsdown.config.ts | 2 + 35 files changed, 2879 insertions(+), 371 deletions(-) create mode 100644 src/config/sessions/session-accessor.sqlite-active-events.test.ts create mode 100644 src/config/sessions/session-accessor.sqlite-active-events.ts create mode 100644 src/config/sessions/session-transcript-projection-rebuild.ts create mode 100644 src/config/sessions/session-transcript-reconcile.ts create mode 100644 src/config/sessions/session-transcript-reconcile.worker.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index fa89d01bfad4..5f081336af01 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -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 diff --git a/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 b/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 index 442bc86a6ac8..a8646cdf263d 100644 --- a/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 +++ b/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 @@ -1 +1 @@ -1754707609cb8b5248307fc7082f3a213e0069883b6be5f392180f80bc0364c5 sqlite-session-transcript-schema-baseline.sql +286d1d7ffb924f2a2fb8e12e5c50ed05daddbc6451549c052540985062272761 sqlite-session-transcript-schema-baseline.sql diff --git a/scripts/lib/sqlite-session-schema-baseline.ts b/scripts/lib/sqlite-session-schema-baseline.ts index c076f83eb7c6..78d0856a09fc 100644 --- a/scripts/lib/sqlite-session-schema-baseline.ts +++ b/scripts/lib/sqlite-session-schema-baseline.ts @@ -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 { diff --git a/scripts/release-check.ts b/scripts/release-check.ts index 6673172eb702..c8832eedd483 100755 --- a/scripts/release-check.ts +++ b/scripts/release-check.ts @@ -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", diff --git a/src/config/sessions/session-accessor.sqlite-active-events.test.ts b/src/config/sessions/session-accessor.sqlite-active-events.test.ts new file mode 100644 index 000000000000..9f7ac19b3d74 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-active-events.test.ts @@ -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((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((resolve) => { + writerEntered = resolve; + }); + const release = new Promise((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((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); +}); diff --git a/src/config/sessions/session-accessor.sqlite-active-events.ts b/src/config/sessions/session-accessor.sqlite-active-events.ts new file mode 100644 index 000000000000..45e7fad8b1c4 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-active-events.ts @@ -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; + state: SessionTranscriptProjectionState; +}; + +const EMPTY_PROJECTION_STATE: SessionTranscriptProjectionState = { + activeEventCount: 0, + activeMessageCount: 0, + indexedSeq: -1, + leafEventId: null, + needsRebuild: false, +}; + +function getActiveTranscriptKysely(database: OpenClawAgentDatabase) { + return getNodeSqliteKysely(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( + 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, + }; + }); +} diff --git a/src/config/sessions/session-accessor.sqlite-transcript-store.ts b/src/config/sessions/session-accessor.sqlite-transcript-store.ts index 2ddd0f221b20..676575b2a70f 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-store.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-store.ts @@ -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); } } diff --git a/src/config/sessions/session-accessor.sqlite-transcript-write.ts b/src/config/sessions/session-accessor.sqlite-transcript-write.ts index ab4904c6b9c7..7a7576705a70 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-write.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-write.ts @@ -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( diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index ea6d7aef702e..33365435a501 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -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, diff --git a/src/config/sessions/session-transcript-index.ts b/src/config/sessions/session-transcript-index.ts index b325844fd829..18095d9fe8d6 100644 --- a/src/config/sessions/session-transcript-index.ts +++ b/src/config/sessions/session-transcript-index.ts @@ -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(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 diff --git a/src/config/sessions/session-transcript-projection-rebuild.ts b/src/config/sessions/session-transcript-projection-rebuild.ts new file mode 100644 index 000000000000..689bd309d851 --- /dev/null +++ b/src/config/sessions/session-transcript-projection-rebuild.ts @@ -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; + }; + session_transcript_fts: OpenClawAgentKyselyDatabase["session_transcript_fts"] & { + rowid: Generated; + }; +}; + +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(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; +} diff --git a/src/config/sessions/session-transcript-reconcile.ts b/src/config/sessions/session-transcript-reconcile.ts new file mode 100644 index 000000000000..aa8ba98966ca --- /dev/null +++ b/src/config/sessions/session-transcript-reconcile.ts @@ -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; +}; + +const runningReconciles = new Map(); + +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 { + 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( + databaseOptions: OpenClawAgentDatabaseOptions, + operationLabel: string, + operation: (database: OpenClawAgentDatabase) => T, +): Promise { + return await runExclusiveSqliteSessionWrite(databaseOptions, async () => + runOpenClawAgentWriteTransaction(operation, databaseOptions, { operationLabel }), + ); +} + +async function claimPreparedSessionTranscriptProjection( + databaseOptions: OpenClawAgentDatabaseOptions, + plan: PreparedSessionTranscriptProjectionMetadata, +): Promise { + 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 { + 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 { + 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 { + 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((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 { + await runningReconciles.get(reconcileKey(params))?.promise; +} diff --git a/src/config/sessions/session-transcript-reconcile.worker.ts b/src/config/sessions/session-transcript-reconcile.worker.ts new file mode 100644 index 000000000000..940bbbe2908d --- /dev/null +++ b/src/config/sessions/session-transcript-reconcile.worker.ts @@ -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; +}; + +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; + 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 { + 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 { + 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 { + 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 { + 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(); diff --git a/src/config/sessions/session-transcript-search.test.ts b/src/config/sessions/session-transcript-search.test.ts index 5740f1f66239..0264021ef07f 100644 --- a/src/config/sessions/session-transcript-search.test.ts +++ b/src/config/sessions/session-transcript-search.test.ts @@ -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(); diff --git a/src/config/sessions/session-transcript-search.ts b/src/config/sessions/session-transcript-search.ts index 4cd445b5bfad..28e102e55e69 100644 --- a/src/config/sessions/session-transcript-search.ts +++ b/src/config/sessions/session-transcript-search.ts @@ -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>(); - -/** - * 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 { - 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>( - 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((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 = diff --git a/src/config/sessions/transcript-visible-events.ts b/src/config/sessions/transcript-visible-events.ts index b021dacdd195..1bdda1d725fe 100644 --- a/src/config/sessions/transcript-visible-events.ts +++ b/src/config/sessions/transcript-visible-events.ts @@ -3,7 +3,7 @@ import { selectSessionTranscriptTreePathNodes, } from "./transcript-tree.js"; -export type VisibleTranscriptEventEntry = { +type VisibleTranscriptEventEntry = { event: T; /** Parent id after active-branch normalization; null when no visible parent exists. */ parentId: string | null; diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 884f294d8c02..9b7a289c9f10 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -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>; + 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({ diff --git a/src/gateway/server-startup-session-migration.test.ts b/src/gateway/server-startup-session-migration.test.ts index f1b6be131119..c1e6a18ce583 100644 --- a/src/gateway/server-startup-session-migration.test.ts +++ b/src/gateway/server-startup-session-migration.test.ts @@ -11,6 +11,9 @@ type ResolveStoreTargets = NonNullable< >; type SweepStoreTemps = NonNullable; type RunDoctorSessionSqlite = NonNullable; +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() + .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().mockResolvedValue({ changes: [], warnings: [] }); + const reconcile = vi + .fn() + .mockResolvedValueOnce({ reconciledSessions: 2 }) + .mockResolvedValueOnce({ reconciledSessions: 1 }); + const cfg = { + agents: { defaults: {}, list: [{ id: "main" }, { id: "ops" }] }, + session: {}, + } as Parameters[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().mockResolvedValue({ changes: [], warnings: [] }); diff --git a/src/gateway/server-startup-session-migration.ts b/src/gateway/server-startup-session-migration.ts index 55cec3076864..d8236577355f 100644 --- a/src/gateway/server-startup-session-migration.ts +++ b/src/gateway/server-startup-session-migration.ts @@ -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[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 { await runSessionStartupMigration(params); await runStartupSessionSqliteImport(params); + await reconcileStartupSessionTranscriptIndexes(params); +} + +async function reconcileStartupSessionTranscriptIndexes(params: { + cfg: OpenClawConfig; + env?: NodeJS.ProcessEnv; + log: SessionStartupMigrationLogger; + deps?: SessionMigrationDeps; +}): Promise { + 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: { diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index 6389394f92a6..dbe3ab815e4f 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -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(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(); diff --git a/src/gateway/session-transcript-anchor-reader.ts b/src/gateway/session-transcript-anchor-reader.ts index ec4b2fe8d9bf..6a902a8f36bc 100644 --- a/src/gateway/session-transcript-anchor-reader.ts +++ b/src/gateway/session-transcript-anchor-reader.ts @@ -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, }; } diff --git a/src/gateway/session-transcript-readers.test.ts b/src/gateway/session-transcript-readers.test.ts index 5a76c6241486..bd32ec736587 100644 --- a/src/gateway/session-transcript-readers.test.ts +++ b/src/gateway/session-transcript-readers.test.ts @@ -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" }), diff --git a/src/gateway/session-transcript-readers.ts b/src/gateway/session-transcript-readers.ts index fed1cc1aaac4..31c05d165f6d 100644 --- a/src/gateway/session-transcript-readers.ts +++ b/src/gateway/session-transcript-readers.ts @@ -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[]): { +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 { return extractMessageRecordsFromEventEntries( - selectVisibleTranscriptEventEntries(await loadTranscriptEvents(toTranscriptReadScope(target))), + readSessionTranscriptMessageEvents(toTranscriptReadScope(target)), ); } @@ -191,57 +187,31 @@ function normalizeRecentSqliteReadOptions(opts?: Partial[], - opts: { maxBytes: number; maxLines: number }, -) { - const selected: VisibleTranscriptEventEntry[] = []; - 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, -): 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 { 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 { 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 { 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 { 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. */ diff --git a/src/gateway/session-utils.fs-anchor.ts b/src/gateway/session-utils.fs-anchor.ts index e7bc82bda46c..6aaa335a1d98 100644 --- a/src/gateway/session-utils.fs-anchor.ts +++ b/src/gateway/session-utils.fs-anchor.ts @@ -19,7 +19,7 @@ type ReadSessionMessagesAroundIdResult = ReadRecentSessionMessagesResult & { offset: number; }; -export function resolveSessionMessageAnchorBounds( +function resolveSessionMessageAnchorBounds( records: readonly { id?: string }[], messageId: string, maxMessages: number, diff --git a/src/gateway/sessions-history-http.revocation.test.ts b/src/gateway/sessions-history-http.revocation.test.ts index c5133c87419c..1748fd1de389 100644 --- a/src/gateway/sessions-history-http.revocation.test.ts +++ b/src/gateway/sessions-history-http.revocation.test.ts @@ -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 }); diff --git a/src/gateway/sessions-history-http.ts b/src/gateway/sessions-history-http.ts index 59921c787de5..c091c2dc96fd 100644 --- a/src/gateway/sessions-history-http.ts +++ b/src/gateway/sessions-history-http.ts @@ -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> + | undefined; + let fullSnapshot: Awaited> | 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, diff --git a/src/state/openclaw-agent-db.generated.d.ts b/src/state/openclaw-agent-db.generated.d.ts index ede360f7c781..11185f249577 100644 --- a/src/state/openclaw-agent-db.generated.d.ts +++ b/src/state/openclaw-agent-db.generated.d.ts @@ -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; + active_message_count: Generated; indexed_seq: number; leaf_event_id: string | null; needs_rebuild: Generated; @@ -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; diff --git a/src/state/openclaw-agent-db.test.ts b/src/state/openclaw-agent-db.test.ts index 1bea38c1140d..d61fbf2b7c0b 100644 --- a/src/state/openclaw-agent-db.test.ts +++ b/src/state/openclaw-agent-db.test.ts @@ -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({ diff --git a/src/state/openclaw-agent-db.ts b/src/state/openclaw-agent-db.ts index e3b976f886eb..de227c55689c 100644 --- a/src/state/openclaw-agent-db.ts +++ b/src/state/openclaw-agent-db.ts @@ -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, diff --git a/src/state/openclaw-agent-schema.generated.ts b/src/state/openclaw-agent-schema.generated.ts index 0bf2ee8c6a81..1720a6e54107 100644 --- a/src/state/openclaw-agent-schema.generated.ts +++ b/src/state/openclaw-agent-schema.generated.ts @@ -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, diff --git a/src/state/openclaw-agent-schema.sql b/src/state/openclaw-agent-schema.sql index a7c17995b8c5..cdf1def05ce2 100644 --- a/src/state/openclaw-agent-schema.sql +++ b/src/state/openclaw-agent-schema.sql @@ -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, diff --git a/src/state/sqlite-query-plan.test.ts b/src/state/sqlite-query-plan.test.ts index 9182eb2bafe8..bd5e1e83ca92 100644 --- a/src/state/sqlite-query-plan.test.ts +++ b/src/state/sqlite-query-plan.test.ts @@ -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"); }); }); diff --git a/test/release-check.test.ts b/test/release-check.test.ts index e8809ce2bf12..69792559bf74 100644 --- a/test/release-check.test.ts +++ b/test/release-check.test.ts @@ -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", diff --git a/test/scripts/sqlite-session-schema-baseline.test.ts b/test/scripts/sqlite-session-schema-baseline.test.ts index 4743492feaeb..bce03112fdea 100644 --- a/test/scripts/sqlite-session-schema-baseline.test.ts +++ b/test/scripts/sqlite-session-schema-baseline.test.ts @@ -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"); diff --git a/tsdown.config.ts b/tsdown.config.ts index d3eef3609b4f..fa11d3c4bd74 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -275,6 +275,8 @@ function buildCoreDistEntries(): Record { "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",