diff --git a/apps/server/src/mcp/toolkits/agent/handlers.ts b/apps/server/src/mcp/toolkits/agent/handlers.ts
index 90f867928b42f..50db55d489cf9 100644
--- a/apps/server/src/mcp/toolkits/agent/handlers.ts
+++ b/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;
});
diff --git a/apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts b/apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts
new file mode 100644
index 0000000000000..25145b2ad0330
--- /dev/null
+++ b/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 = (
+ use: (sql: SqlClient.SqlClient) => Effect.Effect,
+) =>
+ 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) =>
+ ({
+ 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 = [];
+ 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);
+ }),
+ ),
+);
diff --git a/apps/server/src/mcp/toolkits/agent/offLoopReads.ts b/apps/server/src/mcp/toolkits/agent/offLoopReads.ts
new file mode 100644
index 0000000000000..21a721bb2bdfe
--- /dev/null
+++ b/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 = (
+ caller: Caller,
+ threadId: ThreadId,
+ read: Effect.Effect,
+) =>
+ caller.threads
+ .ensureLegacyTranscript(threadId)
+ .pipe(Effect.andThen(read.pipe(NodeSqliteClient.readOnly)));
diff --git a/apps/server/src/mcp/toolkits/agent/pending.test.ts b/apps/server/src/mcp/toolkits/agent/pending.test.ts
index 6763db5d10261..96547474aff5f 100644
--- a/apps/server/src/mcp/toolkits/agent/pending.test.ts
+++ b/apps/server/src/mcp/toolkits/agent/pending.test.ts
@@ -62,6 +62,7 @@ const callerWith = (records: {
const reads: Array = [];
const caller = {
threads: {
+ ensureLegacyTranscript: () => Effect.void,
getThreadRecords: (_threadId: unknown, _fields: unknown, filter: unknown) => {
reads.push(filter);
return Effect.succeed({ runs: [], ...records });
diff --git a/apps/server/src/mcp/toolkits/agent/pending.ts b/apps/server/src/mcp/toolkits/agent/pending.ts
index 8f8d9aa703a93..70d92ab560a3c 100644
--- a/apps/server/src/mcp/toolkits/agent/pending.ts
+++ b/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;