From 9e0d6ecca22a8dc4241a48bf6495640e46a52d4e Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 5 Nov 2026 22:27:13 -0300 Subject: [PATCH 01/11] fix(server): thread shell reads no longer scan every finished run The shell query looked for live runs by walking a thread's whole ordinal index and reading each run row, and sorted every run to find the one that ended last. Use the status index for the live-run lookups and add an index matching the latest-executed-run ordering. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 107 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 4 + apps/server/src/persistence/Migrations.ts | 2 + .../081_ProjectionRunsExecutedIndex.ts | 14 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 5 files changed, 129 insertions(+) create mode 100644 apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c584dd27a8d23..a433de4e6eb74 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6522,6 +6522,113 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect( + "reads the shell of a thread with a long finished run history without scanning it", + () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "shell-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long finished history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (let ordinal = 1; ordinal <= 3; ordinal++) { + const runId = RunId.make(`run:${suffix}:${ordinal}`); + const rootNodeId = NodeId.make(`node:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:run-${ordinal}`), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${suffix}:${ordinal}`), + rootNodeId, + activeAttemptId: null, + status: "completed" as const, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + const shell = yield* projectionStore + .getThreadShell(threadId) + .pipe(Effect.withTracer(tracer)); + assert.isNotNull(shell); + const statement = statements.find((query) => query.includes("AS activity_run_status")); + assert.isDefined(statement); + const plan = yield* sql.unsafe<{ + readonly id: number; + readonly parent: number; + readonly detail: string; + }>(`EXPLAIN QUERY PLAN ${statement}`); + // With no live run, a lookup that walks the ordinal index reads every run the thread has had. + const liveRunLookups = plan.filter((row) => + row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), + ); + assert.isAbove(liveRunLookups.length, 0); + for (const lookup of liveRunLookups) { + assert.include( + lookup.detail, + "orchestration_v2_projection_runs_thread_status_idx (thread_id=? AND status=?)", + ); + } + // The newest run needs no sort, but the run that ended last must come from an ordered index. + const candidateLookups = plan.filter((row) => + row.detail.startsWith("SEARCH candidate USING INDEX orchestration_v2_projection_runs_"), + ); + assert.isAbove(candidateLookups.length, 0); + const sortedCandidates = plan.filter( + (row) => + row.detail === "USE TEMP B-TREE FOR ORDER BY" && + candidateLookups.some((lookup) => lookup.parent === row.parent), + ); + assert.deepEqual(sortedCandidates, []); + }), + ); + it.effect("selects the latest waiting secret only from active runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 23874c6014fe0..46c5879c646c9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6171,6 +6171,7 @@ export const layer: Layer.Layer = ( SELECT r.status FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6185,6 +6186,7 @@ export const layer: Layer.Layer = r.requested_at ) FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6246,6 +6248,7 @@ export const layer: Layer.Layer = SELECT secret.payload_json -- Keep runs outermost so completed history is never scanned for a pending secret. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE r.thread_id = t.thread_id @@ -6294,6 +6297,7 @@ export const layer: Layer.Layer = LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( SELECT r.run_id FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running') AND NOT ${isSilentShellRun("r")} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6a9c57e18cc43..475ec454f5bf9 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -83,6 +83,7 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts"; import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts"; import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; +import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -166,6 +167,7 @@ const migrationEntries = [ [78, "DropOrchestrationEventCorrelationIndex", Migration0078], [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], + [81, "ProjectionRunsExecutedIndex", Migration0081], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts new file mode 100644 index 0000000000000..74838b3f96909 --- /dev/null +++ b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts @@ -0,0 +1,14 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Matches the shell's latest-executed-run ordering, so it reads the newest rows instead of sorting every run. + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_executed_idx + ON orchestration_v2_projection_runs( + thread_id, (completed_at IS NULL), completed_at, ordinal, run_id + ) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 84fa4ddb44c77..5d5f524d0ec96 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -60,6 +60,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -163,6 +164,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 676afc44ec3e826a630a2d61ad2a8743d6e67adf Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 5 Nov 2026 22:36:10 -0300 Subject: [PATCH 02/11] test(server): require every live-run shell lookup to use the status index The plan assertion passed if any one lookup used the index, and a lookup that fell back to a scan was filtered out instead of failing. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a433de4e6eb74..771bc7ec5aad7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6608,7 +6608,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const liveRunLookups = plan.filter((row) => row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), ); - assert.isAbove(liveRunLookups.length, 0); + assert.lengthOf(liveRunLookups, 4); + assert.isFalse(plan.some((row) => /^SCAN (?:r|candidate)\b/.test(row.detail))); for (const lookup of liveRunLookups) { assert.include( lookup.detail, From e7b5c1d9d54f70953c84e408404d4a43a1bcfe0e Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 14:46:05 -0300 Subject: [PATCH 03/11] perf(server): reads of a long thread no longer touch every row On a thread with tens of thousands of items, the shell item count looked up each run in the main runs table, the command read walked every item to find user messages and errors, and the provider-thread read looked each node up through a non-covering index. Add compact covering indexes for those lookups and hint them where the planner prefers the primary key or the ordinal index. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 143 +++++++++++++++++- .../src/orchestration-v2/ProjectionStore.ts | 23 ++- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/082_CommandReadIndexes.ts | 22 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 5 files changed, 184 insertions(+), 8 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 771bc7ec5aad7..f0119e317b0fc 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6605,8 +6605,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { readonly detail: string; }>(`EXPLAIN QUERY PLAN ${statement}`); // With no live run, a lookup that walks the ordinal index reads every run the thread has had. - const liveRunLookups = plan.filter((row) => - row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), + const liveRunLookups = plan.filter( + (row) => + row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_") && + row.detail.includes("(thread_id=?"), ); assert.lengthOf(liveRunLookups, 4); assert.isFalse(plan.some((row) => /^SCAN (?:r|candidate)\b/.test(row.detail))); @@ -6630,6 +6632,143 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("reads a long thread's records for a command without touching every row", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "command-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long command history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (const ordinal of [1, 2]) { + const messageId = MessageId.make(`message:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:message-${ordinal}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId: null, + nodeId: null, + role: "user", + text: `prompt ${ordinal}`, + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:item-${ordinal}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`${suffix}:item:${ordinal}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: "prompt", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId, + inputIntent: "turn_start", + text: `prompt ${ordinal}`, + attachments: [], + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + yield* projectionStore + .getThreadRecords(threadId, ["runs", "nodes", "providerThreads", "messages", "turnItems"], { + turnItemTypes: ["user_message", "error"], + messageRoles: ["user"], + omitSettledLeafNodes: true, + omitSettledSubagents: true, + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); + const planOf = Effect.fnUntraced(function* (marker: string) { + const statement = statements.find((query) => query.includes(marker)); + assert.isDefined(statement, marker); + return (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + )).map((row) => row.detail); + }); + // Each read must reach only the rows it returns; a thread can hold tens of thousands. + const expectations = [ + { + marker: "(type || '') IN (SELECT value FROM json_each", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "MIN(ordinal) AS ordinal", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_run_idx", + }, + { + marker: "owner_node_id IN (", + uses: "COVERING INDEX orchestration_v2_projection_nodes_thread_node_idx (=?)", + avoids: "nodes_thread_run_idx", + }, + { + marker: "AS runless_item_count", + uses: "r USING COVERING INDEX orchestration_v2_projection_runs_id_status_idx (=?)", + avoids: "sqlite_autoindex_orchestration_v2_projection_runs_1", + }, + ] as const; + for (const { marker, uses, avoids } of expectations) { + const plan = (yield* planOf(marker)).join("\n"); + assert.include(plan, uses, marker); + assert.notInclude(plan, avoids, marker); + } + }), + ); + it.effect("selects the latest waiting secret only from active runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 46c5879c646c9..c80840a4b10e7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3139,8 +3139,14 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${ + // The ordinal index would read every item of the thread to find these types. + filter?.turnItemTypes === undefined + ? sql`` + : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + } + WHERE ${filter?.turnItemTypes === undefined ? sql`thread_id = ${threadId}` : sql`(thread_id || '') = ${threadId}`} + ${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)}))`} @@ -3493,9 +3499,10 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_provider_threads WHERE (thread_id = ${threadId} OR owner_node_id IN ( - SELECT node_id + SELECT node_id || '' FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} + INDEXED BY orchestration_v2_projection_nodes_thread_node_idx + WHERE (thread_id || '') = ${threadId} ) OR provider_thread_id IN ( SELECT provider_thread_id @@ -3732,7 +3739,8 @@ export const layer: Layer.Layer = 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') + INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + 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') `; @@ -6398,7 +6406,10 @@ export const layer: Layer.Layer = WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY thread_id, run_id ) counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_id + -- The covering index answers each run from a compact page set instead of the runs table. + LEFT JOIN orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_id_status_idx + ON (r.run_id || '') = (counts.run_id || '') GROUP BY counts.thread_id `; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 475ec454f5bf9..0c80c57fae550 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -84,6 +84,7 @@ import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationInd import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; +import Migration0082 from "./Migrations/082_CommandReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -168,6 +169,7 @@ const migrationEntries = [ [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionRunsExecutedIndex", Migration0081], + [82, "CommandReadIndexes", Migration0082], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts new file mode 100644 index 0000000000000..a7ab4e95074b6 --- /dev/null +++ b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Each lets a read of one long thread skip the rows it does not return. Every column a query + // could match is wrapped in `|| ''`, so only a lookup that writes the same expression can use + // them. Plain indexes were chosen for unrelated statements over their partial indexes. + yield* sql` + CREATE INDEX orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items((thread_id || ''), (type || ''), ordinal, turn_item_id) + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_nodes_thread_node_idx + ON orchestration_v2_projection_nodes((thread_id || ''), (node_id || '')) + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_id_status_idx + ON orchestration_v2_projection_runs((run_id || ''), status) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 5d5f524d0ec96..39cb418baf47a 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -61,6 +61,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -165,6 +166,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 72bd82bc36e0ac446bac9539625519bc94e586ec Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 6 Nov 2026 20:21:18 +0000 Subject: [PATCH 04/11] perf(server): run- and node-scoped turn-item reads keep their own index --- .../src/orchestration-v2/ProjectionStore.test.ts | 11 +++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 16 +++++++++++----- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 4721352fe1411..b15918547a5c0 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6795,6 +6795,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { omitSettledSubagents: true, }) .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["command_execution"], + turnItemRunId: RunId.make(`run:${suffix}`), + }) + .pipe(Effect.withTracer(tracer)); yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); const planOf = Effect.fnUntraced(function* (marker: string) { const statement = statements.find((query) => query.includes(marker)); @@ -6810,6 +6816,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { uses: "turn_items_thread_type_idx (=? AND =?)", avoids: "turn_items_thread_ordinal_idx", }, + { + marker: "AND type IN (SELECT value FROM json_each", + uses: "SEARCH orchestration_v2_projection_turn_items USING INDEX", + avoids: "turn_items_thread_type_idx", + }, { marker: "MIN(ordinal) AS ordinal", uses: "turn_items_thread_type_idx (=? AND =?)", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7106c95ea37f2..cedf45bedf321 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3141,6 +3141,11 @@ export const layer: Layer.Layer = readonly work_last_ordinal?: number | null; readonly work_item_count?: number | null; }; + const byTypeOnly = + filter?.turnItemTypes !== undefined && + filter.turnItemRunId === undefined && + filter.turnItemRunIds === undefined && + filter.turnItemNodeIds === undefined; const boundedTurnItemRows: ReadonlyArray = (fields !== undefined && !fields.includes("turnItems")) || (window === undefined && @@ -3158,13 +3163,14 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_turn_items ${ // The ordinal index would read every item of the thread to find these types. - filter?.turnItemTypes === undefined - ? sql`` - : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + // Run and node filters keep their narrower indexes. + byTypeOnly + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + : sql`` } - WHERE ${filter?.turnItemTypes === undefined ? sql`thread_id = ${threadId}` : sql`(thread_id || '') = ${threadId}`} + WHERE ${byTypeOnly ? sql`(thread_id || '') = ${threadId}` : 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?.turnItemTypes === undefined ? sql`` : byTypeOnly ? sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))` : 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)}))`} From 8f7ed10af92a8fa8c91636f052b171ea6a534902 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 16:32:09 -0300 Subject: [PATCH 05/11] perf(server): turn-start guards read one run instead of the recovery projection isCurrentAttemptInStatus and shouldFinalizeRun loaded and decoded the whole runtime recovery projection to compare a run's status and active attempt, and a turn start calls them about seven times. Read the run by id instead. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 13 ++++++++ .../src/orchestration-v2/ProjectionStore.ts | 20 ++++++++++++ .../ProviderTurnControlService.test.ts | 1 + .../ProviderTurnStartService.test.ts | 5 +++ .../ProviderTurnStartService.ts | 31 ++++++++++--------- 5 files changed, 55 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index b15918547a5c0..98e19055bc76b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6694,6 +6694,19 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { candidateLookups.some((lookup) => lookup.parent === row.parent), ); assert.deepEqual(sortedCandidates, []); + + // A turn-start guard needs one run, not the thread's whole recovery projection. + const runRead = traceSqlStatements(); + const run = yield* projectionStore + .getRun(threadId, RunId.make(`run:${suffix}:2`)) + .pipe(Effect.withTracer(runRead.tracer)); + assert.equal(run?.ordinal, 2); + assert.isFalse( + runRead.statements.some((query) => /turn_items|projection_nodes|messages/.test(query)), + ); + assert.isUndefined( + yield* projectionStore.getRun(threadId, RunId.make(`run:${suffix}:absent`)), + ); }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index cedf45bedf321..5996313397008 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -612,6 +612,10 @@ export interface ProjectionStoreV2Shape { readonly getRuntimeRecoveryProjection: ( threadId: ThreadId, ) => Effect.Effect; + readonly getRun: ( + threadId: ThreadId, + runId: RunId, + ) => Effect.Effect; readonly getPendingNativeUserInputs: ( threadId: ThreadId, providerTurnId: ProviderTurnId, @@ -5507,6 +5511,21 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError(controlReadError(threadId))); + const getRun: ProjectionStoreV2Shape["getRun"] = (threadId, runId) => + sql + .withTransaction( + Effect.gen(function* () { + yield* requireThread(threadId); + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_runs + WHERE run_id = ${runId} AND thread_id = ${threadId}`; + return rows[0] === undefined + ? undefined + : yield* decodeRunPayload(rows[0].payload_json); + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const canStartQueuedRun: ProjectionStoreV2Shape["canStartQueuedRun"] = (threadId) => sql .withTransaction( @@ -7518,6 +7537,7 @@ export const layer: Layer.Layer = getCheckpointCaptureContext: readOnReader(getCheckpointCaptureContext), getCheckpointRollbackContext: readOnReader(getCheckpointRollbackContext), getRunMessage: readOnReader(getRunMessage), + getRun: readOnReader(getRun), canStartQueuedRun: readOnReader(canStartQueuedRun), getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 55e2a9e50e3ef..6a52db7147cec 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -230,6 +230,7 @@ it.effect( getTurnStartContext: () => Effect.die("unused"), getTurnStartHistory: () => Effect.die("unused"), getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), + getRun: () => Effect.die("unused getRun"), getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index d1b5168258874..0859fa15e56cc 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -970,6 +970,11 @@ function makeLocalCommandHarness(input: { (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }), + getRun: (_threadId, id) => + Effect.as( + failReadIfRunning, + projection.runs.find((candidate) => candidate.id === id), + ), getTurnStartHistory: () => input.sideChat !== undefined ? Effect.succeed([]) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 54b16d7aa2aef..2f5cdd03df2bb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -238,12 +238,13 @@ export const layer: Layer.Layer< >; }) => { const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return run?.activeAttemptId === input.attemptId && run.status === expectedStatus; - }), - ); + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => run?.activeAttemptId === input.attemptId && run.status === expectedStatus, + ), + ); return { isCurrentAttemptInStatus, loadInheritedBackgroundTurnItems: () => @@ -261,15 +262,15 @@ export const layer: Layer.Layer< ), shouldStartProviderTurn: () => isCurrentAttemptInStatus("running"), shouldFinalizeRun: () => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return ( - run?.activeAttemptId === input.attemptId && - (run.status === "starting" || run.status === "running") - ); - }), - ), + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => + run?.activeAttemptId === input.attemptId && + (run.status === "starting" || run.status === "running"), + ), + ), hasUnpairedRunInterruptRequest: () => projectionStore .hasUnpairedRunInterruptRequest( From 6d1e4e07986a8f7c15b6c86d312d9cc77f35f7dd Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 18:44:38 -0300 Subject: [PATCH 06/11] perf(server): settling a subagent and MCP calls read less of a long parent thread Settling an app-owned subagent loaded every node, user message, subagent, provider thread/turn and context transfer of the parent, to use one task, its node, its item and its delivery cohort. On a dispatcher thread with 35k nodes and 7k prompts that cost over a second of reads per child. It now reads the records it names, and takes the result-transfer check from the child side. MCP calls that identify their caller loaded the caller thread's provider threads and turns, which no tool reads. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 21 +----- .../DelegatedCompletionDelivery.test.ts | 53 +++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 65 +++++++++++++------ 3 files changed, 101 insertions(+), 38 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 9d15408483a3e..e6d6ac4593729 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1081,15 +1081,7 @@ const make = Effect.gen(function* () { const loadProjection = (threadId: ThreadId, includeUserMessages = true) => loadThreadRecords( threadId, - [ - "runs", - "messages", - "providerThreads", - "providerTurns", - "runtimeRequests", - "contextTransfers", - "turnItems", - ], + ["runs", "messages", "runtimeRequests", "contextTransfers", "turnItems"], { turnItemTypes: [], messageRoles: includeUserMessages ? ["user"] : [] }, ); @@ -1123,16 +1115,7 @@ const make = Effect.gen(function* () { threadManagement .getProjectThreadRecords( { projectId, threadId }, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "runtimeRequests", - "contextTransfers", - "turnItems", - ], + ["runs", "messages", "subagents", "runtimeRequests", "contextTransfers", "turnItems"], { turnItemTypes: [], messageRoles: ["user"] }, ) .pipe(Effect.mapError(threadManagementFailure)); diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index db3a2759e3ac5..6fb060d164db5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2763,6 +2763,59 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect("settles a child without reading all of the parent's nodes", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:bounded-finalize-parent"); + const projectId = ProjectId.make("project:bounded-finalize"); + const runId = RunId.make("run:bounded-finalize-parent"); + const rootNodeId = NodeId.make("node:bounded-finalize-parent"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:bounded-finalize-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "bounded-finalize-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const statements: string[] = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (time, exit) => { + end(time, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer)); + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal(settled.subagents.find((task) => task.id === child.taskId)?.status, "completed"); + // A dispatcher parent holds thousands of nodes; settling one child names its own. + const nodeReads = statements.filter((query) => + query.includes("FROM orchestration_v2_projection_nodes"), + ); + assert.isNotEmpty(nodeReads); + assert.isTrue(nodeReads.every((query) => query.includes("node_id IN"))); + }), + ); + it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..e9076edbefbb5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12803,21 +12803,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; - const parentProjection = yield* projectionStore.getThreadRecords( + // A dispatcher parent holds thousands of tasks, runs and prompts, so read only this task's + // records and the delivery cohort the completion plan joins. + const taskRecords = yield* projectionStore.getThreadRecords( parentThreadId, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "nodes", - "turnItems", - "contextTransfers", - ], - { turnItemTypes: ["subagent"], messageRoles: ["user"] }, + ["subagents", "nodes", "turnItems"], + { + subagentIds: [options.task.id], + nodeIds: [options.task.id], + turnItemNodeIds: [options.task.id], + turnItemTypes: ["subagent"], + }, ); - const task = parentProjection.subagents.find( + const task = taskRecords.subagents.find( (candidate) => candidate.id === options.task.id && candidate.origin === "app_owned" && @@ -12826,7 +12824,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (task === undefined) { return; } - const existingResultTransfer = parentProjection.contextTransfers.find( + // The result transfer runs from the child, so the child's side lists it too. + const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [ + "contextTransfers", + ]); + const existingResultTransfer = contextTransfers.find( (transfer) => transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId && @@ -12840,12 +12842,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; const result = subagentResultForRun(childProjection, { ...childRun, status: outcome }); - const parentRun = - task.runId === null + const taskRunRecords = yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], { + runIds: task.runId === null ? [] : [task.runId], + }); + const parentRun = taskRunRecords.runs.find((candidate) => candidate.id === task.runId); + const delivery = parentRun?.delegatedCompletion?.delivery; + const deliveryRecords = + delivery == null + ? undefined + : yield* projectionStore.getThreadRecords(parentThreadId, ["messages", "subagents"], { + messageIds: [delivery.messageId], + subagentIds: [...new Set([...delivery.taskIds, task.id])], + }); + const deliveryRunId = deliveryRecords?.messages[0]?.runId ?? null; + const deliveryRun = + deliveryRunId === null || deliveryRunId === parentRun?.id ? undefined - : parentProjection.runs.find((candidate) => candidate.id === task.runId); - const parentNode = parentProjection.nodes.find((candidate) => candidate.id === task.id); - const parentTurnItem = parentProjection.turnItems.find( + : (yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], { + runIds: [deliveryRunId], + })).runs[0]; + const parentProjection = { + thread: taskRecords.thread, + runs: [ + ...(parentRun === undefined ? [] : [parentRun]), + ...(deliveryRun ? [deliveryRun] : []), + ], + messages: deliveryRecords?.messages ?? [], + subagents: deliveryRecords?.subagents ?? [task], + providerTurns: [], + }; + const parentNode = taskRecords.nodes.find((candidate) => candidate.id === task.id); + const parentTurnItem = taskRecords.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, ); const updatedTask: OrchestrationV2Subagent = { From 860ff723ae9901a1f75a40fea1c4d36f1b5a9141 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 19:32:01 -0300 Subject: [PATCH 07/11] perf(server): run-, node- and child-scoped reads name their index Without statistics SQLite answers a turn-item read filtered by run or node by walking the thread's whole ordinal index (2.2 s on a thread with tens of thousands of items), whether the filter is a run id, a run id list or node ids. Name the matching index for each, and drop the always-false OR arm from the run list so the run id can be a search key. A restart continuation does this read once per interrupted run. Finalizing a child at startup loaded every task and context transfer of its parent to find its own; read the child's tasks through the child-thread index and its result transfers from the child side. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 13 +++++-- .../orchestration-v2/ProjectionStore.test.ts | 38 +++++++++++++++++-- .../src/orchestration-v2/ProjectionStore.ts | 24 +++++++++--- 3 files changed, 62 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e9076edbefbb5..c040da43a5aef 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13033,8 +13033,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) return; const progress = delegatedTaskProgress(childControls); - const parent = yield* projectionStore.getThreadRecords(parentThreadId, [ - "subagents", + // A dispatcher parent holds thousands of tasks and transfers; this child owns a few of each. + const parent = yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], { + subagentChildThreadId: childThreadId, + }); + const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [ "contextTransfers", ]); for (const task of parent.subagents) { @@ -13044,9 +13047,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (task.id === childControls.thread.forkedFrom.nodeId && progress.state !== "result_available") || (task.result !== null && - parent.contextTransfers.some( + contextTransfers.some( (transfer) => - transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId, + transfer.type === "subagent_result" && + transfer.sourceThreadId === childThreadId && + transfer.targetThreadId === parentThreadId, )) ) continue; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 98e19055bc76b..107ed4ff4bb7f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6814,6 +6814,23 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { turnItemRunId: RunId.make(`run:${suffix}`), }) .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["run_interrupt_request"], + turnItemRunIds: [RunId.make(`run:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["subagent"], + turnItemNodeIds: [NodeId.make(`node:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["subagents"], { + subagentChildThreadId: ThreadId.make(`thread:${suffix}:child`), + }) + .pipe(Effect.withTracer(tracer)); yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); const planOf = Effect.fnUntraced(function* (marker: string) { const statement = statements.find((query) => query.includes(marker)); @@ -6830,9 +6847,24 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { avoids: "turn_items_thread_ordinal_idx", }, { - marker: "AND type IN (SELECT value FROM json_each", - uses: "SEARCH orchestration_v2_projection_turn_items USING INDEX", - avoids: "turn_items_thread_type_idx", + marker: "AND run_id = ?", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND run_id IN (SELECT value FROM json_each", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND node_id IN (SELECT value FROM json_each", + uses: "turn_items_node_ordinal_idx (node_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND child_thread_id = ?", + uses: "subagents_child_thread_idx (child_thread_id=?)", + avoids: "subagents_thread_idx", }, { marker: "MIN(ordinal) AS ordinal", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 5996313397008..0c889ad7dbe41 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -353,6 +353,8 @@ export interface ProjectionRecordFilter { readonly contextTransferStatuses?: ReadonlyArray; readonly subagentIds?: ReadonlyArray; readonly subagentRunIds?: ReadonlyArray; + /** Only the tasks whose child thread this is. */ + readonly subagentChildThreadId?: ThreadId; readonly nodeIds?: ReadonlyArray; readonly turnItemRequestIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; @@ -3145,6 +3147,10 @@ export const layer: Layer.Layer = readonly work_last_ordinal?: number | null; readonly work_item_count?: number | null; }; + const byRun = + filter?.turnItemRunId !== undefined || + (filter?.turnItemRunIds !== undefined && !filter.turnItemRunIds.includes(null)); + const byNode = !byRun && filter?.turnItemNodeIds !== undefined; const byTypeOnly = filter?.turnItemTypes !== undefined && filter.turnItemRunId === undefined && @@ -3166,11 +3172,15 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_turn_items ${ - // The ordinal index would read every item of the thread to find these types. - // Run and node filters keep their narrower indexes. - byTypeOnly - ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` - : sql`` + // Without statistics SQLite walks the thread's ordinal index and reads every + // item of a long thread; each filter names the index that reaches only its rows. + byRun + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx` + : byNode + ? sql`INDEXED BY orchestration_v2_projection_turn_items_node_ordinal_idx` + : byTypeOnly + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + : sql`` } WHERE ${byTypeOnly ? sql`(thread_id || '') = ${threadId}` : sql`thread_id = ${threadId}`} ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} @@ -3178,7 +3188,7 @@ export const layer: Layer.Layer = ${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))`} + ${filter?.turnItemRunIds === undefined ? sql`` : filter.turnItemRunIds.includes(null) ? sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR run_id IS NULL)` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))}))`} ORDER BY ordinal ASC, turn_item_id ASC ` : yield* sql` @@ -3464,7 +3474,9 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent + ${filter?.subagentChildThreadId === undefined ? sql`` : sql`INDEXED BY orchestration_v2_projection_subagents_child_thread_idx`} WHERE thread_id = ${threadId} + ${filter?.subagentChildThreadId === undefined ? sql`` : sql`AND child_thread_id = ${filter.subagentChildThreadId}`} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${ From 4dcd26f420651d400d31cba63e2a201351350279 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Sun, 8 Nov 2026 13:15:21 -0300 Subject: [PATCH 08/11] perf(server): sending a prompt in a compacted thread reads only what follows the compaction A new prompt loaded every run, attempt, node, user message, provider turn and context transfer of its thread before it could start a run, although a thread that was compacted only needs what follows the compaction. On a thread with about 7,000 runs that read took 9-14 s, and the run only showed as working after it finished. `message.dispatch` now reads the runs from the one before the latest completed compaction, plus live runs, fresh-context runs and the runs the command names (`sinceLastCompaction`). Run-scoped records follow those runs. Provider threads are the thread's own, and context transfers are the open ones that target it. A thread with no completed compaction reads in full, as does a fresh-context start or a provider switch, which hand the whole history to the next provider. The "only maintenance messages so far" title check no longer depends on the loaded messages: a window holding only `/compact` would have regenerated the title. It is its own query that stops at the first other user message. Migration 083 adds the indexes the window reads need: fresh-context runs and provider turns by attempt. The settled-subagent and settled-leaf-node filters look up run status in the covering index, which takes the subagent read of a 3,900-task thread from 1.4 s to 15 ms. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 39 ++++- .../orchestration-v2/ProjectionStore.test.ts | 128 +++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 152 ++++++++++++++++-- .../ProviderTurnControlService.test.ts | 1 + apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/083_CommandWindowIndexes.ts | 18 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 7 files changed, 323 insertions(+), 19 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c040da43a5aef..7990dfdb46132 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1272,7 +1272,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); // Runs in every command transaction; avoid reading thousands of finished leaf nodes. - const readCommandProjection = (threadId: ThreadId) => + const readCommandProjection = ( + threadId: ThreadId, + recent?: { keepRunIds: ReadonlyArray }, + ) => projectionStore .getThreadRecords( threadId, @@ -1297,6 +1300,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Commands here read only undelivered or active tasks; a dispatcher thread keeps // thousands of delivered ones, each carrying its full prompt. omitSettledSubagents: true, + ...(recent === undefined ? {} : { sinceLastCompaction: recent }), }, ) .pipe( @@ -1385,9 +1389,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.fnUntraced(function* ( threadId: ThreadId, events: Ref.Ref>, + recent?: { keepRunIds: ReadonlyArray }, ) { const pending = (yield* Ref.get(events)).filter((event) => event.threadId === threadId); - const stored = yield* Effect.option(readCommandProjection(threadId)); + const stored = yield* Effect.option(readCommandProjection(threadId, recent)); let projection: OrchestrationV2ThreadProjection; if (Option.isSome(stored)) { projection = stored.value; @@ -5615,7 +5620,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - let projection = yield* getProjectionWithPendingEvents(command.threadId, events); + // A compacted thread needs only its recent runs, plus any older run the command names. + let projection = yield* getProjectionWithPendingEvents(command.threadId, events, { + keepRunIds: [ + command.manualContinuationOfRunId, + command.restartContinuationOfRunId, + command.usageLimitContinuationOfRunId, + command.delegatedCompletion?.parentRunId, + ].flatMap((runId) => (runId === undefined ? [] : [runId])), + }); + const activeProviderInstanceId = projection.providerThreads.find( + (candidate) => candidate.id === projection.thread.activeProviderThreadId, + )?.providerInstanceId; + // A fresh session or another provider is handed the whole history, so it reads all of it. + if ( + command.freshContext === true || + (activeProviderInstanceId !== undefined && + activeProviderInstanceId !== + (command.modelSelection ?? projection.thread.modelSelection).instanceId) + ) { + projection = yield* getProjectionWithPendingEvents(command.threadId, events); + } if (command.manualContinuationOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId); const limited = latestRootProviderFailure(source ?? null, projection.turnItems); @@ -5788,16 +5813,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); projection = applyToProjection(projection, event); } - const userMessages = projection.messages.filter((message) => message.role === "user"); - const onlyMaintenanceHistory = - userMessages.length > 0 && userMessages.every(isNativeMaintenanceCommand); if ( !isNativeMaintenanceCommand(command) && ((command.titleSeed !== undefined && (yield* projectionStore .getMessageCount(command.threadId) .pipe(mapDispatchError(command))) === 0) || - (command.createdBy === "user" && onlyMaintenanceHistory)) + (command.createdBy === "user" && + (yield* projectionStore + .hasOnlyMaintenanceUserMessages(command.threadId) + .pipe(mapDispatchError(command))))) ) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 107ed4ff4bb7f..063969fb1518a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -706,6 +706,134 @@ it.effect("idle compaction filters keep the latest run, live work and the newest }).pipe(Effect.provide(layerTest)), ); +it.effect("a compacted thread reads its recent runs, live runs, fresh starts and named runs", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("compacted"); + const now = yield* DateTime.now; + const nodeId = NodeId.make("node:compacted"); + const runId = (ordinal: number) => RunId.make(`run:compacted:${ordinal}`); + const read = (keepRunIds: ReadonlyArray) => + store.getThreadRecords(threadId, ["runs", "messages"], { + messageRoles: ["user"], + sinceLastCompaction: { keepRunIds }, + }); + // Run 1 is the helper's rolled-back run. Run 2 is live, run 3 starts a fresh session and + // run 6 is the compaction; the window starts at run 5, the one before it. + for (const [ordinal, status] of [ + [2, "running"], + [3, "completed"], + [4, "completed"], + [5, "completed"], + [6, "completed"], + [7, "completed"], + ] as const) { + yield* store.apply({ + id: EventId.make(`event:compacted:run-${ordinal}`), + type: "run.created", + threadId, + runId: runId(ordinal), + nodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId(ordinal), + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:compacted:${ordinal}`), + rootNodeId: nodeId, + activeAttemptId: null, + status, + requestedAt: now, + startedAt: now, + completedAt: status === "running" ? null : now, + checkpointId: null, + contextHandoffId: null, + ...(ordinal === 3 ? { freshContext: true } : {}), + }, + }); + yield* store.apply({ + id: EventId.make(`event:compacted:message-${ordinal}`), + type: "message.updated", + threadId, + runId: runId(ordinal), + occurredAt: now, + payload: { + id: MessageId.make(`message:compacted:${ordinal}`), + threadId, + runId: runId(ordinal), + nodeId, + createdBy: "user", + creationSource: "web", + role: "user", + text: `Prompt ${ordinal}`, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + } + const rolledBackId = RunId.make("run:compacted:rolled-back"); + const ids = (rows: ReadonlyArray) => + rows.map((row) => row.id); + // Nothing is compacted yet, so the whole history is current. + assert.lengthOf((yield* read([])).runs, 7); + + yield* store.apply({ + id: EventId.make("event:compacted:compaction"), + type: "turn-item.updated", + threadId, + runId: runId(6), + nodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("item:compacted:compaction"), + threadId, + runId: runId(6), + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction", + driver, + }, + }); + const recent = yield* read([rolledBackId]); + assert.deepEqual(ids(recent.runs), [ + rolledBackId, + runId(2), + runId(3), + runId(5), + runId(6), + runId(7), + ]); + assert.deepEqual( + ids(recent.messages), + [2, 3, 5, 6, 7].map((ordinal) => `message:compacted:${ordinal}`), + ); + assert.deepEqual(ids((yield* read([])).runs), [ + runId(2), + runId(3), + runId(5), + runId(6), + runId(7), + ]); + }).pipe(Effect.provide(layerTest)), +); + it.effect("record reads filter notification sources without dropping matching message bodies", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 0c889ad7dbe41..c1e40ae14701b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -365,6 +365,13 @@ export interface ProjectionRecordFilter { * run is active or `except` names them. */ readonly omitSettledSubagents?: true | { readonly except: ReadonlyArray }; + /** + * Reads what a new message needs from a thread that was compacted: the runs from the one before + * the latest completed compaction, plus live runs, fresh-context runs and `keepRunIds`. Run- + * scoped records follow their runs, provider threads are the thread's own, and context transfers + * are the open ones that target it. A thread without a completed compaction reads in full. + */ + readonly sinceLastCompaction?: { readonly keepRunIds: ReadonlyArray }; } // A fork inherits its source's history before it has messages of its own, @@ -465,6 +472,10 @@ export interface ProjectionStoreV2Shape { options: ProjectionTimelinePageOptions, ) => Effect.Effect; readonly getMessageCount: (threadId: ThreadId) => Effect.Effect; + /** True when the thread has user messages and every one is a native maintenance command. */ + readonly hasOnlyMaintenanceUserMessages: ( + threadId: ThreadId, + ) => Effect.Effect; readonly getSubagentCount: (threadId: ThreadId) => Effect.Effect; /** Child threads of the thread's app-owned tasks, in task order, without decoding the tasks. */ readonly listAppOwnedChildThreadIds: ( @@ -3124,6 +3135,47 @@ export const layer: Layer.Layer = `; }); + /** Null when nothing was compacted, so the whole history is current. */ + const runsSinceLastCompaction = Effect.fn("ProjectionStore.runsSinceLastCompaction")(function* ( + threadId: ThreadId, + keepRunIds: ReadonlyArray, + ) { + const [compaction] = yield* sql<{ readonly from_ordinal: number }>` + SELECT COALESCE( + (SELECT MAX(previous.ordinal) FROM orchestration_v2_projection_runs AS previous + WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal), + latest.ordinal + ) AS from_ordinal + FROM ( + SELECT run.ordinal + FROM orchestration_v2_projection_turn_items AS item + INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + INNER JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id + WHERE (item.thread_id || '') = ${threadId} + AND (item.type || '') = 'compaction' AND item.status = 'completed' + ORDER BY item.ordinal DESC + LIMIT 1 + ) AS latest + `; + return compaction === undefined + ? null + : sql` + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND ordinal >= ${compaction.from_ordinal} + UNION + SELECT run_id FROM orchestration_v2_projection_runs + INDEXED BY orchestration_v2_projection_runs_thread_status_idx + WHERE thread_id = ${threadId} + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + UNION + SELECT run_id FROM orchestration_v2_projection_runs + INDEXED BY orchestration_v2_projection_runs_fresh_context_idx + WHERE thread_id = ${threadId} AND json_extract(payload_json, '$.freshContext') = 1 + UNION + SELECT value FROM json_each(${encodeIdList(keepRunIds)}) + `; + }); + const readCanonicalProjection = Effect.fn("ProjectionStore.readCanonicalProjection")( function* ( threadId: ThreadId, @@ -3141,6 +3193,10 @@ export const layer: Layer.Layer = if (!threadRow) { return yield* new ProjectionStoreThreadNotFoundError({ threadId }); } + const recentRuns = + window === undefined && filter?.sinceLastCompaction !== undefined + ? yield* runsSinceLastCompaction(threadId, filter.sinceLastCompaction.keepRunIds) + : null; type WindowPayloadRow = PayloadRow & { readonly work_last_id?: string | null; @@ -3148,10 +3204,12 @@ export const layer: Layer.Layer = readonly work_item_count?: number | null; }; const byRun = + recentRuns !== null || filter?.turnItemRunId !== undefined || (filter?.turnItemRunIds !== undefined && !filter.turnItemRunIds.includes(null)); const byNode = !byRun && filter?.turnItemNodeIds !== undefined; const byTypeOnly = + recentRuns === null && filter?.turnItemTypes !== undefined && filter.turnItemRunId === undefined && filter.turnItemRunIds === undefined && @@ -3187,6 +3245,7 @@ export const layer: Layer.Layer = ${filter?.turnItemTypes === undefined ? sql`` : byTypeOnly ? sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))` : 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}`} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} ${filter?.turnItemRunIds === undefined ? sql`` : filter.turnItemRunIds.includes(null) ? sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR run_id IS NULL)` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))}))`} ORDER BY ordinal ASC, turn_item_id ASC @@ -3376,7 +3435,14 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json - FROM orchestration_v2_projection_runs + FROM ${ + // Looking runs up by key beats filtering the thread's whole run index. + recentRuns === null + ? sql`orchestration_v2_projection_runs` + : sql`(SELECT run_id AS recent_run_id FROM (${recentRuns})) AS recent + CROSS JOIN orchestration_v2_projection_runs + ON orchestration_v2_projection_runs.run_id = recent.recent_run_id` + } WHERE thread_id = ${threadId} ${filter?.runIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.runIds)}))`} ${ @@ -3419,6 +3485,7 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_run_attempts WHERE thread_id = ${threadId} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ORDER BY run_id ASC, attempt_ordinal ASC ` : sql` @@ -3435,7 +3502,9 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_nodes AS node + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_nodes_thread_run_idx`} WHERE node.thread_id = ${threadId} + ${recentRuns === null ? sql`` : sql`AND node.run_id IN (${recentRuns})`} ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} ${ filter?.omitSettledLeafNodes === true @@ -3445,7 +3514,8 @@ export const layer: Layer.Layer = AND node.run_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = node.run_id + INDEXED BY orchestration_v2_projection_runs_id_status_idx + WHERE (run.run_id || '') = (node.run_id || '') AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') ) )` @@ -3474,7 +3544,14 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent - ${filter?.subagentChildThreadId === undefined ? sql`` : sql`INDEXED BY orchestration_v2_projection_subagents_child_thread_idx`} + ${ + filter?.subagentChildThreadId !== undefined + ? sql`INDEXED BY orchestration_v2_projection_subagents_child_thread_idx` + : filter?.omitSettledSubagents !== undefined + ? // Its columns decide settled rows, so only the open ones read their payload. + sql`INDEXED BY orchestration_v2_projection_subagents_settled_idx` + : sql`` + } WHERE thread_id = ${threadId} ${filter?.subagentChildThreadId === undefined ? sql`` : sql`AND child_thread_id = ${filter.subagentChildThreadId}`} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} @@ -3491,7 +3568,8 @@ export const layer: Layer.Layer = IN ('acknowledged', 'delivered', 'disposed') AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = subagent.run_id + INDEXED BY orchestration_v2_projection_runs_id_status_idx + WHERE (run.run_id || '') = (subagent.run_id || '') AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') ) ))` @@ -3538,13 +3616,17 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_provider_threads WHERE (thread_id = ${threadId} - OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) + ${ + recentRuns === null + ? sql`OR owner_node_id IN (${nodesOwningProviderThreads(threadId)}) OR provider_thread_id IN ( SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId} AND provider_thread_id IS NOT NULL - )) + )` + : sql`` + }) ${filter?.providerThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerThreadIds)}))`} ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC ` @@ -3565,7 +3647,15 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_provider_turns_attempt_idx`} WHERE thread_id = ${threadId} + ${ + recentRuns === null + ? sql`` + : sql`AND run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + WHERE thread_id = ${threadId} AND run_id IN (${recentRuns}))` + } ${filter?.providerTurnAttemptIds === undefined ? sql`` : sql`AND run_attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnAttemptIds)}))`} ${filter?.providerTurnThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnThreadIds)}))`} ${ @@ -3611,7 +3701,9 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_messages + ${recentRuns === null ? sql`` : sql`INDEXED BY orchestration_v2_projection_messages_run_idx`} WHERE thread_id = ${threadId} + ${recentRuns === null ? sql`` : sql`AND run_id IN (${recentRuns})`} ${filter?.messageIds === undefined ? sql`` : sql`AND message_id IN (SELECT value FROM json_each(${encodeIdList(filter.messageIds)}))`} ${filter?.messageRoles === undefined ? sql`` : sql`AND role IN (SELECT value FROM json_each(${encodeIdList(filter.messageRoles)}))`} ${filter?.monitorNotificationsOnly === true ? sql`AND json_extract(payload_json, '$.notification.source.kind') = 'monitor'` : sql``} @@ -3715,7 +3807,11 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_context_transfers - WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId}) + WHERE ${ + recentRuns === null + ? sql`(source_thread_id = ${threadId} OR target_thread_id = ${threadId})` + : sql`target_thread_id = ${threadId} AND status IN ('pending', 'running', 'waiting')` + } ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`} ORDER BY rowid ASC ` @@ -3762,7 +3858,13 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - if (fields !== undefined && fields.includes("messages") && messages.length > 1) { + // The recent user turns, loaded above, already order their own messages. + if ( + recentRuns === null && + 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 && @@ -6195,15 +6297,40 @@ export const layer: Layer.Layer = ), ); + // Mirrors JavaScript's `trim().toLowerCase()` of a message's text. + const trimmedLowerText = (payload: Statement.Fragment) => sql`lower(trim( + json_extract(${payload}, '$.text'), + char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, + 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) + ))`; + // Mirrors isNativeMaintenanceCommand over the thread's user messages. + const hasOnlyMaintenanceUserMessages: ProjectionStoreV2Shape["hasOnlyMaintenanceUserMessages"] = + (threadId) => + sql<{ readonly any_user: number; readonly other_user: number }>` + SELECT + EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages + WHERE thread_id = ${threadId} AND role = 'user' + ) AS any_user, + EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages + WHERE thread_id = ${threadId} AND role = 'user' + AND NOT ( + coalesce(json_array_length(payload_json, '$.attachments'), 0) = 0 + AND ${trimmedLowerText(sql.literal("payload_json"))} IN ('/compact', '/logout') + ) + ) AS other_user + `.pipe( + Effect.map(([row]) => row?.any_user === 1 && row.other_user === 0), + Effect.mapError(controlReadError(threadId)), + ); + // Mirror idleCompactionOf, including JavaScript trim whitespace. const isIdleCompactionMessage = (alias: string) => { const payload = sql.literal(`${alias}.payload_json`); return sql`( ${sql.literal(`${alias}.role`)} = 'user' - AND lower(trim(json_extract(${payload}, '$.text'), - char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, - 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) - )) = ${IDLE_COMPACTION_TEXT} + AND ${trimmedLowerText(payload)} = ${IDLE_COMPACTION_TEXT} AND coalesce(json_array_length(${payload}, '$.attachments'), 0) = 0 AND ( json_extract(${payload}, '$.idleCompaction') IS NOT NULL @@ -7556,6 +7683,7 @@ export const layer: Layer.Layer = hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), + hasOnlyMaintenanceUserMessages: readOnReader(hasOnlyMaintenanceUserMessages), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 6a52db7147cec..4b0b7421f7bca 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,6 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), + hasOnlyMaintenanceUserMessages: () => Effect.die("unused maintenance history"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 0c80c57fae550..55043dc2f79cc 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,6 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; import Migration0082 from "./Migrations/082_CommandReadIndexes.ts"; +import Migration0083 from "./Migrations/083_CommandWindowIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -170,6 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionRunsExecutedIndex", Migration0081], [82, "CommandReadIndexes", Migration0082], + [83, "CommandWindowIndexes", Migration0083], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts new file mode 100644 index 0000000000000..b2aac2e30b5bc --- /dev/null +++ b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts @@ -0,0 +1,18 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Reading the runs a new message needs from a compacted thread must not touch the older ones. + // Fresh-context starts are rare, so their partial index stays tiny. + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_fresh_context_idx + ON orchestration_v2_projection_runs(thread_id, ordinal) + WHERE json_extract(payload_json, '$.freshContext') = 1 + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_provider_turns_attempt_idx + ON orchestration_v2_projection_provider_turns(run_attempt_id) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 39cb418baf47a..0b184c580bfc9 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,6 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], [82, "CommandReadIndexes"], + [83, "CommandWindowIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -167,6 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], [82, "CommandReadIndexes"], + [83, "CommandWindowIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 2d7f3dffc13f723bccc22fef9d4710cce22774f5 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Sun, 8 Nov 2026 13:21:36 -0300 Subject: [PATCH 09/11] fix(server): the fresh-context run index accepts legacy rows with unreadable JSON The partial index added for compacted-thread reads evaluated `json_extract` on every run row, so writing a run whose stored payload is not valid JSON failed with "malformed JSON". Guard it with `json_valid`, as the other partial indexes on the run payload do, and insert such a row in the compacted-thread test. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 7 +++++++ apps/server/src/orchestration-v2/ProjectionStore.ts | 3 ++- .../src/persistence/Migrations/083_CommandWindowIndexes.ts | 4 +++- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 063969fb1518a..b023bf20abfa9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -811,6 +811,13 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and driver, }, }); + // Legacy rows can hold unreadable JSON; the fresh-context index must not reject them. + const sql = yield* SqlClient.SqlClient; + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES ('run:compacted:legacy', ${threadId}, 0, 'codex', 'completed', '', '{') + `; const recent = yield* read([rolledBackId]); assert.deepEqual(ids(recent.runs), [ rolledBackId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index c1e40ae14701b..a140d6eebc20d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3170,7 +3170,8 @@ export const layer: Layer.Layer = UNION SELECT run_id FROM orchestration_v2_projection_runs INDEXED BY orchestration_v2_projection_runs_fresh_context_idx - WHERE thread_id = ${threadId} AND json_extract(payload_json, '$.freshContext') = 1 + WHERE thread_id = ${threadId} AND CASE WHEN json_valid(payload_json) + THEN json_extract(payload_json, '$.freshContext') = 1 ELSE 0 END UNION SELECT value FROM json_each(${encodeIdList(keepRunIds)}) `; diff --git a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts index b2aac2e30b5bc..6d48c156f236a 100644 --- a/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts +++ b/apps/server/src/persistence/Migrations/083_CommandWindowIndexes.ts @@ -9,7 +9,9 @@ export default Effect.gen(function* () { yield* sql` CREATE INDEX orchestration_v2_projection_runs_fresh_context_idx ON orchestration_v2_projection_runs(thread_id, ordinal) - WHERE json_extract(payload_json, '$.freshContext') = 1 + WHERE CASE WHEN json_valid(payload_json) + THEN json_extract(payload_json, '$.freshContext') = 1 + ELSE 0 END `; yield* sql` CREATE INDEX orchestration_v2_projection_provider_turns_attempt_idx From 78364104317b36e6027ee4bfc4f5ac40c2b4f67c Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Sun, 8 Nov 2026 13:41:27 -0300 Subject: [PATCH 10/11] fix(server): a compacted thread's read keeps a steer's target run and the last shown prompt Two runs the compaction window dropped were still needed by a new prompt: - An explicit steer or restart names its target run. When that run finished before the window, the target was missing, so the "target already finished, start a new turn" path could not run and the command failed with "Target run was not found". The target is now kept like the other named runs. - A model switch is measured from the last run whose prompt the timeline shows, and `/compact` prompts are not shown. When the run before the compaction was itself a `/compact` run, the window held no baseline and the switch was lost. The window now starts at the last prompt before the compaction that is not a `/compact` run. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 4 + .../orchestration-v2/ProjectionStore.test.ts | 89 ++++++------ .../src/orchestration-v2/ProjectionStore.ts | 33 +++-- .../SteeringCompletion.integration.test.ts | 133 ++++++++++++++++++ 4 files changed, 203 insertions(+), 56 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5f7291d6598bc..209ceb2beb54f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5627,6 +5627,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.restartContinuationOfRunId, command.usageLimitContinuationOfRunId, command.delegatedCompletion?.parentRunId, + ...(command.dispatchMode.type === "steer_active" || + command.dispatchMode.type === "restart_active" + ? [command.dispatchMode.targetRunId] + : []), ].flatMap((runId) => (runId === undefined ? [] : [runId])), }); const activeProviderInstanceId = projection.providerThreads.find( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index b023bf20abfa9..09e4d88b005e7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -718,8 +718,8 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and messageRoles: ["user"], sinceLastCompaction: { keepRunIds }, }); - // Run 1 is the helper's rolled-back run. Run 2 is live, run 3 starts a fresh session and - // run 6 is the compaction; the window starts at run 5, the one before it. + // Run 1 is the helper's rolled-back run. Run 2 is live and run 3 starts a fresh session. + // Runs 6 and 9 are `/compact` runs that hold a compaction; run 8 is one that does not. for (const [ordinal, status] of [ [2, "running"], [3, "completed"], @@ -727,6 +727,8 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and [5, "completed"], [6, "completed"], [7, "completed"], + [8, "completed"], + [9, "completed"], ] as const) { yield* store.apply({ id: EventId.make(`event:compacted:run-${ordinal}`), @@ -770,7 +772,7 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and createdBy: "user", creationSource: "web", role: "user", - text: `Prompt ${ordinal}`, + text: [6, 8, 9].includes(ordinal) ? "/compact" : `Prompt ${ordinal}`, attachments: [], streaming: false, createdAt: now, @@ -781,36 +783,38 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and const rolledBackId = RunId.make("run:compacted:rolled-back"); const ids = (rows: ReadonlyArray) => rows.map((row) => row.id); - // Nothing is compacted yet, so the whole history is current. - assert.lengthOf((yield* read([])).runs, 7); - - yield* store.apply({ - id: EventId.make("event:compacted:compaction"), - type: "turn-item.updated", - threadId, - runId: runId(6), - nodeId, - driver, - occurredAt: now, - payload: { - id: TurnItemId.make("item:compacted:compaction"), + const compact = (ordinal: number) => + store.apply({ + id: EventId.make(`event:compacted:compaction-${ordinal}`), + type: "turn-item.updated", threadId, - runId: runId(6), + runId: runId(ordinal), nodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 2, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "compaction", driver, - }, - }); + occurredAt: now, + payload: { + id: TurnItemId.make(`item:compacted:compaction-${ordinal}`), + threadId, + runId: runId(ordinal), + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction", + driver, + }, + }); + // Nothing is compacted yet, so the whole history is current. + assert.lengthOf((yield* read([])).runs, 9); + + yield* compact(6); // Legacy rows can hold unreadable JSON; the fresh-context index must not reject them. const sql = yield* SqlClient.SqlClient; yield* sql` @@ -818,26 +822,19 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) VALUES ('run:compacted:legacy', ${threadId}, 0, 'codex', 'completed', '', '{') `; + const runIdsOf = (ordinals: ReadonlyArray) => ordinals.map(runId); + // The window starts at run 5, the prompt before the compaction. const recent = yield* read([rolledBackId]); - assert.deepEqual(ids(recent.runs), [ - rolledBackId, - runId(2), - runId(3), - runId(5), - runId(6), - runId(7), - ]); + assert.deepEqual(ids(recent.runs), [rolledBackId, ...runIdsOf([2, 3, 5, 6, 7, 8, 9])]); assert.deepEqual( ids(recent.messages), - [2, 3, 5, 6, 7].map((ordinal) => `message:compacted:${ordinal}`), + [2, 3, 5, 6, 7, 8, 9].map((ordinal) => `message:compacted:${ordinal}`), ); - assert.deepEqual(ids((yield* read([])).runs), [ - runId(2), - runId(3), - runId(5), - runId(6), - runId(7), - ]); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 5, 6, 7, 8, 9])); + + // Run 8 is a `/compact` run, so the model-switch baseline is the prompt before it, run 7. + yield* compact(9); + assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 7, 8, 9])); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a140d6eebc20d..f808027a802c0 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -366,8 +366,8 @@ export interface ProjectionRecordFilter { */ readonly omitSettledSubagents?: true | { readonly except: ReadonlyArray }; /** - * Reads what a new message needs from a thread that was compacted: the runs from the one before - * the latest completed compaction, plus live runs, fresh-context runs and `keepRunIds`. Run- + * Reads what a new message needs from a thread that was compacted: the runs from the last + * prompt before the latest completed compaction, plus live runs, fresh-context runs and `keepRunIds`. Run- * scoped records follow their runs, provider threads are the thread's own, and context transfers * are the open ones that target it. A thread without a completed compaction reads in full. */ @@ -3135,15 +3135,34 @@ export const layer: Layer.Layer = `; }); + // Mirrors JavaScript's `trim().toLowerCase()` of a message's text. + const trimmedLowerText = (payload: Statement.Fragment) => sql`lower(trim( + json_extract(${payload}, '$.text'), + char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, + 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) + ))`; + /** Null when nothing was compacted, so the whole history is current. */ const runsSinceLastCompaction = Effect.fn("ProjectionStore.runsSinceLastCompaction")(function* ( threadId: ThreadId, keepRunIds: ReadonlyArray, ) { + // Starts at the last prompt before the compaction: a model switch is measured from the + // last run whose prompt the timeline shows, and `/compact` runs are not shown. const [compaction] = yield* sql<{ readonly from_ordinal: number }>` SELECT COALESCE( - (SELECT MAX(previous.ordinal) FROM orchestration_v2_projection_runs AS previous - WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal), + (SELECT previous.ordinal FROM orchestration_v2_projection_runs AS previous + WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal + AND previous.status NOT IN ('queued', 'rolled_back') + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages AS prompt + WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') + AND prompt.role = 'user' + AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 + AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} + ) + ORDER BY previous.ordinal DESC + LIMIT 1), latest.ordinal ) AS from_ordinal FROM ( @@ -6298,12 +6317,6 @@ export const layer: Layer.Layer = ), ); - // Mirrors JavaScript's `trim().toLowerCase()` of a message's text. - const trimmedLowerText = (payload: Statement.Fragment) => sql`lower(trim( - json_extract(${payload}, '$.text'), - char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, - 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279) - ))`; // Mirrors isNativeMaintenanceCommand over the thread's user messages. const hasOnlyMaintenanceUserMessages: ProjectionStoreV2Shape["hasOnlyMaintenanceUserMessages"] = (threadId) => diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2b08c0b8345b6..87667ab602c2f 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1105,6 +1105,139 @@ it.effect("starts a steer that missed the turn on the saved next-turn selection" ), ); +it.effect("starts a steer of a finished run from before the last compaction as a new turn", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "steering-before-compaction", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* startFirstTurn; + const finish = (turn: (typeof started)[number]) => + Effect.gen(function* () { + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === turn.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const providerTurn = (yield* orchestrator.getThreadProjection( + threadId, + )).providerTurns.find((candidate) => candidate.runAttemptId === turn.attemptId)!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...providerTurn, + status: "completed", + completedAt: yield* DateTime.now, + }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + runOrdinal: turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + }); + const first = started[0]!; + yield* finish(first); + for (const id of ["second", "third"]) { + const running = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(`message:${id}`), + text: id, + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + yield* finish(started.at(-1)!); + } + // The window starts at the second run, the prompt before the compaction in the third. + const third = started[2]!; + const now = yield* DateTime.now; + yield* sink.write({ + events: [ + { + id: EventId.make("event:compaction"), + type: "turn-item.updated" as const, + threadId, + runId: third.runId, + nodeId: third.rootNodeId, + occurredAt: now, + payload: { + id: TurnItemId.make("item:compaction"), + threadId, + runId: third.runId, + nodeId: third.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1_000, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction" as const, + driver, + }, + }, + ], + }); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "late steer", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.equal(started.length, 4); + assert.equal(started[3]?.message.messageId, MessageId.make("message:steer")); + }).pipe(Effect.provide(layer)); + }), + ), +); + // regression: #26695 cancelling a Copilot prompt to steer it kills its background agents it.effect("queues a steer until the turn ends when the provider cannot steer it", () => Effect.scoped( From 79ed278fd131f12b1a42ca90db1df412b8bfe60c Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Sun, 8 Nov 2026 13:52:29 -0300 Subject: [PATCH 11/11] fix(server): finding a prompt before the compaction skips runs with unreadable JSON The search for the last prompt before the latest compaction read each earlier run's payload with `json_extract`, so walking past a legacy run whose stored payload is not valid JSON failed the whole read and blocked the next message. Guard it with `json_valid`, so such a run is never a window start. A run older than the window is not read at all, and the compacted-thread test now has a thread whose only earlier run is unreadable. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 68 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 4 +- 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 09e4d88b005e7..69f7a3b124d67 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -835,6 +835,74 @@ it.effect("a compacted thread reads its recent runs, live runs, fresh starts and // Run 8 is a `/compact` run, so the model-switch baseline is the prompt before it, run 7. yield* compact(9); assert.deepEqual(ids((yield* read([])).runs), runIdsOf([2, 3, 7, 8, 9])); + + // The search for a prompt before the compaction may walk past an unreadable run. + const legacyThreadId = yield* addRolledBackRecoveryCandidate("compacted-legacy"); + const legacyNodeId = NodeId.make("node:compacted-legacy"); + const compactionRunId = RunId.make("run:compacted-legacy:2"); + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES ('run:compacted-legacy:0', ${legacyThreadId}, 0, 'codex', 'completed', '', '{') + `; + yield* store.apply({ + id: EventId.make("event:compacted-legacy:run"), + type: "run.created", + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: compactionRunId, + threadId: legacyThreadId, + ordinal: 2, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:compacted-legacy:2"), + rootNodeId: legacyNodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + yield* store.apply({ + id: EventId.make("event:compacted-legacy:compaction"), + type: "turn-item.updated", + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("item:compacted-legacy:compaction"), + threadId: legacyThreadId, + runId: compactionRunId, + nodeId: legacyNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "compaction", + driver, + }, + }); + const legacy = yield* store.getThreadRecords(legacyThreadId, ["runs"], { + sinceLastCompaction: { keepRunIds: [] }, + }); + assert.deepEqual(ids(legacy.runs), [compactionRunId]); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index f808027a802c0..de6ec79f2abd0 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3154,13 +3154,13 @@ export const layer: Layer.Layer = (SELECT previous.ordinal FROM orchestration_v2_projection_runs AS previous WHERE previous.thread_id = ${threadId} AND previous.ordinal < latest.ordinal AND previous.status NOT IN ('queued', 'rolled_back') - AND NOT EXISTS ( + AND CASE WHEN json_valid(previous.payload_json) THEN NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_messages AS prompt WHERE prompt.message_id = json_extract(previous.payload_json, '$.userMessageId') AND prompt.role = 'user' AND coalesce(json_array_length(prompt.payload_json, '$.attachments'), 0) = 0 AND ${trimmedLowerText(sql.literal("prompt.payload_json"))} = ${IDLE_COMPACTION_TEXT} - ) + ) ELSE 0 END ORDER BY previous.ordinal DESC LIMIT 1), latest.ordinal