fix(server): turn starts in long threads no longer stall the server

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<string> = [];
+    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<string> = [];
+    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;

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -4368,7 +4368,6 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                     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

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);

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')
+  `;
+});

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:" }))),
   );