import { assert, it } from "@effect/vitest";import { EventId, CheckpointId, CheckpointRef, CheckpointScopeId, MessageId, type ModelSelection, NodeId, ProjectId, ProviderDriverKind, ProviderInstanceId, ProviderSessionId, ProviderThreadId, ProviderTurnId, RunAttemptId, RunId, ThreadId, TurnItemId,} from "@t3tools/contracts";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Schema from "effect/Schema";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";import { isTurnItemAtOrBeforeRun, ProjectionStoreV2, ProjectionStoreThreadNotFoundError, layer as projectionStoreLayer,} from "./ProjectionStore.ts";import { buildBoundedThreadProjection, decodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY,} from "./threadHistoryPaging.ts"; const TestLayer = Layer.mergeAll( projectionStoreLayer.pipe(Layer.provideMerge(SqlitePersistenceMemory)), SqlitePersistenceMemory,);const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4",} satisfies ModelSelection;const driver = ProviderDriverKind.make("codex");const providerInstanceId = modelSelection.instanceId;const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); it("includes imported runless history when selecting fork context through a run", () => { const firstRunId = RunId.make("run:projection-imported-fork:1"); const secondRunId = RunId.make("run:projection-imported-fork:2"); const runOrdinalById = new Map([ [firstRunId, 1], [secondRunId, 2], ]); assert.isTrue( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: null, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isFalse( isTurnItemAtOrBeforeRun({ historyOrigin: undefined, itemRunId: null, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isTrue( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: firstRunId, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isFalse( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: secondRunId, runOrdinalById, sourceRunOrdinal: 1, }), );}); it.layer(TestLayer)("ProjectionStoreV2", (it) => { it.effect("preserves stored provider usage when a terminal update omits it", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; const now = yield* DateTime.now; const threadId = ThreadId.make("thread:provider-usage-reload"); const providerThreadId = ProviderThreadId.make("provider-thread:provider-usage-reload"); const providerTurnId = ProviderTurnId.make("provider-turn:provider-usage-reload"); const nodeId = NodeId.make("node:provider-usage-reload"); const initialUsage = { usedTokens: 12_000, maxTokens: 200_000, inputTokens: 11_000, outputTokens: 1_000, updatedAt: "2026-08-29T12:00:00.000Z", } as const; const replacementUsage = { usedTokens: 18_000, maxTokens: 200_000, inputTokens: 16_000, outputTokens: 2_000, updatedAt: "2026-08-29T12:00:01.000Z", } as const; yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:thread"), type: "thread.created", threadId, occurredAt: now, payload: { createdBy: "user", creationSource: "web", id: threadId, projectId: ProjectId.make("project:provider-usage-reload"), title: "Provider usage reload", 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, }, }); const providerTurn = { id: providerTurnId, providerThreadId, nodeId, runAttemptId: null, nativeTurnRef: null, ordinal: 1, status: "running" as const, startedAt: now, completedAt: null, }; yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:running"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, tokenUsage: initialUsage }, }); yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:completed"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, status: "completed", completedAt: now }, }); const reloaded = yield* projectionStore.getThreadProjection(threadId); assert.deepEqual(reloaded.providerTurns[0]?.tokenUsage, initialUsage); assert.strictEqual(reloaded.providerTurns[0]?.status, "completed"); yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:replacement"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, status: "completed", completedAt: now, tokenUsage: replacementUsage, }, }); const replaced = yield* projectionStore.getThreadProjection(threadId); assert.deepEqual(replaced.providerTurns[0]?.tokenUsage, replacementUsage); }), ); it.effect("reads a fixed SQL turn-item window for long histories and repeated clients", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; const now = yield* DateTime.now; const nowIso = DateTime.formatIso(now); const threadId = ThreadId.make("thread:bounded-sql-history"); yield* projectionStore.apply({ id: EventId.make("event:bounded-sql-history:thread"), type: "thread.created", threadId, occurredAt: now, payload: { createdBy: "user", creationSource: "web", id: threadId, projectId: ProjectId.make("project:bounded-sql-history"), title: "Bounded SQL history", 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, }, }); for (let ordinal = 1; ordinal <= 1_000; ordinal += 1) { const id = `turn-item:bounded-sql-history:${ordinal}`; const runId = `run:bounded-sql-history:${ordinal}`; 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}, ${ordinal}, 'codex', NULL, 'completed', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: runId, threadId, ordinal, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: `message:bounded-sql-history:${ordinal}`, rootNodeId: `node:bounded-sql-history:${ordinal}`, activeAttemptId: null, status: "completed", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; 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}, 'command_execution', 'completed', ${nowIso}, ${encodeUnknownJsonString({ id, threadId, runId, nodeId: null, providerThreadId: null, providerTurnId: null, nativeItemRef: null, parentItemId: null, ordinal, status: "completed", title: `command ${ordinal}`, input: `echo ${ordinal}`, output: "ok", exitCode: 0, startedAt: nowIso, completedAt: nowIso, updatedAt: nowIso, type: "command_execution", })} ) `; } const snapshots = yield* Effect.all( Array.from({ length: 4 }, () => projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 76 }), ), { concurrency: 4 }, ); for (const snapshot of snapshots) { assert.lengthOf(snapshot.projection.turnItems, 76); assert.lengthOf(snapshot.projection.runs, 76); assert.lengthOf(snapshot.projection.visibleTurnItems, 76); assert.strictEqual(snapshot.projection.turnItems[0]?.ordinal, 925); assert.strictEqual(snapshot.projection.turnItems.at(-1)?.ordinal, 1_000); } const retainedRequestPlan = yield* sql<{ readonly detail: string }>` EXPLAIN QUERY PLAN WITH selected AS ( SELECT run_id, type FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} ORDER BY ordinal DESC, turn_item_id DESC LIMIT 77 ) SELECT request.payload_json FROM orchestration_v2_projection_turn_items AS request WHERE request.run_id IN ( SELECT run_id FROM selected WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL ) AND request.type = 'run_interrupt_request' `; assert.isTrue( retainedRequestPlan.some((row) => row.detail.includes("orchestration_v2_projection_turn_items_run_ordinal_idx"), ), ); const older = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 76, anchorItemId: TurnItemId.make("turn-item:bounded-sql-history:925"), }); assert.lengthOf(older.projection.turnItems, 76); assert.lengthOf(older.projection.runs, 76); assert.strictEqual(older.projection.turnItems[0]?.ordinal, 850); assert.strictEqual(older.projection.turnItems.at(-1)?.ordinal, 925); const sqlPageLimit = THREAD_HISTORY_PAGE_POLICY.maxItems + 2; const initialSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, }); const initialPage = buildBoundedThreadProjection({ projection: initialSnapshot.projection, snapshotSequence: initialSnapshot.snapshotSequence, }); const loadedIds = initialPage.projection.visibleTurnItems.map((row) => String(row.sourceItemId), ); let cursor = initialPage.historyCursor; let pageCount = 1; while (cursor !== null) { const anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(cursor).si); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, anchorItemId, }); const page = selectHistoryPageFromCursor({ items: snapshot.projection.visibleTurnItems, cursor, snapshotSequence: snapshot.snapshotSequence, }); loadedIds.push(...page.items.map((row) => String(row.sourceItemId))); pageCount += 1; cursor = page.nextCursor; } assert.isAtLeast(pageCount, 4); assert.lengthOf(loadedIds, 1_000); assert.strictEqual(new Set(loadedIds).size, 1_000); assert.deepEqual( loadedIds.toSorted( (left, right) => Number(left.split(":").at(-1)) - Number(right.split(":").at(-1)), ), Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`), ); const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix"); 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 ( ${hiddenRunId}, ${threadId}, 1001, 'codex', NULL, 'rolled_back', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: hiddenRunId, threadId, ordinal: 1001, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: "message:bounded-sql-history:hidden-suffix", rootNodeId: "node:bounded-sql-history:hidden-suffix", activeAttemptId: null, status: "rolled_back", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; for (let ordinal = 1_001; ordinal <= 1_100; ordinal += 1) { const id = `turn-item:bounded-sql-history:hidden:${ordinal}`; 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}, ${hiddenRunId}, NULL, NULL, NULL, NULL, ${ordinal}, 'command_execution', 'completed', ${nowIso}, ${encodeUnknownJsonString({ id, threadId, runId: hiddenRunId, nodeId: null, providerThreadId: null, providerTurnId: null, nativeItemRef: null, parentItemId: null, ordinal, status: "completed", title: "rolled back", input: "echo hidden", output: "hidden", exitCode: 0, startedAt: nowIso, completedAt: nowIso, updatedAt: nowIso, type: "command_execution", })} ) `; } const hiddenSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, }); const hiddenSuffixPage = buildBoundedThreadProjection({ projection: hiddenSuffixSnapshot.projection, snapshotSequence: hiddenSuffixSnapshot.snapshotSequence, }); assert.lengthOf(hiddenSuffixPage.projection.visibleTurnItems, 75); assert.strictEqual(hiddenSuffixPage.latestLocalTurnOrdinal, 1_100); assert.isTrue(hiddenSuffixPage.hasMoreHistory); assert.isFalse( hiddenSuffixPage.projection.visibleTurnItems.some((row) => String(row.sourceItemId).includes(":hidden:"), ), ); const hiddenSuffixCursor = hiddenSuffixPage.historyCursor; assert.isNotNull(hiddenSuffixCursor); const hiddenSuffixAnchor = TurnItemId.make(decodeThreadHistoryCursor(hiddenSuffixCursor!).si); const hiddenSuffixOlderSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, anchorItemId: hiddenSuffixAnchor, }); const hiddenSuffixOlderPage = selectHistoryPageFromCursor({ items: hiddenSuffixOlderSnapshot.projection.visibleTurnItems, cursor: hiddenSuffixCursor!, snapshotSequence: hiddenSuffixOlderSnapshot.snapshotSequence, }); assert.lengthOf(hiddenSuffixOlderPage.items, 75); assert.isTrue(hiddenSuffixOlderPage.hasMoreHistory); const cancelledRunId = RunId.make("run:bounded-sql-history:cancelled-suffix"); 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 ( ${cancelledRunId}, ${threadId}, 1002, 'codex', NULL, 'cancelled', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: cancelledRunId, threadId, ordinal: 1002, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: "message:bounded-sql-history:cancelled-suffix", rootNodeId: "node:bounded-sql-history:cancelled-suffix", activeAttemptId: null, status: "cancelled", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; for (let ordinal = 1_101; ordinal <= 1_200; ordinal += 1) { const id = `turn-item:bounded-sql-history:cancelled:${ordinal}`;Lines 1–500 of 2925. Load more
import { assert, it } from "@effect/vitest"; import { EventId, CheckpointId, CheckpointRef, CheckpointScopeId, MessageId, type ModelSelection, NodeId, ProjectId, ProviderDriverKind, ProviderInstanceId, ProviderSessionId, ProviderThreadId, ProviderTurnId, RunAttemptId, RunId, ThreadId, TurnItemId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { isTurnItemAtOrBeforeRun, ProjectionStoreV2, ProjectionStoreThreadNotFoundError, layer as projectionStoreLayer, } from "./ProjectionStore.ts"; import { buildBoundedThreadProjection, decodeThreadHistoryCursor, selectHistoryPageFromCursor, THREAD_HISTORY_PAGE_POLICY, } from "./threadHistoryPaging.ts"; const TestLayer = Layer.mergeAll( projectionStoreLayer.pipe(Layer.provideMerge(SqlitePersistenceMemory)), SqlitePersistenceMemory, ); const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4", } satisfies ModelSelection; const driver = ProviderDriverKind.make("codex"); const providerInstanceId = modelSelection.instanceId; const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); it("includes imported runless history when selecting fork context through a run", () => { const firstRunId = RunId.make("run:projection-imported-fork:1"); const secondRunId = RunId.make("run:projection-imported-fork:2"); const runOrdinalById = new Map([ [firstRunId, 1], [secondRunId, 2], ]); assert.isTrue( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: null, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isFalse( isTurnItemAtOrBeforeRun({ historyOrigin: undefined, itemRunId: null, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isTrue( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: firstRunId, runOrdinalById, sourceRunOrdinal: 1, }), ); assert.isFalse( isTurnItemAtOrBeforeRun({ historyOrigin: "v1_import", itemRunId: secondRunId, runOrdinalById, sourceRunOrdinal: 1, }), ); }); it.layer(TestLayer)("ProjectionStoreV2", (it) => { it.effect("preserves stored provider usage when a terminal update omits it", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; const now = yield* DateTime.now; const threadId = ThreadId.make("thread:provider-usage-reload"); const providerThreadId = ProviderThreadId.make("provider-thread:provider-usage-reload"); const providerTurnId = ProviderTurnId.make("provider-turn:provider-usage-reload"); const nodeId = NodeId.make("node:provider-usage-reload"); const initialUsage = { usedTokens: 12_000, maxTokens: 200_000, inputTokens: 11_000, outputTokens: 1_000, updatedAt: "2026-08-29T12:00:00.000Z", } as const; const replacementUsage = { usedTokens: 18_000, maxTokens: 200_000, inputTokens: 16_000, outputTokens: 2_000, updatedAt: "2026-08-29T12:00:01.000Z", } as const; yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:thread"), type: "thread.created", threadId, occurredAt: now, payload: { createdBy: "user", creationSource: "web", id: threadId, projectId: ProjectId.make("project:provider-usage-reload"), title: "Provider usage reload", 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, }, }); const providerTurn = { id: providerTurnId, providerThreadId, nodeId, runAttemptId: null, nativeTurnRef: null, ordinal: 1, status: "running" as const, startedAt: now, completedAt: null, }; yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:running"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, tokenUsage: initialUsage }, }); yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:completed"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, status: "completed", completedAt: now }, }); const reloaded = yield* projectionStore.getThreadProjection(threadId); assert.deepEqual(reloaded.providerTurns[0]?.tokenUsage, initialUsage); assert.strictEqual(reloaded.providerTurns[0]?.status, "completed"); yield* projectionStore.apply({ id: EventId.make("event:provider-usage-reload:replacement"), type: "provider-turn.updated", threadId, nodeId, driver, occurredAt: now, payload: { ...providerTurn, status: "completed", completedAt: now, tokenUsage: replacementUsage, }, }); const replaced = yield* projectionStore.getThreadProjection(threadId); assert.deepEqual(replaced.providerTurns[0]?.tokenUsage, replacementUsage); }), ); it.effect("reads a fixed SQL turn-item window for long histories and repeated clients", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; const now = yield* DateTime.now; const nowIso = DateTime.formatIso(now); const threadId = ThreadId.make("thread:bounded-sql-history"); yield* projectionStore.apply({ id: EventId.make("event:bounded-sql-history:thread"), type: "thread.created", threadId, occurredAt: now, payload: { createdBy: "user", creationSource: "web", id: threadId, projectId: ProjectId.make("project:bounded-sql-history"), title: "Bounded SQL history", 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, }, }); for (let ordinal = 1; ordinal <= 1_000; ordinal += 1) { const id = `turn-item:bounded-sql-history:${ordinal}`; const runId = `run:bounded-sql-history:${ordinal}`; 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}, ${ordinal}, 'codex', NULL, 'completed', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: runId, threadId, ordinal, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: `message:bounded-sql-history:${ordinal}`, rootNodeId: `node:bounded-sql-history:${ordinal}`, activeAttemptId: null, status: "completed", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; 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}, 'command_execution', 'completed', ${nowIso}, ${encodeUnknownJsonString({ id, threadId, runId, nodeId: null, providerThreadId: null, providerTurnId: null, nativeItemRef: null, parentItemId: null, ordinal, status: "completed", title: `command ${ordinal}`, input: `echo ${ordinal}`, output: "ok", exitCode: 0, startedAt: nowIso, completedAt: nowIso, updatedAt: nowIso, type: "command_execution", })} ) `; } const snapshots = yield* Effect.all( Array.from({ length: 4 }, () => projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 76 }), ), { concurrency: 4 }, ); for (const snapshot of snapshots) { assert.lengthOf(snapshot.projection.turnItems, 76); assert.lengthOf(snapshot.projection.runs, 76); assert.lengthOf(snapshot.projection.visibleTurnItems, 76); assert.strictEqual(snapshot.projection.turnItems[0]?.ordinal, 925); assert.strictEqual(snapshot.projection.turnItems.at(-1)?.ordinal, 1_000); } const retainedRequestPlan = yield* sql<{ readonly detail: string }>` EXPLAIN QUERY PLAN WITH selected AS ( SELECT run_id, type FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} ORDER BY ordinal DESC, turn_item_id DESC LIMIT 77 ) SELECT request.payload_json FROM orchestration_v2_projection_turn_items AS request WHERE request.run_id IN ( SELECT run_id FROM selected WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL ) AND request.type = 'run_interrupt_request' `; assert.isTrue( retainedRequestPlan.some((row) => row.detail.includes("orchestration_v2_projection_turn_items_run_ordinal_idx"), ), ); const older = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 76, anchorItemId: TurnItemId.make("turn-item:bounded-sql-history:925"), }); assert.lengthOf(older.projection.turnItems, 76); assert.lengthOf(older.projection.runs, 76); assert.strictEqual(older.projection.turnItems[0]?.ordinal, 850); assert.strictEqual(older.projection.turnItems.at(-1)?.ordinal, 925); const sqlPageLimit = THREAD_HISTORY_PAGE_POLICY.maxItems + 2; const initialSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, }); const initialPage = buildBoundedThreadProjection({ projection: initialSnapshot.projection, snapshotSequence: initialSnapshot.snapshotSequence, }); const loadedIds = initialPage.projection.visibleTurnItems.map((row) => String(row.sourceItemId), ); let cursor = initialPage.historyCursor; let pageCount = 1; while (cursor !== null) { const anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(cursor).si); const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, anchorItemId, }); const page = selectHistoryPageFromCursor({ items: snapshot.projection.visibleTurnItems, cursor, snapshotSequence: snapshot.snapshotSequence, }); loadedIds.push(...page.items.map((row) => String(row.sourceItemId))); pageCount += 1; cursor = page.nextCursor; } assert.isAtLeast(pageCount, 4); assert.lengthOf(loadedIds, 1_000); assert.strictEqual(new Set(loadedIds).size, 1_000); assert.deepEqual( loadedIds.toSorted( (left, right) => Number(left.split(":").at(-1)) - Number(right.split(":").at(-1)), ), Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`), ); const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix"); 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 ( ${hiddenRunId}, ${threadId}, 1001, 'codex', NULL, 'rolled_back', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: hiddenRunId, threadId, ordinal: 1001, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: "message:bounded-sql-history:hidden-suffix", rootNodeId: "node:bounded-sql-history:hidden-suffix", activeAttemptId: null, status: "rolled_back", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; for (let ordinal = 1_001; ordinal <= 1_100; ordinal += 1) { const id = `turn-item:bounded-sql-history:hidden:${ordinal}`; 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}, ${hiddenRunId}, NULL, NULL, NULL, NULL, ${ordinal}, 'command_execution', 'completed', ${nowIso}, ${encodeUnknownJsonString({ id, threadId, runId: hiddenRunId, nodeId: null, providerThreadId: null, providerTurnId: null, nativeItemRef: null, parentItemId: null, ordinal, status: "completed", title: "rolled back", input: "echo hidden", output: "hidden", exitCode: 0, startedAt: nowIso, completedAt: nowIso, updatedAt: nowIso, type: "command_execution", })} ) `; } const hiddenSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, }); const hiddenSuffixPage = buildBoundedThreadProjection({ projection: hiddenSuffixSnapshot.projection, snapshotSequence: hiddenSuffixSnapshot.snapshotSequence, }); assert.lengthOf(hiddenSuffixPage.projection.visibleTurnItems, 75); assert.strictEqual(hiddenSuffixPage.latestLocalTurnOrdinal, 1_100); assert.isTrue(hiddenSuffixPage.hasMoreHistory); assert.isFalse( hiddenSuffixPage.projection.visibleTurnItems.some((row) => String(row.sourceItemId).includes(":hidden:"), ), ); const hiddenSuffixCursor = hiddenSuffixPage.historyCursor; assert.isNotNull(hiddenSuffixCursor); const hiddenSuffixAnchor = TurnItemId.make(decodeThreadHistoryCursor(hiddenSuffixCursor!).si); const hiddenSuffixOlderSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: sqlPageLimit, anchorItemId: hiddenSuffixAnchor, }); const hiddenSuffixOlderPage = selectHistoryPageFromCursor({ items: hiddenSuffixOlderSnapshot.projection.visibleTurnItems, cursor: hiddenSuffixCursor!, snapshotSequence: hiddenSuffixOlderSnapshot.snapshotSequence, }); assert.lengthOf(hiddenSuffixOlderPage.items, 75); assert.isTrue(hiddenSuffixOlderPage.hasMoreHistory); const cancelledRunId = RunId.make("run:bounded-sql-history:cancelled-suffix"); 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 ( ${cancelledRunId}, ${threadId}, 1002, 'codex', NULL, 'cancelled', ${nowIso}, ${nowIso}, ${encodeUnknownJsonString({ id: cancelledRunId, threadId, ordinal: 1002, providerInstanceId, modelSelection, providerThreadId: null, userMessageId: "message:bounded-sql-history:cancelled-suffix", rootNodeId: "node:bounded-sql-history:cancelled-suffix", activeAttemptId: null, status: "cancelled", requestedAt: nowIso, startedAt: nowIso, completedAt: nowIso, checkpointId: null, contextHandoffId: null, })} ) `; for (let ordinal = 1_101; ordinal <= 1_200; ordinal += 1) { const id = `turn-item:bounded-sql-history:cancelled:${ordinal}`;