From 5b4382e8a84699a4854d96904e12d47df95eff3c Mon Sep 17 00:00:00 2001 From: Behzad Motaghi Date: Fri, 30 Oct 2026 10:24:19 +0100 Subject: [PATCH 1/2] perf(server): bound ordinal recovery to selected messages --- .../orchestration-v2/ProjectionStore.test.ts | 96 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 9 ++ 2 files changed, 105 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 1dc7c912e776..671dd54b1834 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -34,6 +34,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -1917,6 +1918,101 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { byteIds, Array.from({ length: 70 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`), ); + + // Selected message reads recover ordinals for only the loaded messages, while 69 + // unloaded assistant messages remain in the thread's history. + const firstMessageId = MessageId.make("message:bounded-sql-history:1"); + const lastMessageId = MessageId.make("message:bounded-sql-history:1000"); + const insertMessageItem = ( + suffix: string, + itemThreadId: string, + ordinal: number, + messageId: string | null | undefined, + ) => 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 ( + ${`turn-item:bounded-sql-history:ordinal-${suffix}`}, ${itemThreadId}, NULL, NULL, NULL, + NULL, NULL, ${ordinal}, 'assistant_message', 'completed', ${nowIso}, + ${encodeUnknownJsonString({ messageId })} + ) + `; + yield* insertMessageItem("duplicate", threadId, 0, lastMessageId); + yield* insertMessageItem( + "other-thread", + "thread:bounded-sql-history:other", + -1, + firstMessageId, + ); + yield* insertMessageItem("null-id", threadId, -2, null); + yield* insertMessageItem("missing-id", threadId, -3, undefined); + const ordinalReads: Array]> = []; + const recordOrdinalReads: Statement.Transformer = (statement) => { + const compiled = statement.compile(); + if (compiled[0].includes("MIN(ordinal) AS ordinal")) ordinalReads.push(compiled); + return Effect.succeed(statement); + }; + const readMessageIds = (filter?: ProjectionStore.ProjectionRecordFilter) => + projectionStore.getThreadRecords(threadId, ["messages"], filter).pipe( + Effect.map(({ messages }) => messages.map((message) => message.id)), + Effect.provideService(Statement.CurrentTransformer, recordOrdinalReads), + ); + // Re-run each recorded ordinal read to count the groups it produced. + const takeOrdinalGroups = () => + Effect.forEach(ordinalReads.splice(0), ([query, params]) => + sql + .unsafe<{ readonly message_id: string | null; readonly ordinal: number }>(query, params) + .pipe( + Effect.map((rows) => + rows.toSorted((left, right) => + String(left.message_id).localeCompare(String(right.message_id)), + ), + ), + ), + ); + assert.deepEqual(yield* readMessageIds(), [lastMessageId, firstMessageId]); + assert.equal(ordinalReads.length, 1); + const unfilteredOrdinalRead = ordinalReads.splice(0)[0]!; + assert.equal(unfilteredOrdinalRead[0].includes("json_each"), false); + assert.deepEqual(yield* readMessageIds({ messageRoles: ["user", "assistant"] }), [ + lastMessageId, + firstMessageId, + ]); + assert.deepEqual(ordinalReads.splice(0), [unfilteredOrdinalRead]); + for (const filter of [ + { + messageIds: [ + firstMessageId, + lastMessageId, + MessageId.make("message:bounded-sql-history:absent"), + ], + }, + { messageRoles: ["user" as const] }, + { + messageRunIds: [ + RunId.make("run:bounded-sql-history:1"), + RunId.make("run:bounded-sql-history:1000"), + ], + }, + ]) { + assert.deepEqual(yield* readMessageIds(filter), [lastMessageId, firstMessageId]); + assert.deepEqual(yield* takeOrdinalGroups(), [ + [ + { message_id: firstMessageId, ordinal: 1 }, + { message_id: lastMessageId, ordinal: 0 }, + ], + ]); + } + assert.deepEqual(yield* readMessageIds({ messageIds: [] }), []); + assert.deepEqual(yield* takeOrdinalGroups(), []); + assert.deepEqual(yield* readMessageIds({ messageIds: [firstMessageId] }), [firstMessageId]); + assert.deepEqual(yield* takeOrdinalGroups(), []); + yield* sql` + DELETE FROM orchestration_v2_projection_turn_items + WHERE turn_item_id LIKE 'turn-item:bounded-sql-history:ordinal-%' + `; + yield* sql` UPDATE orchestration_v2_projection_turn_items SET type = 'command_execution', payload_json = json_set(payload_json, '$.type', 'command_execution') diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 88a9073117dd..a48d1943e7c9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3513,10 +3513,19 @@ export const layer: Layer.Layer = ]); const messageOrdinals = new Map(); if (fields !== undefined && fields.includes("messages") && messages.length > 1) { + // Broad role reads keep the original query to avoid per-ID lookup overhead. + const loadedMessageIds = + filter?.messageIds === undefined && + filter?.messageRunIds === undefined && + (filter?.messageRoles === undefined || + (filter.messageRoles.includes("user") && filter.messageRoles.includes("assistant"))) + ? undefined + : messages.map((message) => message.id); const rows = yield* sql<{ message_id: string; ordinal: number }>` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND type IN ('user_message', 'assistant_message') + ${loadedMessageIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.messageId') IN (SELECT value FROM json_each(${encodeIdList(loadedMessageIds)}))`} GROUP BY json_extract(payload_json, '$.messageId') `; for (const row of rows) messageOrdinals.set(row.message_id, row.ordinal); From 6130b70e7c0cff354248ac1a47f0dcafe6cafc9c Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 01:49:50 +0000 Subject: [PATCH 2/2] test(server): assert ordinal read ordering without sql instrumentation --- .../orchestration-v2/ProjectionStore.test.ts | 39 ++----------------- 1 file changed, 3 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 4036d4111cff..4076e9194212 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -34,7 +34,6 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; -import * as Statement from "effect/sql/Statement"; import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -2127,39 +2126,15 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ); yield* insertMessageItem("null-id", threadId, -2, null); yield* insertMessageItem("missing-id", threadId, -3, undefined); - const ordinalReads: Array]> = []; - const recordOrdinalReads: Statement.Transformer = (statement) => { - const compiled = statement.compile(); - if (compiled[0].includes("MIN(ordinal) AS ordinal")) ordinalReads.push(compiled); - return Effect.succeed(statement); - }; const readMessageIds = (filter?: ProjectionStore.ProjectionRecordFilter) => - projectionStore.getThreadRecords(threadId, ["messages"], filter).pipe( - Effect.map(({ messages }) => messages.map((message) => message.id)), - Effect.provideService(Statement.CurrentTransformer, recordOrdinalReads), - ); - // Re-run each recorded ordinal read to count the groups it produced. - const takeOrdinalGroups = () => - Effect.forEach(ordinalReads.splice(0), ([query, params]) => - sql - .unsafe<{ readonly message_id: string | null; readonly ordinal: number }>(query, params) - .pipe( - Effect.map((rows) => - rows.toSorted((left, right) => - String(left.message_id).localeCompare(String(right.message_id)), - ), - ), - ), - ); + projectionStore + .getThreadRecords(threadId, ["messages"], filter) + .pipe(Effect.map(({ messages }) => messages.map((message) => message.id))); assert.deepEqual(yield* readMessageIds(), [lastMessageId, firstMessageId]); - assert.equal(ordinalReads.length, 1); - const unfilteredOrdinalRead = ordinalReads.splice(0)[0]!; - assert.equal(unfilteredOrdinalRead[0].includes("json_each"), false); assert.deepEqual(yield* readMessageIds({ messageRoles: ["user", "assistant"] }), [ lastMessageId, firstMessageId, ]); - assert.deepEqual(ordinalReads.splice(0), [unfilteredOrdinalRead]); for (const filter of [ { messageIds: [ @@ -2177,17 +2152,9 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }, ]) { assert.deepEqual(yield* readMessageIds(filter), [lastMessageId, firstMessageId]); - assert.deepEqual(yield* takeOrdinalGroups(), [ - [ - { message_id: firstMessageId, ordinal: 1 }, - { message_id: lastMessageId, ordinal: 0 }, - ], - ]); } assert.deepEqual(yield* readMessageIds({ messageIds: [] }), []); - assert.deepEqual(yield* takeOrdinalGroups(), []); assert.deepEqual(yield* readMessageIds({ messageIds: [firstMessageId] }), [firstMessageId]); - assert.deepEqual(yield* takeOrdinalGroups(), []); yield* sql` DELETE FROM orchestration_v2_projection_turn_items WHERE turn_item_id LIKE 'turn-item:bounded-sql-history:ordinal-%'