From 3a4f1d16fc61523b369cfdedff96fdbbff1f379f Mon Sep 17 00:00:00 2001 From: Behzad Motaghi Date: Sat, 31 Oct 2026 22:01:06 +0100 Subject: [PATCH 1/2] fix(server): look up a run's turn-start items by thread and run --- .../orchestration-v2/ProjectionStore.test.ts | 207 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 2 + 2 files changed, 209 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a29d698d76daf..10ee9878d705b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3204,6 +3204,213 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("reads a run's turn-start items by thread and run, not by walking thread history", () => + 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:turn-start-items"); + const otherThreadId = ThreadId.make("thread:turn-start-items:other"); + const selectedRunId = RunId.make("run:turn-start-items:selected"); + const emptyRunId = RunId.make("run:turn-start-items:empty"); + const siblingRunId = RunId.make("run:turn-start-items:sibling"); + + for (const id of [threadId, otherThreadId]) { + yield* projectionStore.apply({ + id: EventId.make(`event:${id}:created`), + type: "thread.created", + threadId: id, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id, + projectId: ProjectId.make("project:turn-start-items"), + title: id, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: id }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + } + for (const [ordinal, runId] of [selectedRunId, emptyRunId, siblingRunId].entries()) { + 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 + 1}, 'codex', NULL, 'completed', ${nowIso}, ${nowIso}, + ${encodeUnknownJsonString({ + id: runId, + threadId, + ordinal: ordinal + 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: `message:${runId}`, + rootNodeId: `node:${runId}`, + activeAttemptId: null, + status: "completed", + requestedAt: nowIso, + startedAt: nowIso, + completedAt: nowIso, + checkpointId: null, + contextHandoffId: null, + })} + ) + `; + } + + const insertItem = ( + id: string, + itemThreadId: ThreadId, + runId: string | null, + ordinal: number, + ) => + 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}, ${itemThreadId}, ${runId}, NULL, NULL, NULL, NULL, ${ordinal}, + 'command_execution', 'completed', ${nowIso}, + ${encodeUnknownJsonString({ + id, + threadId: itemThreadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: id, + input: "echo", + output: "ok", + exitCode: 0, + startedAt: nowIso, + completedAt: nowIso, + updatedAt: nowIso, + type: "command_execution", + })} + ) + `; + const addHistory = (from: number, to: number) => sql` + WITH RECURSIVE history(position) AS ( + SELECT ${from} UNION ALL SELECT position + 1 FROM history WHERE position < ${to} + ) + 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 'item:turn-start-items:history:' || position, ${threadId}, + 'run:turn-start-items:history:' || (position % 100), NULL, NULL, NULL, NULL, position, + 'command_execution', 'completed', ${nowIso}, + json_object( + 'id', 'item:turn-start-items:history:' || position, + 'threadId', ${threadId}, + 'runId', 'run:turn-start-items:history:' || (position % 100), + 'nodeId', NULL, 'providerThreadId', NULL, 'providerTurnId', NULL, + 'nativeItemRef', NULL, 'parentItemId', NULL, 'ordinal', position, + 'status', 'completed', 'title', 'history', 'input', 'echo', 'output', 'ok', + 'exitCode', 0, 'startedAt', ${nowIso}, 'completedAt', ${nowIso}, + 'updatedAt', ${nowIso}, 'type', 'command_execution' + ) + FROM history + `; + + // Twelve ordinals with two items each, inserted in reverse id order. + const selectedIds: Array = []; + for (let slot = 1; slot <= 12; slot += 1) { + for (const suffix of ["b", "a"]) { + yield* insertItem( + `item:turn-start-items:selected:${String(slot).padStart(2, "0")}:${suffix}`, + threadId, + selectedRunId, + slot * 100, + ); + } + selectedIds.push( + `item:turn-start-items:selected:${String(slot).padStart(2, "0")}:a`, + `item:turn-start-items:selected:${String(slot).padStart(2, "0")}:b`, + ); + } + yield* insertItem("item:turn-start-items:no-run", threadId, null, 100); + yield* insertItem("item:turn-start-items:sibling:1", threadId, siblingRunId, 100); + yield* insertItem("item:turn-start-items:sibling:2", threadId, siblingRunId, 101); + yield* insertItem("item:turn-start-items:other-thread:b", otherThreadId, selectedRunId, 100); + yield* insertItem("item:turn-start-items:other-thread:a", otherThreadId, selectedRunId, 100); + + const readItems = Effect.fn("readTurnStartItems")(function* ( + readThreadId: ThreadId, + readRunId: RunId, + ) { + const { statements, tracer } = traceSqlStatements(); + const context = yield* projectionStore + .getTurnStartContext(readThreadId, readRunId) + .pipe(Effect.withTracer(tracer)); + const lookups = statements.filter((statement) => + statement.includes("FROM orchestration_v2_projection_turn_items"), + ); + assert.lengthOf(lookups, 1); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${lookups[0]}`, + [readThreadId, readRunId], + )).map((row) => row.detail); + return { ids: context.turnItems.map((item) => item.id), plan }; + }); + + const lookupPlans: Array> = []; + for (const [from, to] of [ + [1, 1_000], + [1_001, 10_000], + ] as const) { + yield* addHistory(from, to); + const { ids, plan } = yield* readItems(threadId, selectedRunId); + assert.deepEqual(ids, selectedIds); + lookupPlans.push(plan); + } + + assert.deepEqual((yield* readItems(threadId, emptyRunId)).ids, []); + assert.deepEqual( + (yield* readItems(threadId, RunId.make("run:turn-start-items:missing"))).ids, + [], + ); + assert.deepEqual((yield* readItems(threadId, siblingRunId)).ids, [ + "item:turn-start-items:sibling:1", + "item:turn-start-items:sibling:2", + ]); + assert.deepEqual((yield* readItems(otherThreadId, selectedRunId)).ids, [ + "item:turn-start-items:other-thread:a", + "item:turn-start-items:other-thread:b", + ]); + assert.deepEqual((yield* readItems(otherThreadId, siblingRunId)).ids, []); + + for (const plan of lookupPlans) { + const lookup = plan.filter((detail) => + detail.includes("orchestration_v2_projection_turn_items"), + ); + assert.lengthOf(lookup, 1); + assert.match(lookup[0]!, /^SEARCH .*\(thread_id=\? AND run_id=\?\)$/); + } + }), + ); + it.effect( "counts each shell's own runless and live-run items, skipping rolled back and missing runs", () => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2e88fdc119081..ec65e679facbf 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4298,6 +4298,8 @@ export const layer: Layer.Layer = `.pipe(Effect.flatMap(decodeRows(decodeContextTransferPayload, threadId))); const turnItems = yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items + -- Look up one run's items by key, not by walking the thread's whole history in ordinal order. + INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE thread_id = ${threadId} AND run_id = ${runId} ORDER BY ordinal ASC, turn_item_id ASC `.pipe(Effect.flatMap(decodeRows(decodeTurnItemPayload, threadId))); const providerSessions = yield* sql` From dc48a23a36a89a39d1dae5fc9e86bb204e7f3df4 Mon Sep 17 00:00:00 2001 From: Behzad Motaghi Date: Sat, 31 Oct 2026 22:37:41 +0100 Subject: [PATCH 2/2] fix(server): hint turn-start run items to the run ordinal index --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 5 ++++- apps/server/src/orchestration-v2/ProjectionStore.ts | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 10ee9878d705b..15bc0370526ef 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3406,7 +3406,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { detail.includes("orchestration_v2_projection_turn_items"), ); assert.lengthOf(lookup, 1); - assert.match(lookup[0]!, /^SEARCH .*\(thread_id=\? AND run_id=\?\)$/); + assert.match( + lookup[0]!, + /^SEARCH .*\((?:thread_id=\? AND )?run_id=\?(?: AND thread_id=\?)?\)$/, + ); } }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ec65e679facbf..865b455577085 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4299,7 +4299,7 @@ export const layer: Layer.Layer = const turnItems = yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items -- Look up one run's items by key, not by walking the thread's whole history in ordinal order. - INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx + INDEXED BY orchestration_v2_projection_turn_items_run_ordinal_idx WHERE thread_id = ${threadId} AND run_id = ${runId} ORDER BY ordinal ASC, turn_item_id ASC `.pipe(Effect.flatMap(decodeRows(decodeTurnItemPayload, threadId))); const providerSessions = yield* sql`