diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 9b48ecd1ac360bc2c651758e1675134e01be0b94..e8a3c1bcfc08e6a5bf111ee7e8a673a947ba327a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -54,6 +54,23 @@ const driver = ProviderDriverKind.make("codex"); const providerInstanceId = modelSelection.instanceId; const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +const traceSqlStatements = () => { + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + return { statements, tracer }; +}; + const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate")(function* ( suffix: string, ) { @@ -590,19 +607,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`; } - const statements: Array = []; - const tracer = Tracer.make({ - span(options) { - const span = new Tracer.NativeSpan(options); - const end = span.end.bind(span); - span.end = (endTime, exit) => { - end(endTime, exit); - const query = span.attributes.get("db.query.text"); - if (typeof query === "string") statements.push(query); - }; - return span; - }, - }); + const { statements, tracer } = traceSqlStatements(); const initial = yield* Effect.acquireUseRelease( Effect.sync(() => vi.spyOn(JSON, "parse")), (parse) => @@ -635,10 +640,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { /SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/, ); const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail); - assert.include( - topLevel, - "SEARCH item USING INDEX sqlite_autoindex_orchestration_v2_projection_turn_items_1 (turn_item_id=?)", - ); + assert.include(topLevel, "SEARCH item USING INTEGER PRIMARY KEY (rowid=?)"); assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY"); const nodeStatement = statements.find((statement) => statement.includes("WITH RECURSIVE retained"), @@ -779,7 +781,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { nativeItemRef: null, runtimeRequestId: null, checkpointScopeId: null, - startedAt: now, + startedAt: + index === 998 + ? null + : index === 999 + ? DateTime.makeUnsafe("2020-01-01T00:00:00.000Z") + : now, completedAt: index === -1 ? null : now, }, }); @@ -838,7 +845,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { resolvedAt: null, }, }); - const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 75 }); + const { statements, tracer } = traceSqlStatements(); + const snapshot = yield* projectionStore + .getThreadSnapshotWindow(threadId, { rowLimit: 75 }) + .pipe(Effect.withTracer(tracer)); + const nodeStatement = statements.find( + (statement) => + statement.includes("payload_json") && + statement.includes("orchestration_v2_projection_nodes") && + !statement.includes("WITH RECURSIVE"), + ); + assert.isDefined(nodeStatement); + const nodePlan = yield* (yield* SqlClient.SqlClient).unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${nodeStatement}`, + ); + const nodeLookups = nodePlan + .map((row) => row.detail) + .filter((detail) => detail.startsWith("SEARCH ")); + // Cost must follow the retained cohort, even when the thread has + // thousands of completed nodes outside the requested window. + assert.deepEqual(nodeLookups, [ + "SEARCH node USING INDEX sqlite_autoindex_orchestration_v2_projection_nodes_1 (node_id=?)", + ]); + assert.notInclude( + nodePlan.map((row) => row.detail), + "USE TEMP B-TREE FOR ORDER BY", + ); assert.lengthOf(snapshot.projection.visibleTurnItems, 75); assert.lengthOf(snapshot.projection.nodes, 79); const retained = new Set(snapshot.projection.nodes.map((node) => node.id)); @@ -862,6 +894,27 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ); const full = yield* projectionStore.getThreadSnapshot(threadId); assert.lengthOf(full.projection.nodes, 1003); + // Bounded reads keep the full read's payloads and ordering, including + // null start times and equal timestamps. + assert.deepEqual( + snapshot.projection.nodes, + full.projection.nodes.filter((node) => retained.has(node.id)), + ); + const sql = yield* SqlClient.SqlClient; + const duplicateItemId = TurnItemId.make("item:bounded-node-history:duplicate"); + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items + SELECT ${duplicateItemId}, thread_id, run_id, node_id, provider_thread_id, + provider_turn_id, parent_item_id, ordinal + 1, type, status, updated_at, + json_set(payload_json, '$.id', ${duplicateItemId}, '$.ordinal', ordinal + 1) + FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${TurnItemId.make("item:bounded-node-history:999")} + `; + const sharedNodeSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 76, + }); + assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76); + assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes); }), ); @@ -1055,6 +1108,34 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`), ); + // Physical insertion order must not change the ordinal/ID timeline order. + const tiedIds = [ + TurnItemId.make("turn-item:bounded-sql-history:tie:b"), + TurnItemId.make("turn-item:bounded-sql-history:tie:a"), + ]; + for (const id of tiedIds) { + 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 + ) + SELECT ${id}, thread_id, run_id, node_id, provider_thread_id, provider_turn_id, + parent_item_id, 2000, type, status, updated_at, + json_set(payload_json, '$.id', ${id}, '$.ordinal', 2000) + FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${itemId} + `; + } + const tied = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 3 }); + assert.deepEqual( + tied.projection.turnItems.map((item) => item.id), + [TurnItemId.make("turn-item:bounded-sql-history:1000"), tiedIds[1], tiedIds[0]], + ); + yield* sql` + DELETE FROM orchestration_v2_projection_turn_items + WHERE turn_item_id IN ${sql.in(tiedIds)} + `; + const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix"); yield* sql` INSERT INTO orchestration_v2_projection_runs ( @@ -4056,9 +4137,20 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { `; } - const boundedFork = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, { - rowLimit: 2, - }); + const { statements, tracer } = traceSqlStatements(); + const boundedFork = yield* projectionStore + .getThreadSnapshotWindow(targetThreadId, { rowLimit: 2 }) + .pipe(Effect.withTracer(tracer)); + // Only the target's control state is hydrated. Fork ancestors contribute + // timeline rows, runs and attempts, regardless of their own live work. + assert.lengthOf( + statements.filter((statement) => statement.includes("WITH RECURSIVE retained")), + 1, + ); + assert.lengthOf( + statements.filter((statement) => statement.includes("SELECT node.payload_json")), + 1, + ); assert.lengthOf(boundedFork.projection.turnItems, 0); assert.deepEqual( boundedFork.projection.visibleTurnItems.map((row) => [row.visibility, row.item.type]), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 86b2d503c6b237edd609ebf4ea05f9768a85a3af..a6a092ff84b7a15bc1c142b531498f258be64444 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -66,7 +66,6 @@ import { import { createOrchestrationV2TurnItemVisibility, isOrchestrationV2SupersededInterrupt, - isOrchestrationV2TurnItemVisible, } from "@t3tools/shared/orchestrationV2Timeline"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as Context from "effect/Context"; @@ -1182,22 +1181,18 @@ function sortMessagesByTurnItemOrder( function activeLocalTurnItems( projection: OrchestrationV2ThreadProjection, ): Array { - return projection.turnItems - .filter((item) => - isOrchestrationV2TurnItemVisible({ - item, - runs: projection.runs, - attempts: projection.attempts, - items: projection.turnItems, - }), - ) - .map((item, position) => ({ - position, - visibility: "local" as const, - sourceThreadId: item.threadId, - sourceItemId: item.id, - item, - })); + const isVisible = createOrchestrationV2TurnItemVisibility({ + runs: projection.runs, + attempts: projection.attempts, + items: projection.turnItems, + }); + return projection.turnItems.filter(isVisible).map((item, position) => ({ + position, + visibility: "local" as const, + sourceThreadId: item.threadId, + sourceItemId: item.id, + item, + })); } function localVisibleTurnItems( @@ -2722,7 +2717,7 @@ export const layer: Layer.Layer = ` : yield* sql` WITH eligible AS NOT MATERIALIZED ( - SELECT item.payload_json, item.ordinal, item.turn_item_id, + SELECT item.rowid AS item_rowid, item.payload_json, item.ordinal, item.turn_item_id, item.run_id, item.node_id, item.type FROM orchestration_v2_projection_turn_items AS item LEFT JOIN orchestration_v2_projection_runs AS run @@ -2811,7 +2806,7 @@ export const layer: Layer.Layer = -- Choose and sort rows by ID, then fetch payloads in that -- order. A turn window can hold megabytes of tool output, and -- carrying it through the union and sort cost about a second. - SELECT ordinal, turn_item_id, run_id, type + SELECT item_rowid, ordinal, turn_item_id, run_id, type FROM eligible WHERE ordinal >= (SELECT ordinal FROM boundary) ORDER BY ordinal DESC, turn_item_id DESC @@ -2821,9 +2816,9 @@ export const layer: Layer.Layer = ELSE ${window.rowLimit} END ), retained AS MATERIALIZED ( - SELECT ordinal, turn_item_id FROM selected + SELECT item_rowid, ordinal, turn_item_id FROM selected UNION - SELECT request.ordinal, request.turn_item_id + SELECT request.rowid AS item_rowid, request.ordinal, request.turn_item_id FROM orchestration_v2_projection_turn_items AS request WHERE request.run_id IN ( SELECT run_id FROM selected @@ -2831,9 +2826,9 @@ export const layer: Layer.Layer = ) AND request.type = 'run_interrupt_request' UNION - SELECT latest.ordinal, latest.turn_item_id + SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id FROM ( - SELECT ordinal, turn_item_id + SELECT rowid AS item_rowid, ordinal, turn_item_id FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND ${window.anchorItemId ?? null} IS NULL @@ -2844,11 +2839,12 @@ export const layer: Layer.Layer = ORDER BY ordinal ASC, turn_item_id ASC ) -- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite - -- skips sorting again once the payloads are attached. + -- skips sorting again once the payloads are attached. Carry the + -- rowid through selection to avoid a second string-key index lookup. SELECT item.payload_json FROM retained CROSS JOIN orchestration_v2_projection_turn_items AS item - ON item.turn_item_id = retained.turn_item_id + ON item.rowid = retained.item_rowid ORDER BY retained.ordinal ASC, retained.turn_item_id ASC `; // Reuse the decoded items for cohort IDs and the resulting projection. @@ -2865,7 +2861,11 @@ export const layer: Layer.Layer = // A run can contain thousands of completed nodes. Load the visible // nodes and live control dependencies, then walk only their ancestors. const cohortNodeIds = - window === undefined + window === undefined || + (fields !== undefined && + fields.every( + (field) => field === "runs" || field === "attempts" || field === "turnItems", + )) ? cohortJson("nodeId") : encodeIdList( (yield* sql<{ readonly node_id: string }>` @@ -2900,7 +2900,14 @@ export const layer: Layer.Layer = INNER JOIN retained ON parent.node_id = retained.node_id WHERE parent.thread_id = ${threadId} AND parent.parent_node_id IS NOT NULL ) - SELECT node_id FROM retained WHERE node_id IS NOT NULL + -- Sort IDs before fetching bodies, so large payloads never enter + -- SQLite's temporary sort. The node read restores this order by key. + SELECT retained.node_id + FROM retained + LEFT JOIN orchestration_v2_projection_nodes AS node + ON node.node_id = retained.node_id AND node.thread_id = ${threadId} + WHERE retained.node_id IS NOT NULL + ORDER BY COALESCE(node.started_at, ''), retained.node_id ASC `).map((row) => row.node_id), ); const cohortProviderThreadIds = cohortJson("providerThreadId"); @@ -2970,13 +2977,20 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} ORDER BY COALESCE(started_at, ''), node_id ASC - ` - : sql` - SELECT payload_json FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} - AND node_id IN (SELECT value FROM json_each(${cohortNodeIds})) - ORDER BY COALESCE(started_at, ''), node_id ASC - `, + ` + : sql` + -- Keep retained IDs outermost so SQLite looks up each node by its + -- primary key instead of scanning every historical node in the thread. + SELECT node.payload_json, retained.key AS position + FROM json_each(${cohortNodeIds}) AS retained + CROSS JOIN orchestration_v2_projection_nodes AS node + ON node.node_id = retained.value + WHERE node.thread_id = ${threadId} + `.pipe( + Effect.map((rows) => + rows.toSorted((left, right) => left.position - right.position), + ), + ), fields !== undefined && !fields.includes("subagents") ? Effect.succeed([]) : window === undefined @@ -3257,7 +3271,7 @@ export const layer: Layer.Layer = visibleTurnItems: [], updatedAt: thread.updatedAt, } satisfies OrchestrationV2ThreadProjection; - return fields === undefined ? withLocalVisibleTurnItems(projection) : projection; + return projection; }).pipe( Effect.mapError((cause) => isProjectionStoreThreadNotFoundError(cause) @@ -3289,7 +3303,15 @@ export const layer: Layer.Layer = (window?.historyAnchor !== undefined && window.historyAnchor.threadId !== threadId) ? { ...window, rowLimit: 0, anchorItemId: undefined } : window; - const projection = yield* readCanonicalProjection(threadId, localWindow); + // Ancestors contribute timeline items, not their provider/control state. + // Keep their runs and attempts for fork boundaries and visibility. + const projection = yield* readCanonicalProjection( + threadId, + localWindow, + window !== undefined && seenThreadIds.size > 0 + ? ["runs", "attempts", "turnItems"] + : undefined, + ); const forkedFrom = projection.thread.forkedFrom; if (forkedFrom?.type !== "run" || seenThreadIds.has(forkedFrom.threadId)) { return withLocalVisibleTurnItems(projection); diff --git a/apps/server/src/orchestration-v2/WireProjection.test.ts b/apps/server/src/orchestration-v2/WireProjection.test.ts index ba4998341db17ee59543a531aba2026cac6b2627..a44dc0fe817771ac2648e7897b176e0947948c72 100644 --- a/apps/server/src/orchestration-v2/WireProjection.test.ts +++ b/apps/server/src/orchestration-v2/WireProjection.test.ts @@ -142,6 +142,7 @@ describe("orchestration V2 wire projection", () => { deletedAt: null, }, runs: [], + attempts: [], runtimeRequests: [], messages: [ { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 9c1115ca0df7fc716d71971a85cc89bbe17517df..57a6bf86da587167185fac51d53a1ec3c1c9c302 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -634,7 +634,7 @@ describe("threadHistoryPaging", () => { }); it("preserves oversized actionable state and reports the budget exception", () => { - const projection = makeProjection([]); + const projection = makeProjection([makeRow(0)]); const actionable = { ...projection, plans: [ @@ -654,6 +654,7 @@ describe("threadHistoryPaging", () => { expect(bounded.payloadBudgetExceeded).toBe(true); expect(bounded.projection.plans[0]).toEqual(actionable.plans[0]); + expect(bounded.projection.visibleTurnItems).toEqual(actionable.visibleTurnItems); }); it("keeps paged historical plan detail in the turn item and only status in its artifact", () => { diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 8ee7c2641bd3028fbd96bb4b0c2f426cb5be0d0a..ce7fccfd7d26ce0c53791208c80bf7b3002c71e3 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -567,6 +567,9 @@ export function buildBoundedThreadProjection(input: { // Always from the full projection so inherited-only windows still carry a // watermark for partial live reducers. latestLocalTurnOrdinal, - payloadBudgetExceeded: bytesOfJson(projection) > policy.maxEncodedBytes, + // Adding timeline rows and historical summaries cannot shrink control state. + // Avoid serializing the whole timeline when control state already exceeds the cap. + payloadBudgetExceeded: + controlBytes > policy.maxEncodedBytes || bytesOfJson(projection) > policy.maxEncodedBytes, }; }