From aed7d49cc5ca04571a5493af8739715cfbba3350 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:22:27 +0800 Subject: [PATCH 01/14] feat(storage): maintain lightweight transcript turn positions Generated-by: Codex --- .../sqlite-session-metadata-store.test.ts | 312 +++++++++++++++ packages/storage/src/execution-stores.ts | 5 + packages/storage/src/session-store.ts | 40 ++ .../src/sqlite-session-metadata-schema.ts | 26 +- .../src/sqlite-session-metadata-store.ts | 56 ++- .../src/sqlite-session-turn-positions.ts | 358 ++++++++++++++++++ 6 files changed, 792 insertions(+), 5 deletions(-) create mode 100644 packages/storage/src/sqlite-session-turn-positions.ts diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 8aea6a22be..b76d21891e 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -501,6 +501,287 @@ describe('SqliteSessionMetadataStore', () => { } }); + test('indexes one durable position per user Turn without decoding transcript pages', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-turn-positions' })); + await store.appendMessages( + 'session-turn-positions', + [ + { + type: 'user', + id: 'message-turn-1-root', + turnId: 'turn-1', + ts: 10, + text: 'first prompt', + }, + { + type: 'assistant', + id: 'message-turn-1-assistant', + turnId: 'turn-1', + ts: 11, + text: 'first answer', + modelId: 'fake-model', + }, + { + type: 'user', + id: 'message-turn-1-steering', + turnId: 'turn-1', + ts: 12, + text: 'steering', + }, + { + type: 'user', + id: 'message-turn-2-root', + turnId: 'turn-2', + ts: 20, + text: 'second prompt', + }, + ], + { lastMessageAt: 20, lastMessagePreview: 'second prompt' }, + ); + + assert.deepEqual( + await store.readTurnPositions('session-turn-positions', { + direction: 'newer', + throughSequence: 3, + anchorSequence: null, + maxPositions: 128, + }), + { + kind: 'page', + throughSequence: 3, + revision: 0, + positions: [ + { turnId: 'turn-1', firstSequence: 0 }, + { turnId: 'turn-2', firstSequence: 3 }, + ], + hasOlder: false, + hasNewer: false, + }, + ); + } finally { + store.close(); + } + }); + + test('backfills at most 1,024 legacy records per step and resumes after reopen', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-turn-position-backfill-')); + const path = join(root, 'state.sqlite'); + try { + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(fullHeader({ id: 'session-position-backfill' })); + await setup.appendMessages( + 'session-position-backfill', + Array.from({ length: 1_025 }, (_, index) => ({ + type: 'user' as const, + id: `message-${index}`, + turnId: `turn-${index}`, + ts: index, + text: `prompt ${index}`, + })), + { lastMessageAt: 1_024, lastMessagePreview: 'prompt 1024' }, + ); + } finally { + setup.close(); + } + + const legacy = new DatabaseSync(path); + try { + legacy.exec(` + DELETE FROM session_turn_positions; + UPDATE session_turn_position_state SET built_through_sequence = NULL; + `); + } finally { + legacy.close(); + } + + const firstPass = createSqliteSessionMetadataStore(path); + try { + assert.deepEqual( + await firstPass.readTurnPositions('session-position-backfill', { + direction: 'older', + throughSequence: 1_024, + anchorSequence: null, + maxPositions: 2, + }), + { + kind: 'building', + throughSequence: 1_024, + indexedThroughSequence: 1_023, + }, + ); + } finally { + firstPass.close(); + } + + const resumed = createSqliteSessionMetadataStore(path); + try { + assert.deepEqual( + await resumed.readTurnPositions('session-position-backfill', { + direction: 'older', + throughSequence: 1_024, + anchorSequence: null, + maxPositions: 2, + }), + { + kind: 'page', + throughSequence: 1_024, + revision: 0, + positions: [ + { turnId: 'turn-1023', firstSequence: 1_023 }, + { turnId: 'turn-1024', firstSequence: 1_024 }, + ], + hasOlder: true, + hasNewer: false, + }, + ); + } finally { + resumed.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('lets one oversized legacy record advance the position build watermark alone', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-turn-position-oversized-')); + const path = join(root, 'state.sqlite'); + try { + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(fullHeader({ id: 'session-position-oversized' })); + await setup.appendMessages( + 'session-position-oversized', + [ + { + type: 'user', + id: 'message-oversized', + turnId: 'turn-oversized', + ts: 1, + text: 'x'.repeat(4 * 1024 * 1024 + 1), + }, + { + type: 'user', + id: 'message-after-oversized', + turnId: 'turn-after-oversized', + ts: 2, + text: 'after', + }, + ], + { lastMessageAt: 2, lastMessagePreview: 'after' }, + ); + } finally { + setup.close(); + } + + const legacy = new DatabaseSync(path); + try { + legacy.exec(` + DELETE FROM session_turn_positions; + UPDATE session_turn_position_state SET built_through_sequence = NULL; + `); + } finally { + legacy.close(); + } + + const store = createSqliteSessionMetadataStore(path); + try { + assert.deepEqual( + await store.readTurnPositions('session-position-oversized', { + direction: 'newer', + throughSequence: 1, + anchorSequence: null, + maxPositions: 128, + }), + { + kind: 'building', + throughSequence: 1, + indexedThroughSequence: 0, + }, + ); + const completed = await store.readTurnPositions('session-position-oversized', { + direction: 'newer', + throughSequence: 1, + anchorSequence: null, + maxPositions: 128, + }); + assert.equal(completed.kind, 'page'); + assert.deepEqual(completed.kind === 'page' ? completed.positions : [], [ + { turnId: 'turn-oversized', firstSequence: 0 }, + { turnId: 'turn-after-oversized', firstSequence: 1 }, + ]); + } finally { + store.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('lazily indexes a v36 transcript record above the legacy 16 KiB bootstrap size', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-turn-position-v36-')); + const path = join(root, 'state.sqlite'); + try { + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(fullHeader({ id: 'session-position-v36' })); + await setup.appendMessages( + 'session-position-v36', + [ + { + type: 'user', + id: 'message-v36', + turnId: 'turn-v36', + ts: 1, + text: 'v'.repeat(32 * 1024), + }, + ], + { lastMessageAt: 1, lastMessagePreview: 'legacy prompt' }, + ); + } finally { + setup.close(); + } + + const legacy = new DatabaseSync(path); + try { + legacy.exec(` + DROP TABLE session_turn_position_state; + DROP TABLE session_turn_positions; + UPDATE session_metadata_schema SET version = 36 WHERE scope = 'session_metadata'; + `); + } finally { + legacy.close(); + } + + const migrated = createSqliteSessionMetadataStore(path); + try { + assert.equal(migrated.schemaVersion(), SQLITE_SESSION_METADATA_SCHEMA_VERSION); + assert.deepEqual( + await migrated.readTurnPositions('session-position-v36', { + direction: 'newer', + throughSequence: 0, + anchorSequence: null, + maxPositions: 128, + }), + { + kind: 'page', + throughSequence: 0, + revision: 0, + positions: [{ turnId: 'turn-v36', firstSequence: 0 }], + hasOlder: false, + hasNewer: false, + }, + ); + } finally { + migrated.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + test('materializes a proven Root message when its admission is absent', async () => { const store = createSqliteSessionMetadataStore(':memory:'); try { @@ -573,6 +854,18 @@ describe('SqliteSessionMetadataStore', () => { { lastMessageAt: 30, lastMessagePreview: 'newest preview' }, ); + assert.equal( + ( + await store.readTurnPositions('session-legacy-order', { + direction: 'newer', + throughSequence: 2, + anchorSequence: null, + maxPositions: 128, + }) + ).kind, + 'page', + ); + await markMessagesHandedOffWithProvenRoots(store, { sessionId: 'session-legacy-order', messageIds: ['message-legacy-followup', 'message-legacy-steering'], @@ -614,6 +907,25 @@ describe('SqliteSessionMetadataStore', () => { (await store.readCatalogRecord('session-legacy-order')).lastMessagePreview, 'newest preview', ); + assert.deepEqual( + await store.readTurnPositions('session-legacy-order', { + direction: 'newer', + throughSequence: 4, + anchorSequence: null, + maxPositions: 128, + }), + { + kind: 'page', + throughSequence: 4, + revision: 1, + positions: [ + { turnId: 'turn-legacy-order', firstSequence: 1 }, + { turnId: 'turn-newer', firstSequence: 4 }, + ], + hasOlder: false, + hasNewer: false, + }, + ); const audit = new DatabaseSync(path, { readOnly: true }); try { assert.deepEqual(audit.prepare('PRAGMA foreign_key_check').all(), []); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index dc62748e68..fc138ede1c 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -136,6 +136,9 @@ export type { SessionTranscriptRecordScanRequest, SessionTranscriptStoragePage, SessionTranscriptStorageFragment, + SessionTurnPosition, + SessionTurnPositionPageRequest, + SessionTurnPositionPageResult, } from './session-store.js'; export type ExecutionSessionWriter = SessionAuthorityStore; @@ -410,6 +413,8 @@ async function createExecutionStoresForWrite sessionStore.readTranscriptMessagesSnapshot(sessionId, request)), readTranscriptHighWaterSnapshot: (sessionId) => run(() => sessionStore.readTranscriptHighWaterSnapshot(sessionId)), + readTurnPositionsSnapshot: (sessionId, request) => + run(() => sessionStore.readTurnPositionsSnapshot(sessionId, request)), readTurnContributionsSnapshot: (sessionId, throughSequence, position, maxContributions) => run(() => sessionStore.readTurnContributionsSnapshot( diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 1831eb21b9..c886a0a56e 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -306,6 +306,34 @@ export interface SessionTurnLandmarkSnapshot { readonly landmarks: readonly SessionTurnLandmark[]; } +export interface SessionTurnPosition { + readonly turnId: string; + readonly firstSequence: number; +} + +export interface SessionTurnPositionPageRequest { + readonly direction: 'older' | 'newer'; + readonly throughSequence: number | null; + /** Exclusive sequence boundary. */ + readonly anchorSequence: number | null; + readonly maxPositions: number; +} + +export type SessionTurnPositionPageResult = + | { + readonly kind: 'page'; + readonly throughSequence: number | null; + readonly revision: number; + readonly positions: readonly SessionTurnPosition[]; + readonly hasOlder: boolean; + readonly hasNewer: boolean; + } + | { + readonly kind: 'building'; + readonly throughSequence: number; + readonly indexedThroughSequence: number | null; + }; + export interface SessionStore { create(input: CreateSessionInput, initialBoundary?: ExecutionBoundary): Promise; list(filter?: SessionListFilter): Promise; @@ -360,6 +388,10 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto sessionId: string, request: SessionTranscriptMessageLookupRequest, ): Promise; + readTurnPositionsSnapshot( + sessionId: string, + request: SessionTurnPositionPageRequest, + ): Promise; /** Observe successful durable ledger appends. Listeners must not throw. */ subscribeTranscriptChanges(listener: (sessionId: string) => void): () => void; /** Wait until the SQLite authority is ready for cross-domain transactions. */ @@ -980,6 +1012,14 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.metadata.readTranscriptHighWater(sessionId); } + async readTurnPositionsSnapshot( + sessionId: string, + request: SessionTurnPositionPageRequest, + ): Promise { + await this.ensureReady(); + return this.metadata.readTurnPositions(sessionId, request); + } + async readTurnContributionsSnapshot( sessionId: string, throughSequence: number | null, diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index b4e3c037a5..0e83a1dee4 100644 --- a/packages/storage/src/sqlite-session-metadata-schema.ts +++ b/packages/storage/src/sqlite-session-metadata-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 36; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 37; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1242,6 +1242,30 @@ const MIGRATIONS: ReadonlyMap = new Map([ SELECT 1; `, ], + [ + 37, + ` + CREATE TABLE IF NOT EXISTS session_turn_positions ( + session_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + first_sequence INTEGER NOT NULL CHECK (first_sequence >= 0), + PRIMARY KEY(session_id, turn_id), + FOREIGN KEY(session_id) REFERENCES session_metadata(session_id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS session_turn_positions_by_sequence + ON session_turn_positions(session_id, first_sequence, turn_id); + + CREATE TABLE IF NOT EXISTS session_turn_position_state ( + session_id TEXT PRIMARY KEY, + built_through_sequence INTEGER CHECK ( + built_through_sequence IS NULL OR built_through_sequence >= 0 + ), + structural_revision INTEGER NOT NULL CHECK (structural_revision >= 0), + FOREIGN KEY(session_id) REFERENCES session_metadata(session_id) ON DELETE CASCADE + ); + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 770147b5d3..c16b4e056b 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -144,6 +144,8 @@ import { type SessionTurnContribution, type SessionTurnContributionPage, type SessionTurnLandmarkSnapshot, + type SessionTurnPositionPageRequest, + type SessionTurnPositionPageResult, } from './session-store.js'; import { isDiscardableConversationCopy, @@ -167,12 +169,18 @@ import { sqliteOrdinarySessionRolePredicate, sqliteRecoverableSessionRolePredicate, } from './sqlite-session-role-scope.js'; +import { + initializeTurnPositionState, + readTurnPositions as readSqliteTurnPositions, + recordTurnPositions, + shiftTurnPositions, + SQLITE_TURN_POSITION_MAX_SOURCE_BYTES, + SQLITE_TURN_POSITION_MAX_SOURCE_MESSAGES, +} from './sqlite-session-turn-positions.js'; export { SQLITE_SESSION_METADATA_SCHEMA_VERSION } from './sqlite-session-metadata-schema.js'; const SQLITE_TRANSCRIPT_MESSAGE_LOOKUP_BATCH_SIZE = 256; -const SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_MESSAGES = 1_024; -const SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_BYTES = 4 * 1024 * 1024; const SQLITE_TURN_LANDMARK_LEGACY_NEIGHBOR_MESSAGES = 32; function decodeStoredMessage(value: unknown): StoredMessage { @@ -2763,6 +2771,42 @@ export class SqliteSessionMetadataStore { return nullableStoredMessageSequence(row.high_water, sessionId); } + async readTurnPositions( + sessionId: string, + request: SessionTurnPositionPageRequest, + ): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + if ( + (request.direction !== 'older' && request.direction !== 'newer') || + (request.throughSequence !== null && + (!Number.isSafeInteger(request.throughSequence) || request.throughSequence < 0)) || + (request.anchorSequence !== null && + (!Number.isSafeInteger(request.anchorSequence) || request.anchorSequence < 0)) || + !Number.isSafeInteger(request.maxPositions) || + request.maxPositions < 1 || + request.maxPositions > 128 + ) { + throw new Error('Invalid Session Turn position request'); + } + return this.transaction(() => { + if (!this.readRecordSync(sessionId)) throw new SessionNotFoundError(sessionId); + return readSqliteTurnPositions(this.db, sessionId, request, (sequences) => { + const messages = new Map(); + for (const row of readStoredMessageRows(this.db, sessionId, sequences)) { + try { + messages.set(row.sequence, decodeStoredMessage(JSON.parse(row.recordJson) as unknown)); + } catch (error) { + throw new StoredSessionMessageIncompatibleError(sessionId, row.sequence, { + cause: error, + }); + } + } + return messages; + }); + }); + } + async readTurnContributions( sessionId: string, throughSequence: number | null, @@ -2823,8 +2867,8 @@ export class SqliteSessionMetadataStore { const recordBytes = storedMessageRecordBytes(row, sessionId, sequence); if ( sourceMessages > 0 && - (sourceMessages >= SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_MESSAGES || - sourceBytes + recordBytes > SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_BYTES) + (sourceMessages >= SQLITE_TURN_POSITION_MAX_SOURCE_MESSAGES || + sourceBytes + recordBytes > SQLITE_TURN_POSITION_MAX_SOURCE_BYTES) ) { return { throughSequence: fixedThrough, @@ -4753,6 +4797,7 @@ export class SqliteSessionMetadataStore { committedAt, ); if (result.changes !== 1) return undefined; + initializeTurnPositionState(this.db, header.id); this.options.failpoint?.('after_session_row_write'); this.ensureGenesisExecutionBoundary(header, initialBoundary); return { header, metadataVersion, committedAt }; @@ -5243,6 +5288,8 @@ export class SqliteSessionMetadataStore { } if (sequences.length === 0) return; + shiftTurnPositions(this.db, sessionId, firstSequence, amount); + this.db.exec('PRAGMA defer_foreign_keys = ON'); const moveChunks = this.db.prepare( 'UPDATE session_message_chunks SET sequence = ? WHERE session_id = ? AND sequence = ?', @@ -5330,6 +5377,7 @@ export class SqliteSessionMetadataStore { ); } } + recordTurnPositions(this.db, sessionId, firstSequence, entries); } private replaceSessionMessageSync( diff --git a/packages/storage/src/sqlite-session-turn-positions.ts b/packages/storage/src/sqlite-session-turn-positions.ts new file mode 100644 index 0000000000..47c774dcde --- /dev/null +++ b/packages/storage/src/sqlite-session-turn-positions.ts @@ -0,0 +1,358 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { DatabaseSync } from 'node:sqlite'; +import type { StoredMessage } from '@maka/core/session'; +import type { + SessionTurnPositionPageRequest, + SessionTurnPositionPageResult, +} from './session-store.js'; + +export const SQLITE_TURN_POSITION_MAX_SOURCE_MESSAGES = 1_024; +export const SQLITE_TURN_POSITION_MAX_SOURCE_BYTES = 4 * 1024 * 1024; + +interface PositionStateRow { + readonly built_through_sequence?: unknown; + readonly structural_revision?: unknown; +} + +export interface TurnPositionSourceRow { + readonly sequence?: unknown; + readonly message_type?: unknown; + readonly stored_bytes?: unknown; + readonly admission_turn_id?: unknown; +} + +export function initializeTurnPositionState(db: DatabaseSync, sessionId: string): void { + db.prepare( + `INSERT INTO session_turn_position_state( + session_id, built_through_sequence, structural_revision + ) VALUES (?, NULL, 0) + ON CONFLICT(session_id) DO NOTHING`, + ).run(sessionId); +} + +export function recordTurnPositions( + db: DatabaseSync, + sessionId: string, + firstSequence: number, + entries: readonly { readonly message: StoredMessage }[], +): void { + initializeTurnPositionState(db, sessionId); + const upsert = db.prepare(` + INSERT INTO session_turn_positions(session_id, turn_id, first_sequence) + VALUES (?, ?, ?) + ON CONFLICT(session_id, turn_id) DO UPDATE SET + first_sequence = MIN(first_sequence, excluded.first_sequence) + `); + entries.forEach(({ message }, index) => { + if (message.type === 'user' && message.turnId.length > 0) { + upsert.run(sessionId, message.turnId, firstSequence + index); + } + }); + + if (entries.length === 0) return; + const state = readState(db, sessionId); + if ( + (state.builtThroughSequence === null && firstSequence === 0) || + state.builtThroughSequence === firstSequence - 1 + ) { + db.prepare( + `UPDATE session_turn_position_state + SET built_through_sequence = ? WHERE session_id = ?`, + ).run(firstSequence + entries.length - 1, sessionId); + } +} + +export function shiftTurnPositions( + db: DatabaseSync, + sessionId: string, + firstSequence: number, + amount: number, +): void { + initializeTurnPositionState(db, sessionId); + const positions = db + .prepare( + `SELECT turn_id, first_sequence FROM session_turn_positions + WHERE session_id = ? AND first_sequence >= ? + ORDER BY first_sequence DESC, turn_id DESC`, + ) + .all(sessionId, firstSequence) as Array<{ + readonly turn_id?: unknown; + readonly first_sequence?: unknown; + }>; + const move = db.prepare( + `UPDATE session_turn_positions SET first_sequence = ? + WHERE session_id = ? AND turn_id = ? AND first_sequence = ?`, + ); + for (const row of positions) { + if (typeof row.turn_id !== 'string' || !isCount(row.first_sequence)) { + throw new Error(`Invalid Session Turn position for ${sessionId}`); + } + move.run(row.first_sequence + amount, sessionId, row.turn_id, row.first_sequence); + } + const state = readState(db, sessionId); + if (state.structuralRevision === Number.MAX_SAFE_INTEGER) { + throw new Error(`Session Turn position revision overflow for ${sessionId}`); + } + db.prepare( + `UPDATE session_turn_position_state + SET built_through_sequence = CASE + WHEN built_through_sequence IS NOT NULL AND built_through_sequence >= ? + THEN built_through_sequence + ? + ELSE built_through_sequence + END, + structural_revision = structural_revision + 1 + WHERE session_id = ?`, + ).run(firstSequence, amount, sessionId); +} + +export function readTurnPositions( + db: DatabaseSync, + sessionId: string, + request: SessionTurnPositionPageRequest, + readMessages: (sequences: readonly number[]) => ReadonlyMap, +): SessionTurnPositionPageResult { + initializeTurnPositionState(db, sessionId); + const actualThrough = readHighWater(db, sessionId); + const throughSequence = request.throughSequence; + if (throughSequence === null) { + if (actualThrough !== null) { + throw new Error(`Session Turn position watermark is behind durable storage: ${sessionId}`); + } + const state = readState(db, sessionId); + return { + kind: 'page', + throughSequence: null, + revision: state.structuralRevision, + positions: [], + hasOlder: false, + hasNewer: false, + }; + } + if (actualThrough === null || throughSequence > actualThrough) { + throw new Error(`Session Turn position watermark is ahead of durable storage: ${sessionId}`); + } + + let state = readState(db, sessionId); + if (state.builtThroughSequence === null || state.builtThroughSequence < throughSequence) { + buildTurnPositionStep(db, sessionId, state.builtThroughSequence, throughSequence, readMessages); + state = readState(db, sessionId); + if (state.builtThroughSequence === null || state.builtThroughSequence < throughSequence) { + return { + kind: 'building', + throughSequence, + indexedThroughSequence: state.builtThroughSequence, + }; + } + } + + const comparison = request.direction === 'older' ? '<' : '>'; + const order = request.direction === 'older' ? 'DESC' : 'ASC'; + const boundary = + request.anchorSequence ?? (request.direction === 'older' ? throughSequence + 1 : -1); + const raw = db + .prepare( + `SELECT turn_id, first_sequence FROM session_turn_positions + WHERE session_id = ? AND first_sequence <= ? AND first_sequence ${comparison} ? + ORDER BY first_sequence ${order}, turn_id ${order} + LIMIT ?`, + ) + .all(sessionId, throughSequence, boundary, request.maxPositions + 1) as Array<{ + readonly turn_id?: unknown; + readonly first_sequence?: unknown; + }>; + const selected = raw.slice(0, request.maxPositions).map((row) => { + if (typeof row.turn_id !== 'string' || !isCount(row.first_sequence)) { + throw new Error(`Invalid Session Turn position for ${sessionId}`); + } + return { turnId: row.turn_id, firstSequence: row.first_sequence }; + }); + if (request.direction === 'older') selected.reverse(); + const first = selected[0]?.firstSequence; + const last = selected.at(-1)?.firstSequence; + return { + kind: 'page', + throughSequence, + revision: state.structuralRevision, + positions: selected, + hasOlder: first === undefined ? false : hasPosition(db, sessionId, throughSequence, '<', first), + hasNewer: last === undefined ? false : hasPosition(db, sessionId, throughSequence, '>', last), + }; +} + +function buildTurnPositionStep( + db: DatabaseSync, + sessionId: string, + builtThroughSequence: number | null, + throughSequence: number, + readMessages: (sequences: readonly number[]) => ReadonlyMap, +): void { + const position = builtThroughSequence === null ? 0 : builtThroughSequence + 1; + const admissionTable = db + .prepare(`SELECT 1 AS found FROM sqlite_schema WHERE type = 'table' AND name = ?`) + .get('core_root_turn_admissions'); + const admissionProjection = admissionTable + ? `( + SELECT admission.turn_id FROM core_root_turn_admissions AS admission + WHERE admission.session_id = message.session_id + AND json_extract(admission.record_json, '$.userMessageId') = message.message_id + LIMIT 1 + )` + : 'NULL'; + const rows = db + .prepare( + `SELECT message.sequence, message.message_type, + coalesce(payload.record_bytes, length(CAST(message.record_json AS BLOB))) AS stored_bytes, + ${admissionProjection} AS admission_turn_id + FROM session_messages AS message + LEFT JOIN session_message_payloads AS payload + ON payload.session_id = message.session_id AND payload.sequence = message.sequence + WHERE message.session_id = ? AND message.sequence >= ? AND message.sequence <= ? + ORDER BY message.sequence ASC + LIMIT ${SQLITE_TURN_POSITION_MAX_SOURCE_MESSAGES}`, + ) + .all(sessionId, position, throughSequence) as TurnPositionSourceRow[]; + if (rows.length === 0) { + throw new Error(`Session Turn position build did not advance: ${sessionId}`); + } + + const upsert = db.prepare(` + INSERT INTO session_turn_positions(session_id, turn_id, first_sequence) + VALUES (?, ?, ?) + ON CONFLICT(session_id, turn_id) DO UPDATE SET + first_sequence = MIN(first_sequence, excluded.first_sequence) + `); + let sourceBytes = 0; + let lastSequence: number | null = null; + const selected: Array<{ + readonly sequence: number; + readonly messageType: unknown; + readonly admissionTurnId: unknown; + }> = []; + for (const row of rows) { + const sequence = requireCount(row.sequence, `Session Turn position sequence for ${sessionId}`); + const recordBytes = requireCount( + row.stored_bytes, + `Session Turn position bytes for ${sessionId}`, + ); + if ( + lastSequence !== null && + sourceBytes + recordBytes > SQLITE_TURN_POSITION_MAX_SOURCE_BYTES + ) { + break; + } + sourceBytes += recordBytes; + selected.push({ + sequence, + messageType: row.message_type, + admissionTurnId: row.admission_turn_id, + }); + lastSequence = sequence; + } + const legacyUserSequences = selected.flatMap((row) => + row.messageType === 'user' && typeof row.admissionTurnId !== 'string' ? [row.sequence] : [], + ); + const legacyMessages = readMessages(legacyUserSequences); + for (const row of selected) { + if (row.messageType === 'user') { + const turnId = + typeof row.admissionTurnId === 'string' + ? row.admissionTurnId + : (() => { + const message = legacyMessages.get(row.sequence); + if (message?.type !== 'user') { + throw new Error(`Session Turn position identity changed for ${sessionId}`); + } + return message.turnId; + })(); + if (turnId.length > 0) upsert.run(sessionId, turnId, row.sequence); + } + } + if (lastSequence === null) { + throw new Error(`Session Turn position build did not advance: ${sessionId}`); + } + db.prepare( + `UPDATE session_turn_position_state SET built_through_sequence = ? WHERE session_id = ?`, + ).run(lastSequence, sessionId); +} + +function readState( + db: DatabaseSync, + sessionId: string, +): { readonly builtThroughSequence: number | null; readonly structuralRevision: number } { + const row = db + .prepare( + `SELECT built_through_sequence, structural_revision + FROM session_turn_position_state WHERE session_id = ?`, + ) + .get(sessionId) as PositionStateRow | undefined; + if (!row) throw new Error(`Missing Session Turn position state for ${sessionId}`); + const builtThroughSequence = + row.built_through_sequence === null + ? null + : requireCount( + row.built_through_sequence, + `Session Turn position watermark for ${sessionId}`, + ); + return { + builtThroughSequence, + structuralRevision: requireCount( + row.structural_revision, + `Session Turn position revision for ${sessionId}`, + ), + }; +} + +function readHighWater(db: DatabaseSync, sessionId: string): number | null { + const row = db + .prepare('SELECT MAX(sequence) AS high_water FROM session_messages WHERE session_id = ?') + .get(sessionId) as { readonly high_water?: unknown }; + return row.high_water === null + ? null + : requireCount(row.high_water, `Session transcript watermark for ${sessionId}`); +} + +function hasPosition( + db: DatabaseSync, + sessionId: string, + throughSequence: number, + comparison: '<' | '>', + boundary: number, +): boolean { + return Boolean( + db + .prepare( + `SELECT 1 AS found FROM session_turn_positions + WHERE session_id = ? AND first_sequence <= ? AND first_sequence ${comparison} ? LIMIT 1`, + ) + .get(sessionId, throughSequence, boundary), + ); +} + +function requireCount(value: unknown, label: string): number { + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + throw new Error(`Invalid ${label}`); + } + return value; +} + +function isCount(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} From 8fce25c28506ea019ab6631c6b2d9789eb4107c6 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:22:31 +0800 Subject: [PATCH 02/14] feat(runtime-host): page transcript turn positions Generated-by: Codex --- .../runtime-host-session-driver.test.ts | 4 + .../src/__tests__/connection-session.test.ts | 14 ++ .../fixtures/session-transcript-reader.ts | 33 +++ .../session-transcript-pager.test.ts | 198 +++++++++++++++ .../session-transcript-protocol.test.ts | 67 +++++ .../runtime-host/src/client/connection.ts | 1 + .../src/client/session-subscription.ts | 52 ++++ packages/runtime-host/src/protocol/index.ts | 4 +- .../runtime-host/src/protocol/operations.ts | 1 + .../src/protocol/session-transcript.ts | 236 ++++++++++++++++++ .../src/server/connection-session.ts | 11 +- .../server/session-continuity-coordinator.ts | 69 +++++ .../src/server/session-transcript-pager.ts | 29 +-- .../session-transcript-position-pager.ts | 171 +++++++++++++ .../src/server/session-transcript-reader.ts | 8 + .../src/server/transcript-signed-cursor.ts | 57 +++++ 16 files changed, 929 insertions(+), 26 deletions(-) create mode 100644 packages/runtime-host/src/server/session-transcript-position-pager.ts create mode 100644 packages/runtime-host/src/server/transcript-signed-cursor.ts diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index a925f859b9..22b4a88c5c 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -2818,6 +2818,10 @@ class FakeSubscription implements RuntimeHostSessionSubscription, AsyncIterator< throw new Error('Fake subscription does not expose transcript pages'); } + async loadTranscriptPositionsPage(): Promise { + throw new Error('Fake subscription does not expose transcript positions'); + } + async close(): Promise { this.#closed = true; for (const waiter of this.#waiters.splice(0)) { diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index ff6b96769b..68a60e55cb 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -461,6 +461,20 @@ test('flushes concurrent subscription opens before activating their live frame s ok: false, error: { code: 'operation_unavailable', message: 'not used' }, }), + 'session.transcript.positions.page': async (input) => ({ + ok: true, + result: { + kind: 'page', + sessionId: 'session-1', + direction: input.direction, + throughSequence: input.throughSequence, + revision: input.revision ?? 0, + positions: [], + hasOlder: false, + hasNewer: false, + nextCursor: null, + }, + }), }, attachConnection: (_connectionId, attachedSink) => { sink = attachedSink; diff --git a/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts b/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts index 8d2dbecb96..e9cc3f1cc5 100644 --- a/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts +++ b/packages/runtime-host/src/__tests__/fixtures/session-transcript-reader.ts @@ -142,6 +142,39 @@ export function transcriptReader( (message, sequence) => sequence <= request.throughSequence! && request.messageIds.includes(message.id), ), + readDurableTurnPositions: async (_sessionId, request) => { + const positions = durable.flatMap((message, firstSequence) => + message.type === 'user' && + (request.throughSequence === null || firstSequence <= request.throughSequence) && + !durable + .slice(0, firstSequence) + .some((candidate) => candidate.type === 'user' && candidate.turnId === message.turnId) + ? [{ turnId: message.turnId, firstSequence }] + : [], + ); + const boundary = + request.anchorSequence ?? + (request.direction === 'older' ? (request.throughSequence ?? -1) + 1 : -1); + const candidates = positions.filter(({ firstSequence }) => + request.direction === 'older' ? firstSequence < boundary : firstSequence > boundary, + ); + const selected = + request.direction === 'older' + ? candidates.slice(-request.maxPositions) + : candidates.slice(0, request.maxPositions); + return { + kind: 'page' as const, + throughSequence: request.throughSequence, + revision: 0, + positions: selected, + hasOlder: + selected.length > 0 && + positions.some(({ firstSequence }) => firstSequence < selected[0]!.firstSequence), + hasNewer: + selected.length > 0 && + positions.some(({ firstSequence }) => firstSequence > selected.at(-1)!.firstSequence), + }; + }, readActiveOverlay: async () => overlay, }; } diff --git a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts index 564dbfd27e..a8f8b0a13e 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts @@ -33,10 +33,208 @@ import { TranscriptPageRequestError, updateSubscriberTranscriptHighWater, } from '../server/session-transcript-pager.js'; +import { readSessionTranscriptPositionsPage } from '../server/session-transcript-position-pager.js'; import type { SessionTranscriptReader } from '../server/session-transcript-reader.js'; import { projectSharedSessionTranscriptMessage } from '../server/shared-session-transcript.js'; import { transcriptReader } from './fixtures/session-transcript-reader.js'; +test('pages owner Turn positions with a subscription-bound stateless cursor', async () => { + const reader = transcriptReader(Array.from({ length: 4 }, (_, index) => userMessage(index))); + const { state } = await createSessionTranscriptBootstrap({ + reader, + sessionId: 'session-1', + subscriptionId: 'subscription-1', + throughSequence: 3, + rootTurn: null, + activeAssistantStreams: [], + maxBytes: 16 * 1024, + projection: 'owner', + }); + const first = await readSessionTranscriptPositionsPage({ + reader, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'newer', + throughSequence: 3, + revision: null, + cursor: null, + anchorSequence: null, + maxPositions: 2, + }, + }); + assert.equal(first.kind, 'page'); + if (first.kind !== 'page') return; + assert.deepEqual(first.positions, [ + { turnId: 'turn-0', firstSequence: 0 }, + { turnId: 'turn-1', firstSequence: 1 }, + ]); + assert.ok(first.nextCursor); + + const second = await readSessionTranscriptPositionsPage({ + reader, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'newer', + throughSequence: 3, + revision: null, + cursor: first.nextCursor, + anchorSequence: null, + maxPositions: 2, + }, + }); + assert.equal(second.kind, 'page'); + if (second.kind !== 'page') return; + assert.deepEqual(second.positions, [ + { turnId: 'turn-2', firstSequence: 2 }, + { turnId: 'turn-3', firstSequence: 3 }, + ]); + assert.equal(second.nextCursor, null); + const tampered = `${first.nextCursor!.slice(0, -1)}${first.nextCursor!.endsWith('A') ? 'B' : 'A'}`; + await assert.rejects( + readSessionTranscriptPositionsPage({ + reader, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'newer', + throughSequence: 3, + revision: null, + cursor: tampered, + anchorSequence: null, + maxPositions: 2, + }, + }), + TranscriptPageRequestError, + ); + await assert.rejects( + readSessionTranscriptPositionsPage({ + reader, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'older', + throughSequence: 3, + revision: null, + cursor: first.nextCursor, + anchorSequence: null, + maxPositions: 2, + }, + }), + TranscriptPageRequestError, + ); + await assert.rejects( + readSessionTranscriptPositionsPage({ + reader, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'newer', + throughSequence: 2, + revision: null, + cursor: first.nextCursor, + anchorSequence: null, + maxPositions: 2, + }, + }), + TranscriptPageRequestError, + ); + + const { state: otherState } = await createSessionTranscriptBootstrap({ + reader, + sessionId: 'session-1', + subscriptionId: 'subscription-2', + throughSequence: 3, + rootTurn: null, + activeAssistantStreams: [], + maxBytes: 16 * 1024, + projection: 'owner', + }); + await assert.rejects( + readSessionTranscriptPositionsPage({ + reader, + state: otherState, + request: { + subscriptionId: 'subscription-2', + direction: 'newer', + throughSequence: 3, + revision: null, + cursor: first.nextCursor, + anchorSequence: null, + maxPositions: 2, + }, + }), + TranscriptPageRequestError, + ); +}); + +test('reports position build progress and structural revision changes', async () => { + const base = transcriptReader([userMessage(0)]); + const { state } = await createSessionTranscriptBootstrap({ + reader: base, + sessionId: 'session-1', + subscriptionId: 'subscription-1', + throughSequence: 0, + rootTurn: null, + activeAssistantStreams: [], + maxBytes: 16 * 1024, + projection: 'owner', + }); + const building = await readSessionTranscriptPositionsPage({ + reader: { + ...base, + readDurableTurnPositions: async () => ({ + kind: 'building', + throughSequence: 0, + indexedThroughSequence: null, + }), + }, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'older', + throughSequence: 0, + revision: null, + cursor: null, + anchorSequence: null, + maxPositions: 128, + }, + }); + assert.deepEqual(building, { + kind: 'building', + sessionId: 'session-1', + throughSequence: 0, + indexedThroughSequence: null, + retryAfterMs: 25, + }); + + const stale = await readSessionTranscriptPositionsPage({ + reader: { + ...base, + readDurableTurnPositions: async () => ({ + kind: 'page', + throughSequence: 0, + revision: 2, + positions: [{ turnId: 'turn-0', firstSequence: 0 }], + hasOlder: false, + hasNewer: false, + }), + }, + state, + request: { + subscriptionId: 'subscription-1', + direction: 'older', + throughSequence: 0, + revision: 1, + cursor: null, + anchorSequence: null, + maxPositions: 128, + }, + }); + assert.deepEqual(stale, { kind: 'stale', sessionId: 'session-1', currentRevision: 2 }); +}); + test('reads newly durable messages forward from an announced watermark', async () => { const durable = [userMessage(0), userMessage(1)]; const reader = transcriptReader(durable); diff --git a/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts b/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts index da33e1ce1f..02b978959b 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-protocol.test.ts @@ -24,6 +24,8 @@ import { decodeSessionTranscriptBootstrap, decodeSessionTranscriptPage, decodeSessionTranscriptPageInput, + decodeSessionTranscriptPositionsPageInput, + decodeSessionTranscriptPositionsPageResult, encodeProtocolMessage, HOST_OPERATION_SPECS, RUNTIME_HOST_MAX_MESSAGE_BYTES, @@ -104,6 +106,71 @@ test('Session transcript protocol accepts bounded correlated pages and bootstrap ); }); +test('Session transcript position protocol accepts bounded pages and build progress', () => { + const positionsInput = { + subscriptionId: 'subscription-1', + direction: 'newer' as const, + throughSequence: 20, + revision: null, + cursor: null, + anchorSequence: 4, + maxPositions: 128, + }; + const positionsPage = { + kind: 'page' as const, + sessionId: 'session-1', + direction: 'newer' as const, + throughSequence: 20, + revision: 3, + positions: [ + { turnId: 'turn-2', firstSequence: 8 }, + { turnId: 'turn-3', firstSequence: 15 }, + ], + hasOlder: true, + hasNewer: false, + nextCursor: null, + }; + assert.deepEqual(decodeSessionTranscriptPositionsPageInput(positionsInput), positionsInput); + assert.deepEqual(decodeSessionTranscriptPositionsPageResult(positionsPage), positionsPage); + assert.deepEqual( + decodeSessionTranscriptPositionsPageResult({ + kind: 'building', + sessionId: 'session-1', + throughSequence: 20, + indexedThroughSequence: 7, + retryAfterMs: 25, + }), + { + kind: 'building', + sessionId: 'session-1', + throughSequence: 20, + indexedThroughSequence: 7, + retryAfterMs: 25, + }, + ); + assert.throws( + () => + decodeSessionTranscriptPositionsPageInput({ + ...positionsInput, + cursor: 'continuation', + anchorSequence: null, + revision: 3, + }), + isProtocolError, + ); + assert.throws( + () => + decodeSessionTranscriptPositionsPageResult({ + ...positionsPage, + positions: Array.from({ length: 129 }, (_, index) => ({ + turnId: `turn-${index}`, + firstSequence: index, + })), + }), + isProtocolError, + ); +}); + test('a maximum single-fragment continuation remains transport safe', () => { const data = Buffer.alloc(SESSION_TRANSCRIPT_PAGE_MAX_BYTES, 0x61); const result = { diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index abbd00d4f5..4acfbf3cca 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -614,6 +614,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { throw error; } }, + (query) => this.request('session.transcript.positions.page', query, timeoutMs), ); this.#subscriptions.set(result.subscriptionId, subscription); return subscription; diff --git a/packages/runtime-host/src/client/session-subscription.ts b/packages/runtime-host/src/client/session-subscription.ts index a458dc9743..fbd054975c 100644 --- a/packages/runtime-host/src/client/session-subscription.ts +++ b/packages/runtime-host/src/client/session-subscription.ts @@ -31,6 +31,8 @@ import { type SessionTranscriptFragment, type SessionTranscriptPage, type SessionTranscriptPageInput, + type SessionTranscriptPositionsPageInput, + type SessionTranscriptPositionsPageResult, } from '../protocol/index.js'; const MAX_CLIENT_QUEUED_FRAMES = 32; @@ -81,6 +83,9 @@ export interface RuntimeHostSessionSubscription extends AsyncIterable, ): Promise; + loadTranscriptPositionsPage( + input: Omit, + ): Promise; close(): Promise; } @@ -109,6 +114,9 @@ export class ClientSessionSubscription readonly #readTranscriptPage: ( input: SessionTranscriptPageInput, ) => Promise; + readonly #readTranscriptPositionsPage: ( + input: SessionTranscriptPositionsPageInput, + ) => Promise; readonly #releaseTranscriptOverlay: () => Promise; readonly #expectedSessionId: string; readonly #queue: QueuedFrame[] = []; @@ -136,6 +144,14 @@ export class ClientSessionSubscription requestClose: () => Promise, readTranscriptPage: (input: SessionTranscriptPageInput) => Promise, releaseTranscriptOverlay: () => Promise = async () => undefined, + readTranscriptPositionsPage: ( + input: SessionTranscriptPositionsPageInput, + ) => Promise = async () => { + throw new RuntimeHostSubscriptionError( + 'correlation_changed', + 'Session transcript positions are unavailable', + ); + }, ) { this.hostEpoch = result.hostEpoch; this.subscriptionId = result.subscriptionId; @@ -148,6 +164,7 @@ export class ClientSessionSubscription this.#latestTranscriptThroughSequence = result.transcript?.throughSequence ?? null; this.#requestClose = requestClose; this.#readTranscriptPage = readTranscriptPage; + this.#readTranscriptPositionsPage = readTranscriptPositionsPage; this.#releaseTranscriptOverlay = releaseTranscriptOverlay; } @@ -334,6 +351,41 @@ export class ClientSessionSubscription }); } + loadTranscriptPositionsPage( + input: Omit, + ): Promise { + this.#assertTranscriptReadable(); + if ( + input.throughSequence !== null && + (this.#latestTranscriptThroughSequence === null || + input.throughSequence > this.#latestTranscriptThroughSequence) + ) { + return Promise.reject( + new RuntimeHostSubscriptionError( + 'correlation_changed', + 'Session transcript watermark has not been announced', + ), + ); + } + return this.#readTranscriptPositionsPage({ + subscriptionId: this.subscriptionId, + ...input, + }).then((result) => { + this.#assertTranscriptReadable(); + if ( + result.sessionId !== this.#expectedSessionId || + (result.kind !== 'stale' && result.throughSequence !== input.throughSequence) || + (result.kind === 'page' && result.direction !== input.direction) + ) { + throw new RuntimeHostSubscriptionError( + 'correlation_changed', + 'Session transcript position page changed identity', + ); + } + return result; + }); + } + async #loadTranscript(): Promise { this.#assertTranscriptReadable(); const bootstrap = this.transcriptBootstrap; diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 83883daec7..cb929d935e 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -100,7 +100,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 94 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 95 as const; +// 95: Owners can page bounded durable Turn positions through an existing +// transcript subscription without adding a second transcript body protocol. // 94: A failed Turn snapshot no longer carries contextBudgetExhaustedDetail; the // retired outcome reads as context_overflow at the ledger boundary, and an older // Host still sending the field fails a newer client's closed snapshot decode. diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 0a03ab137e..9a160bbee9 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -322,6 +322,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'session.remove.preview', 'session.revision.abandon', 'session.revision.create', + 'session.transcript.positions.page', 'session.transcript.page', 'session.transcript.overlay.release', 'session.turn_landmarks.query', diff --git a/packages/runtime-host/src/protocol/session-transcript.ts b/packages/runtime-host/src/protocol/session-transcript.ts index e50b0d03e9..412f291843 100644 --- a/packages/runtime-host/src/protocol/session-transcript.ts +++ b/packages/runtime-host/src/protocol/session-transcript.ts @@ -37,6 +37,9 @@ export const SESSION_TRANSCRIPT_RANGE_MAX_MESSAGES = SESSION_TRANSCRIPT_PAGE_MAX export const SESSION_TRANSCRIPT_OVERLAY_MAX_MESSAGES = 4_096; export const SESSION_TRANSCRIPT_PAGE_RESULT_MAX_BYTES = 744 * 1024; export const SESSION_TRANSCRIPT_CURSOR_MAX_BYTES = 1024; +export const SESSION_TRANSCRIPT_POSITION_MAX_ITEMS = 128; +export const SESSION_TRANSCRIPT_POSITION_RESULT_MAX_BYTES = 64 * 1024; +export const SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS = 25 as const; export type SessionTranscriptPageSource = 'durable' | 'overlay'; export type SessionTranscriptPageDirection = 'older' | 'newer'; @@ -100,6 +103,46 @@ export interface SessionTranscriptOverlayReleaseResult { readonly subscriptionId: string; } +export interface SessionTranscriptPositionsPageInput { + readonly subscriptionId: string; + readonly direction: SessionTranscriptPageDirection; + readonly throughSequence: number | null; + readonly revision: number | null; + readonly cursor: string | null; + readonly anchorSequence: number | null; + readonly maxPositions: number; +} + +export interface SessionTranscriptTurnPosition { + readonly turnId: string; + readonly firstSequence: number; +} + +export type SessionTranscriptPositionsPageResult = + | { + readonly kind: 'page'; + readonly sessionId: string; + readonly direction: SessionTranscriptPageDirection; + readonly throughSequence: number | null; + readonly revision: number; + readonly positions: readonly SessionTranscriptTurnPosition[]; + readonly hasOlder: boolean; + readonly hasNewer: boolean; + readonly nextCursor: string | null; + } + | { + readonly kind: 'building'; + readonly sessionId: string; + readonly throughSequence: number | null; + readonly indexedThroughSequence: number | null; + readonly retryAfterMs: typeof SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS; + } + | { + readonly kind: 'stale'; + readonly sessionId: string; + readonly currentRevision: number; + }; + const QUERY_ERRORS = [ 'host_not_ready', 'host_draining', @@ -112,6 +155,14 @@ const QUERY_ERRORS = [ ] as const; export const SESSION_TRANSCRIPT_OPERATION_SPECS = { + 'session.transcript.positions.page': defineOperation({ + mode: 'query', + availability: 'ready', + errors: QUERY_ERRORS, + decodeInput: decodeSessionTranscriptPositionsPageInput, + decodeOutput: decodeSessionTranscriptPositionsPageResult, + assertOutputForInput: assertSessionTranscriptPositionsPageOutput, + }), 'session.transcript.page': defineOperation({ mode: 'query', availability: 'ready', @@ -134,6 +185,186 @@ export const SESSION_TRANSCRIPT_OPERATION_SPECS = { }), } as const; +export function decodeSessionTranscriptPositionsPageInput( + value: unknown, +): SessionTranscriptPositionsPageInput { + const input = requireExactRecord(value, 'Session transcript positions page input', [ + 'subscriptionId', + 'direction', + 'throughSequence', + 'revision', + 'cursor', + 'anchorSequence', + 'maxPositions', + ]); + const cursor = + input.cursor === null + ? null + : requireUtf8String( + input.cursor, + 'Session transcript position cursor', + SESSION_TRANSCRIPT_CURSOR_MAX_BYTES, + ); + const revision = + input.revision === null + ? null + : requireCount(input.revision, 'Session transcript position revision'); + const anchorSequence = + input.anchorSequence === null + ? null + : requireCount(input.anchorSequence, 'Session transcript position anchor sequence'); + if (cursor !== null && (anchorSequence !== null || revision !== null)) { + throw invalidProtocolFrame( + 'Session transcript position cursor, anchor, and revision are mutually exclusive', + ); + } + const maxPositions = requireCount(input.maxPositions, 'Session transcript position page limit'); + if (maxPositions < 1 || maxPositions > SESSION_TRANSCRIPT_POSITION_MAX_ITEMS) { + throw invalidProtocolFrame('Invalid Session transcript position page limit'); + } + return { + subscriptionId: requireId(input.subscriptionId, 'subscriptionId'), + direction: decodeDirection(input.direction), + throughSequence: + input.throughSequence === null + ? null + : requireCount(input.throughSequence, 'Session transcript position watermark'), + revision, + cursor, + anchorSequence, + maxPositions, + }; +} + +export function decodeSessionTranscriptPositionsPageResult( + value: unknown, +): SessionTranscriptPositionsPageResult { + requireEncodedByteLimit( + value, + 'Session transcript positions page result', + SESSION_TRANSCRIPT_POSITION_RESULT_MAX_BYTES, + ); + const result = requireRecord(value, 'Session transcript positions page result'); + if (result.kind === 'building') { + const exact = requireExactRecord(result, 'Session transcript positions build result', [ + 'kind', + 'sessionId', + 'throughSequence', + 'indexedThroughSequence', + 'retryAfterMs', + ]); + if (exact.retryAfterMs !== SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS) { + throw invalidProtocolFrame('Invalid Session transcript position retry delay'); + } + return { + kind: 'building', + sessionId: requireEntityId(exact.sessionId, 'sessionId'), + throughSequence: + exact.throughSequence === null + ? null + : requireCount(exact.throughSequence, 'Session transcript position watermark'), + indexedThroughSequence: + exact.indexedThroughSequence === null + ? null + : requireCount( + exact.indexedThroughSequence, + 'Session transcript indexed position watermark', + ), + retryAfterMs: SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS, + }; + } + if (result.kind === 'stale') { + const exact = requireExactRecord(result, 'Session transcript positions stale result', [ + 'kind', + 'sessionId', + 'currentRevision', + ]); + return { + kind: 'stale', + sessionId: requireEntityId(exact.sessionId, 'sessionId'), + currentRevision: requireCount( + exact.currentRevision, + 'Session transcript current position revision', + ), + }; + } + const exact = requireExactRecord(result, 'Session transcript positions page result', [ + 'kind', + 'sessionId', + 'direction', + 'throughSequence', + 'revision', + 'positions', + 'hasOlder', + 'hasNewer', + 'nextCursor', + ]); + if (exact.kind !== 'page' || !Array.isArray(exact.positions)) { + throw invalidProtocolFrame('Invalid Session transcript position page kind'); + } + if (exact.positions.length > SESSION_TRANSCRIPT_POSITION_MAX_ITEMS) { + throw invalidProtocolFrame('Session transcript position page exceeds its item limit'); + } + const throughSequence = + exact.throughSequence === null + ? null + : requireCount(exact.throughSequence, 'Session transcript position watermark'); + const positions = exact.positions.map((value) => { + const position = requireExactRecord(value, 'Session transcript Turn position', [ + 'turnId', + 'firstSequence', + ]); + return { + turnId: requireEntityId(position.turnId, 'turnId'), + firstSequence: requireCount(position.firstSequence, 'Session transcript Turn first sequence'), + }; + }); + for (let index = 0; index < positions.length; index += 1) { + const position = positions[index]!; + if ( + (throughSequence !== null && position.firstSequence > throughSequence) || + (index > 0 && positions[index - 1]!.firstSequence >= position.firstSequence) + ) { + throw invalidProtocolFrame('Invalid Session transcript Turn position order'); + } + } + return { + kind: 'page', + sessionId: requireEntityId(exact.sessionId, 'sessionId'), + direction: decodeDirection(exact.direction), + throughSequence, + revision: requireCount(exact.revision, 'Session transcript position revision'), + positions, + hasOlder: requireBoolean(exact.hasOlder, 'Session transcript older position coverage'), + hasNewer: requireBoolean(exact.hasNewer, 'Session transcript newer position coverage'), + nextCursor: + exact.nextCursor === null + ? null + : requireUtf8String( + exact.nextCursor, + 'Session transcript position cursor', + SESSION_TRANSCRIPT_CURSOR_MAX_BYTES, + ), + }; +} + +function assertSessionTranscriptPositionsPageOutput( + input: SessionTranscriptPositionsPageInput, + output: SessionTranscriptPositionsPageResult, +): void { + if (output.kind === 'stale') return; + if (output.throughSequence !== input.throughSequence) { + throw invalidProtocolFrame('Session transcript position watermark changed'); + } + if ( + output.kind === 'page' && + (output.direction !== input.direction || + (input.revision !== null && output.revision !== input.revision)) + ) { + throw invalidProtocolFrame('Session transcript position page does not match request'); + } +} + function decodeSessionTranscriptOverlayReleaseInput( value: unknown, ): SessionTranscriptOverlayReleaseInput { @@ -446,6 +677,11 @@ function decodeDirection(value: unknown): SessionTranscriptPageDirection { return value; } +function requireBoolean(value: unknown, label: string): boolean { + if (typeof value !== 'boolean') throw invalidProtocolFrame(`Invalid ${label}`); + return value; +} + function requirePageByteLimit(value: unknown): number { const limit = requireCount(value, 'Session transcript page byte limit'); if (limit < 1 || limit > SESSION_TRANSCRIPT_PAGE_MAX_BYTES) { diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts index 9d5141a7ed..ff67e34d21 100644 --- a/packages/runtime-host/src/server/connection-session.ts +++ b/packages/runtime-host/src/server/connection-session.ts @@ -176,7 +176,8 @@ export class RuntimeHostConnectionSession { #dispatch(frame: RequestFrame): void { if (frame.operation === 'host.status') this.#inFlightStatusRequests += 1; const handling = - frame.operation === 'session.transcript.page' + frame.operation === 'session.transcript.page' || + frame.operation === 'session.transcript.positions.page' ? this.#transcriptPageTail.then(() => this.#handleRequest(frame)) : this.#handleRequest(frame); const task = handling @@ -188,7 +189,10 @@ export class RuntimeHostConnectionSession { } }); this.#requests.set(frame.requestId, task); - if (frame.operation === 'session.transcript.page') { + if ( + frame.operation === 'session.transcript.page' || + frame.operation === 'session.transcript.positions.page' + ) { this.#transcriptPageTail = task.catch(() => undefined); } } @@ -221,7 +225,8 @@ export class RuntimeHostConnectionSession { const continuity = frame.operation === 'subscription.open' || frame.operation === 'subscription.close' || - frame.operation === 'session.transcript.page' + frame.operation === 'session.transcript.page' || + frame.operation === 'session.transcript.positions.page' ? this.#ensureContinuity() : undefined; const response = await dispatchOperation(frame, this.#options.resolveHandlers(), { diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index 2596e1a8af..171f69ab34 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -46,6 +46,7 @@ import { type SessionToolEvent, type SessionTranscriptAdvancedFrame, type SessionTranscriptPageInput, + type SessionTranscriptPositionsPageInput, type OperationOutcome, type SubscriptionFrame, type SubscriptionOpenInput, @@ -77,6 +78,7 @@ import { TranscriptPageRequestError, updateSubscriberTranscriptHighWater, } from './session-transcript-pager.js'; +import { readSessionTranscriptPositionsPage } from './session-transcript-position-pager.js'; import { ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES, type SessionTranscriptReader, @@ -253,6 +255,8 @@ export class SessionContinuityCoordinator implements SessionContinuityService { }, 'session.transcript.page': (input, context) => this.#readTranscriptPage(context.connectionId, input), + 'session.transcript.positions.page': (input, context) => + this.#readTranscriptPositionsPage(context.connectionId, input), 'session.transcript.overlay.release': async (input, context) => { const existing = this.#subscriptions.get(input.subscriptionId); if (!existing) { @@ -1150,6 +1154,64 @@ export class SessionContinuityCoordinator implements SessionContinuityService { }); } + async #readTranscriptPositionsPage( + connectionId: string, + input: SessionTranscriptPositionsPageInput, + ): Promise> { + const subscriber = this.#ownedSubscriber(connectionId, input.subscriptionId); + if (!subscriber) return transcriptPositionsSubscriptionNotFound(); + if (!this.#transcriptReader || !subscriber.transcript) { + return { + ok: false, + error: { code: 'operation_unavailable', message: 'Session transcript is unavailable' }, + }; + } + if (subscriber.transcript.projection !== 'owner') { + return { + ok: false, + error: { code: 'operation_unavailable', message: 'Transcript positions require an owner' }, + }; + } + const connection = this.#connections.get(connectionId); + if (!connection || !this.#canObserve(subscriber, subscriber.sessionId)) { + this.#closeSubscriber(subscriber, 'access_revoked'); + return transcriptPositionsSubscriptionNotFound(); + } + const transcript = subscriber.transcript; + return this.sessionAdmission.run(subscriber.sessionId, async () => { + if ( + this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber || + this.#connections.get(connectionId) !== connection || + !this.#canObserve(subscriber, subscriber.sessionId) + ) { + return transcriptPositionsSubscriptionNotFound(); + } + try { + const page = await readSessionTranscriptPositionsPage({ + reader: this.#transcriptReader!, + state: transcript, + request: input, + }); + if ( + this.#ownedSubscriber(connectionId, input.subscriptionId) !== subscriber || + this.#connections.get(connectionId) !== connection || + !this.#canObserve(subscriber, subscriber.sessionId) + ) { + return transcriptPositionsSubscriptionNotFound(); + } + return { ok: true, result: page }; + } catch (error) { + if (error instanceof TranscriptPageRequestError) { + return { ok: false, error: { code: 'invalid_request', message: error.message } }; + } + return { + ok: false, + error: { code: 'persistence_failed', message: 'Session transcript is unavailable' }, + }; + } + }); + } + #prepareTranscriptOverlay( state: SessionProjectionState, sessionId: string, @@ -1875,6 +1937,13 @@ function transcriptSubscriptionNotFound(): OperationOutcome<'session.transcript. }; } +function transcriptPositionsSubscriptionNotFound(): OperationOutcome<'session.transcript.positions.page'> { + return { + ok: false, + error: { code: 'not_found', message: 'Session subscription was not found' }, + }; +} + function terminalFrameByteBudget(subscriber: Subscriber, hostEpoch: string): number { return Math.max( slowConsumerFrameBytes(subscriber, hostEpoch), diff --git a/packages/runtime-host/src/server/session-transcript-pager.ts b/packages/runtime-host/src/server/session-transcript-pager.ts index b6cd126825..5799082dc7 100644 --- a/packages/runtime-host/src/server/session-transcript-pager.ts +++ b/packages/runtime-host/src/server/session-transcript-pager.ts @@ -17,7 +17,7 @@ * under the License. */ -import { createHmac, randomBytes, timingSafeEqual } from 'node:crypto'; +import { randomBytes } from 'node:crypto'; import type { StoredMessage } from '@maka/core/session'; import { SESSION_TRANSCRIPT_PAGE_MAX_MESSAGES, @@ -37,6 +37,10 @@ import { type SessionTranscriptReader, } from './session-transcript-reader.js'; import { projectSharedSessionTranscriptMessage } from './shared-session-transcript.js'; +import { + decodeTranscriptSignedCursor, + encodeTranscriptSignedCursor, +} from './transcript-signed-cursor.js'; type SessionTranscriptProjection = 'owner' | 'shared'; @@ -791,28 +795,13 @@ function emptyPage( } function encodeCursor(cursor: TranscriptCursorState, secret: Buffer): string { - const payload = Buffer.from(JSON.stringify(cursor), 'utf8').toString('base64url'); - return `${payload}.${signCursor(payload, secret).toString('base64url')}`; + return encodeTranscriptSignedCursor(cursor, secret); } function decodeCursor(value: string, secret: Buffer): TranscriptCursorState { let decoded: unknown; try { - const parts = value.split('.'); - if (parts.length !== 2) throw new Error('invalid cursor envelope'); - const [payload, signatureValue] = parts as [string, string]; - const bytes = Buffer.from(payload, 'base64url'); - const signature = Buffer.from(signatureValue, 'base64url'); - const expected = signCursor(payload, secret); - if ( - bytes.toString('base64url') !== payload || - signature.toString('base64url') !== signatureValue || - signature.byteLength !== expected.byteLength || - !timingSafeEqual(signature, expected) - ) { - throw new Error('invalid cursor signature'); - } - decoded = JSON.parse(bytes.toString('utf8')) as unknown; + decoded = decodeTranscriptSignedCursor(value, secret); } catch (cause) { throw new TranscriptPageRequestError('Invalid transcript cursor', { cause }); } @@ -853,10 +842,6 @@ function decodeCursor(value: string, secret: Buffer): TranscriptCursorState { return cursor as unknown as TranscriptCursorState; } -function signCursor(payload: string, secret: Buffer): Buffer { - return createHmac('sha256', secret).update(payload, 'utf8').digest(); -} - function mergeActiveAssistantStreams( overlay: readonly StoredMessage[], prefixes: Iterable, diff --git a/packages/runtime-host/src/server/session-transcript-position-pager.ts b/packages/runtime-host/src/server/session-transcript-position-pager.ts new file mode 100644 index 0000000000..b6181b70f2 --- /dev/null +++ b/packages/runtime-host/src/server/session-transcript-position-pager.ts @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS, + type SessionTranscriptPositionsPageInput, + type SessionTranscriptPositionsPageResult, +} from '../protocol/index.js'; +import type { SessionTranscriptReader } from './session-transcript-reader.js'; +import { + type SubscriberTranscriptState, + TranscriptPageRequestError, +} from './session-transcript-pager.js'; +import { + decodeTranscriptSignedCursor, + encodeTranscriptSignedCursor, +} from './transcript-signed-cursor.js'; + +const POSITION_CURSOR_DOMAIN = 'session-transcript-turn-positions-v1'; + +interface PositionCursorState { + readonly version: 1; + readonly subscriptionId: string; + readonly sessionId: string; + readonly direction: 'older' | 'newer'; + readonly throughSequence: number | null; + readonly revision: number; + readonly boundarySequence: number; +} + +export async function readSessionTranscriptPositionsPage(input: { + reader: SessionTranscriptReader; + state: SubscriberTranscriptState; + request: SessionTranscriptPositionsPageInput; +}): Promise { + const { reader, state, request } = input; + if (state.projection !== 'owner') { + throw new TranscriptPageRequestError('Transcript positions require an owner subscription'); + } + if ( + request.subscriptionId !== state.subscriptionId || + request.throughSequence !== state.durableThroughSequence + ) { + throw new TranscriptPageRequestError('Transcript position request does not match subscription'); + } + + let revision = request.revision; + let anchorSequence = request.anchorSequence; + if (request.cursor !== null) { + const cursor = decodePositionCursor(request.cursor, state.cursorSecret); + if ( + cursor.subscriptionId !== state.subscriptionId || + cursor.sessionId !== state.sessionId || + cursor.direction !== request.direction || + cursor.throughSequence !== request.throughSequence + ) { + throw new TranscriptPageRequestError('Transcript position cursor does not match request'); + } + revision = cursor.revision; + anchorSequence = cursor.boundarySequence; + } + + const page = await reader.readDurableTurnPositions(state.sessionId, { + direction: request.direction, + throughSequence: request.throughSequence, + anchorSequence, + maxPositions: request.maxPositions, + }); + if (page.kind === 'building') { + return { + kind: 'building', + sessionId: state.sessionId, + throughSequence: page.throughSequence, + indexedThroughSequence: page.indexedThroughSequence, + retryAfterMs: SESSION_TRANSCRIPT_POSITION_RETRY_AFTER_MS, + }; + } + if (revision !== null && page.revision !== revision) { + return { kind: 'stale', sessionId: state.sessionId, currentRevision: page.revision }; + } + const hasContinuation = request.direction === 'older' ? page.hasOlder : page.hasNewer; + const edge = + request.direction === 'older' + ? page.positions[0]?.firstSequence + : page.positions.at(-1)?.firstSequence; + if (hasContinuation && edge === undefined) { + throw new Error('Session transcript position page has an empty continuation'); + } + return { + kind: 'page', + sessionId: state.sessionId, + direction: request.direction, + throughSequence: page.throughSequence, + revision: page.revision, + positions: page.positions, + hasOlder: page.hasOlder, + hasNewer: page.hasNewer, + nextCursor: + hasContinuation && edge !== undefined + ? encodeTranscriptSignedCursor( + { + version: 1, + subscriptionId: state.subscriptionId, + sessionId: state.sessionId, + direction: request.direction, + throughSequence: page.throughSequence, + revision: page.revision, + boundarySequence: edge, + } satisfies PositionCursorState, + state.cursorSecret, + POSITION_CURSOR_DOMAIN, + ) + : null, + }; +} + +function decodePositionCursor(value: string, secret: Buffer): PositionCursorState { + let decoded: unknown; + try { + decoded = decodeTranscriptSignedCursor(value, secret, POSITION_CURSOR_DOMAIN); + } catch (cause) { + throw new TranscriptPageRequestError('Invalid transcript position cursor', { cause }); + } + if (!decoded || typeof decoded !== 'object' || Array.isArray(decoded)) { + throw new TranscriptPageRequestError('Invalid transcript position cursor'); + } + const cursor = decoded as Record; + const keys = [ + 'version', + 'subscriptionId', + 'sessionId', + 'direction', + 'throughSequence', + 'revision', + 'boundarySequence', + ]; + if ( + Object.keys(cursor).length !== keys.length || + keys.some((key) => !Object.hasOwn(cursor, key)) || + cursor.version !== 1 || + typeof cursor.subscriptionId !== 'string' || + typeof cursor.sessionId !== 'string' || + (cursor.direction !== 'older' && cursor.direction !== 'newer') || + (cursor.throughSequence !== null && !isCount(cursor.throughSequence)) || + !isCount(cursor.revision) || + !isCount(cursor.boundarySequence) + ) { + throw new TranscriptPageRequestError('Invalid transcript position cursor fields'); + } + return cursor as unknown as PositionCursorState; +} + +function isCount(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} diff --git a/packages/runtime-host/src/server/session-transcript-reader.ts b/packages/runtime-host/src/server/session-transcript-reader.ts index 442d3b8851..cb9242ce61 100644 --- a/packages/runtime-host/src/server/session-transcript-reader.ts +++ b/packages/runtime-host/src/server/session-transcript-reader.ts @@ -35,6 +35,8 @@ import type { SessionTranscriptRecordScanPage, SessionTranscriptRecordScanRequest, SessionTranscriptStoragePage, + SessionTurnPositionPageRequest, + SessionTurnPositionPageResult, } from '@maka/storage/execution-stores'; import { SESSION_TRANSCRIPT_OVERLAY_MAX_MESSAGES, type TurnSnapshot } from '../protocol/index.js'; @@ -57,6 +59,8 @@ export function createSessionTranscriptReader(input: { input.stores.sessionStore.readTranscriptRecordsSnapshot(sessionId, request), readDurableMessagesById: (sessionId, request) => input.stores.sessionStore.readTranscriptMessagesSnapshot(sessionId, request), + readDurableTurnPositions: (sessionId, request) => + input.stores.sessionStore.readTurnPositionsSnapshot(sessionId, request), readActiveOverlay: async (sessionId, rootTurn) => { if (!rootTurn || isTerminalTurn(rootTurn)) return []; @@ -93,6 +97,10 @@ export interface SessionTranscriptReader { sessionId: string, request: SessionTranscriptMessageLookupRequest, ): Promise; + readDurableTurnPositions( + sessionId: string, + request: SessionTurnPositionPageRequest, + ): Promise; readActiveOverlay( sessionId: string, rootTurn: TurnSnapshot | null, diff --git a/packages/runtime-host/src/server/transcript-signed-cursor.ts b/packages/runtime-host/src/server/transcript-signed-cursor.ts new file mode 100644 index 0000000000..df05538d9a --- /dev/null +++ b/packages/runtime-host/src/server/transcript-signed-cursor.ts @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHmac, timingSafeEqual } from 'node:crypto'; + +export function encodeTranscriptSignedCursor( + value: unknown, + secret: Buffer, + domain?: string, +): string { + const payload = Buffer.from(JSON.stringify(value), 'utf8').toString('base64url'); + return `${payload}.${sign(payload, secret, domain).toString('base64url')}`; +} + +export function decodeTranscriptSignedCursor( + value: string, + secret: Buffer, + domain?: string, +): unknown { + const parts = value.split('.'); + if (parts.length !== 2) throw new Error('invalid cursor envelope'); + const [payload, signatureValue] = parts as [string, string]; + const bytes = Buffer.from(payload, 'base64url'); + const signature = Buffer.from(signatureValue, 'base64url'); + const expected = sign(payload, secret, domain); + if ( + bytes.toString('base64url') !== payload || + signature.toString('base64url') !== signatureValue || + signature.byteLength !== expected.byteLength || + !timingSafeEqual(signature, expected) + ) { + throw new Error('invalid cursor signature'); + } + return JSON.parse(bytes.toString('utf8')) as unknown; +} + +function sign(payload: string, secret: Buffer, domain?: string): Buffer { + const hmac = createHmac('sha256', secret); + if (domain !== undefined) hmac.update(domain, 'utf8').update('\0', 'utf8'); + return hmac.update(payload, 'utf8').digest(); +} From 3f6509b8f4ace8b65361143d11682bc723703f8e Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:22:36 +0800 Subject: [PATCH 03/14] feat(desktop): project bounded transcript positions Generated-by: Codex --- .../desktop-transcript-range-store.test.ts | 174 ++++++++++++++++++ .../__tests__/runtime-host-client.test.ts | 3 + .../runtime-host-session-observer.test.ts | 10 + .../runtime-host-session-test-fixture.ts | 13 ++ .../__tests__/transcript-identity.test.ts | 1 + .../__tests__/workhub-session-port.test.ts | 18 +- .../src/main/desktop-transcript-ipc.ts | 25 +++ .../src/main/desktop-transcript-replica.ts | 107 ++++++++++- apps/desktop/src/main/runtime-host-client.ts | 11 ++ .../src/main/runtime-host-session-observer.ts | 11 ++ .../src/preload/transcript-contract.ts | 59 ++++++ .../desktop-transcript-range-store.ts | 26 +++ 12 files changed, 455 insertions(+), 3 deletions(-) diff --git a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts index ba076c2fc0..5c5475ac25 100644 --- a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts @@ -246,6 +246,46 @@ test('keeps unchanged message references stable across immutable range snapshots assert.deepEqual(second.messages, [firstMessage, secondMessage]); }); +test('keeps a null position sidecar as an explicit no-update batch', () => { + const identity = { + sessionId: 'session-1', + generation: 'generation-1', + hostEpoch: 'host-1', + }; + const store = transcriptStore(); + for (const batch of encodeDesktopTranscriptSnapshot({ + ...identity, + durableThrough: 1, + durable: [{ sequence: 1, message: userMessage('first', 'user-1') }], + overlay: [], + positionRange: { + state: 'ready', + throughSequence: 1, + revision: 2, + positions: [{ turnId: 'turn-user-1', firstSequence: 1 }], + hasOlder: false, + hasNewer: false, + }, + hasOlder: false, + hasNewer: false, + })) store.accept(batch); + const positions = store.snapshot().positionRange; + + const batches = [...encodeDesktopTranscriptChange(identity, { + durableThrough: 1, + durableUpserts: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + positionRange: null, + hasOlder: false, + hasNewer: false, + })]; + assert.equal(batches.length, 1); + assert.equal(batches[0]!.positionRange, null); + assert.equal(store.accept(batches[0]!), false); + assert.strictEqual(store.snapshot().positionRange, positions); +}); + test('bounds the default active transcript range by Turn identities', async () => { const messages = Array.from({ length: 200 }, (_, sequence) => ({ identity: sequence, @@ -287,6 +327,140 @@ test('bounds the default active transcript range by Turn identities', async () = assert.equal(snapshot.hasNewer, false); }); +test('keeps a bounded Turn position sidecar on the existing transcript replica', async () => { + const messages = [ + { identity: 0, message: userMessage('first', 'user-1') }, + { identity: 1, message: assistantMessage('answer', 'assistant-1') }, + { identity: 2, message: userMessage('second', 'user-2') }, + ]; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 2, + durableCoverage: 'complete', + overlayMessageCount: 0, + durable: transcriptPage('older', null, 2), + overlay: { ...transcriptPage('older', null, 2), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async () => ({ messages, nextCursor: null }), + loadTranscriptPositionsPage: async (input) => ({ + kind: 'page', + sessionId: 'session-1', + direction: input.direction, + throughSequence: input.throughSequence, + revision: 0, + positions: [ + { turnId: 'turn-1', firstSequence: 0 }, + { turnId: 'turn-2', firstSequence: 2 }, + ], + hasOlder: false, + hasNewer: false, + nextCursor: null, + }), + async close() {}, + }); + + const replica = await DesktopTranscriptReplica.prepare(handle); + + assert.deepEqual(replica.snapshot().positionRange, { + state: 'ready', + throughSequence: 2, + revision: 0, + positions: [ + { turnId: 'turn-1', firstSequence: 0 }, + { turnId: 'turn-2', firstSequence: 2 }, + ], + hasOlder: false, + hasNewer: false, + }); +}); + +test('retries a building position sidecar without withholding durable bodies', async () => { + const messages = [{ identity: 0, message: userMessage('first', 'user-1') }]; + let positionReads = 0; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 0, + durableCoverage: 'complete', + overlayMessageCount: 0, + durable: transcriptPage('older', null, 0), + overlay: { ...transcriptPage('older', null, 0), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async () => ({ messages, nextCursor: null }), + loadTranscriptPositionsPage: async (input) => { + positionReads += 1; + if (positionReads === 1) { + return { + kind: 'building' as const, + sessionId: 'session-1', + throughSequence: input.throughSequence, + indexedThroughSequence: null, + retryAfterMs: 25 as const, + }; + } + return { + kind: 'page' as const, + sessionId: 'session-1', + direction: input.direction, + throughSequence: input.throughSequence, + revision: 0, + positions: [{ turnId: 'turn-1', firstSequence: 0 }], + hasOlder: false, + hasNewer: false, + nextCursor: null, + }; + }, + async close() {}, + }); + + const replica = await DesktopTranscriptReplica.prepare(handle); + assert.deepEqual(replica.snapshot().durable, messages.map((entry) => ({ + sequence: entry.identity, + message: entry.message, + }))); + assert.equal(replica.snapshot().positionRange?.state, 'building'); + + await new Promise((resolve) => setTimeout(resolve, 50)); + assert.equal(replica.snapshot().positionRange?.state, 'ready'); + assert.equal(positionReads, 2); + await replica.close(); +}); + +test('degrades only the position sidecar when its pager fails', async () => { + const messages = [{ identity: 0, message: userMessage('first', 'user-1') }]; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 0, + durableCoverage: 'complete', + overlayMessageCount: 0, + durable: transcriptPage('older', null, 0), + overlay: { ...transcriptPage('older', null, 0), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async () => ({ messages, nextCursor: null }), + loadTranscriptPositionsPage: async () => { + throw new Error('position pager unavailable'); + }, + async close() {}, + }); + + const replica = await DesktopTranscriptReplica.prepare(handle); + + assert.equal(replica.snapshot().durable.length, 1); + assert.equal(replica.snapshot().positionRange?.state, 'unavailable'); + await replica.close(); +}); + test('bounds the default active transcript range by presentation bytes', async () => { const messages = syntheticLargeTranscript(); const bootstrapPage = transcriptPage('older', null, messages.length - 1); diff --git a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts index 0b82aef122..14e0770fe5 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts @@ -194,6 +194,9 @@ function subscription( loadTranscriptPage: async () => { throw new Error('Fake subscription does not expose transcript pages'); }, + loadTranscriptPositionsPage: async () => { + throw new Error('Fake subscription does not expose transcript positions'); + }, close: async () => { lifecycle.push(`${sessionId}:close`); }, diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index 1b713542e9..14f0affe7c 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -39,6 +39,15 @@ import { type DesktopTranscriptBatch, type DesktopTranscriptOpenResult, } from '../../preload/transcript-contract.js'; + +const UNAVAILABLE_POSITION_RANGE = { + state: 'unavailable', + throughSequence: null, + revision: null, + positions: [], + hasOlder: false, + hasNewer: false, +} as const; import { RuntimeHostSessionObservationRegistry } from "../runtime-host-session-observation-registry.js"; import { RuntimeHostSessionObserver, @@ -438,6 +447,7 @@ test('restores transcript consumers across Host replacement', async () => { fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: true, diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts b/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts index 8a22cd60a3..94edd32b80 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-test-fixture.ts @@ -36,6 +36,7 @@ export function runtimeHostSessionFixture(input: { loadTranscriptOverlay?: DesktopRuntimeHostSession['loadTranscriptOverlay']; decodeTranscriptPage?: DesktopRuntimeHostSession['decodeTranscriptPage']; loadTranscriptPage?: DesktopRuntimeHostSession['loadTranscriptPage']; + loadTranscriptPositionsPage?: DesktopRuntimeHostSession['loadTranscriptPositionsPage']; close(): Promise; }): DesktopRuntimeHostSession { const sessionId = input.snapshot.session.sessionId; @@ -61,6 +62,18 @@ export function runtimeHostSessionFixture(input: { })), loadTranscriptPage: input.loadTranscriptPage ?? (async () => emptyPage(sessionId, 'durable')), + loadTranscriptPositionsPage: input.loadTranscriptPositionsPage ?? + (async (request) => ({ + kind: 'page', + sessionId, + direction: request.direction, + throughSequence: request.throughSequence, + revision: request.revision ?? 0, + positions: [], + hasOlder: false, + hasNewer: false, + nextCursor: null, + })), close: input.close, }; } diff --git a/apps/desktop/src/main/__tests__/transcript-identity.test.ts b/apps/desktop/src/main/__tests__/transcript-identity.test.ts index bd853f7083..46ee94d292 100644 --- a/apps/desktop/src/main/__tests__/transcript-identity.test.ts +++ b/apps/desktop/src/main/__tests__/transcript-identity.test.ts @@ -34,6 +34,7 @@ function batch(overrides: Partial = {}): DesktopTranscri fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: null, hasOlder: false, hasNewer: false, reset: false, diff --git a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts index b982a265d2..7a2c160dc2 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -33,6 +33,15 @@ import { projectWorkHubCoordinationTurns, } from '../../renderer/workhub-coordination-port.js'; +const UNAVAILABLE_POSITION_RANGE = { + state: 'unavailable', + throughSequence: null, + revision: null, + positions: [], + hasOlder: false, + hasNewer: false, +} as const; + function desktopSession( id: string, overrides: Partial = {}, @@ -88,6 +97,7 @@ function transcriptsWith(messages: readonly StoredMessage[]) { fragments, evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: true, @@ -375,6 +385,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: true, @@ -471,6 +482,7 @@ test('Coordination transcript reset rebuilds active linkage outside the resident fragments: [fragment(recent, 1)], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: true, hasNewer: false, reset: true, @@ -492,6 +504,7 @@ test('Coordination transcript reset rebuilds active linkage outside the resident fragments: [fragment(assignment, 0)], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: false, @@ -536,6 +549,7 @@ test('Coordination transcript reset rebuilds active linkage outside the resident fragments: [fragment(recent, 1)], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: true, hasNewer: false, reset: true, @@ -555,6 +569,7 @@ test('Coordination transcript reset rebuilds active linkage outside the resident fragments: [fragment(recent, 1)], evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: false, @@ -682,6 +697,7 @@ test('desktop adapter rebuilds recent turns from the Session transcript and clos fragments, evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: true, @@ -751,7 +767,7 @@ test('desktop adapter cancels an unavailable transcript without hiding ready Ses source: 'durable', identity: 0, order: null, byteOffset: 0, totalBytes: data.byteLength, data, }], - evictedDurableSequences: [], completedOverlayMessageIds: [], + evictedDurableSequences: [], completedOverlayMessageIds: [], positionRange: UNAVAILABLE_POSITION_RANGE, hasOlder: false, hasNewer: false, reset: true, ready: true, }); return { diff --git a/apps/desktop/src/main/desktop-transcript-ipc.ts b/apps/desktop/src/main/desktop-transcript-ipc.ts index 86bf93495b..a94450d46f 100644 --- a/apps/desktop/src/main/desktop-transcript-ipc.ts +++ b/apps/desktop/src/main/desktop-transcript-ipc.ts @@ -22,6 +22,7 @@ import { DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, type DesktopTranscriptBatchPayload, type DesktopTranscriptFragment, + type DesktopTranscriptPositionRange, } from '../preload/transcript-contract.js'; import type { DesktopSequencedTranscriptMessage, @@ -41,6 +42,7 @@ interface TranscriptBatchContent { readonly overlay: readonly StoredMessage[]; readonly evictedDurableSequences: readonly number[]; readonly completedOverlayMessageIds: readonly string[]; + readonly positionRange: DesktopTranscriptPositionRange | null; readonly hasOlder: boolean; readonly hasNewer: boolean; readonly reset: boolean; @@ -55,6 +57,9 @@ export function encodeDesktopTranscriptSnapshot( overlay: snapshot.overlay, evictedDurableSequences: [], completedOverlayMessageIds: [], + positionRange: + snapshot.positionRange ?? + fallbackPositionRange(snapshot.durableThrough, snapshot.hasOlder, snapshot.hasNewer), hasOlder: snapshot.hasOlder, hasNewer: snapshot.hasNewer, reset: true, @@ -71,6 +76,10 @@ export function encodeDesktopTranscriptChange( overlay: [], evictedDurableSequences: change.evictedDurableSequences, completedOverlayMessageIds: change.completedOverlayMessageIds, + positionRange: + change.positionRange === undefined + ? fallbackPositionRange(change.durableThrough, change.hasOlder, change.hasNewer) + : change.positionRange, hasOlder: change.hasOlder, hasNewer: change.hasNewer, reset: false, @@ -134,6 +143,7 @@ function* encodeDesktopTranscriptBatches( fragments: batchFragments, evictedDurableSequences, completedOverlayMessageIds, + positionRange: first || ready ? content.positionRange : null, hasOlder: content.hasOlder, hasNewer: content.hasNewer, reset: content.reset && first, @@ -143,6 +153,21 @@ function* encodeDesktopTranscriptBatches( } } +function fallbackPositionRange( + throughSequence: number | null, + hasOlder: boolean, + hasNewer: boolean, +): DesktopTranscriptPositionRange { + return { + state: 'unavailable', + throughSequence, + revision: null, + positions: [], + hasOlder, + hasNewer, + }; +} + function* encodeMessages(content: TranscriptBatchContent): Generator { for (const entry of content.durable) { yield* encodeMessage('durable', entry.sequence, null, entry.message); diff --git a/apps/desktop/src/main/desktop-transcript-replica.ts b/apps/desktop/src/main/desktop-transcript-replica.ts index 195ccc7630..e6083e35fb 100644 --- a/apps/desktop/src/main/desktop-transcript-replica.ts +++ b/apps/desktop/src/main/desktop-transcript-replica.ts @@ -32,6 +32,7 @@ import { DESKTOP_TRANSCRIPT_ACTIVE_RANGE_MAX_TURNS, DESKTOP_TRANSCRIPT_OVERLAY_CACHE_MAX_BYTES, DESKTOP_TRANSCRIPT_RANGE_MAX_BYTES, + type DesktopTranscriptPositionRange, } from '../preload/transcript-contract.js'; import type { DesktopRuntimeHostSession } from './runtime-host-client.js'; @@ -60,6 +61,7 @@ export interface DesktopTranscriptReplicaSnapshot { readonly durableThrough: number | null; readonly durable: readonly DesktopSequencedTranscriptMessage[]; readonly overlay: readonly StoredMessage[]; + readonly positionRange?: DesktopTranscriptPositionRange; readonly hasOlder: boolean; readonly hasNewer: boolean; } @@ -69,6 +71,7 @@ export interface DesktopTranscriptReplicaChange { readonly durableUpserts: readonly DesktopSequencedTranscriptMessage[]; readonly evictedDurableSequences: readonly number[]; readonly completedOverlayMessageIds: readonly string[]; + readonly positionRange?: DesktopTranscriptPositionRange | null; readonly hasOlder: boolean; readonly hasNewer: boolean; } @@ -100,6 +103,7 @@ export class DesktopTranscriptReplica { #targetThrough: number | null; #hasOlder: boolean; #hasNewer = false; + #positionRange: DesktopTranscriptPositionRange; #resident = true; #residentExternallyAccounted = true; #closed = false; @@ -127,6 +131,14 @@ export class DesktopTranscriptReplica { this.#durableThrough = handle.transcriptBootstrap.throughSequence; this.#targetThrough = this.#durableThrough; this.#hasOlder = handle.transcriptBootstrap.durable.nextCursor !== null; + this.#positionRange = { + state: 'unavailable', + throughSequence: this.#durableThrough, + revision: null, + positions: [], + hasOlder: this.#hasOlder, + hasNewer: this.#hasNewer, + }; } static async prepare( @@ -151,6 +163,7 @@ export class DesktopTranscriptReplica { replica.#durableThrough ?? undefined, ); + await replica.#refreshPositionRange('older', null); if (replica.#overlayBytes > replica.#maxOverlayBytes) { throw new RangeError('Desktop transcript overlay exceeds the session cache limit'); } @@ -194,6 +207,7 @@ export class DesktopTranscriptReplica { durableThrough: this.#durableThrough, durable: this.#orderedDurable(false), overlay: [...this.#overlay.values()], + positionRange: this.#positionRange, hasOlder: this.#hasOlder, hasNewer: this.#hasNewer, }; @@ -246,7 +260,7 @@ export class DesktopTranscriptReplica { anchorSequence: anchor, maxBytes, }); - await this.#withDecodedPage(page, (decoded) => { + await this.#withDecodedPage(page, async (decoded) => { this.#assertOpen(); // Same post-await `#resident` invariant as `#replaceWithRange` and the // paged catch-up: a concurrent `discard()` may have reclaimed this @@ -268,6 +282,7 @@ export class DesktopTranscriptReplica { 'newest', anchor ?? undefined, ); + await this.#refreshPositionRange('older', anchor); this.#publish(decoded.messages, completedOverlayMessageIds, evictedDurableSequences); }); } @@ -297,7 +312,7 @@ export class DesktopTranscriptReplica { anchorSequence: loadTail ? sequence + 1 : sequence === 0 ? null : sequence - 1, maxBytes, }); - await this.#withDecodedPage(page, (decoded) => { + await this.#withDecodedPage(page, async (decoded) => { this.#assertOpen(); // `#resident` can flip to false across the `await` above (a concurrent // `discard()` reclaims memory for a non-visible session while the page is @@ -328,6 +343,10 @@ export class DesktopTranscriptReplica { loadTail ? (page.protectedTurnSequence ?? sequence) : sequence, ), ); + await this.#refreshPositionRange( + loadTail ? 'older' : 'newer', + loadTail ? null : sequence === 0 ? null : sequence - 1, + ); this.#publish(decoded.messages, completedOverlayMessageIds, evictedDurableSequences); }); } @@ -457,6 +476,7 @@ export class DesktopTranscriptReplica { throw correlationError('Desktop transcript catch-up ended before its watermark'); } this.#durableThrough = target; + await this.#refreshPositionRange('older', null); this.#publish([], [], []); } } @@ -550,11 +570,94 @@ export class DesktopTranscriptReplica { (sequence) => !this.#durable.has(sequence), ), completedOverlayMessageIds, + positionRange: this.#positionRange, hasOlder: this.#hasOlder, hasNewer: this.#hasNewer, }; } + async #refreshPositionRange( + direction: 'older' | 'newer', + anchorSequence: number | null, + ): Promise { + if (this.#closed || !this.#resident) return; + const throughSequence = this.#durableThrough; + try { + let result = await this.#handle.loadTranscriptPositionsPage({ + direction, + throughSequence, + revision: this.#positionRange.revision, + cursor: null, + anchorSequence, + maxPositions: 128, + }); + if (result.kind === 'stale') { + result = await this.#handle.loadTranscriptPositionsPage({ + direction, + throughSequence, + revision: null, + cursor: null, + anchorSequence, + maxPositions: 128, + }); + } + if (result.kind === 'page') { + this.#positionRange = { + state: 'ready', + throughSequence: result.throughSequence, + revision: result.revision, + positions: result.positions, + hasOlder: result.hasOlder, + hasNewer: result.hasNewer, + }; + return; + } + if (result.kind === 'building') { + this.#positionRange = { + state: 'building', + throughSequence: result.throughSequence, + revision: null, + positions: [], + hasOlder: this.#hasOlder, + hasNewer: this.#hasNewer, + }; + this.#schedulePositionRetry(direction, anchorSequence, result.retryAfterMs); + return; + } + this.#positionRange = { + state: 'unavailable', + throughSequence, + revision: result.currentRevision, + positions: [], + hasOlder: this.#hasOlder, + hasNewer: this.#hasNewer, + }; + } catch { + this.#positionRange = { + state: 'unavailable', + throughSequence, + revision: null, + positions: [], + hasOlder: this.#hasOlder, + hasNewer: this.#hasNewer, + }; + } + } + + #schedulePositionRetry( + direction: 'older' | 'newer', + anchorSequence: number | null, + retryAfterMs: number, + ): void { + globalThis.setTimeout(() => { + if (this.#closed || !this.#resident || this.#positionRange.state !== 'building') return; + void this.#enqueue(async () => { + await this.#refreshPositionRange(direction, anchorSequence); + if (!this.#closed && this.#resident) this.#publish([], [], []); + }).catch(() => undefined); + }, retryAfterMs); + } + #evictToBudget( budget: number | undefined = undefined, edge: 'oldest' | 'newest' = 'oldest', diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 54e5932172..bdf251bd6e 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -126,6 +126,8 @@ import { type SessionTranscriptBootstrap, type SessionTranscriptPage, type SessionTranscriptPageInput, + type SessionTranscriptPositionsPageInput, + type SessionTranscriptPositionsPageResult, mergeSessionTurnContributions, projectSessionTurnContribution, type SessionConversationCopyInput, @@ -219,6 +221,9 @@ export interface DesktopRuntimeHostSession { loadTranscriptPage( input: Omit, ): Promise; + loadTranscriptPositionsPage( + input: Omit, + ): Promise; close(): Promise; } @@ -1819,6 +1824,12 @@ class DesktopSessionHandle implements DesktopRuntimeHostSession { return this.subscription.loadTranscriptPage(input); } + loadTranscriptPositionsPage( + input: Omit, + ): Promise { + return this.subscription.loadTranscriptPositionsPage(input); + } + close(): Promise { this.#closeTask ??= this.subscription.close().finally(this.onClose); return this.#closeTask; diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts index acf5e12377..cbb474be16 100644 --- a/apps/desktop/src/main/runtime-host-session-observer.ts +++ b/apps/desktop/src/main/runtime-host-session-observer.ts @@ -43,6 +43,7 @@ import { type DesktopTranscriptBatch, type DesktopTranscriptBatchPayload, type DesktopTranscriptOpenResult, + type DesktopTranscriptPositionRange, type DesktopTranscriptRangeRequest, } from '../preload/transcript-contract.js'; import { @@ -144,6 +145,7 @@ interface PendingTranscriptChange { readonly durableUpserts: Map; readonly evictedDurableSequences: Set; readonly completedOverlayMessageIds: Set; + positionRange: DesktopTranscriptPositionRange | null; hasOlder: boolean; hasNewer: boolean; encodedBytes: number; @@ -1155,6 +1157,7 @@ export class RuntimeHostSessionObserver { durableUpserts: [...pending.durableUpserts.values()].map(({ entry }) => entry), evictedDurableSequences: [...pending.evictedDurableSequences], completedOverlayMessageIds: [...pending.completedOverlayMessageIds], + positionRange: pending.positionRange, hasOlder: pending.hasOlder, hasNewer: pending.hasNewer, }, @@ -1194,6 +1197,7 @@ export class RuntimeHostSessionObserver { durableUpserts: new Map(), evictedDurableSequences: new Set(), completedOverlayMessageIds: new Set(), + positionRange: change.positionRange ?? null, hasOlder: change.hasOlder, hasNewer: change.hasNewer, encodedBytes: 0, @@ -1202,6 +1206,13 @@ export class RuntimeHostSessionObserver { pending.durableThrough = change.durableThrough; pending.hasOlder = change.hasOlder; pending.hasNewer = change.hasNewer; + if (change.positionRange !== undefined && change.positionRange !== null) { + const previousBytes = pending.positionRange + ? Buffer.byteLength(JSON.stringify(pending.positionRange), 'utf8') + : 0; + pending.positionRange = change.positionRange; + byteDelta += Buffer.byteLength(JSON.stringify(change.positionRange), 'utf8') - previousBytes; + } for (const entry of change.durableUpserts) { const previous = pending.durableUpserts.get(entry.sequence); if (previous) byteDelta -= previous.encodedBytes; diff --git a/apps/desktop/src/preload/transcript-contract.ts b/apps/desktop/src/preload/transcript-contract.ts index 41f00924c7..558fcff75f 100644 --- a/apps/desktop/src/preload/transcript-contract.ts +++ b/apps/desktop/src/preload/transcript-contract.ts @@ -23,6 +23,20 @@ export const DESKTOP_TRANSCRIPT_ACTIVE_RANGE_MAX_TURNS = 10; export const DESKTOP_TRANSCRIPT_OVERLAY_CACHE_MAX_BYTES = 16 * 1024 * 1024; export const DESKTOP_TRANSCRIPT_GLOBAL_CACHE_MAX_BYTES = 64 * 1024 * 1024; +export interface DesktopTranscriptTurnPosition { + readonly turnId: string; + readonly firstSequence: number; +} + +export interface DesktopTranscriptPositionRange { + readonly state: 'ready' | 'building' | 'unavailable'; + readonly throughSequence: number | null; + readonly revision: number | null; + readonly positions: readonly DesktopTranscriptTurnPosition[]; + readonly hasOlder: boolean; + readonly hasNewer: boolean; +} + export interface DesktopTranscriptFragment { readonly source: 'durable' | 'overlay'; readonly identity: number | string; @@ -40,6 +54,8 @@ export interface DesktopTranscriptBatchPayload { readonly fragments: readonly DesktopTranscriptFragment[]; readonly evictedDurableSequences: readonly number[]; readonly completedOverlayMessageIds: readonly string[]; + /** Null means this batch does not change the bounded position sidecar. */ + readonly positionRange: DesktopTranscriptPositionRange | null; readonly hasOlder: boolean; readonly hasNewer: boolean; readonly reset: boolean; @@ -91,6 +107,10 @@ export function assertDesktopTranscriptBatch(value: unknown): DesktopTranscriptB (messageId) => typeof messageId === 'string' && messageId.length > 0 && messageId.length <= 256, ) || batch.completedOverlayMessageIds.length > 256 || + !isPositionRange(batch.positionRange) || + (batch.positionRange !== null && + batch.positionRange.throughSequence !== batch.durableThrough) || + (batch.reset && batch.positionRange === null) || typeof batch.hasOlder !== 'boolean' || typeof batch.hasNewer !== 'boolean' || typeof batch.reset !== 'boolean' || @@ -135,6 +155,45 @@ export function assertDesktopTranscriptBatch(value: unknown): DesktopTranscriptB return value as DesktopTranscriptBatch; } +function isPositionRange(value: unknown): value is DesktopTranscriptPositionRange | null { + if (value === null) return true; + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const range = value as Record; + if ( + (range.state !== 'ready' && range.state !== 'building' && range.state !== 'unavailable') || + (range.throughSequence !== null && !isSequence(range.throughSequence)) || + (range.revision !== null && !isSequence(range.revision)) || + !Array.isArray(range.positions) || + range.positions.length > 128 || + (range.state === 'ready' && range.revision === null) || + (range.state !== 'ready' && range.positions.length > 0) || + typeof range.hasOlder !== 'boolean' || + typeof range.hasNewer !== 'boolean' + ) { + return false; + } + let previous = -1; + const turnIds = new Set(); + for (const value of range.positions) { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const position = value as Record; + if ( + typeof position.turnId !== 'string' || + position.turnId.length === 0 || + position.turnId.length > 128 || + !isSequence(position.firstSequence) || + (range.throughSequence !== null && position.firstSequence > range.throughSequence) || + position.firstSequence <= previous || + turnIds.has(position.turnId) + ) { + return false; + } + previous = position.firstSequence; + turnIds.add(position.turnId); + } + return true; +} + function isSequence(value: unknown): value is number { return Number.isSafeInteger(value) && (value as number) >= 0; } diff --git a/apps/desktop/src/renderer/desktop-transcript-range-store.ts b/apps/desktop/src/renderer/desktop-transcript-range-store.ts index 6cee40e944..3a4b1ed7e1 100644 --- a/apps/desktop/src/renderer/desktop-transcript-range-store.ts +++ b/apps/desktop/src/renderer/desktop-transcript-range-store.ts @@ -23,6 +23,7 @@ import type { DesktopTranscriptBatchPayload, DesktopTranscriptFragment, DesktopTranscriptHandle, + DesktopTranscriptPositionRange, } from '../preload/transcript-contract.js'; import { projectDesktopStoredMessage } from '../shared/desktop-session-projection.js'; import { parseDesktopSessionKey } from '../shared/runtime-host-identity.js'; @@ -125,6 +126,7 @@ export interface DesktopTranscriptRangeState { readonly newestSequence: number | null; readonly hasOlder: boolean; readonly hasNewer: boolean; + readonly positionRange: DesktopTranscriptPositionRange; readonly ready: boolean; } @@ -150,6 +152,14 @@ export class DesktopTranscriptRangeStore { #newestUserSequence: number | null = null; #hasOlder = false; #hasNewer = false; + #positionRange: DesktopTranscriptPositionRange = { + state: 'unavailable', + throughSequence: null, + revision: null, + positions: [], + hasOlder: false, + hasNewer: false, + }; #ready = false; #batchChanged = false; #snapshot: DesktopTranscriptRangeSnapshot | undefined; @@ -179,6 +189,10 @@ export class DesktopTranscriptRangeStore { this.#durableThrough = batch.durableThrough; this.#hasOlder = batch.hasOlder; this.#hasNewer = batch.hasNewer; + if (batch.positionRange !== null && !samePositionRange(this.#positionRange, batch.positionRange)) { + this.#positionRange = freezeTranscriptValue(structuredClone(batch.positionRange)); + changed = true; + } for (const sequence of batch.evictedDurableSequences) { if (this.#durable.delete(sequence)) { removeOrdered(this.#durableOrder, sequence); @@ -235,6 +249,7 @@ export class DesktopTranscriptRangeStore { newestSequence: this.#newestSequence, hasOlder: this.#hasOlder, hasNewer: this.#hasNewer, + positionRange: this.#positionRange, ready: this.#ready, }; } @@ -292,6 +307,10 @@ export class DesktopTranscriptRangeStore { this.#newestUserSequence = null; this.#hasOlder = batch.hasOlder; this.#hasNewer = batch.hasNewer; + if (batch.positionRange === null) { + throw new Error('Desktop transcript reset omitted its position range'); + } + this.#positionRange = freezeTranscriptValue(structuredClone(batch.positionRange)); this.#ready = false; this.#batchChanged = false; this.#snapshot = undefined; @@ -407,6 +426,13 @@ export class DesktopTranscriptRangeStore { } } +function samePositionRange( + left: DesktopTranscriptPositionRange, + right: DesktopTranscriptPositionRange, +): boolean { + return JSON.stringify(left) === JSON.stringify(right); +} + function freezeTranscriptValue(value: T): T { if (typeof value !== 'object' || value === null || Object.isFrozen(value)) return value; for (const child of Object.values(value)) freezeTranscriptValue(child); From 2ad9f62825a64d7e1a71222c7fa281e103960a6a Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:22:41 +0800 Subject: [PATCH 04/14] feat(ui): represent unloaded transcript ranges Generated-by: Codex --- apps/desktop/src/renderer/app-shell.tsx | 6 +- .../src/renderer/chat-message-surface.tsx | 20 ++- .../src/renderer/styles/chat-message.css | 7 + .../transcript-history-notice.test.tsx | 21 ++- .../transcript-row-projection.test.ts | 126 ++++++++++++++ packages/ui/src/chat-view.tsx | 110 +++++++++++- packages/ui/src/conversation-copy.ts | 7 + packages/ui/src/transcript-row-projection.ts | 163 ++++++++++++++++++ 8 files changed, 449 insertions(+), 11 deletions(-) create mode 100644 packages/ui/src/__tests__/transcript-row-projection.test.ts create mode 100644 packages/ui/src/transcript-row-projection.ts diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index 84a2984fa1..cf5afb2749 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -3085,11 +3085,11 @@ function AppShellContent({ } > {navSelection.section === 'sessions' ? ( - loadTranscriptHistory('earlier', anchorTurnId)} diff --git a/apps/desktop/src/renderer/chat-message-surface.tsx b/apps/desktop/src/renderer/chat-message-surface.tsx index d9684f3263..601f469bac 100644 --- a/apps/desktop/src/renderer/chat-message-surface.tsx +++ b/apps/desktop/src/renderer/chat-message-surface.tsx @@ -41,6 +41,12 @@ import { useExternalStoreSelector } from './use-external-store-selector'; import { useDeepResearchRun } from './use-deep-research-run'; import { ChatRecoveryNotice, SessionHealthRecoveryNotice } from './chat-recovery-notice'; +type TranscriptRangeProjection = Readonly<{ + hasOlder: boolean; + hasNewer: boolean; + positionRange: ComponentProps['transcriptPositionRange']; +}>; + const selectShellRunRecord = (state: AppShellSessionUiState, sessionId: string | undefined) => sessionId ? state.shellRunUpdatesBySession[sessionId] : undefined; @@ -89,8 +95,8 @@ interface ChatMessageSurfaceProps extends Omit< connections: LlmConnection[]; onRefreshConnections: () => Promise | void; onSkip: () => Promise | void; - hasOlderHistory: boolean; - hasNewerHistory: boolean; + olderHistoryRange: TranscriptRangeProjection | false | undefined; + newerHistoryRange: TranscriptRangeProjection | false | undefined; historyLoadPending: boolean; onLoadEarlierHistory: (anchorTurnId?: string) => Promise | void; onReturnToLatestHistory: () => Promise | void; @@ -126,14 +132,15 @@ export function ChatMessageSurface({ connections, onRefreshConnections, onSkip, - hasOlderHistory, - hasNewerHistory, + olderHistoryRange, + newerHistoryRange, historyLoadPending, onLoadEarlierHistory, onReturnToLatestHistory, ...chatViewRest }: ChatMessageSurfaceProps) { const locale = useUiLocale(); + const transcriptRange = olderHistoryRange || newerHistoryRange || undefined; const copy = getShellCopy(locale).app; const transcriptCopy = getDesktopConversationCopy(locale).actions; // Configuration notices share the Settings label; identity recovery supplies @@ -247,9 +254,10 @@ export function ChatMessageSurface({ deepResearchRun={deepResearchRun} emptyOverride={emptyOverride} goalIndicator={goalProjection.goalIndicator} - hasOlderHistory={hasOlderHistory} + hasOlderHistory={transcriptRange?.hasOlder === true} onLoadEarlierHistory={onLoadEarlierHistory} - returnToLatest={hasNewerHistory ? { + transcriptPositionRange={transcriptRange?.positionRange} + returnToLatest={transcriptRange?.hasNewer ? { title: transcriptCopy.partialHistoryTitle, label: transcriptCopy.returnLatest, isPending: historyLoadPending, diff --git a/apps/desktop/src/renderer/styles/chat-message.css b/apps/desktop/src/renderer/styles/chat-message.css index ebe039fd44..f0b7b92c7f 100644 --- a/apps/desktop/src/renderer/styles/chat-message.css +++ b/apps/desktop/src/renderer/styles/chat-message.css @@ -332,3 +332,10 @@ width: min(var(--maka-reading-measure), calc(100% - (2 * var(--space-6)))); margin: var(--space-2) auto 0; } + +.maka-transcript-gap-row { + width: min(var(--maka-reading-measure), calc(100% - (2 * var(--space-6)))); + margin: var(--space-2) auto; + padding-block: var(--space-1); + border-block: 1px solid var(--border-subtle); +} diff --git a/packages/ui/src/__tests__/transcript-history-notice.test.tsx b/packages/ui/src/__tests__/transcript-history-notice.test.tsx index 2a26d78b30..541585d537 100644 --- a/packages/ui/src/__tests__/transcript-history-notice.test.tsx +++ b/packages/ui/src/__tests__/transcript-history-notice.test.tsx @@ -20,7 +20,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { renderToStaticMarkup } from 'react-dom/server'; -import { TranscriptHistoryNotice } from '../chat-view.js'; +import { TranscriptGapRow, TranscriptHistoryNotice } from '../chat-view.js'; function renderNotice(isPending: boolean): string { return renderToStaticMarkup( @@ -53,3 +53,22 @@ test('keeps the position status visible while return-to-latest is pending', () = assert.doesNotMatch(markup, /saved|loaded/); assert.match(markup, /disabled/); }); + +test('renders an unloaded range as one in-transcript row without scroll machinery', () => { + const markup = renderToStaticMarkup( + undefined} + />, + ); + + assert.match(markup, /data-transcript-gap="internal"/); + assert.match(markup, /data-missing-turn-count="2"/); + assert.match(markup, /2 unloaded turns/); + assert.match(markup, /Load this range/); + assert.doesNotMatch(markup, /height|resize|scroll/iu); +}); diff --git a/packages/ui/src/__tests__/transcript-row-projection.test.ts b/packages/ui/src/__tests__/transcript-row-projection.test.ts new file mode 100644 index 0000000000..c0f3d05ad4 --- /dev/null +++ b/packages/ui/src/__tests__/transcript-row-projection.test.ts @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { projectTranscriptRows } from '../transcript-row-projection.js'; + +const positionRange = { + state: 'ready' as const, + throughSequence: 8, + revision: 0, + positions: [ + { turnId: 'turn-1', firstSequence: 0 }, + { turnId: 'turn-2', firstSequence: 2 }, + { turnId: 'turn-3', firstSequence: 4 }, + { turnId: 'turn-4', firstSequence: 6 }, + ], + hasOlder: false, + hasNewer: false, +}; + +describe('bounded transcript row projection', () => { + test('keeps unloaded durable turns between an old range and the active turn visible as one gap', () => { + const rows = projectTranscriptRows({ + turns: [{ turnId: 'turn-1' }, { turnId: 'turn-4' }], + positionRange, + activeTurnId: 'turn-4', + }); + + assert.deepEqual(rows, [ + { kind: 'turn', turn: { turnId: 'turn-1' } }, + { + kind: 'gap', + direction: 'internal', + missingCount: 2, + firstMissing: { turnId: 'turn-2', firstSequence: 2 }, + }, + { kind: 'turn', turn: { turnId: 'turn-4' } }, + ]); + }); + + test('keeps one Turn identity when the active overlay settles to durable data', () => { + const first = { turnId: 'turn-1' }; + const active = { turnId: 'turn-4' }; + const liveRows = projectTranscriptRows({ + turns: [first, active], + positionRange, + activeTurnId: active.turnId, + }); + const settledRows = projectTranscriptRows({ + turns: [first, active], + positionRange, + }); + + assert.deepEqual(settledRows.map((row) => row.kind), ['turn', 'gap', 'turn']); + assert.strictEqual(liveRows[0]?.kind === 'turn' ? liveRows[0].turn : null, first); + assert.strictEqual(liveRows[2]?.kind === 'turn' ? liveRows[2].turn : null, active); + assert.strictEqual(settledRows[2]?.kind === 'turn' ? settledRows[2].turn : null, active); + assert.equal(settledRows.filter( + (row) => row.kind === 'turn' && row.turn.turnId === active.turnId, + ).length, 1); + }); + + test('uses one generic gap before an active overlay while positions are unavailable', () => { + const rows = projectTranscriptRows({ + turns: [{ turnId: 'turn-1' }, { turnId: 'turn-4' }], + positionRange: { + ...positionRange, + state: 'unavailable', + positions: [], + hasNewer: true, + }, + activeTurnId: 'turn-4', + }); + + assert.deepEqual(rows.map((row) => row.kind), ['turn', 'gap', 'turn']); + assert.deepEqual(rows[1], { + kind: 'gap', + direction: 'newer', + missingCount: null, + firstMissing: null, + }); + }); + + test('merges known missing positions with an adjacent unknown boundary', () => { + const rows = projectTranscriptRows({ + turns: [{ turnId: 'turn-2' }], + positionRange: { + ...positionRange, + positions: positionRange.positions.slice(0, 2), + hasOlder: true, + hasNewer: true, + }, + }); + + assert.equal(rows.filter((row) => row.kind === 'gap').length, 2); + assert.deepEqual(rows[0], { + kind: 'gap', + direction: 'older', + missingCount: null, + firstMissing: { turnId: 'turn-1', firstSequence: 0 }, + }); + assert.deepEqual(rows.at(-1), { + kind: 'gap', + direction: 'newer', + missingCount: null, + firstMissing: null, + }); + }); +}); diff --git a/packages/ui/src/chat-view.tsx b/packages/ui/src/chat-view.tsx index b090851041..ca7deb0b66 100644 --- a/packages/ui/src/chat-view.tsx +++ b/packages/ui/src/chat-view.tsx @@ -46,6 +46,11 @@ import { useChatLayoutContext } from '@astryxdesign/core/Chat'; import { useLayer } from '@astryxdesign/core/Layer'; import { materializeChat } from './materialize.js'; import { useTranscriptProjection } from './use-transcript-projection.js'; +import { + projectTranscriptRows, + type TranscriptPositionRange, + type TranscriptRow, +} from './transcript-row-projection.js'; import type { LiveTurnProjection } from './live-turn-projection.js'; import { ModelProviderRetryIndicator, @@ -123,6 +128,49 @@ export function TranscriptHistoryNotice({ ); } +export interface TranscriptGapRowProps { + direction: 'older' | 'internal' | 'newer'; + missingCount: number | null; + description: string; + actionLabel: string; + isPending: boolean; + onActivate(): Promise | void; +} + +/** The sole new visual primitive: a real row for a deliberately unloaded range. */ +export function TranscriptGapRow({ + direction, + missingCount, + description, + actionLabel, + isPending, + onActivate, +}: TranscriptGapRowProps) { + return ( + + {description} +