perf(server): reuse stored turn item positions

apps/server/src/orchestration-v2/TurnItemPositionStore.test.ts

@@ -0,0 +1,58 @@
+import { assert, it } from "@effect/vitest";
+import { RunId, ThreadId, TurnItemId } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
+
+const layerTest = TurnItemPositionStore.layer.pipe(
+  Layer.provideMerge(SqlitePersistence.layerMemory),
+);
+
+it.effect("reuses an item's position without attempting another allocation insert", () =>
+  Effect.gen(function* () {
+    const sql = yield* SqlClient.SqlClient;
+    const positions = yield* TurnItemPositionStore.TurnItemPositionStoreV2;
+    const input = {
+      threadId: ThreadId.make("thread:position-reuse"),
+      turnItemId: TurnItemId.make("item:streaming"),
+      runId: null,
+    };
+    const first = yield* positions.allocate(input);
+    assert.equal(first, 1);
+
+    // BEFORE INSERT runs even when ON CONFLICT would discard the allocation.
+    yield* sql`
+      CREATE TEMP TRIGGER reject_redundant_position_insert
+      BEFORE INSERT ON orchestration_v2_turn_item_positions
+      WHEN EXISTS (
+        SELECT 1 FROM orchestration_v2_turn_item_positions
+        WHERE thread_id = NEW.thread_id AND turn_item_id = NEW.turn_item_id
+      )
+      BEGIN
+        SELECT RAISE(ABORT, 'Existing item positions must be reused');
+      END
+    `;
+
+    const next = yield* positions.allocate({
+      ...input,
+      turnItemId: TurnItemId.make("item:next"),
+    });
+    assert.equal(next, first + 1);
+    assert.equal(yield* positions.allocate(input), first);
+    assert.equal(
+      yield* positions.allocate({
+        ...input,
+        runId: RunId.make("run:moved"),
+        runOrdinal: 2,
+      }),
+      first,
+    );
+    assert.equal(
+      yield* positions.allocate({ ...input, threadId: ThreadId.make("thread:other") }),
+      1,
+    );
+  }).pipe(Effect.provide(layerTest)),
+);

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

@@ -46,6 +46,14 @@ export const layer: Layer.Layer<TurnItemPositionStoreV2, never, SqlClient.SqlCli
       runOrdinal: suppliedRunOrdinal,
     }) =>
       Effect.gen(function* () {
+        const existing = yield* sql<{ readonly ordinal: number }>`
+          SELECT ordinal
+          FROM orchestration_v2_turn_item_positions
+          WHERE thread_id = ${threadId} AND turn_item_id = ${turnItemId}
+          LIMIT 1
+        `;
+        if (existing[0] !== undefined) return existing[0].ordinal;
+
         const runRows =
           runId === null
             ? []