You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/09/07 00:26:25 UTC
(maka) 10/15: fix(storage): seek transcript pages before projecting ledger messages
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch refactor/4791-single-transcript-authority
in repository https://gitbox.apache.org/repos/asf/maka.git
commit a5544ad4694b4c27b12cac283bf43362f9fcddde
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Sun Sep 6 20:07:46 2026 +0800
fix(storage): seek transcript pages before projecting ledger messages
---
.../runtime-host/src/__tests__/protocol.test.ts | 4 +
.../__tests__/session-transcript-reader.test.ts | 285 +++++++++++++-
packages/runtime-host/src/protocol/index.ts | 3 +-
.../src/server/session-transcript-reader.ts | 315 +++++++--------
packages/runtime/src/runtime-event-read-model.ts | 47 ++-
.../__tests__/invocation-opening-backfill.test.ts | 5 +-
.../src/__tests__/sqlite-runtime-schema.test.ts | 2 +-
packages/storage/src/execution-stores.ts | 13 +
packages/storage/src/runtime-transcript-query.ts | 436 +++++++++++++++++++++
packages/storage/src/sqlite-runtime-schema.ts | 19 +-
packages/storage/src/sqlite-runtime-store.ts | 70 +++-
11 files changed, 1006 insertions(+), 193 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts
index b025966e9d..a323a02a46 100644
--- a/packages/runtime-host/src/__tests__/protocol.test.ts
+++ b/packages/runtime-host/src/__tests__/protocol.test.ts
@@ -443,6 +443,10 @@ describe('Runtime Host bootstrap protocol', () => {
assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 117);
});
+ test('publishes a new compatibility epoch for event-addressed transcript cursors', () => {
+ assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 118);
+ });
+
test('selects the highest mutually supported protocol and rejects a gap', () => {
assert.equal(negotiateProtocol({ min: 0, max: 0 }, { min: 0, max: 0 }), 0);
assert.equal(negotiateProtocol({ min: 1, max: 3 }, { min: 2, max: 4 }), 3);
diff --git a/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts b/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts
index 69922cc93f..512b04a4d5 100644
--- a/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts
+++ b/packages/runtime-host/src/__tests__/session-transcript-reader.test.ts
@@ -26,6 +26,9 @@ import { seedInvocation, testInvocationOpening } from '@maka/runtime/test-only/i
import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation';
import type { RuntimeEvent } from '@maka/core/runtime-event';
import type { StoredMessage } from '@maka/core/session';
+import { projectRuntimeEventsToStoredMessages } from '@maka/runtime/runtime-event-read-model';
+import { foldTurnContribution } from '@maka/storage/session-message-projection';
+import type { SessionTurnContribution } from '@maka/storage/execution-stores';
import {
type ExecutionStoresWriter,
openInteractiveExecutionStoresForWrite,
@@ -251,7 +254,8 @@ test('keeps durable history separate from the canonical active overlay', async (
maxBytes: 1024,
maxMessages: 10,
});
- assert.equal(durable.throughSequence, 1);
+ assert.equal(durable.throughSequence, await read.readDurableHighWater(session.id));
+ assert.ok(durable.throughSequence !== null);
assert.deepEqual(
durable.fragments.map((fragment) => {
const message = JSON.parse(fragment.data.toString('utf8')) as StoredMessage;
@@ -268,6 +272,285 @@ test('keeps durable history separate from the canonical active overlay', async (
}
});
+test('pages the ledger without materializing off-page Turns or messages', async (t) => {
+ const base = await mkdtemp(join(tmpdir(), 'maka-transcript-seek-'));
+ const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' });
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ try {
+ const stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ const session = await stores.sessionStore.create({
+ cwd: capability.canonicalPath,
+ llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc',
+ llmConnectionSlug: 'fake',
+ model: 'fake-model',
+ permissionMode: 'ask',
+ });
+ const expected: StoredMessage[] = [];
+ for (let turn = 0; turn < 5; turn++) {
+ const runId = `run-${turn}`;
+ const turnId = `turn-${turn}`;
+ await seedInvocation(stores.runtimeEventStore, {
+ sessionId: session.id,
+ runId,
+ turnId,
+ openedAt: turn,
+ });
+ let count = 0;
+ const append = (overrides: Partial<RuntimeEvent>) =>
+ stores.runtimeEventStore.appendRuntimeEvent(
+ session.id,
+ runId,
+ runtimeEvent(session.id, {
+ id: `${runId}-event-${count++}`,
+ invocationId: runId,
+ runId,
+ turnId,
+ ...overrides,
+ }),
+ );
+ await append({
+ role: 'user',
+ author: 'user',
+ content: { kind: 'text', text: `prompt ${turn}` },
+ });
+ if (turn === 4) {
+ // More than 5 MiB in a single Turn, outside a tiny head/tail page.
+ for (let index = 0; index < 180; index++) {
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: 'x'.repeat(32 * 1024) },
+ });
+ }
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'function_call', id: 'tool-1', name: 'Read', args: {} },
+ refs: { toolCallId: 'tool-1', stepId: 'assistant-final' },
+ });
+ await append({
+ actions: {
+ permissionRequest: {
+ kind: 'tool_permission',
+ requestId: 'request-1',
+ toolUseId: 'tool-1',
+ toolName: 'Read',
+ category: 'read',
+ reason: 'custom',
+ args: {},
+ rememberForTurnAllowed: true,
+ hint: 'original permission hint',
+ },
+ },
+ });
+ await append({
+ actions: {
+ permissionDecision: {
+ requestId: 'request-1',
+ decision: 'allow',
+ rememberForTurn: true,
+ },
+ },
+ refs: { toolCallId: 'tool-1' },
+ });
+ await append({
+ role: 'tool',
+ author: 'tool',
+ content: {
+ kind: 'function_response',
+ id: 'tool-1',
+ name: 'Read',
+ result: { kind: 'text', text: 'result' },
+ isError: true,
+ },
+ });
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'thinking', text: 'before text' },
+ refs: { providerEventId: 'assistant-final' },
+ });
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: 'final answer 中文' },
+ refs: { storedMessageId: 'assistant-final' },
+ });
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'thinking', text: ' after text' },
+ refs: { providerEventId: 'assistant-final', storedMessageId: 'usage-final' },
+ actions: { tokenUsage: { input: 100, output: 25 } },
+ });
+ await append({ content: { kind: 'system_note', note: 'step_limit' } });
+ } else {
+ await append({
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: '\u3000\u00a0' },
+ });
+ }
+ await append({
+ status: 'failed',
+ actions: { endInvocation: true, stateDelta: { failureClass: 'tool_step_cap_reached' } },
+ });
+ const invocation = await stores.runtimeEventStore.readRunInvocation(session.id, runId);
+ assert.ok(invocation);
+ const projection = projectRuntimeEventsToStoredMessages(
+ await stores.runtimeEventStore.readRuntimeEvents(session.id, runId),
+ { invocations: [invocation] },
+ );
+ assert.deepEqual(projection.diagnostics, []);
+ expected.push(...projection.messages);
+ }
+ const read = createSessionTranscriptReader({
+ stores,
+ canonicalPermissionOutcomes: { readPermissionOutcome: async () => undefined },
+ });
+ // Measure actual JSON decoded, not only the eventual response size. Neither
+ // a small page, the Turn index, nor a lookup miss may decode the 5 MiB Turn.
+ let decodedBytes = 0;
+ const parse = JSON.parse;
+ const measured = t.mock.method(JSON, 'parse', (...args: Parameters<typeof JSON.parse>) => {
+ decodedBytes += Buffer.byteLength(args[0]);
+ return parse(...args);
+ });
+ const through = await read.readDurableHighWater(session.id);
+ const tail = await read.readDurablePage(session.id, {
+ direction: 'older',
+ maxBytes: 1024,
+ maxMessages: 1,
+ });
+ assert.equal(JSON.parse(tail.fragments[0]!.data.toString()).type, 'system_note');
+ const head = await read.readDurablePage(session.id, {
+ direction: 'newer',
+ maxBytes: 1024,
+ maxMessages: 1,
+ });
+ assert.equal(JSON.parse(head.fragments[0]!.data.toString()).text, 'prompt 0');
+ assert.deepEqual(
+ await read.readDurableMessagesById(session.id, {
+ throughSequence: through,
+ messageIds: ['missing-stream'],
+ maxBytes: 1024,
+ maxMessages: 1,
+ }),
+ [],
+ );
+ const landmarks = await read.readDurableTurnLandmarks(session.id, 3);
+ assert.deepEqual(
+ landmarks.landmarks.map((item) => item.label),
+ ['prompt 0', 'prompt 2', 'prompt 4'],
+ );
+ const contributions: SessionTurnContribution[] = [];
+ let contributionPosition = 0;
+ for (;;) {
+ const page = await read.readDurableTurnContributions(
+ session.id,
+ through,
+ contributionPosition,
+ 2,
+ );
+ contributions.push(...page.contributions);
+ if (page.nextPosition === null) break;
+ contributionPosition = page.nextPosition;
+ }
+ assert.ok(decodedBytes < 512 * 1024, `decoded ${decodedBytes} bytes for bounded reads`);
+ measured.mock.restore();
+
+ const records: Array<{ sequence: number; message: StoredMessage }> = [];
+ let position = 0;
+ for (;;) {
+ const page = await read.readDurableRecords(session.id, {
+ direction: 'newer',
+ throughSequence: through,
+ position,
+ maxMessages: 2,
+ maxStoredBytes: 128 * 1024,
+ });
+ records.push(...page.records);
+ if (page.nextPosition === null) break;
+ position = page.nextPosition;
+ }
+ assert.deepEqual(
+ records.map((record) => record.message),
+ expected,
+ );
+ const folded = new Map<string, SessionTurnContribution>();
+ for (const record of records) {
+ if (!('turnId' in record.message) || !record.message.turnId) continue;
+ const turnId = record.message.turnId;
+ folded.set(
+ turnId,
+ foldTurnContribution(folded.get(turnId), turnId, record.sequence, record.message),
+ );
+ }
+ assert.deepEqual(contributions, [...folded.values()]);
+ const assistant = records.find((record) => record.message.id === 'assistant-final');
+ assert.ok(assistant);
+ assert.deepEqual(
+ await read.readDurableMessagesById(session.id, {
+ throughSequence: through,
+ messageIds: ['assistant-final'],
+ maxBytes: 4096,
+ maxMessages: 1,
+ }),
+ [assistant.message],
+ );
+ // Reassemble the same multibyte message in either direction, inside one row.
+ for (const direction of ['older', 'newer'] as const) {
+ let byteOffset: number | undefined;
+ const chunks: Buffer[] = [];
+ for (;;) {
+ const page = await read.readDurablePage(session.id, {
+ direction,
+ throughSequence: through,
+ position: assistant.sequence,
+ ...(byteOffset === undefined ? {} : { byteOffset }),
+ maxBytes: 37,
+ maxMessages: 1,
+ });
+ chunks.push(page.fragments[0]!.data);
+ if (page.next?.position !== assistant.sequence) break;
+ assert.notEqual(page.next.byteOffset, null);
+ byteOffset = page.next.byteOffset!;
+ }
+ if (direction === 'older') chunks.reverse();
+ assert.deepEqual(JSON.parse(Buffer.concat(chunks).toString()), assistant.message);
+ }
+ // A later sealed Turn must not alter a previously issued snapshot.
+ await seedInvocation(stores.runtimeEventStore, {
+ sessionId: session.id,
+ runId: 'later',
+ turnId: 'later',
+ openedAt: 99,
+ });
+ await stores.runtimeEventStore.appendRuntimeEvent(
+ session.id,
+ 'later',
+ runtimeEvent(session.id, {
+ id: 'later-terminal',
+ invocationId: 'later',
+ runId: 'later',
+ turnId: 'later',
+ status: 'completed',
+ }),
+ );
+ const frozen = await read.readDurablePage(session.id, {
+ direction: 'older',
+ throughSequence: through,
+ maxBytes: 1024,
+ maxMessages: 1,
+ });
+ assert.deepEqual(frozen.fragments, tail.fragments);
+ } finally {
+ await owner.close();
+ await rm(base, { recursive: true, force: true });
+ }
+});
+
test('stops scanning a control-only ledger at the cumulative immutable event limit', async () => {
const sessionId = 'session-1';
const events = Array.from({ length: 8_193 }, (_, index) =>
diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts
index 4ec1d4da53..a1c40a7709 100644
--- a/packages/runtime-host/src/protocol/index.ts
+++ b/packages/runtime-host/src/protocol/index.ts
@@ -101,7 +101,8 @@ 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 = 121 as const;
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 122 as const;
+// 122: Durable transcript cursors seek Session event ordinals instead of run indexes.
// 121: Host diagnostics report `upgradeBlockingActivity`, the Host's
// authoritative activity answer for maintenance probes, computed by the same
// authority that gates `host.upgrade.prepare`. Older Clients reject the
diff --git a/packages/runtime-host/src/server/session-transcript-reader.ts b/packages/runtime-host/src/server/session-transcript-reader.ts
index 19c5eec280..87ea53765b 100644
--- a/packages/runtime-host/src/server/session-transcript-reader.ts
+++ b/packages/runtime-host/src/server/session-transcript-reader.ts
@@ -18,21 +18,19 @@
*/
import type { RuntimeEvent } from '@maka/core/runtime-event';
+import { readRunInvocation } from '@maka/core/runtime-event-store';
import type { StoredMessage } from '@maka/core/session';
import {
activePresentationRuntimeEvents,
affectsRuntimeEventStoredMessageProjection,
isHardRuntimeEventReadModelDiagnostic,
projectRuntimeEventsToStoredMessages,
+ projectRuntimeEventUserMessage,
} from '@maka/runtime/runtime-event-read-model';
import {
type CanonicalPermissionOutcomeReader,
type CanonicalPermissionOutcomeRecord,
} from '@maka/runtime/interaction-authority';
-import {
- isSessionInlineInvocation,
- type RuntimeInvocationRecord,
-} from '@maka/core/runtime-invocation';
import type {
ExecutionStoresWriter,
SessionTranscriptMessageLookupRequest,
@@ -45,13 +43,14 @@ import type {
SessionTurnContributionPage,
SessionTurnLandmark,
SessionTurnLandmarkSnapshot,
+ RuntimeTranscriptSource,
} from '@maka/storage/execution-stores';
import { foldTurnContribution } from '@maka/storage/session-message-projection';
import { SESSION_TRANSCRIPT_OVERLAY_MAX_MESSAGES, type TurnSnapshot } from '../protocol/index.js';
const PERMISSION_OUTCOME_READ_CONCURRENCY = 8;
-/** Sequence room reserved for one invocation's projected transcript rows. */
-const RUN_SEQUENCE_STRIDE = 1 << 20;
+/** One event can emit content, a permission, usage, and terminal/notice rows. */
+const EVENT_SEQUENCE_STRIDE = 8;
export const ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES = SESSION_TRANSCRIPT_OVERLAY_MAX_MESSAGES;
export const ACTIVE_TRANSCRIPT_OVERLAY_MAX_BYTES = 16 * 1024 * 1024;
const ACTIVE_TRANSCRIPT_SOURCE_MAX_EVENTS = ACTIVE_TRANSCRIPT_OVERLAY_MAX_MESSAGES * 2;
@@ -104,7 +103,11 @@ export function createSessionTranscriptReader(input: {
readActiveOverlay: async (sessionId, rootTurn) => {
if (!rootTurn || isTerminalTurn(rootTurn)) return [];
- const invocations = await input.stores.runtimeEventStore.listSessionInvocations(sessionId);
+ const invocation = await readRunInvocation(
+ input.stores.runtimeEventStore,
+ sessionId,
+ rootTurn.runId,
+ );
const events = await readActiveProjectionEvents(input.stores, sessionId, rootTurn.runId);
const canonicalPermissionOutcomes = await readCanonicalPermissionOutcomes(
events,
@@ -113,7 +116,7 @@ export function createSessionTranscriptReader(input: {
const projected = projectRuntimeEventsToStoredMessages(
activePresentationRuntimeEvents(events),
{
- invocations: invocations.filter((invocation) => invocation.runId === rootTurn.runId),
+ invocations: invocation ? [invocation] : [],
canonicalPermissionOutcomes,
},
);
@@ -157,89 +160,48 @@ export interface SessionTranscriptReader {
}
/**
- * The settled part of a Session's transcript, read off the RuntimeEvent ledger.
- *
- * A page is bounded by reading one ended invocation at a time: a Session with a
- * thousand Turns costs the same per page as one with three. Sequences are
- * `runIndex * RUN_SEQUENCE_STRIDE + indexWithinRun`, which is monotone in the
- * order the read model presents runs and derivable from the invocation list
- * alone — so locating a page never has to project the Turns before it. They are
- * stable for as long as a subscription lives, which is exactly as long as the
- * signed cursors that carry them.
+ * Pages seek immutable Session event ordinals before decoding payloads. Each
+ * event is projected with just its indexed message context, so neither a long
+ * Session nor a long Turn has to be loaded to serve a page. The low sequence
+ * bits distinguish the few rows one event can emit.
*/
function createDurableLedgerTranscriptReader(input: {
stores: ExecutionStoresWriter<'interactive'>;
canonicalPermissionOutcomes: CanonicalPermissionOutcomeReader;
}) {
- const endedInvocations = async (sessionId: string): Promise<RuntimeInvocationRecord[]> =>
- (await input.stores.runtimeEventStore.listSessionInvocations(sessionId)).filter(
- (invocation) => isSessionInlineInvocation(invocation.opening) && invocation.terminalEvent,
- );
-
- const projectRun = async (
- sessionId: string,
- invocation: RuntimeInvocationRecord,
+ const store = input.stores.runtimeEventStore;
+ const highWater = async (sessionId: string): Promise<number | null> => {
+ const ordinal = await store.readTranscriptSourceHighWater(sessionId);
+ return ordinal === null ? null : ordinal * EVENT_SEQUENCE_STRIDE + EVENT_SEQUENCE_STRIDE - 1;
+ };
+ const projectSource = async (
+ source: RuntimeTranscriptSource,
): Promise<readonly StoredMessage[]> => {
- const events = await input.stores.runtimeEventStore.readRuntimeEvents(
- sessionId,
- invocation.runId,
- );
- const projected = projectRuntimeEventsToStoredMessages(events, {
- invocations: [invocation],
+ const event = source.event;
+ const projected = projectRuntimeEventsToStoredMessages(source.events, {
+ invocations: [source.invocation],
canonicalPermissionOutcomes: await readCanonicalPermissionOutcomes(
- events,
+ source.events,
input.canonicalPermissionOutcomes,
),
+ context: {
+ messageId: event.refs?.storedMessageId ?? event.refs?.providerEventId ?? event.id,
+ ...(source.contentOrder ? { contentOrder: source.contentOrder } : {}),
+ ...(source.permissionRequest ? { permissionRequest: source.permissionRequest } : {}),
+ ...(source.toolName ? { toolName: source.toolName } : {}),
+ ...(event.refs?.toolCallId ? { toolUseId: event.refs.toolCallId } : {}),
+ hasRetainedOutput: source.hasRetainedOutput,
+ },
});
if (projected.diagnostics.some(isHardRuntimeEventReadModelDiagnostic)) {
throw new Error('Durable RuntimeEvent transcript projection is incomplete');
}
- if (projected.messages.length > RUN_SEQUENCE_STRIDE) {
- throw new Error('Durable Session Turn exceeds its transcript sequence stride');
+ if (projected.messages.length > EVENT_SEQUENCE_STRIDE) {
+ throw new Error('RuntimeEvent exceeds its transcript sequence stride');
}
return projected.messages;
};
- /**
- * The runs a page must visit, in traversal order, already clipped to the
- * watermark and the caller's position.
- */
- const traversal = (
- invocations: readonly RuntimeInvocationRecord[],
- direction: 'older' | 'newer',
- throughSequence: number,
- position: number,
- ): Array<{ runIndex: number; invocation: RuntimeInvocationRecord }> => {
- const highestRunIndex = Math.min(runIndexOf(throughSequence), invocations.length - 1);
- const startRunIndex = Math.min(runIndexOf(position), highestRunIndex);
- const runs: Array<{ runIndex: number; invocation: RuntimeInvocationRecord }> = [];
- if (direction === 'older') {
- for (let index = startRunIndex; index >= 0; index -= 1) {
- const invocation = invocations[index];
- if (invocation) runs.push({ runIndex: index, invocation });
- }
- return runs;
- }
- for (let index = Math.max(0, startRunIndex); index <= highestRunIndex; index += 1) {
- const invocation = invocations[index];
- if (invocation) runs.push({ runIndex: index, invocation });
- }
- return runs;
- };
-
- const highWaterOf = async (
- sessionId: string,
- invocations: readonly RuntimeInvocationRecord[],
- ): Promise<number | null> => {
- for (let index = invocations.length - 1; index >= 0; index -= 1) {
- const invocation = invocations[index]!;
- const messages = await projectRun(sessionId, invocation);
- if (messages.length > 0) return index * RUN_SEQUENCE_STRIDE + messages.length - 1;
- }
- return null;
- };
-
- /** Every projected record of the requested page, ordered for its direction. */
const scan = async function* (
sessionId: string,
request: {
@@ -248,38 +210,37 @@ function createDurableLedgerTranscriptReader(input: {
position?: number;
},
): AsyncGenerator<{ sequence: number; message: StoredMessage }> {
- const invocations = await endedInvocations(sessionId);
const throughSequence =
- request.throughSequence === undefined
- ? await highWaterOf(sessionId, invocations)
- : request.throughSequence;
+ request.throughSequence === undefined ? await highWater(sessionId) : request.throughSequence;
if (throughSequence === null) return;
const position = request.position ?? (request.direction === 'older' ? throughSequence : 0);
- for (const { runIndex, invocation } of traversal(
- invocations,
- request.direction,
- throughSequence,
- position,
- )) {
- const messages = await projectRun(sessionId, invocation);
- const indexed = messages.map((message, index) => ({
- sequence: runIndex * RUN_SEQUENCE_STRIDE + index,
- message,
- }));
- const selected = indexed.filter(
- ({ sequence }) =>
- sequence <= throughSequence &&
- (request.direction === 'older' ? sequence <= position : sequence >= position),
- );
- if (request.direction === 'older') selected.reverse();
- yield* selected;
+ let ordinal = ordinalOf(position);
+ while (ordinal >= 0 && ordinal <= ordinalOf(throughSequence)) {
+ const source = await store.readTranscriptSource(sessionId, {
+ direction: request.direction,
+ throughOrdinal: ordinalOf(throughSequence),
+ position: ordinal,
+ });
+ if (!source) return;
+ const messages = await projectSource(source);
+ const records = messages
+ .map((message, index) => ({
+ sequence: source.ordinal * EVENT_SEQUENCE_STRIDE + index,
+ message,
+ }))
+ .filter(
+ ({ sequence }) =>
+ sequence <= throughSequence &&
+ (request.direction === 'older' ? sequence <= position : sequence >= position),
+ );
+ if (request.direction === 'older') records.reverse();
+ yield* records;
+ ordinal = source.ordinal + (request.direction === 'older' ? -1 : 1);
}
};
return {
- async readHighWater(sessionId: string): Promise<number | null> {
- return highWaterOf(sessionId, await endedInvocations(sessionId));
- },
+ readHighWater: highWater,
async readPage(
sessionId: string,
@@ -363,77 +324,95 @@ function createDurableLedgerTranscriptReader(input: {
return { throughSequence, records, nextPosition };
},
- /**
- * What each Turn contributed, one ended invocation at a time.
- *
- * An invocation is a Turn, so the run listing is the index: a page costs the
- * runs it actually summarizes, never a scan of the Turns before them.
- */
+ /** Fold indexed Turn facts, loading only the prompt and terminal payloads. */
async readTurnContributions(
sessionId: string,
throughSequence: number | null,
position: number,
maxContributions: number,
): Promise<SessionTurnContributionPage> {
- const invocations = await endedInvocations(sessionId);
- const watermark = throughSequence ?? (await highWaterOf(sessionId, invocations));
+ const watermark = throughSequence ?? (await highWater(sessionId));
if (watermark === null) {
return { throughSequence: null, contributions: [], nextPosition: null };
}
+ const turns = await store.readTranscriptTurns(
+ sessionId,
+ ordinalOf(watermark),
+ ordinalOf(position),
+ maxContributions + 1,
+ );
const contributions: SessionTurnContribution[] = [];
- const lastRunIndex = Math.min(runIndexOf(watermark), invocations.length - 1);
- let runIndex = Math.max(0, runIndexOf(position));
- for (; runIndex <= lastRunIndex; runIndex += 1) {
- if (contributions.length >= maxContributions) {
- return {
- throughSequence: watermark,
- contributions,
- nextPosition: runIndex * RUN_SEQUENCE_STRIDE,
- };
+ for (const turn of turns.slice(0, maxContributions)) {
+ let contribution: SessionTurnContribution = {
+ turnId: turn.invocation.turnId,
+ firstSequence: Math.max(position, turn.firstOrdinal * EVENT_SEQUENCE_STRIDE),
+ latestState: null,
+ userPromptPreview: null,
+ hasAssistantMessage: turn.hasAssistantMessage,
+ hasAssistantOutput: turn.hasAssistantOutput,
+ hasToolResult: turn.hasToolResult,
+ hasFailedToolResult: turn.hasFailedToolResult,
+ hasAbortNote: turn.hasAbortNote,
+ };
+ if (turn.user) {
+ const user = projectRuntimeEventUserMessage(turn.user.event, turn.user.event.id);
+ if (user)
+ contribution = foldTurnContribution(
+ contribution,
+ turn.invocation.turnId,
+ turn.user.ordinal * EVENT_SEQUENCE_STRIDE,
+ user,
+ );
}
- const invocation = invocations[runIndex];
- if (!invocation) continue;
- const messages = await projectRun(sessionId, invocation);
- let contribution: SessionTurnContribution | undefined;
- for (const [index, message] of messages.entries()) {
- const sequence = runIndex * RUN_SEQUENCE_STRIDE + index;
- if (sequence > watermark || sequence < position) continue;
- contribution = foldTurnContribution(contribution, invocation.turnId, sequence, message);
+ const source = await store.readTranscriptSource(sessionId, {
+ direction: 'newer',
+ throughOrdinal: ordinalOf(watermark),
+ position: turn.terminalOrdinal,
+ });
+ if (source?.ordinal === turn.terminalOrdinal) {
+ const messages = await projectSource(source);
+ for (const [index, message] of messages.entries()) {
+ const sequence = source.ordinal * EVENT_SEQUENCE_STRIDE + index;
+ if (sequence < position || sequence > watermark) continue;
+ contribution = foldTurnContribution(
+ contribution,
+ turn.invocation.turnId,
+ sequence,
+ message,
+ );
+ }
}
- if (contribution) contributions.push(contribution);
+ contributions.push(contribution);
}
- return { throughSequence: watermark, contributions, nextPosition: null };
+ const next = turns[maxContributions];
+ return {
+ throughSequence: watermark,
+ contributions,
+ nextPosition: next ? next.firstOrdinal * EVENT_SEQUENCE_STRIDE : null,
+ };
},
- /** Evenly spaced Turn starts, sampled from the run listing itself. */
+ /** Evenly spaced Turn starts, selected in SQL before loading their prompts. */
async readTurnLandmarks(
sessionId: string,
maxLandmarks: number,
): Promise<SessionTurnLandmarkSnapshot> {
- const invocations = await endedInvocations(sessionId);
- const throughSequence = await highWaterOf(sessionId, invocations);
+ const throughSequence = await highWater(sessionId);
if (throughSequence === null) return { throughSequence: null, landmarks: [] };
- const lastRunIndex = Math.min(runIndexOf(throughSequence), invocations.length - 1);
- const count = Math.min(maxLandmarks, lastRunIndex + 1);
- const sampled =
- count <= 0
- ? []
- : Array.from({ length: count }, (_, index) =>
- count === 1 ? lastRunIndex : Math.floor((lastRunIndex * index) / (count - 1)),
- );
+ const turns = await store.readTranscriptLandmarks(
+ sessionId,
+ ordinalOf(throughSequence),
+ maxLandmarks,
+ );
const landmarks: SessionTurnLandmark[] = [];
- for (const runIndex of [...new Set(sampled)]) {
- const invocation = invocations[runIndex];
- if (!invocation) continue;
- const messages = await projectRun(sessionId, invocation);
- const index = messages.findIndex((message) => message.type === 'user');
- const message = index < 0 ? undefined : messages[index];
- if (message?.type !== 'user') continue;
- const label = (message.displayText ?? message.text).trim();
+ for (const turn of turns) {
+ if (!turn.user) continue;
+ const message = projectRuntimeEventUserMessage(turn.user.event, turn.user.event.id);
+ const label = (message?.displayText ?? message?.text ?? '').trim();
if (!label) continue;
landmarks.push({
- turnId: invocation.turnId,
- sequence: runIndex * RUN_SEQUENCE_STRIDE + index,
+ turnId: turn.invocation.turnId,
+ sequence: turn.user.ordinal * EVENT_SEQUENCE_STRIDE,
label,
});
}
@@ -445,33 +424,33 @@ function createDurableLedgerTranscriptReader(input: {
request: SessionTranscriptMessageLookupRequest,
): Promise<StoredMessage[]> {
if (request.throughSequence === null || request.messageIds.length === 0) return [];
- const wanted = new Set(request.messageIds);
- const found: StoredMessage[] = [];
+ const found: Array<{ sequence: number; message: StoredMessage }> = [];
let bytes = 0;
- // Callers look up streams that were active a moment ago, so a durable copy
- // can only be in the run that just sealed. Without this bound the ordinary
- // miss — the run is still open — reprojects every Turn in the Session.
- let newestRunIndex: number | undefined;
- for await (const record of scan(sessionId, {
- direction: 'older',
- throughSequence: request.throughSequence,
- })) {
- newestRunIndex ??= runIndexOf(record.sequence);
- if (runIndexOf(record.sequence) < newestRunIndex) break;
- if (!wanted.delete(record.message.id)) continue;
- bytes += Buffer.byteLength(JSON.stringify(record.message), 'utf8');
- if (found.length >= request.maxMessages || bytes > request.maxBytes) break;
- found.push(record.message);
- if (wanted.size === 0) break;
+ for (const messageId of new Set(request.messageIds)) {
+ const source = await store.readTranscriptSource(sessionId, {
+ direction: 'older',
+ throughOrdinal: ordinalOf(request.throughSequence),
+ position: ordinalOf(request.throughSequence),
+ messageId,
+ });
+ if (!source) continue;
+ for (const [index, message] of (await projectSource(source)).entries()) {
+ const sequence = source.ordinal * EVENT_SEQUENCE_STRIDE + index;
+ if (message.id !== messageId || sequence > request.throughSequence) continue;
+ bytes += Buffer.byteLength(JSON.stringify(message), 'utf8');
+ if (found.length >= request.maxMessages || bytes > request.maxBytes) {
+ return found.sort((a, b) => a.sequence - b.sequence).map((record) => record.message);
+ }
+ found.push({ sequence, message });
+ }
}
- // Restore transcript order: the scan walked backwards to find them.
- return found.reverse();
+ return found.sort((a, b) => a.sequence - b.sequence).map((record) => record.message);
},
};
}
-function runIndexOf(sequence: number): number {
- return Math.floor(sequence / RUN_SEQUENCE_STRIDE);
+function ordinalOf(sequence: number): number {
+ return Math.floor(sequence / EVENT_SEQUENCE_STRIDE);
}
function assertActiveOverlayBounded(messages: readonly StoredMessage[]): void {
diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts
index 9aa5f21deb..819d19fceb 100644
--- a/packages/runtime/src/runtime-event-read-model.ts
+++ b/packages/runtime/src/runtime-event-read-model.ts
@@ -152,6 +152,15 @@ export interface ProjectRuntimeEventsToStoredMessagesOptions {
| readonly RuntimeInvocationRecord[]
| Readonly<Record<string, RuntimeInvocationRecord>>;
canonicalPermissionOutcomes?: ReadonlyMap<string, CanonicalPermissionOutcomeRecord>;
+ /** Facts read by indexed lookup when projecting one durable message. */
+ context?: {
+ messageId: string;
+ contentOrder?: readonly AssistantStepContentKind[];
+ permissionRequest?: RuntimeEvent;
+ toolName?: string;
+ toolUseId?: string;
+ hasRetainedOutput: boolean;
+ };
}
export interface ArchivedToolResultReadModelStatus {
@@ -204,6 +213,7 @@ interface ProjectionState {
*/
thinkingByMessageId: Map<string, PendingThinking[]>;
contentOrderByMessageId: Map<string, AssistantStepContentKind[]>;
+ hasRetainedOutput?: boolean;
}
interface PendingThinking {
@@ -228,6 +238,29 @@ export function projectRuntimeEventsToStoredMessages(
};
const messages: StoredMessage[] = [];
+ const context = options.context;
+ if (context) {
+ state.hasRetainedOutput = context.hasRetainedOutput;
+ if (context.contentOrder)
+ state.contentOrderByMessageId.set(context.messageId, [...context.contentOrder]);
+ if (context.toolName && context.toolUseId)
+ state.toolNameByUseId.set(context.toolUseId, context.toolName);
+ const requestEvent = context.permissionRequest;
+ const request = requestEvent?.actions?.permissionRequest;
+ if (request && requestEvent) {
+ state.permissionRequestById.set(request.requestId, {
+ requestId: request.requestId,
+ toolUseId: request.toolUseId,
+ toolName: request.toolName,
+ sessionId: requestEvent.sessionId,
+ runId: requestEvent.runId,
+ turnId: requestEvent.turnId,
+ ...(request.hint !== undefined ? { hint: request.hint } : {}),
+ });
+ state.toolNameByUseId.set(request.toolUseId, request.toolName);
+ }
+ }
+
for (const event of events) {
recordStepContentOrder(event, state);
if (isPartialRuntimeEvent(event)) {
@@ -1240,12 +1273,14 @@ function projectTerminalTurnState(
}
const abortSource = status === 'aborted' ? abortSourceFromRuntime(event) : undefined;
const failureClass = status === 'failed' ? failureClassFromRuntimeEvent(event) : undefined;
- const partialOutputRetained = messages.some(
- (message) =>
- message.turnId === event.turnId &&
- ((message.type === 'assistant' && message.text.trim().length > 0) ||
- message.type === 'tool_result'),
- );
+ const partialOutputRetained =
+ state.hasRetainedOutput ??
+ messages.some(
+ (message) =>
+ message.turnId === event.turnId &&
+ ((message.type === 'assistant' && message.text.trim().length > 0) ||
+ message.type === 'tool_result'),
+ );
messages.push({
type: 'turn_state',
id: stableMessageId(event, state, 'turn_state'),
diff --git a/packages/storage/src/__tests__/invocation-opening-backfill.test.ts b/packages/storage/src/__tests__/invocation-opening-backfill.test.ts
index 3e8a5d1610..9f5e1b3a01 100644
--- a/packages/storage/src/__tests__/invocation-opening-backfill.test.ts
+++ b/packages/storage/src/__tests__/invocation-opening-backfill.test.ts
@@ -462,7 +462,7 @@ describe('invocation opening fact backfill', () => {
'INSERT INTO core_agent_runs(session_id, run_id, created_at, record_json) VALUES (?, ?, ?, ?)',
).run(corrupt.sessionId, corrupt.runId, corrupt.createdAt, JSON.stringify(corrupt));
assert.throws(() => migrateSqliteRuntimeDatabase(db), /session-1\/run-corrupt-root/);
- assert.equal(readUserVersion(db), SQLITE_RUNTIME_SCHEMA_VERSION - 1);
+ assert.equal(readUserVersion(db), 15);
const openings = db
.prepare(
"SELECT COUNT(*) AS total FROM runtime_events WHERE event_kind = 'invocation_opened'",
@@ -509,7 +509,8 @@ function rewindToHeaderEra(db: DatabaseSync): void {
'ALTER TABLE runtime_continuation_claims RENAME COLUMN target_opening_json TO target_run_header_json',
);
db.exec('ALTER TABLE core_agent_runs ADD COLUMN record_json TEXT');
- db.exec(`PRAGMA user_version = ${SQLITE_RUNTIME_SCHEMA_VERSION - 1}`);
+ // Opening facts were introduced in v16, regardless of the current version.
+ db.exec('PRAGMA user_version = 15');
}
function readUserVersion(db: DatabaseSync): number {
diff --git a/packages/storage/src/__tests__/sqlite-runtime-schema.test.ts b/packages/storage/src/__tests__/sqlite-runtime-schema.test.ts
index 5bdd3984b7..b4e3548893 100644
--- a/packages/storage/src/__tests__/sqlite-runtime-schema.test.ts
+++ b/packages/storage/src/__tests__/sqlite-runtime-schema.test.ts
@@ -155,7 +155,7 @@ describe('SQLite runtime schema migration', () => {
migrateSqliteRuntimeDatabase(db);
- assert.equal(SQLITE_RUNTIME_SCHEMA_VERSION, 16);
+ assert.equal(SQLITE_RUNTIME_SCHEMA_VERSION, 17);
assert.equal(
(
db
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index 0343df2cc1..1488d2ec72 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -20,6 +20,7 @@
import type { AgentRunEvent, AgentRunEventType, AgentRunProjectionKey } from '@maka/core/agent-run';
import type { RuntimeEvent, ToolBoundaryProtocol } from '@maka/core/runtime-event';
import type { RuntimeContinuationAuthorityStore } from '@maka/core/runtime-event-store';
+import type { RuntimeTranscriptQueries } from './runtime-transcript-query.js';
import type {
RuntimeInvocationPageInput,
RuntimeInvocationPageResult,
@@ -135,8 +136,10 @@ export type {
} from './session-store.js';
export type ExecutionSessionWriter = SessionAuthorityStore;
+export type { RuntimeTranscriptSource, RuntimeTranscriptTurn } from './runtime-transcript-query.js';
export type ExecutionAgentRunWriter = DurableAgentRunStore;
export type ExecutionRuntimeEventWriter = DurableRuntimeEventStore &
+ RuntimeTranscriptQueries &
RuntimeContinuationAuthorityStore & {
readonly toolBoundaryProtocol: ToolBoundaryProtocol;
commitToolPrepared(input: CommitToolPreparedInput): Promise<ToolCommitResult>;
@@ -569,6 +572,16 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
run(() => runtimeEventStore.readSessionRuntimeEvents(sessionId)),
readSessionRuntimeEventEntries: (sessionId) =>
run(() => runtimeEventStore.readSessionRuntimeEventEntries(sessionId)),
+ readTranscriptSourceHighWater: (sessionId) =>
+ run(() => runtimeEventStore.readTranscriptSourceHighWater(sessionId)),
+ readTranscriptSource: (sessionId, request) =>
+ run(() => runtimeEventStore.readTranscriptSource(sessionId, request)),
+ readTranscriptTurns: (sessionId, throughOrdinal, position, limit) =>
+ run(() =>
+ runtimeEventStore.readTranscriptTurns(sessionId, throughOrdinal, position, limit),
+ ),
+ readTranscriptLandmarks: (sessionId, throughOrdinal, limit) =>
+ run(() => runtimeEventStore.readTranscriptLandmarks(sessionId, throughOrdinal, limit)),
claimContinuation: (input) => run(() => runtimeEventStore.claimContinuation(input)),
readContinuationClaimByBoundary: (boundaryDigest) =>
run(() => runtimeEventStore.readContinuationClaimByBoundary(boundaryDigest)),
diff --git a/packages/storage/src/runtime-transcript-query.ts b/packages/storage/src/runtime-transcript-query.ts
new file mode 100644
index 0000000000..ee05fd4809
--- /dev/null
+++ b/packages/storage/src/runtime-transcript-query.ts
@@ -0,0 +1,436 @@
+/*
+ * 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 {
+ decodeRuntimeEvent,
+ isTerminalRuntimeEvent,
+ type RuntimeEvent,
+} from '@maka/core/runtime-event';
+import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation';
+import type { AssistantStepContentKind } from '@maka/core/session';
+
+/** SQL counterpart of isTerminalRuntimeEvent; shared with the ledger store. */
+export const TERMINAL_RUNTIME_EVENT_SQL = `(
+ json_extract(payload_json, '$.actions.endInvocation') = 1
+ OR json_extract(payload_json, '$.status') IN ('completed', 'failed', 'aborted', 'cancelled')
+)`;
+
+export const TRANSCRIPT_MESSAGE_KEY_SQL = `CASE WHEN event_kind = 'function_call'
+ THEN json_extract(payload_json, '$.refs.stepId')
+ ELSE COALESCE(json_extract(payload_json, '$.refs.providerEventId'), json_extract(payload_json, '$.refs.storedMessageId'), event_id) END`;
+export const TRANSCRIPT_STORED_ID_SQL = `COALESCE(json_extract(payload_json, '$.refs.storedMessageId'), json_extract(payload_json, '$.refs.providerEventId'), json_extract(payload_json, '$.content.id'), event_id)`;
+/** Small indexed facts needed by a terminal row and the Turn index. */
+export const TRANSCRIPT_OUTPUT_SHAPE_SQL = `CASE
+ WHEN event_kind = 'text' AND json_extract(payload_json, '$.role') = 'model'
+ THEN CASE WHEN TRIM(json_extract(payload_json, '$.content.text'), char(9,10,11,12,13,32,160,5760,8192,8193,8194,8195,8196,8197,8198,8199,8200,8201,8202,8232,8233,8239,8287,12288,65279)) <> '' THEN 3 ELSE 1 END
+ WHEN event_kind = 'function_response' THEN CASE WHEN json_extract(payload_json, '$.content.isError') = 1 THEN 12 ELSE 4 END
+ ELSE 0 END`;
+
+export interface RuntimeTranscriptSource {
+ readonly ordinal: number;
+ readonly event: RuntimeEvent;
+ /** Only this message's thinking, in ledger order around its text event. */
+ readonly events: readonly RuntimeEvent[];
+ readonly invocation: RuntimeInvocationRecord;
+ readonly contentOrder?: readonly AssistantStepContentKind[];
+ readonly permissionRequest?: RuntimeEvent;
+ readonly toolName?: string;
+ readonly hasRetainedOutput: boolean;
+}
+
+export interface RuntimeTranscriptPosition {
+ readonly direction: 'older' | 'newer';
+ readonly throughOrdinal: number;
+ readonly position: number;
+ /** Exact lookup for the bounded live-to-durable handoff. */
+ readonly messageId?: string;
+}
+
+export interface RuntimeTranscriptTurn {
+ readonly firstOrdinal: number;
+ readonly terminalOrdinal: number;
+ readonly invocation: RuntimeInvocationRecord;
+ readonly user?: { ordinal: number; event: RuntimeEvent };
+ readonly hasAssistantMessage: boolean;
+ readonly hasAssistantOutput: boolean;
+ readonly hasToolResult: boolean;
+ readonly hasFailedToolResult: boolean;
+ readonly hasAbortNote: boolean;
+}
+
+export interface RuntimeTranscriptQueries {
+ readTranscriptSourceHighWater(sessionId: string): Promise<number | null>;
+ readTranscriptSource(
+ sessionId: string,
+ request: RuntimeTranscriptPosition,
+ ): Promise<RuntimeTranscriptSource | null>;
+ readTranscriptTurns(
+ sessionId: string,
+ throughOrdinal: number,
+ position: number,
+ limit: number,
+ ): Promise<RuntimeTranscriptTurn[]>;
+ readTranscriptLandmarks(
+ sessionId: string,
+ throughOrdinal: number,
+ limit: number,
+ ): Promise<RuntimeTranscriptTurn[]>;
+}
+
+const messageKey = (alias: string) =>
+ TRANSCRIPT_MESSAGE_KEY_SQL.replaceAll('event_kind', `${alias}.event_kind`)
+ .replaceAll('payload_json', `${alias}.payload_json`)
+ .replaceAll('event_id', `${alias}.event_id`);
+const terminal = (alias: string) =>
+ TERMINAL_RUNTIME_EVENT_SQL.replaceAll('payload_json', `${alias}.payload_json`);
+const opening = `COALESCE(json_extract(opened.payload_json, '$.content'), legacy.opening_json)`;
+const joins = `
+ FROM runtime_session_event_ordinals o
+ JOIN runtime_events e ON e.event_id = o.event_id
+ LEFT JOIN runtime_events opened ON opened.invocation_id = e.invocation_id AND opened.event_kind = 'invocation_opened'
+ LEFT JOIN runtime_legacy_invocation_openings legacy ON legacy.invocation_id = e.invocation_id`;
+const settledInline = `
+ ${opening} IS NOT NULL
+ AND (json_extract(${opening}, '$.lineage.parentRunId') IS NULL
+ OR (json_extract(${opening}, '$.source.kind') = 'continuation'
+ AND json_extract(${opening}, '$.lineage.agentId') IS NULL))
+ AND EXISTS (
+ SELECT 1 FROM runtime_events ended
+ JOIN runtime_session_event_ordinals ending ON ending.event_id = ended.event_id
+ WHERE ended.invocation_id = e.invocation_id AND ${terminal('ended')}
+ AND ending.ordinal <= :throughOrdinal
+ )`;
+// Thinking is context for its text row. An orphan must still reach the
+// projector, which reports the missing text instead of silently dropping it.
+const transcriptSource = `(
+ (json_extract(e.payload_json, '$.content') IS NOT NULL
+ AND e.event_kind <> 'invocation_opened'
+ AND (e.event_kind <> 'thinking' OR NOT EXISTS (
+ SELECT 1 FROM runtime_events text
+ WHERE text.invocation_id = e.invocation_id AND text.event_kind = 'text'
+ AND json_extract(text.payload_json, '$.role') = 'model'
+ AND COALESCE(json_extract(text.payload_json, '$.refs.storedMessageId'), json_extract(text.payload_json, '$.refs.providerEventId'), text.event_id) = ${messageKey('e')}
+ )))
+ OR json_extract(e.payload_json, '$.actions.permissionDecision') IS NOT NULL
+ OR json_extract(e.payload_json, '$.actions.permissionAnswerAccepted') IS NOT NULL
+ OR json_extract(e.payload_json, '$.actions.tokenUsage') IS NOT NULL
+ OR ${terminal('e')}
+)`;
+
+type SourceRow = {
+ ordinal: number;
+ event_id: string;
+ run_id: string;
+ invocation_id: string;
+ event_seq: number;
+};
+
+/** Queries select ledger positions before loading any message payload. No transcript is persisted. */
+export class RuntimeTranscriptQuery {
+ constructor(
+ private readonly db: DatabaseSync,
+ private readonly invocation: (sessionId: string, runId: string) => RuntimeInvocationRecord,
+ ) {}
+
+ highWater(sessionId: string): number | null {
+ return (
+ this.sourceRow(sessionId, {
+ direction: 'older',
+ throughOrdinal: Number.MAX_SAFE_INTEGER,
+ position: Number.MAX_SAFE_INTEGER,
+ })?.ordinal ?? null
+ );
+ }
+
+ private sourceRow(sessionId: string, request: RuntimeTranscriptPosition): SourceRow | undefined {
+ assertOrdinal(request.throughOrdinal);
+ assertOrdinal(request.position);
+ if (request.direction !== 'older' && request.direction !== 'newer')
+ throw new Error('Invalid transcript direction');
+ return this.db
+ .prepare(`
+ SELECT o.ordinal, e.event_id, e.run_id, e.invocation_id, e.event_seq ${joins}
+ WHERE o.session_id = :sessionId AND o.ordinal <= :throughOrdinal
+ ${
+ request.messageId === undefined
+ ? ''
+ : `AND e.event_id IN (
+ SELECT event_id FROM runtime_events WHERE session_id = :sessionId AND (${TRANSCRIPT_STORED_ID_SQL}) = :messageId
+ UNION SELECT event_id FROM runtime_events WHERE session_id = :sessionId AND event_id = :noticeEventId
+ )`
+ }
+ AND o.ordinal ${request.direction === 'older' ? '<=' : '>='} :position
+ AND ${settledInline} AND ${transcriptSource}
+ ORDER BY o.ordinal ${request.direction === 'older' ? 'DESC' : 'ASC'} LIMIT 1
+ `)
+ .get({
+ sessionId,
+ throughOrdinal: request.throughOrdinal,
+ position: request.position,
+ ...(request.messageId === undefined
+ ? {}
+ : {
+ messageId: request.messageId,
+ noticeEventId: request.messageId.endsWith(':step-limit-notice')
+ ? request.messageId.slice(0, -':step-limit-notice'.length)
+ : null,
+ }),
+ }) as SourceRow | undefined;
+ }
+
+ source(sessionId: string, request: RuntimeTranscriptPosition): RuntimeTranscriptSource | null {
+ const row = this.sourceRow(sessionId, request);
+ if (!row) return null;
+ const event = this.event(row.event_id);
+ const invocation = this.invocation(sessionId, row.run_id);
+ let primary = event;
+ if (event.content?.kind === 'thinking') {
+ const text = this.db
+ .prepare(`
+ SELECT 1 FROM runtime_events WHERE invocation_id = ? AND event_kind = 'text'
+ AND json_extract(payload_json, '$.role') = 'model'
+ AND COALESCE(json_extract(payload_json, '$.refs.storedMessageId'), json_extract(payload_json, '$.refs.providerEventId'), event_id) = ? LIMIT 1
+ `)
+ .get(
+ row.invocation_id,
+ event.refs?.providerEventId ?? event.refs?.storedMessageId ?? event.id,
+ );
+ if (text) {
+ // Its thinking is attached at the text position; any actions still own
+ // their rows at this event's position, exactly once.
+ const { content: _content, ...actionsOnly } = event;
+ primary = actionsOnly;
+ }
+ }
+ const events: Array<{ event: RuntimeEvent; sequence: number }> = [
+ { event: primary, sequence: row.event_seq },
+ ];
+ let contentOrder: AssistantStepContentKind[] | undefined;
+ if (event.role === 'model' && event.content?.kind === 'text') {
+ const id = event.refs?.storedMessageId ?? event.refs?.providerEventId ?? event.id;
+ const thinking = this.db
+ .prepare(`
+ SELECT e.event_id, e.event_seq FROM runtime_events e
+ WHERE e.invocation_id = ? AND e.event_kind = 'thinking' AND ${messageKey('e')} = ?
+ ORDER BY e.event_seq
+ `)
+ .all(row.invocation_id, id) as Array<{ event_id: string; event_seq: number }>;
+ for (const item of thinking) {
+ const { actions: _actions, status: _status, ...context } = this.event(item.event_id);
+ events.push({ event: context, sequence: item.event_seq });
+ }
+ const kinds = this.db
+ .prepare(`
+ SELECT CASE e.event_kind WHEN 'function_call' THEN 'tools' ELSE e.event_kind END AS kind,
+ MIN(e.event_seq) AS first_sequence FROM runtime_events e
+ WHERE e.invocation_id = :invocationId AND e.event_seq <= :sequence
+ AND e.event_kind IN ('text', 'thinking', 'function_call') AND json_extract(e.payload_json, '$.role') = 'model'
+ AND ${messageKey('e')} = :messageId
+ GROUP BY kind ORDER BY first_sequence
+ `)
+ .all({ invocationId: row.invocation_id, sequence: row.event_seq, messageId: id }) as Array<{
+ kind: AssistantStepContentKind;
+ }>;
+ contentOrder = kinds.map((item) => item.kind);
+ }
+ const requestId =
+ event.actions?.permissionDecision?.requestId ??
+ event.actions?.permissionAnswerAccepted?.requestId;
+ const permissionRow = requestId
+ ? (this.db
+ .prepare(`
+ SELECT event_id FROM runtime_events
+ WHERE invocation_id = ? AND event_seq <= ? AND json_extract(payload_json, '$.actions.permissionRequest.requestId') = ?
+ ORDER BY event_seq DESC LIMIT 1
+ `)
+ .get(row.invocation_id, row.event_seq, requestId) as { event_id: string } | undefined)
+ : undefined;
+ const permissionRequest = permissionRow ? this.event(permissionRow.event_id) : undefined;
+ const toolUseId =
+ event.refs?.toolCallId ?? permissionRequest?.actions?.permissionRequest?.toolUseId;
+ const toolRow =
+ requestId && toolUseId
+ ? (this.db
+ .prepare(`
+ SELECT json_extract(payload_json, '$.content.name') AS name FROM runtime_events
+ WHERE invocation_id = ? AND event_seq <= ? AND json_extract(payload_json, '$.content.id') = ?
+ AND event_kind IN ('function_call', 'function_response')
+ ORDER BY event_seq DESC LIMIT 1
+ `)
+ .get(row.invocation_id, row.event_seq, toolUseId) as { name: string } | undefined)
+ : undefined;
+ const hasRetainedOutput =
+ isTerminalRuntimeEvent(event) && this.hasShape(row.invocation_id, row.ordinal, 0, '3,4,12');
+ return {
+ ordinal: row.ordinal,
+ event,
+ invocation,
+ events: events.sort((a, b) => a.sequence - b.sequence).map((item) => item.event),
+ ...(contentOrder ? { contentOrder } : {}),
+ ...(permissionRequest ? { permissionRequest } : {}),
+ ...(toolRow?.name ? { toolName: toolRow.name } : {}),
+ hasRetainedOutput,
+ };
+ }
+
+ turns(
+ sessionId: string,
+ throughOrdinal: number,
+ position: number,
+ limit: number,
+ ): RuntimeTranscriptTurn[] {
+ assertOrdinal(throughOrdinal);
+ assertOrdinal(position);
+ const rows = this.db
+ .prepare(`
+ SELECT e.run_id, e.invocation_id, o.ordinal ${joins}
+ WHERE o.session_id = :sessionId AND o.ordinal <= :throughOrdinal
+ AND e.event_seq = 1 AND ${settledInline}
+ AND EXISTS (SELECT 1 FROM runtime_events tail JOIN runtime_session_event_ordinals t ON t.event_id = tail.event_id
+ WHERE tail.invocation_id = e.invocation_id AND t.ordinal >= :position AND t.ordinal <= :throughOrdinal AND ${transcriptSource.replaceAll('e.', 'tail.')})
+ ORDER BY o.ordinal LIMIT :limit
+ `)
+ .all({ sessionId, throughOrdinal, position, limit }) as SourceRow[];
+ return rows.map((row) => this.turn(sessionId, row, throughOrdinal, position));
+ }
+
+ landmarks(sessionId: string, throughOrdinal: number, limit: number): RuntimeTranscriptTurn[] {
+ assertOrdinal(throughOrdinal);
+ if (limit < 1) return [];
+ const rows = this.db
+ .prepare(`
+ WITH candidates AS (
+ SELECT e.run_id, e.invocation_id, o.ordinal,
+ ROW_NUMBER() OVER (ORDER BY o.ordinal) - 1 AS rank, COUNT(*) OVER () AS total
+ ${joins} WHERE o.session_id = :sessionId AND o.ordinal <= :throughOrdinal
+ AND e.event_seq = 1 AND ${settledInline}
+ ), samples(n) AS (
+ SELECT 0 UNION ALL SELECT n + 1 FROM samples WHERE n + 1 < :limit
+ )
+ SELECT DISTINCT run_id, invocation_id, ordinal FROM candidates
+ JOIN samples ON rank = CASE WHEN :limit = 1 THEN total - 1
+ ELSE CAST(n * (total - 1) / (:limit - 1) AS INTEGER) END
+ ORDER BY ordinal
+ `)
+ .all({ sessionId, throughOrdinal, limit }) as SourceRow[];
+ return rows.map((row) => this.turn(sessionId, row, throughOrdinal, 0));
+ }
+
+ private turn(
+ sessionId: string,
+ row: SourceRow,
+ throughOrdinal: number,
+ position: number,
+ ): RuntimeTranscriptTurn {
+ const bounds = this.db
+ .prepare(`
+ SELECT o.ordinal AS first
+ FROM runtime_events e JOIN runtime_session_event_ordinals o ON o.event_id = e.event_id
+ WHERE e.invocation_id = ? AND o.ordinal >= ? AND o.ordinal <= ? AND ${transcriptSource}
+ ORDER BY e.event_seq LIMIT 1
+ `)
+ .get(row.invocation_id, position, throughOrdinal) as { first: number };
+ const user = this.db
+ .prepare(`
+ SELECT o.ordinal, e.event_id FROM runtime_events e JOIN runtime_session_event_ordinals o ON o.event_id = e.event_id
+ WHERE e.invocation_id = ? AND o.ordinal >= ? AND o.ordinal <= ?
+ AND e.event_kind = 'text' AND json_extract(e.payload_json, '$.role') = 'user'
+ ORDER BY e.event_seq LIMIT 1
+ `)
+ .get(row.invocation_id, position, throughOrdinal) as
+ | { ordinal: number; event_id: string }
+ | undefined;
+ const ended = this.db
+ .prepare(`
+ SELECT o.ordinal FROM runtime_events e JOIN runtime_session_event_ordinals o ON o.event_id = e.event_id
+ WHERE e.invocation_id = ? AND o.ordinal <= ? AND ${terminal('e')}
+ ORDER BY o.ordinal DESC LIMIT 1
+ `)
+ .get(row.invocation_id, throughOrdinal) as { ordinal: number };
+ return {
+ firstOrdinal: bounds.first,
+ terminalOrdinal: ended.ordinal,
+ invocation: this.invocation(sessionId, row.run_id),
+ ...(user ? { user: { ordinal: user.ordinal, event: this.event(user.event_id) } } : {}),
+ ...this.flags(row.invocation_id, throughOrdinal, position),
+ };
+ }
+
+ private flags(invocationId: string, throughOrdinal: number, position: number) {
+ return {
+ hasAssistantMessage: this.hasShape(invocationId, throughOrdinal, position, '1,3'),
+ hasAssistantOutput: this.hasShape(invocationId, throughOrdinal, position, '3'),
+ hasToolResult: this.hasShape(invocationId, throughOrdinal, position, '4,12'),
+ hasFailedToolResult: this.hasShape(invocationId, throughOrdinal, position, '12'),
+ hasAbortNote: false,
+ };
+ }
+
+ private hasShape(
+ invocationId: string,
+ throughOrdinal: number,
+ position: number,
+ shapes: string,
+ ): boolean {
+ return (
+ this.db
+ .prepare(`
+ SELECT 1 FROM runtime_events JOIN runtime_session_event_ordinals o USING (event_id)
+ WHERE invocation_id = ? AND (${TRANSCRIPT_OUTPUT_SHAPE_SQL}) IN (${shapes})
+ AND o.ordinal >= ? AND o.ordinal <= ? LIMIT 1
+ `)
+ .get(invocationId, position, throughOrdinal) !== undefined
+ );
+ }
+
+ private event(id: string): RuntimeEvent {
+ const row = this.db
+ .prepare(
+ 'SELECT event_id, session_id, invocation_id, run_id, turn_id, payload_json FROM runtime_events WHERE event_id = ?',
+ )
+ .get(id) as
+ | {
+ event_id: string;
+ session_id: string;
+ invocation_id: string;
+ run_id: string;
+ turn_id: string;
+ payload_json: string;
+ }
+ | undefined;
+ if (!row) throw new Error(`Transcript RuntimeEvent ${id} is missing`);
+ const event = decodeRuntimeEvent(JSON.parse(row.payload_json));
+ if (
+ event.id !== row.event_id ||
+ event.sessionId !== row.session_id ||
+ event.invocationId !== row.invocation_id ||
+ event.runId !== row.run_id ||
+ event.turnId !== row.turn_id
+ ) {
+ throw new Error(`Transcript RuntimeEvent ${id} has inconsistent storage identity`);
+ }
+ return event;
+ }
+}
+
+function assertOrdinal(value: number): void {
+ if (!Number.isSafeInteger(value) || value < 0)
+ throw new Error('Invalid transcript event ordinal');
+}
diff --git a/packages/storage/src/sqlite-runtime-schema.ts b/packages/storage/src/sqlite-runtime-schema.ts
index a10532f026..3419761fa9 100644
--- a/packages/storage/src/sqlite-runtime-schema.ts
+++ b/packages/storage/src/sqlite-runtime-schema.ts
@@ -25,12 +25,18 @@ import {
} from './legacy-run-header.js';
import type { RuntimeEvent } from '@maka/core/runtime-event';
import { encodeCanonicalRuntimeEvent } from '@maka/core/canonical-runtime-event';
+import {
+ TERMINAL_RUNTIME_EVENT_SQL,
+ TRANSCRIPT_MESSAGE_KEY_SQL,
+ TRANSCRIPT_OUTPUT_SHAPE_SQL,
+ TRANSCRIPT_STORED_ID_SQL,
+} from './runtime-transcript-query.js';
import {
buildInvocationOpenedEvent,
buildSyntheticTerminalRuntimeEvent,
} from '@maka/core/runtime-invocation';
-export const SQLITE_RUNTIME_SCHEMA_VERSION = 16;
+export const SQLITE_RUNTIME_SCHEMA_VERSION = 17;
export const RUNTIME_RECOVERY_AUTHORITY_CAPABILITY = 'runtime_recovery_authority';
export const RUNTIME_RECOVERY_AUTHORITY_CAPABILITY_VERSION = 1;
export const RUNTIME_CONTINUATION_AUTHORITY_CAPABILITY = 'runtime_continuation_authority';
@@ -579,6 +585,17 @@ const MIGRATIONS: ReadonlyMap<number, string> = new Map([
ALTER TABLE runtime_continuation_claims_v16 RENAME TO runtime_continuation_claims;
`,
],
+ [
+ 17,
+ `
+ CREATE INDEX IF NOT EXISTS runtime_events_transcript_message ON runtime_events(invocation_id, (${TRANSCRIPT_MESSAGE_KEY_SQL}), event_seq);
+ CREATE INDEX IF NOT EXISTS runtime_events_transcript_output ON runtime_events(invocation_id, (${TRANSCRIPT_OUTPUT_SHAPE_SQL}), event_seq);
+ CREATE INDEX IF NOT EXISTS runtime_events_transcript_request ON runtime_events(invocation_id, json_extract(payload_json, '$.actions.permissionRequest.requestId'), event_seq);
+ CREATE INDEX IF NOT EXISTS runtime_events_transcript_tool ON runtime_events(invocation_id, json_extract(payload_json, '$.content.id'), event_seq);
+ CREATE INDEX IF NOT EXISTS runtime_events_terminal ON runtime_events(invocation_id, event_seq) WHERE ${TERMINAL_RUNTIME_EVENT_SQL};
+ CREATE INDEX IF NOT EXISTS runtime_events_transcript_stored_id ON runtime_events(session_id, (${TRANSCRIPT_STORED_ID_SQL}));
+ `,
+ ],
]);
/**
diff --git a/packages/storage/src/sqlite-runtime-store.ts b/packages/storage/src/sqlite-runtime-store.ts
index c71dcbfb01..c54fdcf934 100644
--- a/packages/storage/src/sqlite-runtime-store.ts
+++ b/packages/storage/src/sqlite-runtime-store.ts
@@ -129,24 +129,18 @@ import {
import type { OperationalStateDatabaseLease } from './operational-state-store.js';
import { immutableSteeringMessageId, isRuntimeStorageSafeId } from './runtime-event-invariants.js';
import { assertNoReservedWorkspaceAuthorityAppend } from './runtime-event-authority.js';
+import {
+ RuntimeTranscriptQuery,
+ TERMINAL_RUNTIME_EVENT_SQL,
+ type RuntimeTranscriptPosition,
+ type RuntimeTranscriptSource,
+ type RuntimeTranscriptTurn,
+} from './runtime-transcript-query.js';
export { SQLITE_RUNTIME_SCHEMA_VERSION } from './sqlite-runtime-schema.js';
export type { ToolRecoveryMode } from '@maka/core/runtime-event';
-/**
- * `isTerminalRuntimeEvent` asked in SQL.
- *
- * The TypeScript predicate stays the authority; this only lets a query find the
- * terminal event without decoding every row it passes over. Both have to say the
- * same thing, so the SQL half is written once here instead of at each query.
- */
-const TERMINAL_RUNTIME_EVENT_SQL = `(
- json_extract(payload_json, '$.actions.endInvocation') = 1
- OR json_extract(payload_json, '$.status')
- IN ('completed', 'failed', 'aborted', 'cancelled')
- )`;
-
const RUNTIME_EVENT_SCAN_BATCH_SIZE = 128;
const RUNTIME_PARTIAL_SEGMENT_TARGET_BYTES = 64 * 1024;
@@ -548,6 +542,52 @@ export class SqliteRuntimeStore
return this.readRuntimeEventsSync(sessionId, runId);
}
+ private transcriptQuery(): RuntimeTranscriptQuery {
+ return new RuntimeTranscriptQuery(this.db, (sessionId, runId) => {
+ const opening = this.readInvocationOpeningsSync(sessionId, { direction: 'asc', runId }).at(0);
+ if (!opening) throw new Error(`Transcript invocation ${runId} is missing`);
+ return this.completeInvocationRecordSync(opening);
+ });
+ }
+
+ async readTranscriptSourceHighWater(sessionId: string): Promise<number | null> {
+ assertRuntimeStorageSafeId(sessionId, 'Invalid session id');
+ return this.readTransaction(() => this.transcriptQuery().highWater(sessionId));
+ }
+
+ async readTranscriptSource(
+ sessionId: string,
+ request: RuntimeTranscriptPosition,
+ ): Promise<RuntimeTranscriptSource | null> {
+ assertRuntimeStorageSafeId(sessionId, 'Invalid session id');
+ return this.readTransaction(() => this.transcriptQuery().source(sessionId, request));
+ }
+
+ async readTranscriptTurns(
+ sessionId: string,
+ throughOrdinal: number,
+ position: number,
+ limit: number,
+ ): Promise<RuntimeTranscriptTurn[]> {
+ assertRuntimeStorageSafeId(sessionId, 'Invalid session id');
+ assertInvocationSearchLimit(limit);
+ return this.readTransaction(() =>
+ this.transcriptQuery().turns(sessionId, throughOrdinal, position, limit),
+ );
+ }
+
+ async readTranscriptLandmarks(
+ sessionId: string,
+ throughOrdinal: number,
+ limit: number,
+ ): Promise<RuntimeTranscriptTurn[]> {
+ assertRuntimeStorageSafeId(sessionId, 'Invalid session id');
+ assertInvocationSearchLimit(limit);
+ return this.readTransaction(() =>
+ this.transcriptQuery().landmarks(sessionId, throughOrdinal, limit),
+ );
+ }
+
/**
* Enumerate a Session's invocations: the opening fact names each one, and its
* highest-sequence event says whether it ended.
@@ -688,6 +728,8 @@ export class SqliteRuntimeStore
1 AS from_events
FROM runtime_events
WHERE session_id = :sessionId AND event_kind = 'invocation_opened'
+ ${options.runId === undefined ? '' : 'AND run_id = :runId'}
+ ${options.invocationId === undefined ? '' : 'AND invocation_id = :invocationId'}
UNION ALL
SELECT
NULL,
@@ -699,6 +741,8 @@ export class SqliteRuntimeStore
0
FROM runtime_legacy_invocation_openings AS legacy
WHERE legacy.session_id = :sessionId
+ ${options.runId === undefined ? '' : 'AND legacy.run_id = :runId'}
+ ${options.invocationId === undefined ? '' : 'AND legacy.invocation_id = :invocationId'}
AND NOT EXISTS (
SELECT 1 FROM runtime_events
WHERE runtime_events.invocation_id = legacy.invocation_id
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org