diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce388..929015d3330b6 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Tracer from "effect/Tracer"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -369,6 +370,44 @@ it.effect("selects unfinished recovery work without reading settled thread histo }).pipe(Effect.provide(layerTest)), ); +it.effect("looks up a thread's unfinished turn items without reading its finished history", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = yield* createThread("unfinished-item-lookups"); + yield* createRun(threadId, "running"); + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + yield* projections.getRuntimeRecoveryProjection(threadId).pipe(Effect.withTracer(tracer)); + + const itemLookups: Array = []; + for (const statement of statements) { + const plan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + ); + itemLookups.push( + ...plan.map((row) => row.detail).filter((detail) => /^(?:SEARCH|SCAN) item\b/.test(detail)), + ); + } + // Partial indexes hold only unfinished items, so a long thread's history stays unread. + assert.equal(itemLookups.length, 7); + for (const lookup of itemLookups) { + assert.match(lookup, /turn_items_(?:active|recovery)_idx\b/); + } + }).pipe(Effect.provide(layerTest)), +); + it.effect("recovers terminal subagent results until their cross-thread transfer exists", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 92ef73ebf1ba4..709e34a4e5965 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4368,7 +4368,6 @@ export const layer: Layer.Layer = WHERE item.thread_id = ${threadId} AND item.type IN ('command_execution', 'dynamic_tool', 'subagent') AND item.status IN ('pending', 'running', 'waiting') - AND item.run_id IS NOT NULL ) ) ORDER BY run.ordinal ASC diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5b8ca15241ba4..058b935c8fdbf 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -71,6 +71,7 @@ import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts"; import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts"; import Migration0067 from "./Migrations/067_WebhookDedupe.ts"; import Migration0068 from "./Migrations/068_ShellStopRequestIndex.ts"; +import Migration0069 from "./Migrations/069_TurnItemsActiveStatusIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -142,6 +143,7 @@ const migrationEntries = [ [66, "RepairClientSettlementTimestamps", Migration0066], [67, "WebhookDedupe", Migration0067], [68, "ShellStopRequestIndex", Migration0068], + [69, "TurnItemsActiveStatusIndex", Migration0069], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/069_TurnItemsActiveStatusIndex.ts b/apps/server/src/persistence/Migrations/069_TurnItemsActiveStatusIndex.ts new file mode 100644 index 0000000000000..1001100e8d793 --- /dev/null +++ b/apps/server/src/persistence/Migrations/069_TurnItemsActiveStatusIndex.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; + + // ProjectionStore's runtime recovery reads look up a thread's unfinished + // items of any type; without it they read the thread's whole history. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_active_idx + ON orchestration_v2_projection_turn_items(thread_id, ordinal, turn_item_id) + WHERE status IN ('pending', 'running', 'waiting') + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 64f1d9e3ae2f6..f994552535034 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -48,6 +48,7 @@ describe("V2 preview upgrade", () => { [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], [68, "ShellStopRequestIndex"], + [69, "TurnItemsActiveStatusIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -139,6 +140,7 @@ describe("V2 preview upgrade", () => { [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], [68, "ShellStopRequestIndex"], + [69, "TurnItemsActiveStatusIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );