diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2ba22350330b7..041d4d0946b12 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2929,6 +2929,16 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const now = yield* DateTime.now; const nowIso = DateTime.formatIso(now); const threadId = ThreadId.make("thread:bounded-sql-history"); + // The raw SQL edits below stand in for committed writes, which always append a thread event. + let revision = 0; + const historyPage = (...args: Parameters) => + sql`INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`event:bounded-sql-history:revision:${++revision}`}, 'thread', ${threadId}, ${revision}, + 'turn-item.updated', ${nowIso}, 'user', '{}', '{}', 2 + )`.pipe(Effect.andThen(projectionStore.getThreadHistoryPage(...args))); yield* projectionStore.apply({ id: EventId.make("event:bounded-sql-history:thread"), type: "thread.created", @@ -3152,7 +3162,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { sourceItemId: TurnItemId.make("turn-item:bounded-sql-history:71"), position: 70, }); - const bytePage = yield* projectionStore.getThreadHistoryPage(threadId, byteCursor); + const bytePage = yield* historyPage(threadId, byteCursor); assert.isAbove(bytePage.items.length, 0); assert.isBelow(bytePage.items.length, 70); assert.isAtMost( @@ -3163,7 +3173,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { let byteNext = bytePage.nextCursor; for (let pages = 0; byteNext !== null; pages += 1) { assert.isBelow(pages, 70); - const page = yield* projectionStore.getThreadHistoryPage(threadId, byteNext); + const page = yield* historyPage(threadId, byteNext); assert.isAbove(page.items.length, 0); byteIds.unshift(...page.items.map((row) => String(row.sourceItemId))); byteNext = page.nextCursor; @@ -3269,7 +3279,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { '$.attachments', json('[]')) WHERE thread_id = ${threadId} AND ordinal BETWEEN 600 AND 624 `; - const targetedPage = yield* projectionStore.getThreadHistoryPage( + const targetedPage = yield* historyPage( threadId, encodeThreadHistoryCursor({ snapshotSequence: 0, @@ -3299,7 +3309,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { '$.type', CASE ordinal WHEN 500 THEN 'run_interrupt_request' ELSE 'run_interrupt_result' END) WHERE thread_id = ${threadId} AND ordinal IN (500, 580) `; - const interruptPage = yield* projectionStore.getThreadHistoryPage( + const interruptPage = yield* historyPage( threadId, encodeThreadHistoryCursor({ snapshotSequence: 0, @@ -3446,7 +3456,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { let workCursor: string | null = anchor.item.deferredWork!.cursor; for (let pages = 0; workCursor !== null; pages += 1) { assert.isBelow(pages, 20); - const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore.getThreadHistoryPage( + const page: OrchestrationV2ThreadHistoryPage = yield* historyPage( threadId, workCursor, anchor.sourceItemId, @@ -3475,7 +3485,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { window.projection.visibleTurnItems.find((row) => row.item.deferredWork)?.item.ordinal, 3, ); - const page = yield* projectionStore.getThreadHistoryPage( + const page = yield* historyPage( threadId, encodeThreadHistoryCursor({ snapshotSequence: 0, @@ -3532,14 +3542,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const loaded = new Set(); for (let pages = 0; cursor !== null; pages += 1) { assert.isBelow(pages, 20); - const page: OrchestrationV2ThreadHistoryPage = - yield* projectionStore.getThreadHistoryPage( - threadId, - cursor, - oldWork!.sourceItemId, - false, - "work", - ); + const page: OrchestrationV2ThreadHistoryPage = yield* historyPage( + threadId, + cursor, + oldWork!.sourceItemId, + false, + "work", + ); for (const row of page.items) loaded.add(row.sourceItemId); cursor = page.nextCursor; } @@ -4470,6 +4479,276 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("pages a whole thread without re-reading its full timeline for every page", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const nowIso = DateTime.formatIso(now); + const threadId = ThreadId.make("thread:paged-scrollback"); + const runId = RunId.make("run:paged-scrollback"); + const itemCount = 600; + yield* projectionStore.apply({ + id: EventId.make("event:paged-scrollback:thread"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:paged-scrollback"), + title: "Paged scrollback", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + yield* sql` + INSERT INTO orchestration_v2_projection_runs ( + run_id, thread_id, ordinal, provider, provider_thread_id, status, + requested_at, completed_at, payload_json + ) VALUES ( + ${runId}, ${threadId}, 1, 'codex', NULL, 'completed', ${nowIso}, ${nowIso}, + ${encodeUnknownJsonString({ + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: "message:paged-scrollback", + rootNodeId: "node:paged-scrollback", + activeAttemptId: null, + status: "completed", + requestedAt: nowIso, + startedAt: nowIso, + completedAt: nowIso, + checkpointId: null, + contextHandoffId: null, + })} + ) + `; + for (let ordinal = 0; ordinal < itemCount; ordinal += 1) { + const id = `turn-item:paged-scrollback:${ordinal}`; + const message = ordinal % 9 === 0; + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id, + parent_item_id, ordinal, type, status, updated_at, payload_json + ) VALUES ( + ${id}, ${threadId}, ${runId}, NULL, NULL, NULL, NULL, ${ordinal}, + ${message ? "user_message" : "command_execution"}, 'completed', ${nowIso}, + ${encodeUnknownJsonString({ + id, + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: `item ${ordinal}`, + startedAt: nowIso, + completedAt: nowIso, + updatedAt: nowIso, + ...(message + ? { + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId: `message:paged-scrollback:${ordinal}`, + inputIntent: "turn_start", + text: `Turn ${ordinal}`, + attachments: [], + } + : { type: "command_execution", input: "echo", output: "ok", exitCode: 0 }), + })} + ) + `; + } + // An unbounded read of every turn item in the thread costs O(thread) per page, so a full + // scrollback that repeats it on each page is quadratic in thread length. + const wholeThreadReads = (statements: ReadonlyArray) => + statements.filter( + (statement) => + /FROM orchestration_v2_projection_turn_items(\s+AS item)?\s+WHERE (item\.)?thread_id = \?\s*(ORDER BY|$)/.test( + statement, + ) && !/\bLIMIT\b/.test(statement), + ).length; + const snapshot = yield* projectionStore.getThreadSnapshot(threadId); + + const history = traceSqlStatements(); + let cursor: string | null = encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "absent", + position: itemCount, + }); + const historyIds: Array = []; + let historyPages = 0; + while (cursor !== null) { + const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore + .getThreadHistoryPage(threadId, cursor) + .pipe(Effect.withTracer(history.tracer)); + historyIds.push(...page.items.map((row) => String(row.sourceItemId))); + historyPages += 1; + cursor = page.hasMoreHistory ? page.nextCursor : null; + } + assert.strictEqual(new Set(historyIds).size, itemCount); + assert.isAtLeast(historyPages, 4); + + const timeline = traceSqlStatements(); + const timelineIds: Array = []; + let afterPosition = -1; + let timelinePages = 0; + for (let hasMore = true; hasMore;) { + const page = yield* projectionStore + .getTimelinePage(threadId, { view: "activity", afterPosition, limit: 100 }) + .pipe(Effect.withTracer(timeline.tracer)); + timelineIds.push(...page.items.map((row) => String(row.sourceItemId))); + afterPosition = page.items.at(-1)?.position ?? afterPosition; + timelinePages += 1; + hasMore = page.hasMore; + } + assert.strictEqual(new Set(timelineIds).size, itemCount); + assert.isAtLeast(timelinePages, 4); + + assert.isAtMost( + wholeThreadReads(history.statements), + 1, + `getThreadHistoryPage read the whole thread on each of ${historyPages} pages`, + ); + assert.isAtMost( + wholeThreadReads(timeline.statements), + 1, + `getTimelinePage read the whole thread on each of ${timelinePages} pages`, + ); + + // A write to the thread a fork inherits from must reach the fork's next page. + const forkThreadId = ThreadId.make("thread:paged-scrollback:fork"); + yield* projectionStore.apply({ + id: EventId.make("event:paged-scrollback:fork"), + type: "thread.created", + threadId: forkThreadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: forkThreadId, + projectId: ProjectId.make("project:paged-scrollback"), + title: "Paged scrollback fork", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: threadId, + relationshipToParent: "fork", + rootThreadId: threadId, + }, + forkedFrom: { type: "run", threadId, runId }, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + const forkPage = () => + projectionStore.getTimelinePage(forkThreadId, { view: "activity", limit: 1 }); + const forkTotal = (yield* forkPage()).totalItems; + assert.strictEqual((yield* forkPage()).totalItems, forkTotal); + yield* projectionStore.apply({ + id: EventId.make("event:paged-scrollback:late-item"), + type: "turn-item.updated", + threadId, + runId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:paged-scrollback:late"), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: itemCount, + status: "completed", + title: "late", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "command_execution", + input: "echo", + }, + }); + assert.strictEqual((yield* forkPage()).totalItems, forkTotal + 1); + + // Another store instance's write reaches this one only through the committed thread event. + const directId = "turn-item:paged-scrollback:direct"; + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id, + parent_item_id, ordinal, type, status, updated_at, payload_json + ) VALUES ( + ${directId}, ${threadId}, ${runId}, NULL, NULL, NULL, NULL, ${itemCount + 1}, + 'command_execution', 'completed', ${nowIso}, + ${encodeUnknownJsonString({ + id: directId, + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: itemCount + 1, + status: "completed", + title: "direct", + startedAt: nowIso, + completedAt: nowIso, + updatedAt: nowIso, + type: "command_execution", + input: "echo", + })} + ) + `; + yield* sql` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + 'event:paged-scrollback:direct', 'thread', ${threadId}, 1, + 'turn-item.updated', ${nowIso}, 'user', '{}', '{}', 2 + ) + `; + assert.strictEqual((yield* forkPage()).totalItems, forkTotal + 2); + }), + ); + it.effect("scopes actionable provider state to the requested bounded thread", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; @@ -4781,6 +5060,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { number: 42, url: "https://github.com/acme/project/pull/42", })})) WHERE thread_id = ${readThreadId}`; + const [latest] = yield* sql<{ readonly sequence: number }>` + SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events`; const trace = traceSqlStatements(); const discovery = yield* projectionStore .getPullRequestSnapshot({}) @@ -4791,7 +5072,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { .filter((id) => id === readThreadId || id === laterThreadId), [readThreadId], ); - assert.strictEqual(discovery.snapshotSequence, 0); + assert.strictEqual(discovery.snapshotSequence, latest?.sequence); assert.strictEqual(trace.queryTransactions.length, 2); assert.isNotNull(trace.queryTransactions[0]); assert.strictEqual(new Set(trace.queryTransactions).size, 1); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9e27210350642..4368e2fc7de9e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -2060,10 +2060,50 @@ export const layer: Layer.Layer = END `; + type TimelineIndexItem = Pick< + OrchestrationV2TurnItem, + "id" | "threadId" | "runId" | "nodeId" | "parentItemId" | "type" | "status" | "ordinal" + > & { + readonly inputIntent?: Extract< + OrchestrationV2TurnItem, + { type: "user_message" } + >["inputIntent"]; + readonly createdBy?: "user" | "agent" | "system"; + readonly messageId?: MessageId; + readonly runStatus?: string | undefined; + readonly attemptId?: string | null; + readonly attemptStatus?: string | null; + readonly persistentOutput?: boolean; + readonly deferredWork?: OrchestrationV2TurnItem["deferredWork"]; + }; + type TimelineIndexRow = { + readonly sourceThreadId: ThreadId; + readonly sourceItemId: TurnItemId; + readonly visibility: OrchestrationV2ProjectedTurnItem["visibility"]; + readonly item: TimelineIndexItem; + readonly synthetic?: OrchestrationV2TurnItem; + }; + type TimelineIndex = { + readonly records: ProjectionRecords<"runs" | "attempts">; + /** The thread and every fork source the index read. */ + readonly threadIds: ReadonlyArray; + readonly local: ReadonlyArray; + readonly visible: ReadonlyArray; + }; + // Least recently used first; a few cover clients scrolling side by side without pinning every thread. + const TIMELINE_INDEX_CACHE_LIMIT = 8; + const timelineIndexes = new Map< + string, + { readonly sequence: number; readonly index: TimelineIndex } + >(); + const apply: ProjectionStoreV2Shape["apply"] = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced( function* (event) { + for (const [key, cached] of timelineIndexes) { + if (cached.index.threadIds.includes(event.threadId)) timelineIndexes.delete(key); + } switch (event.type) { case "thread.pull-request-link-updated": { const thread = yield* getThread(event.threadId); @@ -5902,41 +5942,13 @@ export const layer: Layer.Layer = ), ); - type TimelineIndexItem = Pick< - OrchestrationV2TurnItem, - "id" | "threadId" | "runId" | "nodeId" | "parentItemId" | "type" | "status" | "ordinal" - > & { - readonly inputIntent?: Extract< - OrchestrationV2TurnItem, - { type: "user_message" } - >["inputIntent"]; - readonly createdBy?: "user" | "agent" | "system"; - readonly messageId?: MessageId; - readonly runStatus?: string | undefined; - readonly attemptId?: string | null; - readonly attemptStatus?: string | null; - readonly persistentOutput?: boolean; - readonly deferredWork?: OrchestrationV2TurnItem["deferredWork"]; - }; - type TimelineIndexRow = { - readonly sourceThreadId: ThreadId; - readonly sourceItemId: TurnItemId; - readonly visibility: OrchestrationV2ProjectedTurnItem["visibility"]; - readonly item: TimelineIndexItem; - readonly synthetic?: OrchestrationV2TurnItem; - }; const readTimelineIndex: ( threadId: ThreadId, seen: ReadonlySet, includePersistentOutput?: boolean, - ) => Effect.Effect< - { - readonly records: ProjectionRecords<"runs" | "attempts">; - readonly local: ReadonlyArray; - readonly visible: ReadonlyArray; - }, - ProjectionStoreV2Error - > = Effect.fn("ProjectionStore.readTimelineIndex")( + ) => Effect.Effect = Effect.fn( + "ProjectionStore.readTimelineIndex", + )( function* (threadId: ThreadId, seen: ReadonlySet, includePersistentOutput = false) { const records = yield* getThreadRecords(threadId, ["runs", "attempts"]); const rows = yield* sql<{ @@ -6006,7 +6018,8 @@ export const layer: Layer.Layer = }); const visible = local.filter((row) => isVisible(row.item)); const fork = inheritedRunFork(records.thread); - if (fork === null || seen.has(fork.threadId)) return { records, local, visible }; + if (fork === null || seen.has(fork.threadId)) + return { records, threadIds: [threadId], local, visible }; const source = yield* readTimelineIndex( fork.threadId, new Set([...seen, threadId]), @@ -6044,6 +6057,7 @@ export const layer: Layer.Layer = }); return { records, + threadIds: [threadId, ...source.threadIds], local, visible: [ ...inherited, @@ -6068,6 +6082,39 @@ export const layer: Layer.Layer = ), ); + // Every committed projection write appends a thread event, so the newest event across the + // index's threads identifies its snapshot; apply also drops entries for replayed writes. Call + // inside the read transaction. Without the index hint SQLite walks every newer V2 event. + const latestThreadEventSequence = (threadId: ThreadId, threadIds: ReadonlyArray) => + sql<{ sequence: number }>`SELECT COALESCE(MAX(sequence), 0) AS sequence + FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' AND stream_id IN ${sql.in(threadIds)}`.pipe( + Effect.map((rows) => rows[0]?.sequence ?? 0), + Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })), + ); + const readSnapshotTimelineIndex = Effect.fn("ProjectionStore.readSnapshotTimelineIndex")( + function* (threadId: ThreadId, includePersistentOutput: boolean) { + const key = `${includePersistentOutput ? "work" : "plain"}:${threadId}`; + const cached = timelineIndexes.get(key); + const index = + cached !== undefined && + (yield* latestThreadEventSequence(threadId, cached.index.threadIds)) === cached.sequence + ? cached.index + : yield* readTimelineIndex(threadId, new Set(), includePersistentOutput); + const sequence = + index === cached?.index + ? cached.sequence + : yield* latestThreadEventSequence(threadId, index.threadIds); + timelineIndexes.delete(key); + timelineIndexes.set(key, { sequence, index }); + const oldest = timelineIndexes.keys().next(); + if (timelineIndexes.size > TIMELINE_INDEX_CACHE_LIMIT && oldest.done !== true) + timelineIndexes.delete(oldest.value); + return index; + }, + ); + const { searchThread, searchThreadStream } = yield* makeThreadFind((threadId) => readTimelineIndex(threadId, new Set()).pipe(Effect.map((index) => index.visible)), ); @@ -6082,7 +6129,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set(), workMode !== undefined); + const index = yield* readSnapshotTimelineIndex(threadId, workMode !== undefined); if (index.records.thread.deletedAt !== null) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); const revisions = yield* sql<{ @@ -6199,7 +6246,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set()); + const index = yield* readSnapshotTimelineIndex(threadId, false); const attachmentOwners = options.attachmentId === undefined ? undefined