perf(server): phone reconnects no longer stall on counting every turn item

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

@@ -5988,27 +5988,22 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
 
     const selectShellRunItemCounts = (threadIds: ReadonlyArray<ThreadId>) =>
       sql<ShellRunItemCountRow>`
-        SELECT thread_id, run_id, COUNT(*) AS item_count
-        FROM orchestration_v2_projection_turn_items
-        WHERE run_id IS NOT NULL AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))
-        GROUP BY thread_id, run_id
+        SELECT thread_id, run_key AS run_id, item_count
+        FROM orchestration_v2_projection_turn_item_counts
+        WHERE run_key <> '' AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))
       `;
 
-    // Count all selected threads in one index walk; fork sources still need per-run counts.
+    // Fork sources still need per-run counts.
     const selectShellItemCounts = (threadIds: ReadonlyArray<ThreadId>) =>
       sql<ShellItemCountRow>`
         SELECT counts.thread_id,
-          SUM(CASE WHEN counts.run_id IS NULL OR r.status <> 'rolled_back'
+          SUM(CASE WHEN counts.run_key = '' OR r.status <> 'rolled_back'
             THEN counts.item_count ELSE 0 END) AS item_count,
-          SUM(CASE WHEN counts.run_id IS NULL THEN counts.item_count ELSE 0 END)
+          SUM(CASE WHEN counts.run_key = '' THEN counts.item_count ELSE 0 END)
             AS runless_item_count
-        FROM (
-          SELECT thread_id, run_id, COUNT(*) AS item_count
-          FROM orchestration_v2_projection_turn_items
-          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
+        FROM orchestration_v2_projection_turn_item_counts counts
+        LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_key
+        WHERE counts.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))
         GROUP BY counts.thread_id
       `;
 

apps/server/src/persistence/Migrations.ts

@@ -76,6 +76,7 @@ import Migration0070 from "./Migrations/070_PeerLinks.ts";
 import Migration0071 from "./Migrations/071_PeerLinkFence.ts";
 import Migration0072 from "./Migrations/072_WaitingLinkRequestIndex.ts";
 import Migration0073 from "./Migrations/073_ActiveTurnItemIndexes.ts";
+import Migration0074 from "./Migrations/074_TurnItemCounts.ts";
 
 const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -152,6 +153,7 @@ const migrationEntries = [
   [71, "PeerLinkFence", Migration0071],
   [72, "WaitingLinkRequestIndex", Migration0072],
   [73, "ActiveTurnItemIndexes", Migration0073],
+  [74, "TurnItemCounts", Migration0074],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/074_TurnItemCounts.test.ts

@@ -0,0 +1,60 @@
+import { assert, it } from "@effect/vitest";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import { runMigrations } from "../Migrations.ts";
+
+const layer = it.layer(NodeSqliteClient.layer({ filename: ":memory:" }));
+
+layer("074_TurnItemCounts", (it) => {
+  it.effect("backfills item counts, then follows upserts, moves, and deletes", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      yield* runMigrations({ toMigrationInclusive: 73 });
+      const upsert = (id: string, threadId: string, runId: string | null) => sql`
+        INSERT INTO orchestration_v2_projection_turn_items (
+          turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json
+        )
+        VALUES (
+          ${id}, ${threadId}, ${runId}, 1, 'assistant_message', 'completed',
+          '2026-10-29T00:00:00.000Z', '{}'
+        )
+        ON CONFLICT(turn_item_id) DO UPDATE SET
+          thread_id = excluded.thread_id,
+          run_id = excluded.run_id,
+          status = excluded.status
+      `;
+      yield* upsert("runless-1", "thread-1", null);
+      yield* upsert("runless-2", "thread-1", null);
+      yield* upsert("run-1", "thread-1", "run-1");
+
+      yield* runMigrations();
+
+      const counts = () =>
+        sql<{ readonly thread_id: string; readonly run_key: string; readonly item_count: number }>`
+          SELECT thread_id, run_key, item_count
+          FROM orchestration_v2_projection_turn_item_counts
+          ORDER BY thread_id, run_key
+        `;
+      assert.deepEqual(yield* counts(), [
+        { thread_id: "thread-1", run_key: "", item_count: 2 },
+        { thread_id: "thread-1", run_key: "run-1", item_count: 1 },
+      ]);
+
+      yield* upsert("run-1", "thread-1", "run-1");
+      yield* upsert("run-2", "thread-2", "run-2");
+      yield* upsert("runless-2", "thread-1", "run-1");
+      assert.deepEqual(yield* counts(), [
+        { thread_id: "thread-1", run_key: "", item_count: 1 },
+        { thread_id: "thread-1", run_key: "run-1", item_count: 2 },
+        { thread_id: "thread-2", run_key: "run-2", item_count: 1 },
+      ]);
+
+      yield* sql`DELETE FROM orchestration_v2_projection_turn_items WHERE thread_id = 'thread-1'`;
+      assert.deepEqual(yield* counts(), [
+        { thread_id: "thread-2", run_key: "run-2", item_count: 1 },
+      ]);
+    }),
+  );
+});

apps/server/src/persistence/Migrations/074_TurnItemCounts.ts

@@ -0,0 +1,61 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+/**
+ * Shell snapshots read item counts from here instead of walking every turn item, which on a
+ * large history takes seconds once the index is out of the page cache. Triggers keep it exact
+ * for every writer, including projection rebuilds and dev-database pruning.
+ */
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  // run_key is '' for items without a run; NULLs never conflict in a primary key.
+  yield* sql`
+    CREATE TABLE orchestration_v2_projection_turn_item_counts (
+      thread_id TEXT NOT NULL,
+      run_key TEXT NOT NULL,
+      item_count INTEGER NOT NULL,
+      PRIMARY KEY (thread_id, run_key)
+    ) WITHOUT ROWID
+  `;
+  yield* sql`
+    INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count)
+    SELECT thread_id, COALESCE(run_id, ''), COUNT(*)
+    FROM orchestration_v2_projection_turn_items
+    GROUP BY thread_id, run_id
+  `;
+  // An upsert that hits an existing item fires only the update trigger.
+  yield* sql`
+    CREATE TRIGGER orchestration_v2_turn_item_counts_insert
+    AFTER INSERT ON orchestration_v2_projection_turn_items
+    BEGIN
+      INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count)
+      VALUES (new.thread_id, COALESCE(new.run_id, ''), 1)
+      ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1;
+    END
+  `;
+  yield* sql`
+    CREATE TRIGGER orchestration_v2_turn_item_counts_update
+    AFTER UPDATE OF thread_id, run_id ON orchestration_v2_projection_turn_items
+    WHEN old.thread_id IS NOT new.thread_id OR old.run_id IS NOT new.run_id
+    BEGIN
+      UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1
+      WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '');
+      DELETE FROM orchestration_v2_projection_turn_item_counts
+      WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0;
+      INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count)
+      VALUES (new.thread_id, COALESCE(new.run_id, ''), 1)
+      ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1;
+    END
+  `;
+  yield* sql`
+    CREATE TRIGGER orchestration_v2_turn_item_counts_delete
+    AFTER DELETE ON orchestration_v2_projection_turn_items
+    BEGIN
+      UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1
+      WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '');
+      DELETE FROM orchestration_v2_projection_turn_item_counts
+      WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0;
+    END
+  `;
+});

apps/server/src/persistence/reconcileV2PreviewMigration.test.ts

@@ -53,6 +53,7 @@ describe("V2 preview upgrade", () => {
         [71, "PeerLinkFence"],
         [72, "WaitingLinkRequestIndex"],
         [73, "ActiveTurnItemIndexes"],
+        [74, "TurnItemCounts"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
       assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
@@ -149,6 +150,7 @@ describe("V2 preview upgrade", () => {
         [71, "PeerLinkFence"],
         [72, "WaitingLinkRequestIndex"],
         [73, "ActiveTurnItemIndexes"],
+        [74, "TurnItemCounts"],
       ]);
     }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
   );

packages/shared/src/nodeSqliteClient.test.ts

@@ -259,6 +259,24 @@ it.effect("serves another read while a read-only transaction is open", () =>
   ),
 );
 
+it.effect("serves a read that starts with a read-only transaction before any reader is up", () =>
+  withReaderClient((sql) =>
+    Effect.gen(function* () {
+      const release = yield* Deferred.make<void>();
+      const holder = sql
+        .withTransaction(countEntries(sql).pipe(Effect.andThen(Deferred.await(release))))
+        .pipe(SqliteClient.readOnly);
+      const read = countEntries(sql).pipe(
+        SqliteClient.readOnly,
+        Effect.tap(() => Deferred.succeed(release, undefined)),
+      );
+
+      const [, count] = yield* Effect.all([holder, read], { concurrency: "unbounded" });
+      assert.equal(count, 1);
+    }),
+  ),
+);
+
 // regression: #19226 read-only transactions run on the separate reader connection
 it.effect("rejects a write in a read-only transaction", () =>
   withReaderClient((sql) =>