diff --git a/apps/server/src/orchestration-v2/TurnItemPositionStore.test.ts b/apps/server/src/orchestration-v2/TurnItemPositionStore.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..579c82f03ee4029d06c05caec0175dc27761e00c --- /dev/null +++ b/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)), +); diff --git a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts index 016fe4f8401745a8a4c0f7106944d1547c5b0b32..9e60c0f0fdfcf40899959ce3dd18cee49df8dfb2 100644 --- a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts +++ b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts @@ -46,6 +46,14 @@ export const layer: Layer.Layer 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 ? []