authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-08 22:23:20-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 22:23:20-05:00
log42c6623d492850c747b7ba06c1d0f1e1559d0ad5
tree8f159eb26f27c3e872287b9adb9de780016e9fdb
parentb707eeb052782cfd8b1ff0445f84b2aaf01da38b
signature Signed by PGP key B5690EEEBB952194

fix(server): speed up long thread message sync (#17387)

5 files changed, 179 insertions(+), 60 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+114-22
......@@ -54,6 +54,23 @@ const driver = ProviderDriverKind.make("codex");
5454const providerInstanceId = modelSelection.instanceId;
5555const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
5656
57const traceSqlStatements = () => {
58 const statements: Array<string> = [];
59 const tracer = Tracer.make({
60 span(options) {
61 const span = new Tracer.NativeSpan(options);
62 const end = span.end.bind(span);
63 span.end = (endTime, exit) => {
64 end(endTime, exit);
65 const query = span.attributes.get("db.query.text");
66 if (typeof query === "string") statements.push(query);
67 };
68 return span;
69 },
70 });
71 return { statements, tracer };
72};
73
5774const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate")(function* (
5875 suffix: string,
5976) {
......@@ -590,19 +607,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
590607 });
591608 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
592609 }
593 const statements: Array<string> = [];
594 const tracer = Tracer.make({
595 span(options) {
596 const span = new Tracer.NativeSpan(options);
597 const end = span.end.bind(span);
598 span.end = (endTime, exit) => {
599 end(endTime, exit);
600 const query = span.attributes.get("db.query.text");
601 if (typeof query === "string") statements.push(query);
602 };
603 return span;
604 },
605 });
610 const { statements, tracer } = traceSqlStatements();
606611 const initial = yield* Effect.acquireUseRelease(
607612 Effect.sync(() => vi.spyOn(JSON, "parse")),
608613 (parse) =>
......@@ -635,10 +640,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
635640 /SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/,
636641 );
637642 const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail);
638 assert.include(
639 topLevel,
640 "SEARCH item USING INDEX sqlite_autoindex_orchestration_v2_projection_turn_items_1 (turn_item_id=?)",
641 );
643 assert.include(topLevel, "SEARCH item USING INTEGER PRIMARY KEY (rowid=?)");
642644 assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY");
643645 const nodeStatement = statements.find((statement) =>
644646 statement.includes("WITH RECURSIVE retained"),
......@@ -779,7 +781,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
779781 nativeItemRef: null,
780782 runtimeRequestId: null,
781783 checkpointScopeId: null,
782 startedAt: now,
784 startedAt:
785 index === 998
786 ? null
787 : index === 999
788 ? DateTime.makeUnsafe("2020-01-01T00:00:00.000Z")
789 : now,
783790 completedAt: index === -1 ? null : now,
784791 },
785792 });
......@@ -838,7 +845,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
838845 resolvedAt: null,
839846 },
840847 });
841 const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 75 });
848 const { statements, tracer } = traceSqlStatements();
849 const snapshot = yield* projectionStore
850 .getThreadSnapshotWindow(threadId, { rowLimit: 75 })
851 .pipe(Effect.withTracer(tracer));
852 const nodeStatement = statements.find(
853 (statement) =>
854 statement.includes("payload_json") &&
855 statement.includes("orchestration_v2_projection_nodes") &&
856 !statement.includes("WITH RECURSIVE"),
857 );
858 assert.isDefined(nodeStatement);
859 const nodePlan = yield* (yield* SqlClient.SqlClient).unsafe<{ readonly detail: string }>(
860 `EXPLAIN QUERY PLAN ${nodeStatement}`,
861 );
862 const nodeLookups = nodePlan
863 .map((row) => row.detail)
864 .filter((detail) => detail.startsWith("SEARCH "));
865 // Cost must follow the retained cohort, even when the thread has
866 // thousands of completed nodes outside the requested window.
867 assert.deepEqual(nodeLookups, [
868 "SEARCH node USING INDEX sqlite_autoindex_orchestration_v2_projection_nodes_1 (node_id=?)",
869 ]);
870 assert.notInclude(
871 nodePlan.map((row) => row.detail),
872 "USE TEMP B-TREE FOR ORDER BY",
873 );
842874 assert.lengthOf(snapshot.projection.visibleTurnItems, 75);
843875 assert.lengthOf(snapshot.projection.nodes, 79);
844876 const retained = new Set(snapshot.projection.nodes.map((node) => node.id));
......@@ -862,6 +894,27 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
862894 );
863895 const full = yield* projectionStore.getThreadSnapshot(threadId);
864896 assert.lengthOf(full.projection.nodes, 1003);
897 // Bounded reads keep the full read's payloads and ordering, including
898 // null start times and equal timestamps.
899 assert.deepEqual(
900 snapshot.projection.nodes,
901 full.projection.nodes.filter((node) => retained.has(node.id)),
902 );
903 const sql = yield* SqlClient.SqlClient;
904 const duplicateItemId = TurnItemId.make("item:bounded-node-history:duplicate");
905 yield* sql`
906 INSERT INTO orchestration_v2_projection_turn_items
907 SELECT ${duplicateItemId}, thread_id, run_id, node_id, provider_thread_id,
908 provider_turn_id, parent_item_id, ordinal + 1, type, status, updated_at,
909 json_set(payload_json, '$.id', ${duplicateItemId}, '$.ordinal', ordinal + 1)
910 FROM orchestration_v2_projection_turn_items
911 WHERE turn_item_id = ${TurnItemId.make("item:bounded-node-history:999")}
912 `;
913 const sharedNodeSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
914 rowLimit: 76,
915 });
916 assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76);
917 assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes);
865918 }),
866919 );
867920
......@@ -1055,6 +1108,34 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
10551108 Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`),
10561109 );
10571110
1111 // Physical insertion order must not change the ordinal/ID timeline order.
1112 const tiedIds = [
1113 TurnItemId.make("turn-item:bounded-sql-history:tie:b"),
1114 TurnItemId.make("turn-item:bounded-sql-history:tie:a"),
1115 ];
1116 for (const id of tiedIds) {
1117 yield* sql`
1118 INSERT INTO orchestration_v2_projection_turn_items (
1119 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1120 parent_item_id, ordinal, type, status, updated_at, payload_json
1121 )
1122 SELECT ${id}, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1123 parent_item_id, 2000, type, status, updated_at,
1124 json_set(payload_json, '$.id', ${id}, '$.ordinal', 2000)
1125 FROM orchestration_v2_projection_turn_items
1126 WHERE turn_item_id = ${itemId}
1127 `;
1128 }
1129 const tied = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 3 });
1130 assert.deepEqual(
1131 tied.projection.turnItems.map((item) => item.id),
1132 [TurnItemId.make("turn-item:bounded-sql-history:1000"), tiedIds[1], tiedIds[0]],
1133 );
1134 yield* sql`
1135 DELETE FROM orchestration_v2_projection_turn_items
1136 WHERE turn_item_id IN ${sql.in(tiedIds)}
1137 `;
1138
10581139 const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix");
10591140 yield* sql`
10601141 INSERT INTO orchestration_v2_projection_runs (
......@@ -4056,9 +4137,20 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
40564137 `;
40574138 }
40584139
4059 const boundedFork = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
4060 rowLimit: 2,
4061 });
4140 const { statements, tracer } = traceSqlStatements();
4141 const boundedFork = yield* projectionStore
4142 .getThreadSnapshotWindow(targetThreadId, { rowLimit: 2 })
4143 .pipe(Effect.withTracer(tracer));
4144 // Only the target's control state is hydrated. Fork ancestors contribute
4145 // timeline rows, runs and attempts, regardless of their own live work.
4146 assert.lengthOf(
4147 statements.filter((statement) => statement.includes("WITH RECURSIVE retained")),
4148 1,
4149 );
4150 assert.lengthOf(
4151 statements.filter((statement) => statement.includes("SELECT node.payload_json")),
4152 1,
4153 );
40624154 assert.lengthOf(boundedFork.projection.turnItems, 0);
40634155 assert.deepEqual(
40644156 boundedFork.projection.visibleTurnItems.map((row) => [row.visibility, row.item.type]),
apps/server/src/orchestration-v2/ProjectionStore.ts+58-36
......@@ -66,7 +66,6 @@ import {
6666import {
6767 createOrchestrationV2TurnItemVisibility,
6868 isOrchestrationV2SupersededInterrupt,
69 isOrchestrationV2TurnItemVisible,
7069} from "@t3tools/shared/orchestrationV2Timeline";
7170import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
7271import * as Context from "effect/Context";
......@@ -1182,22 +1181,18 @@ function sortMessagesByTurnItemOrder(
11821181function activeLocalTurnItems(
11831182 projection: OrchestrationV2ThreadProjection,
11841183): Array<OrchestrationV2ProjectedTurnItem> {
1185 return projection.turnItems
1186 .filter((item) =>
1187 isOrchestrationV2TurnItemVisible({
1188 item,
1189 runs: projection.runs,
1190 attempts: projection.attempts,
1191 items: projection.turnItems,
1192 }),
1193 )
1194 .map((item, position) => ({
1195 position,
1196 visibility: "local" as const,
1197 sourceThreadId: item.threadId,
1198 sourceItemId: item.id,
1199 item,
1200 }));
1184 const isVisible = createOrchestrationV2TurnItemVisibility({
1185 runs: projection.runs,
1186 attempts: projection.attempts,
1187 items: projection.turnItems,
1188 });
1189 return projection.turnItems.filter(isVisible).map((item, position) => ({
1190 position,
1191 visibility: "local" as const,
1192 sourceThreadId: item.threadId,
1193 sourceItemId: item.id,
1194 item,
1195 }));
12011196}
12021197
12031198function localVisibleTurnItems(
......@@ -2722,7 +2717,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
27222717 `
27232718 : yield* sql<PayloadRow>`
27242719 WITH eligible AS NOT MATERIALIZED (
2725 SELECT item.payload_json, item.ordinal, item.turn_item_id,
2720 SELECT item.rowid AS item_rowid, item.payload_json, item.ordinal, item.turn_item_id,
27262721 item.run_id, item.node_id, item.type
27272722 FROM orchestration_v2_projection_turn_items AS item
27282723 LEFT JOIN orchestration_v2_projection_runs AS run
......@@ -2811,7 +2806,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28112806 -- Choose and sort rows by ID, then fetch payloads in that
28122807 -- order. A turn window can hold megabytes of tool output, and
28132808 -- carrying it through the union and sort cost about a second.
2814 SELECT ordinal, turn_item_id, run_id, type
2809 SELECT item_rowid, ordinal, turn_item_id, run_id, type
28152810 FROM eligible
28162811 WHERE ordinal >= (SELECT ordinal FROM boundary)
28172812 ORDER BY ordinal DESC, turn_item_id DESC
......@@ -2821,9 +2816,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28212816 ELSE ${window.rowLimit}
28222817 END
28232818 ), retained AS MATERIALIZED (
2824 SELECT ordinal, turn_item_id FROM selected
2819 SELECT item_rowid, ordinal, turn_item_id FROM selected
28252820 UNION
2826 SELECT request.ordinal, request.turn_item_id
2821 SELECT request.rowid AS item_rowid, request.ordinal, request.turn_item_id
28272822 FROM orchestration_v2_projection_turn_items AS request
28282823 WHERE request.run_id IN (
28292824 SELECT run_id FROM selected
......@@ -2831,9 +2826,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28312826 )
28322827 AND request.type = 'run_interrupt_request'
28332828 UNION
2834 SELECT latest.ordinal, latest.turn_item_id
2829 SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id
28352830 FROM (
2836 SELECT ordinal, turn_item_id
2831 SELECT rowid AS item_rowid, ordinal, turn_item_id
28372832 FROM orchestration_v2_projection_turn_items
28382833 WHERE thread_id = ${threadId}
28392834 AND ${window.anchorItemId ?? null} IS NULL
......@@ -2844,11 +2839,12 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28442839 ORDER BY ordinal ASC, turn_item_id ASC
28452840 )
28462841 -- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite
2847 -- skips sorting again once the payloads are attached.
2842 -- skips sorting again once the payloads are attached. Carry the
2843 -- rowid through selection to avoid a second string-key index lookup.
28482844 SELECT item.payload_json
28492845 FROM retained
28502846 CROSS JOIN orchestration_v2_projection_turn_items AS item
2851 ON item.turn_item_id = retained.turn_item_id
2847 ON item.rowid = retained.item_rowid
28522848 ORDER BY retained.ordinal ASC, retained.turn_item_id ASC
28532849 `;
28542850 // Reuse the decoded items for cohort IDs and the resulting projection.
......@@ -2865,7 +2861,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
28652861 // A run can contain thousands of completed nodes. Load the visible
28662862 // nodes and live control dependencies, then walk only their ancestors.
28672863 const cohortNodeIds =
2868 window === undefined
2864 window === undefined ||
2865 (fields !== undefined &&
2866 fields.every(
2867 (field) => field === "runs" || field === "attempts" || field === "turnItems",
2868 ))
28692869 ? cohortJson("nodeId")
28702870 : encodeIdList(
28712871 (yield* sql<{ readonly node_id: string }>`
......@@ -2900,7 +2900,14 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
29002900 INNER JOIN retained ON parent.node_id = retained.node_id
29012901 WHERE parent.thread_id = ${threadId} AND parent.parent_node_id IS NOT NULL
29022902 )
2903 SELECT node_id FROM retained WHERE node_id IS NOT NULL
2903 -- Sort IDs before fetching bodies, so large payloads never enter
2904 -- SQLite's temporary sort. The node read restores this order by key.
2905 SELECT retained.node_id
2906 FROM retained
2907 LEFT JOIN orchestration_v2_projection_nodes AS node
2908 ON node.node_id = retained.node_id AND node.thread_id = ${threadId}
2909 WHERE retained.node_id IS NOT NULL
2910 ORDER BY COALESCE(node.started_at, ''), retained.node_id ASC
29042911 `).map((row) => row.node_id),
29052912 );
29062913 const cohortProviderThreadIds = cohortJson("providerThreadId");
......@@ -2970,13 +2977,20 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
29702977 FROM orchestration_v2_projection_nodes
29712978 WHERE thread_id = ${threadId}
29722979 ORDER BY COALESCE(started_at, ''), node_id ASC
2973 `
2974 : sql<PayloadRow>`
2975 SELECT payload_json FROM orchestration_v2_projection_nodes
2976 WHERE thread_id = ${threadId}
2977 AND node_id IN (SELECT value FROM json_each(${cohortNodeIds}))
2978 ORDER BY COALESCE(started_at, ''), node_id ASC
2979 `,
2980 `
2981 : sql<PayloadRow & { readonly position: number }>`
2982 -- Keep retained IDs outermost so SQLite looks up each node by its
2983 -- primary key instead of scanning every historical node in the thread.
2984 SELECT node.payload_json, retained.key AS position
2985 FROM json_each(${cohortNodeIds}) AS retained
2986 CROSS JOIN orchestration_v2_projection_nodes AS node
2987 ON node.node_id = retained.value
2988 WHERE node.thread_id = ${threadId}
2989 `.pipe(
2990 Effect.map((rows) =>
2991 rows.toSorted((left, right) => left.position - right.position),
2992 ),
2993 ),
29802994 fields !== undefined && !fields.includes("subagents")
29812995 ? Effect.succeed([])
29822996 : window === undefined
......@@ -3257,7 +3271,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
32573271 visibleTurnItems: [],
32583272 updatedAt: thread.updatedAt,
32593273 } satisfies OrchestrationV2ThreadProjection;
3260 return fields === undefined ? withLocalVisibleTurnItems(projection) : projection;
3274 return projection;
32613275 }).pipe(
32623276 Effect.mapError((cause) =>
32633277 isProjectionStoreThreadNotFoundError(cause)
......@@ -3289,7 +3303,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
32893303 (window?.historyAnchor !== undefined && window.historyAnchor.threadId !== threadId)
32903304 ? { ...window, rowLimit: 0, anchorItemId: undefined }
32913305 : window;
3292 const projection = yield* readCanonicalProjection(threadId, localWindow);
3306 // Ancestors contribute timeline items, not their provider/control state.
3307 // Keep their runs and attempts for fork boundaries and visibility.
3308 const projection = yield* readCanonicalProjection(
3309 threadId,
3310 localWindow,
3311 window !== undefined && seenThreadIds.size > 0
3312 ? ["runs", "attempts", "turnItems"]
3313 : undefined,
3314 );
32933315 const forkedFrom = projection.thread.forkedFrom;
32943316 if (forkedFrom?.type !== "run" || seenThreadIds.has(forkedFrom.threadId)) {
32953317 return withLocalVisibleTurnItems(projection);
apps/server/src/orchestration-v2/WireProjection.test.ts+1
......@@ -142,6 +142,7 @@ describe("orchestration V2 wire projection", () => {
142142 deletedAt: null,
143143 },
144144 runs: [],
145 attempts: [],
145146 runtimeRequests: [],
146147 messages: [
147148 {
apps/server/src/orchestration-v2/threadHistoryPaging.test.ts+2-1
......@@ -634,7 +634,7 @@ describe("threadHistoryPaging", () => {
634634 });
635635
636636 it("preserves oversized actionable state and reports the budget exception", () => {
637 const projection = makeProjection([]);
637 const projection = makeProjection([makeRow(0)]);
638638 const actionable = {
639639 ...projection,
640640 plans: [
......@@ -654,6 +654,7 @@ describe("threadHistoryPaging", () => {
654654
655655 expect(bounded.payloadBudgetExceeded).toBe(true);
656656 expect(bounded.projection.plans[0]).toEqual(actionable.plans[0]);
657 expect(bounded.projection.visibleTurnItems).toEqual(actionable.visibleTurnItems);
657658 });
658659
659660 it("keeps paged historical plan detail in the turn item and only status in its artifact", () => {
apps/server/src/orchestration-v2/threadHistoryPaging.ts+4-1
......@@ -567,6 +567,9 @@ export function buildBoundedThreadProjection(input: {
567567 // Always from the full projection so inherited-only windows still carry a
568568 // watermark for partial live reducers.
569569 latestLocalTurnOrdinal,
570 payloadBudgetExceeded: bytesOfJson(projection) > policy.maxEncodedBytes,
570 // Adding timeline rows and historical summaries cannot shrink control state.
571 // Avoid serializing the whole timeline when control state already exceeds the cap.
572 payloadBudgetExceeded:
573 controlBytes > policy.maxEncodedBytes || bytesOfJson(projection) > policy.maxEncodedBytes,
571574 };
572575}