diff --git a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.expbkt3.test.ts b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.expbkt3.test.ts new file mode 100644 index 0000000000000..58db3067a5123 --- /dev/null +++ b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.expbkt3.test.ts @@ -0,0 +1,130 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { EventId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import type * as Tracer from "effect/Tracer"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; + +import { layer as OrchestrationEventStoreLive } from "../../persistence/OrchestrationEventStore.ts"; +import * as SqlitePersistence from "../../persistence/Sqlite.ts"; +import { ProjectService } from "../../project/ProjectService.ts"; +import * as EventSink from "../EventSink.ts"; +import * as EventStore from "../EventStore.ts"; +import * as LegacyV1ThreadImporter from "../legacy/LegacyV1ThreadImporter.ts"; +import * as ProjectionStore from "../ProjectionStore.ts"; +import * as ProjectStore from "../ProjectStore.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { ThreadSearch } from "../ThreadSearch.ts"; +import { ProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; + +// t3_get_session reads a whole session through getThreadDetailSnapshot. That read +// takes seconds on a large session, so it must not hold the writer or the event loop. + +const threadId = ThreadId.make("thread:legacy-detail-snapshot"); + +/** The connection a statement ran on: its own span, or the transaction span above it. */ +const connectionOf = (span: Tracer.Span): unknown => { + let current: Tracer.AnySpan | undefined = span; + while (current !== undefined) { + if (current._tag === "Span") { + const connection = current.attributes.get("db.connection"); + if (connection !== undefined) return connection; + } + current = current._tag === "Span" ? Option.getOrUndefined(current.parent) : undefined; + } + return undefined; +}; + +const withSnapshotQuery = ( + use: Effect.Effect< + A, + E, + ProjectionSnapshotQuery | LegacyV1ThreadImporter.LegacyV1ThreadImporter | SqlClient.SqlClient + >, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-detail-snapshot-" }); + const stores = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe( + Layer.provideMerge(SqlitePersistence.layerFromPath(path.join(directory, "state.sqlite"))), + ); + const importer = LegacyV1ThreadImporter.layer.pipe( + Layer.provideMerge(EventSink.layer.pipe(Layer.provideMerge(stores))), + ); + const query = ProjectionSnapshotQueryLive.pipe( + Layer.provide( + Layer.mergeAll( + OrchestrationEventStoreLive, + Layer.mock(ProjectService)({}), + Layer.mock(ThreadSearch)({}), + ), + ), + Layer.provideMerge(importer), + ); + return yield* use.pipe(Effect.provide(query)); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +it.effect("imports a v1 transcript on the writer, then reads the session on the reader", () => + withSnapshotQuery( + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; + const snapshots = yield* ProjectionSnapshotQuery; + yield* sql`INSERT INTO projection_projects (project_id,title,workspace_root,scripts_json,created_at,updated_at,deleted_at) VALUES ('project:detail-snapshot','Detail snapshot','/tmp/detail-snapshot','[]','2026-01-01T00:00:00Z','2026-01-02T00:00:00Z',NULL)`; + yield* sql`INSERT INTO projection_threads (thread_id,project_id,title,model_selection_json,runtime_mode,interaction_mode,created_at,updated_at,deleted_at) VALUES (${threadId},'project:detail-snapshot','Legacy session','{"instanceId":"codex","model":"gpt-5.4"}','full-access','default','2026-01-01T00:00:00Z','2026-01-02T00:00:00Z',NULL)`; + yield* sql`INSERT INTO projection_thread_activities (activity_id,thread_id,tone,kind,summary,payload_json,sequence,created_at) VALUES ('activity:legacy',${threadId},'info','context-window.updated','Retained context','{"currentTokens":21}',7,'2026-01-01T01:00:00Z')`; + yield* legacyImporter.reconcileShells; + + // Read each span's connection after the call: a statement outside a + // transaction is tagged when it takes its connection, after the transformer. + const recordStatements = (effect: Effect.Effect) => { + const spans: Array<{ readonly query: string; readonly span: Tracer.Span }> = []; + const record: Statement.Transformer = (statement, _sql, _fiber, span) => { + spans.push({ query: statement.compile()[0], span }); + return Effect.succeed(statement); + }; + return effect.pipe( + Effect.provideService(Statement.CurrentTransformer, record), + Effect.provideService(Statement.SpanPropagationEnabled, true), + Effect.map((value) => ({ + value, + statements: spans.map(({ query, span }) => ({ query, connection: connectionOf(span) })), + })), + ); + }; + + // The first read imports the v1 transcript, which writes. On the reader that + // write would fail with "attempt to write a readonly database". + const first = yield* recordStatements(snapshots.getThreadDetailSnapshot(threadId)); + assert.isTrue(Option.isSome(first.value)); + const writes = first.statements.filter(({ query }) => /\b(INSERT|UPDATE)\b/i.test(query)); + assert.isAbove(writes.length, 0); + assert.deepEqual([...new Set(writes.map(({ connection }) => connection))], ["writer"]); + + // The transcript is in place now, so the import check costs nothing and every + // statement of the session read runs on the reader. + const second = yield* recordStatements(snapshots.getThreadDetailSnapshot(threadId)); + assert.isTrue(Option.isSome(second.value)); + if (Option.isNone(second.value)) return; + assert.include( + second.value.value.thread.activities.map((activity) => activity.id), + EventId.make("activity:legacy"), + ); + assert.isTrue( + second.statements.some(({ query }) => + /FROM\s+orchestration_v2_projection_turn_items\b/i.test(query), + ), + ); + assert.deepEqual( + [...new Set(second.statements.map(({ connection }) => connection))], + ["reader"], + ); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts index 044f25e2fce00..0e3448297bcee 100644 --- a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts @@ -303,16 +303,21 @@ const make = Effect.gen(function* () { wrap( Effect.gen(function* () { yield* importer.ensureTranscript(threadId); - return yield* sql.withTransaction( - Effect.gen(function* () { - const thread = yield* detail(threadId, false); - if (Option.isNone(thread)) return Option.none(); - return Option.some({ - snapshotSequence: yield* events.latestApplicationSequence, - thread: thread.value, - }); - }), - ); + // The whole-session read takes seconds on a large session: run it on the + // SQLite reader worker, off the event loop. The transcript import above can + // write, so it stays on the writer, and `detail(threadId, false)` skips it. + return yield* sql + .withTransaction( + Effect.gen(function* () { + const thread = yield* detail(threadId, false); + if (Option.isNone(thread)) return Option.none(); + return Option.some({ + snapshotSequence: yield* events.latestApplicationSequence, + thread: thread.value, + }); + }), + ) + .pipe(NodeSqliteClient.readOnly); }), ), getThreadRuntimeContext: (threadId) =>