From 5ce35b414d91fc1ecbdf9b50d0a6aed2bb7a4a2d Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Thu, 22 Oct 2026 18:59:18 +0000 Subject: [PATCH 1/2] fix(server): agent toolkit session reads no longer freeze the server On bkt3 (2026-10-22 18:37-18:41 UTC) the remaining event-loop stalls were agent MCP toolkit reads of one large session on the main SQLite connection: agentToolkit.goalOf -> readTimelineIndex (1.8-2.2 s) and agentToolkit.loadPendingDetail -> readCanonicalProjection (2.5 s), during agents' brief/inbox calls such as the Manager's 15-minute run. Both reads now run on the SQLite reader worker. Every thread read first ensures a migrated v1 transcript, which can write and fails on the reader, so a read that fails there runs again on the writer. Co-Authored-By: Claude Opus 5.5 --- .../server/src/mcp/toolkits/agent/handlers.ts | 3 +- .../mcp/toolkits/agent/offLoopReads.test.ts | 66 +++++++++++++++++++ .../src/mcp/toolkits/agent/offLoopReads.ts | 18 +++++ apps/server/src/mcp/toolkits/agent/pending.ts | 3 +- 4 files changed, 88 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts create mode 100644 apps/server/src/mcp/toolkits/agent/offLoopReads.ts diff --git a/apps/server/src/mcp/toolkits/agent/handlers.ts b/apps/server/src/mcp/toolkits/agent/handlers.ts index 90f867928b42f..b123ed40b71bc 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 { readOffEventLoop } from "./offLoopReads.ts"; import { EMPTY_PENDING, loadPendingDetail, @@ -973,7 +974,7 @@ const goalOf = Effect.fn("agentToolkit.goalOf")(function* (caller: Caller, shell 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); + .pipe(readOffEventLoop, 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..3776a18d8326e --- /dev/null +++ b/apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts @@ -0,0 +1,66 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +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 { readOffEventLoop } from "./offLoopReads.ts"; + +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), + ); + +it.effect("a toolkit read runs once and returns what it read", () => + withReaderClient((sql) => + Effect.gen(function* () { + let attempts = 0; + const read = yield* readOffEventLoop( + Effect.suspend(() => { + attempts += 1; + return sql.withTransaction(names(sql)); + }), + ); + + assert.deepEqual(read, ["first"]); + assert.equal(attempts, 1); + }), + ), +); + +it.effect("a toolkit read that has to write runs again where the write lands", () => + withReaderClient((sql) => + Effect.gen(function* () { + let attempts = 0; + // Stands in for the v1 transcript import that every thread read may run first. + yield* readOffEventLoop( + Effect.suspend(() => { + attempts += 1; + return sql.withTransaction(sql`INSERT INTO entries VALUES ('imported')`); + }), + ); + + assert.equal(attempts, 2); + assert.deepEqual(yield* names(sql), ["first", "imported"]); + }), + ), +); 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..bf09675691c92 --- /dev/null +++ b/apps/server/src/mcp/toolkits/agent/offLoopReads.ts @@ -0,0 +1,18 @@ +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; + +/** + * 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. A write fails on the reader, so a read that fails there + * runs again on the writer, where the import can happen. + */ +export const readOffEventLoop = (effect: Effect.Effect): Effect.Effect => + effect.pipe( + NodeSqliteClient.readOnly, + Effect.orElse(() => effect), + ); diff --git a/apps/server/src/mcp/toolkits/agent/pending.ts b/apps/server/src/mcp/toolkits/agent/pending.ts index 8f8d9aa703a93..37d17cde6e208 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 { readOffEventLoop } from "./offLoopReads.ts"; import { shortRef, toIso, type SessionShell } from "./view.ts"; export interface PendingQuestion { @@ -182,7 +183,7 @@ export const loadPendingDetail = Effect.fn("agentToolkit.loadPendingDetail")(fun .getThreadRecords(shell.id, ["runtimeRequests", "turnItems"], { turnItemTypes: [...types], }) - .pipe(Effect.option); + .pipe(readOffEventLoop, Effect.option); if (Option.isNone(records)) return EMPTY_PENDING; const { runtimeRequests, turnItems } = records.value; From 4ccdc5fb6be5cc513519e60afbb11f42a17aca60 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Thu, 22 Oct 2026 19:07:08 +0000 Subject: [PATCH 2/2] fix(server): import the v1 transcript on the writer before the off-loop read Review of 5ce35b414d: Effect.orElse does not exist in Effect v4, and a blanket retry on the writer brought the freeze back for any read error. The toolkit now runs caller.threads.ensureLegacyTranscript on the writer first, then the session read under NodeSqliteClient.readOnly with no retry. Tests cover the import-then-read order, that the read itself runs on the read-only reader, and that a failed import skips the read. Co-Authored-By: Claude Opus 5.5 --- .../server/src/mcp/toolkits/agent/handlers.ts | 10 ++- .../mcp/toolkits/agent/offLoopReads.test.ts | 76 ++++++++++++++----- .../src/mcp/toolkits/agent/offLoopReads.ts | 21 +++-- .../src/mcp/toolkits/agent/pending.test.ts | 1 + apps/server/src/mcp/toolkits/agent/pending.ts | 12 +-- 5 files changed, 85 insertions(+), 35 deletions(-) diff --git a/apps/server/src/mcp/toolkits/agent/handlers.ts b/apps/server/src/mcp/toolkits/agent/handlers.ts index b123ed40b71bc..50db55d489cf9 100644 --- a/apps/server/src/mcp/toolkits/agent/handlers.ts +++ b/apps/server/src/mcp/toolkits/agent/handlers.ts @@ -47,7 +47,7 @@ import { type VisibleSessions, } from "./context.ts"; import { modelSelectionFor, resolveModel } from "./models.ts"; -import { readOffEventLoop } from "./offLoopReads.ts"; +import { readSessionOffEventLoop } from "./offLoopReads.ts"; import { EMPTY_PENDING, loadPendingDetail, @@ -972,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(readOffEventLoop, 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 index 3776a18d8326e..25145b2ad0330 100644 --- a/apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts +++ b/apps/server/src/mcp/toolkits/agent/offLoopReads.test.ts @@ -1,12 +1,16 @@ 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 { readOffEventLoop } from "./offLoopReads.ts"; +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, @@ -30,37 +34,71 @@ const names = (sql: SqlClient.SqlClient) => rows.map((row) => row.name), ); -it.effect("a toolkit read runs once and returns what it read", () => - withReaderClient((sql) => - Effect.gen(function* () { - let attempts = 0; - const read = yield* readOffEventLoop( +/** 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(() => { - attempts += 1; - return sql.withTransaction(names(sql)); + 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(read, ["first"]); - assert.equal(attempts, 1); + assert.deepEqual(imports, [threadId]); + assert.deepEqual(read, ["first", "imported"]); }), ), ); -it.effect("a toolkit read that has to write runs again where the write lands", () => +it.effect("runs the read itself on the read-only reader", () => withReaderClient((sql) => Effect.gen(function* () { - let attempts = 0; - // Stands in for the v1 transcript import that every thread read may run first. - yield* readOffEventLoop( + 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(() => { - attempts += 1; - return sql.withTransaction(sql`INSERT INTO entries VALUES ('imported')`); + reads += 1; + return sql.withTransaction(names(sql)); }), - ); + ).pipe(Effect.flip); - assert.equal(attempts, 2); - assert.deepEqual(yield* names(sql), ["first", "imported"]); + 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 index bf09675691c92..21a721bb2bdfe 100644 --- a/apps/server/src/mcp/toolkits/agent/offLoopReads.ts +++ b/apps/server/src/mcp/toolkits/agent/offLoopReads.ts @@ -1,6 +1,9 @@ +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 @@ -8,11 +11,15 @@ import * as Effect from "effect/Effect"; * 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. A write fails on the reader, so a read that fails there - * runs again on the writer, where the import can happen. + * 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 readOffEventLoop = (effect: Effect.Effect): Effect.Effect => - effect.pipe( - NodeSqliteClient.readOnly, - Effect.orElse(() => effect), - ); +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 37d17cde6e208..70d92ab560a3c 100644 --- a/apps/server/src/mcp/toolkits/agent/pending.ts +++ b/apps/server/src/mcp/toolkits/agent/pending.ts @@ -27,7 +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 { readOffEventLoop } from "./offLoopReads.ts"; +import { readSessionOffEventLoop } from "./offLoopReads.ts"; import { shortRef, toIso, type SessionShell } from "./view.ts"; export interface PendingQuestion { @@ -179,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(readOffEventLoop, Effect.option); + }), + ).pipe(Effect.option); if (Option.isNone(records)) return EMPTY_PENDING; const { runtimeRequests, turnItems } = records.value;