fix(server): agent toolkit session reads no longer freeze the server

apps/server/src/mcp/toolkits/agent/handlers.ts

@@ -47,6 +47,7 @@ import {
   type VisibleSessions,
 } from "./context.ts";
 import { modelSelectionFor, resolveModel } from "./models.ts";
+import { readSessionOffEventLoop } from "./offLoopReads.ts";
 import {
   EMPTY_PENDING,
   loadPendingDetail,
@@ -971,9 +972,11 @@ const correctPlanFlags = Effect.fn("agentToolkit.correctPlanFlags")(function* (
 /** The session's goal: the provider's own, else the first message. */
 const goalOf = Effect.fn("agentToolkit.goalOf")(function* (caller: Caller, shell: SessionShell) {
   if (shell.goal?.objective !== undefined) return shell.goal.objective;
-  const page = yield* caller.threads
-    .getTimelinePage(shell.id, { afterPosition: -1, limit: 1, view: "messages" })
-    .pipe(Effect.option);
+  const page = yield* readSessionOffEventLoop(
+    caller,
+    shell.id,
+    caller.threads.getTimelinePage(shell.id, { afterPosition: -1, limit: 1, view: "messages" }),
+  ).pipe(Effect.option);
   const first = Option.isSome(page) ? page.value.items[0]?.item : undefined;
   return first?.type === "user_message" ? first.text : null;
 });

apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts

@@ -0,0 +1,104 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import { ThreadId } from "@t3tools/contracts";
+import * as SqliteClient from "@t3tools/shared/nodeSqliteClient";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Path from "effect/Path";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import type { Caller } from "../../threadAccess.ts";
+import { readSessionOffEventLoop } from "./offLoopReads.ts";
+
+const threadId = ThreadId.make("toolkit-read-thread");
+
+const withReaderClient = <A, E>(
+  use: (sql: SqlClient.SqlClient) => Effect.Effect<A, E, SqlClient.SqlClient>,
+) =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-toolkit-reads-" });
+    const filename = path.join(directory, "state.sqlite");
+    return yield* Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      yield* sql`PRAGMA journal_mode = WAL`;
+      yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
+      yield* sql`INSERT INTO entries VALUES ('first')`;
+      return yield* use(sql);
+    }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
+  }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
+
+const names = (sql: SqlClient.SqlClient) =>
+  Effect.map(sql<{ name: string }>`SELECT name FROM entries ORDER BY rowid`, (rows) =>
+    rows.map((row) => row.name),
+  );
+
+/** A caller whose transcript import writes a row, as a v1 import would. */
+const importingCaller = (sql: SqlClient.SqlClient, imports: Array<ThreadId>) =>
+  ({
+    threads: {
+      ensureLegacyTranscript: (id: ThreadId) =>
+        Effect.suspend(() => {
+          imports.push(id);
+          return sql`INSERT INTO entries VALUES ('imported')`.pipe(Effect.asVoid, Effect.orDie);
+        }),
+    },
+  }) as unknown as Caller;
+
+it.effect("imports the transcript on the writer, then reads it", () =>
+  withReaderClient((sql) =>
+    Effect.gen(function* () {
+      const imports: Array<ThreadId> = [];
+      const read = yield* readSessionOffEventLoop(
+        importingCaller(sql, imports),
+        threadId,
+        sql.withTransaction(names(sql)),
+      );
+
+      assert.deepEqual(imports, [threadId]);
+      assert.deepEqual(read, ["first", "imported"]);
+    }),
+  ),
+);
+
+it.effect("runs the read itself on the read-only reader", () =>
+  withReaderClient((sql) =>
+    Effect.gen(function* () {
+      const error = yield* readSessionOffEventLoop(
+        importingCaller(sql, []),
+        threadId,
+        sql.withTransaction(sql`INSERT INTO entries VALUES ('blocked')`),
+      ).pipe(Effect.flip);
+
+      assert.equal(error._tag, "SqlError");
+      assert.include(
+        String(error._tag === "SqlError" ? error.reason.cause : error),
+        "readonly database",
+      );
+      assert.deepEqual(yield* names(sql), ["first", "imported"]);
+    }),
+  ),
+);
+
+it.effect("does not read when the transcript import fails", () =>
+  withReaderClient((sql) =>
+    Effect.gen(function* () {
+      let reads = 0;
+      const caller = {
+        threads: { ensureLegacyTranscript: () => Effect.fail("import failed" as const) },
+      } as unknown as Caller;
+      const error = yield* readSessionOffEventLoop(
+        caller,
+        threadId,
+        Effect.suspend(() => {
+          reads += 1;
+          return sql.withTransaction(names(sql));
+        }),
+      ).pipe(Effect.flip);
+
+      assert.equal(error as unknown, "import failed");
+      assert.equal(reads, 0);
+    }),
+  ),
+);

apps/server/src/mcp/toolkits/agent/offLoopReads.ts

@@ -0,0 +1,25 @@
+import type { ThreadId } from "@t3tools/contracts";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
+import * as Effect from "effect/Effect";
+
+import type { Caller } from "../../threadAccess.ts";
+
+/**
+ * Runs a toolkit read of one session on the SQLite reader worker, off the
+ * event loop. Agents read sessions through these tools all day (a brief, the
+ * inbox); on bkt3 a cold read of one large session took 2–2.5 s, and on the
+ * event loop it froze the whole server for that long.
+ *
+ * Every thread read first makes sure a migrated v1 transcript is imported,
+ * which can write, and a write fails on the reader. So that check runs on the
+ * writer first (one indexed read, none once the thread is known), and the read
+ * then finds the transcript in place.
+ */
+export const readSessionOffEventLoop = <A, E, R>(
+  caller: Caller,
+  threadId: ThreadId,
+  read: Effect.Effect<A, E, R>,
+) =>
+  caller.threads
+    .ensureLegacyTranscript(threadId)
+    .pipe(Effect.andThen(read.pipe(NodeSqliteClient.readOnly)));

apps/server/src/mcp/toolkits/agent/pending.test.ts

@@ -62,6 +62,7 @@ const callerWith = (records: {
   const reads: Array<unknown> = [];
   const caller = {
     threads: {
+      ensureLegacyTranscript: () => Effect.void,
       getThreadRecords: (_threadId: unknown, _fields: unknown, filter: unknown) => {
         reads.push(filter);
         return Effect.succeed({ runs: [], ...records });

apps/server/src/mcp/toolkits/agent/pending.ts

@@ -27,6 +27,7 @@ import { findForkAsyncRequest } from "../../../orchestration-v2/forkAsyncRespons
 import { runEndedWithoutAnswer } from "../../../orchestration-v2/orphanedRuntimeRequest.expbkt3.ts";
 import type { Caller } from "../../threadAccess.ts";
 import type { QuestionView } from "./answers.ts";
+import { readSessionOffEventLoop } from "./offLoopReads.ts";
 import { shortRef, toIso, type SessionShell } from "./view.ts";
 
 export interface PendingQuestion {
@@ -178,11 +179,13 @@ export const loadPendingDetail = Effect.fn("agentToolkit.loadPendingDetail")(fun
   if (options.includePlan && shell.hasActionableProposedPlan) types.add("proposed_plan");
   if (types.size === 0) return EMPTY_PENDING;
 
-  const records = yield* caller.threads
-    .getThreadRecords(shell.id, ["runtimeRequests", "turnItems"], {
+  const records = yield* readSessionOffEventLoop(
+    caller,
+    shell.id,
+    caller.threads.getThreadRecords(shell.id, ["runtimeRequests", "turnItems"], {
       turnItemTypes: [...types],
-    })
-    .pipe(Effect.option);
+    }),
+  ).pipe(Effect.option);
   if (Option.isNone(records)) return EMPTY_PENDING;
   const { runtimeRequests, turnItems } = records.value;