diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index f7773665aa9b0..21afe0b0574c8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -28,6 +28,7 @@ import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import { orchestrationCommandDuration, orchestrationCommandsTotal, @@ -2199,3 +2200,293 @@ it.effect("a user message dispatched first stays run 1 when its lineage read is ); }).pipe(Effect.provide(layerGatedLineageRead)); }); + +it.effect("queued-message commands read a long thread's history without walking all of it", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:long-history-reads"); + const driver = ProviderDriverKind.make("codex"); + const now = yield* DateTime.now; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-long-history-reads"), + threadId, + projectId: ProjectId.make("project:long-history-reads"), + title: "Long history", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + const providerThreadId = ProviderThreadId.make("provider-thread:long-history-reads"); + yield* projections.apply({ + id: EventId.make("event:long-history-reads:provider-thread"), + type: "provider-thread.updated", + threadId, + driver, + providerInstanceId: instanceId, + occurredAt: now, + payload: { + id: providerThreadId, + appThreadId: threadId, + ownerNodeId: null, + driver, + providerInstanceId: instanceId, + providerSessionId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 4, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + pendingBackgroundTasks: [], + }, + }); + const dispatch = (name: string) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`dispatch:long-history-reads:${name}`), + threadId, + messageId: MessageId.make(`message:long-history-reads:${name}`), + // A compacting target lets promote skip it without a live provider session. + text: name === "active" ? "/compact" : `Request ${name}`, + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const nextOrdinal = Effect.gen(function* () { + const [row] = yield* sql<{ readonly ordinal: number }>` + SELECT COALESCE(MAX(ordinal), 0) + 1 AS ordinal + FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId}`; + return row!.ordinal; + }); + // Settled history: each run leaves completed leaf nodes, tool output and a reply. + for (const name of ["1", "2", "3", "active"]) { + yield* dispatch(name); + const run = (yield* projections.getThreadProjection(threadId)).runs.at(-1)!; + const active = name === "active"; + yield* projections.apply({ + id: EventId.make(`event:long-history-reads:${name}:run`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...run, + providerThreadId, + status: active ? "running" : "completed", + queuePosition: null, + startedAt: now, + completedAt: active ? null : now, + }, + }); + if (active) continue; + for (const kind of ["tool_call", "assistant_message"] as const) { + const nodeId = NodeId.make(`node:long-history-reads:${name}:${kind}`); + yield* projections.apply({ + id: EventId.make(`event:${nodeId}`), + type: "node.updated", + threadId, + runId: run.id, + nodeId, + driver, + occurredAt: now, + payload: { + id: nodeId, + threadId, + runId: run.id, + rootNodeId: run.rootNodeId!, + parentNodeId: run.rootNodeId, + kind, + status: "completed", + countsForRun: false, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }, + }); + } + const replyId = MessageId.make(`message:long-history-reads:${name}:reply`); + yield* projections.apply({ + id: EventId.make(`event:${replyId}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: replyId, + threadId, + runId: run.id, + nodeId: NodeId.make(`node:long-history-reads:${name}:assistant_message`), + createdBy: "agent", + creationSource: "server", + role: "assistant", + text: `Reply ${name}`, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + const itemBase = { + threadId, + runId: run.id, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + status: "completed" as const, + startedAt: now, + completedAt: now, + updatedAt: now, + }; + for (const item of [ + { + ...itemBase, + id: TurnItemId.make(`item:long-history-reads:${name}:command`), + nodeId: NodeId.make(`node:long-history-reads:${name}:tool_call`), + ordinal: yield* nextOrdinal, + title: "command", + type: "command_execution" as const, + input: "cat build.log", + }, + { + ...itemBase, + id: TurnItemId.make(`item:long-history-reads:${name}:reply`), + nodeId: NodeId.make(`node:long-history-reads:${name}:assistant_message`), + ordinal: yield* nextOrdinal, + title: null, + type: "assistant_message" as const, + messageId: replyId, + text: `Reply ${name}`, + streaming: false, + }, + ]) { + yield* projections.apply({ + id: EventId.make(`event:${item.id}`), + type: "turn-item.updated", + threadId, + runId: run.id, + nodeId: item.nodeId, + driver, + occurredAt: now, + payload: item, + }); + } + } + const activeRun = (yield* projections.getThreadProjection(threadId)).runs.at(-1)!; + // Stop needs a live provider turn to interrupt. + yield* projections.apply({ + id: EventId.make("event:long-history-reads:active-turn"), + type: "provider-turn.updated", + threadId, + runId: activeRun.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:long-history-reads"), + providerThreadId, + nodeId: activeRun.rootNodeId!, + runAttemptId: activeRun.activeAttemptId!, + nativeTurnRef: { driver, nativeId: "native:long-history-reads", strength: "strong" }, + ordinal: 4, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + + const statements: Array["compile"]>> = []; + const observe: Statement.Transformer = (statement) => { + const compiled = statement.compile(); + // Row reads and the message-ordering pass; a MAX(ordinal) probe reads one index entry. + if (/_(nodes|turn_items)\b/.test(compiled[0]) && /payload_json|GROUP BY/.test(compiled[0])) { + statements.push(compiled); + } + return Effect.succeed(statement); + }; + const historyWalks = Effect.fnUntraced(function* (commandType: string) { + const walks: Array = []; + for (const [query, params] of statements.splice(0)) { + const plan = yield* sql.unsafe<{ detail: string }>(`EXPLAIN QUERY PLAN ${query}`, params); + for (const { detail } of plan) { + // A history-sized read: an ordering pass over every message item for a narrowed + // message read, or a node/turn-item row search bounded by nothing but the thread. + // Walking only a covering index's ids reads no payloads. Stop still loads every + // message, so its ordering pass is no larger than that read. + if ( + (detail.includes("TEMP B-TREE FOR GROUP BY") && query.includes("json_each")) || + (/^SEARCH .*(node|turn_item).* \(thread_id=\?\)$/.test(detail) && + !detail.includes("USING COVERING INDEX")) || + // Every active run in the environment, not only this thread's. + /^SEARCH .*runs.* \(status=\?\)$/.test(detail) || + // A status-filtered read that fetches every typed row to check its status. + (/^SEARCH .*turn_items USING INDEX .* \(thread_id=\? AND type=\?\)$/.test(detail) && + query.includes("status IN")) + ) { + walks.push(`${commandType}: ${detail}\n in: ${query.replace(/\s+/g, " ").trim()}`); + } + } + } + return walks; + }); + const observed = (effect: Effect.Effect) => + effect.pipe(Effect.provideService(Statement.CurrentTransformer, observe)); + + yield* observed(dispatch("queued")); + const walks = yield* historyWalks("message.dispatch"); + const queuedRun = (yield* projections.getThreadProjection(threadId)).runs.at(-1)!; + assert.equal(queuedRun.status, "queued"); + yield* observed( + orchestrator.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make("edit:long-history-reads"), + threadId, + runId: queuedRun.id, + text: "Request queued, edited", + }), + ); + walks.push(...(yield* historyWalks("queued-run.edit"))); + yield* observed( + orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("promote:long-history-reads"), + threadId, + queuedRunId: queuedRun.id, + targetRunId: activeRun.id, + }), + ); + walks.push(...(yield* historyWalks("queued-message.promote-to-steer"))); + yield* observed( + orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("interrupt:long-history-reads"), + threadId, + runId: activeRun.id, + }), + ); + walks.push(...(yield* historyWalks("run.interrupt"))); + + const after = yield* projections.getThreadProjection(threadId); + assert.deepEqual( + after.messages.filter((message) => message.role === "user").map((message) => message.text), + ["Request 1", "Request 2", "Request 3", "/compact", "Request queued, edited"], + ); + assert.deepEqual( + after.providerThreads.map((providerThread) => providerThread.id), + [providerThreadId], + ); + assert.deepEqual(walks, [], walks.join("\n")); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bd06785814cf1..a3ad41ec9eb4a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10839,8 +10839,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "runtimeRequests", "subagents", ], - // User messages measure a restart's model switch. - { turnItemTypes: ["user_message"], messageRoles: ["user"] }, + // User messages measure a restart's model switch. Steering reads root and live nodes only. + { turnItemTypes: ["user_message"], messageRoles: ["user"], omitSettledLeafNodes: true }, ); if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) { return yield* new OrchestratorDispatchError({ @@ -12644,6 +12644,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "reasoning", ], turnItemStatuses: ["pending", "running", "waiting"], + // Stop reads root nodes and ends open ones; a settled leaf has nothing left to end. + omitSettledLeafNodes: true, }, ); const run = projection.runs.find((candidate) => candidate.id === command.runId); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 23d64f289623b..9fc4e7be00c7a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -755,7 +755,7 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }) .pipe(Effect.withTracer(command.tracer)); for (const [table, seeks] of [ - ["turn_items", ["USING INDEX orchestration_v2_projection_turn_items_conversation_idx"]], + ["turn_items", ["turn_items_thread_type_idx (thread_id=? AND type=?)"]], ["nodes", ["nodes_settled_leaf_idx (thread_id=? AND =?)"]], ["subagents", ["USING COVERING INDEX orchestration_v2_projection_subagents_settled_idx"]], [ @@ -1292,9 +1292,9 @@ it.effect("record reads filter notification sources without dropping matching me `EXPLAIN QUERY PLAN ${ordinalLookup}`, [threadId, "[]"], )).map((row) => row.detail); - // User-only reads must stay on the partial index instead of scanning every turn item. + // User-only reads must stay on a type-bounded index instead of scanning every turn item. assert.isTrue( - plan.some((detail) => detail.includes("turn_items_user_message_idx")), + plan.some((detail) => /turn_items_(user_message|thread_type)_idx/.test(detail)), plan.join("\n"), ); // With every user item loaded, the items give the order and the lookup does not run. diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9ad158dd533bb..841cd0a0b5b7d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1151,15 +1151,6 @@ const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema const matchesNothing = (...filters: ReadonlyArray | undefined>) => filters.some((values) => values?.length === 0); -// A typed item read that names no tool type fits in migration 082's conversation index. A run or -// node filter is left alone: it already reads through a narrower index. -const readsConversationItemsOnly = (filter: ProjectionRecordFilter | undefined) => - filter?.turnItemTypes !== undefined && - filter.turnItemRunId === undefined && - filter.turnItemRunIds === undefined && - filter.turnItemNodeIds === undefined && - filter.turnItemTypes.every((type) => DEFERRED_WORK_ITEM_TYPES.every((tool) => tool !== type)); - const encodeThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); @@ -3490,6 +3481,16 @@ export const layer: Layer.Layer = `; }); + const turnItemMatch = (threadId: ThreadId, filter: ProjectionRecordFilter | undefined) => sql` + thread_id = ${threadId} + ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} + ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} + ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} + ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} + ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} + `; + // An id filter reads by primary key. `+` keeps SQLite off the thread index, which walks // every row of the thread to find the few ids. const threadColumn = (ids: ReadonlyArray | undefined, column = "thread_id") => @@ -3590,19 +3591,14 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_turn_items ${ - // A type list of conversation items reads them from the conversation index. - // Any other plan fetches every tool call of the thread to check its type. - readsConversationItemsOnly(filter) - ? sql`INDEXED BY orchestration_v2_projection_turn_items_conversation_idx - WHERE thread_id = ${threadId} AND type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES}` - : sql`WHERE thread_id = ${threadId}` + filter?.turnItemTypes === undefined + ? sql`WHERE ${turnItemMatch(threadId, filter)}` + : // Seeks the type index. Sorted, SQLite would walk the thread by ordinal instead. + sql`WHERE rowid IN ( + SELECT rowid FROM orchestration_v2_projection_turn_items + WHERE ${turnItemMatch(threadId, filter)} + )` } - ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} - ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} - ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} - ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} - ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} ORDER BY ordinal ASC, turn_item_id ASC ` : yield* sql` diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index ff799863c64c6..6399ee7b85950 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -95,6 +95,7 @@ import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; import Migration0086 from "./Migrations/086_QueuedTurnInputIndex.ts"; import Migration0087 from "./Migrations/087_ProjectionProjectsCustomTitle.ts"; +import Migration0088 from "./Migrations/088_TurnItemTypeStatusIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -185,6 +186,7 @@ const migrationEntries = [ [85, "TurnItemCounts", Migration0085], [86, "QueuedTurnInputIndex", Migration0086], [87, "ProjectionProjectsCustomTitle", Migration0087], + [88, "TurnItemTypeStatusIndex", Migration0088], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/088_TurnItemTypeStatusIndex.ts b/apps/server/src/persistence/Migrations/088_TurnItemTypeStatusIndex.ts new file mode 100644 index 0000000000000..09e0b3273d6cf --- /dev/null +++ b/apps/server/src/persistence/Migrations/088_TurnItemTypeStatusIndex.ts @@ -0,0 +1,13 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Typed reads of tool items (Stop's active-item read) otherwise walk every item of the thread. + // Status keeps an active-item read from fetching every settled row to reject it. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items(thread_id, type, status) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index e046cc4133dc6..a2bba7d790eaf 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -67,6 +67,7 @@ describe("V2 preview upgrade", () => { [85, "TurnItemCounts"], [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], + [88, "TurnItemTypeStatusIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -177,6 +178,7 @@ describe("V2 preview upgrade", () => { [85, "TurnItemCounts"], [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], + [88, "TurnItemTypeStatusIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );