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;