From 62f9dc451a6b34854c41dafee2de2b1620b5dbd3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 10 Nov 2026 18:05:19 -0800 Subject: [PATCH 1/3] fix(server): Find frees searched thread text after one idle minute The parsed-text caches used a one-minute TTL, but Effect's Cache only expires an entry when that key is read again. The text of the last searched threads stayed in memory until the server restarted. Hold both caches in an RcRef with a one-minute idle time. They live while a search runs and are dropped once Find is idle. Entries are keyed by message content, so they never go stale and the per-entry TTL is gone. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/ThreadFind.test.ts | 24 +++++++++++++++++ .../server/src/orchestration-v2/ThreadFind.ts | 27 ++++++++++--------- 2 files changed, 39 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 6793421a6433..f73b9cae4898 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -1,4 +1,5 @@ import { assert, it, describe } from "@effect/vitest"; +import { searchableMessageSegments } from "@t3tools/shared/threadFindText"; import { EventId, MessageId, @@ -19,6 +20,8 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { vi } from "vite-plus/test"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -31,6 +34,8 @@ import { } from "./threadHistoryPaging.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; +vi.mock("@t3tools/shared/threadFindText", { spy: true }); + const layerTest = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe( Layer.provideMerge(SqlitePersistence.layerMemory), ); @@ -460,6 +465,25 @@ describe("V2 thread find", () => { }).pipe(Effect.provide(layerTest)), ); + it.effect("keeps parsed text while Find is in use and drops it after one idle minute", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([item("kept", 1, "idle needle")]); + const parses = vi.mocked(searchableMessageSegments).mock.calls; + const search = (query: string) => projection.searchThread({ threadId, query }); + yield* search("a"); + const first = parses.length; + yield* TestClock.adjust("40 seconds"); + yield* search("b"); + yield* TestClock.adjust("40 seconds"); + yield* search("c"); + assert.equal(parses.length, first); + yield* TestClock.adjust("1 minute"); + yield* search("d"); + assert.equal(parses.length, first + 1); + }).pipe(Effect.provide(layerTest)), + ); + it.effect("invalidates cached counts when streamed text changes or a run rolls back", () => Effect.gen(function* () { const projection = yield* setup; diff --git a/apps/server/src/orchestration-v2/ThreadFind.ts b/apps/server/src/orchestration-v2/ThreadFind.ts index ead68680a21c..431a27034250 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.ts @@ -12,6 +12,7 @@ import { countThreadSearchOccurrences } from "@t3tools/shared/threadSearch"; import * as Cache from "effect/Cache"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; +import * as RcRef from "effect/RcRef"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; @@ -232,17 +233,18 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( }); }); }); - // Scans visit every message; a cache smaller than the thread churns without hits. - const textCache = yield* Cache.make({ - capacity: 8192, - timeToLive: "1 minute", - lookup: (key: TextKey) => Effect.sync(() => parseText(key)), - }); - // Bounded separately because each entry can hold a large body; typing reuses the parse. - const largeTextCache = yield* Cache.make({ - capacity: 32, - timeToLive: "1 minute", - lookup: (key: TextKey) => Effect.sync(() => parseText(key)), + const makeTextCache = (capacity: number) => + Cache.make({ capacity, lookup: (key: TextKey) => Effect.sync(() => parseText(key)) }); + // Keys are message content, so entries never go stale. Cache TTL only expires on access and + // would keep the text of the last searched threads forever, so drop it all once Find is idle. + const textCaches = yield* RcRef.make({ + acquire: Effect.all({ + // Scans visit every message; a cache smaller than the thread churns without hits. + text: makeTextCache(8192), + // Bounded separately because each entry can hold a large body; typing reuses the parse. + largeText: makeTextCache(32), + }), + idleTimeToLive: "1 minute", }); const scans = new Map< string, @@ -330,6 +332,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( const data = yield* snapshot(input); if (data.result) return Stream.succeed(data.result); const { payloads, cwd, sequence, cacheKey, sourceThreadIds } = data; + const textCache = yield* RcRef.get(textCaches); const decode = (row: FindPayload) => decodeItem(row.payload).pipe( Effect.map((item) => ({ ...row, item })), @@ -357,7 +360,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( const decoded = yield* decode(row); const key = textKey(decoded.item, cwd, input.skills ?? []); const segments = yield* Cache.get( - key.text.length <= 32_768 ? textCache : largeTextCache, + key.text.length <= 32_768 ? textCache.text : textCache.largeText, key, ); const document = documentFor(decoded, countSegments(segments, input.query)); From 2a8b4c96e5ca60d9ee1f834f83dd595e7c000435 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 10 Nov 2026 18:36:22 -0800 Subject: [PATCH 2/3] fix(server): every Find search keeps its caches alive A repeated query is answered from the completed-scan map and returned before it borrowed the parsed-text caches, so stepping through matches for a minute dropped them. The completed-scan map was also never freed. Borrow the caches at the start of each search and move the completed-scan map into the same idle-released resource. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/ThreadFind.test.ts | 32 +++++++++++++------ .../server/src/orchestration-v2/ThreadFind.ts | 30 +++++++++-------- 2 files changed, 38 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index f73b9cae4898..a825cd109339 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -465,22 +465,34 @@ describe("V2 thread find", () => { }).pipe(Effect.provide(layerTest)), ); - it.effect("keeps parsed text while Find is in use and drops it after one idle minute", () => + it.effect("keeps its caches while Find is in use and drops them after one idle minute", () => Effect.gen(function* () { const projection = yield* setup; - yield* putItems([item("kept", 1, "idle needle")]); - const parses = vi.mocked(searchableMessageSegments).mock.calls; + yield* putItems([item("one", 1, "idle needle one"), item("two", 2, "idle needle two")]); + const parses = () => + vi + .mocked(searchableMessageSegments) + .mock.calls.filter(([message]) => message.text.startsWith("idle needle")).length; const search = (query: string) => projection.searchThread({ threadId, query }); - yield* search("a"); - const first = parses.length; + // A search still in progress holds the caches past the idle time. + yield* projection.searchThreadStream({ threadId, query: "needle" }).pipe( + Stream.tap((frame) => + frame.complete === false ? TestClock.adjust("1 minute") : Effect.void, + ), + Stream.runDrain, + ); + assert.equal(parses(), 2); + yield* TestClock.adjust("40 seconds"); + yield* search("idle"); yield* TestClock.adjust("40 seconds"); - yield* search("b"); + // A repeated query is answered from the completed scan and still counts as use. + yield* search("needle"); yield* TestClock.adjust("40 seconds"); - yield* search("c"); - assert.equal(parses.length, first); + yield* search("one"); + assert.equal(parses(), 2); yield* TestClock.adjust("1 minute"); - yield* search("d"); - assert.equal(parses.length, first + 1); + yield* search("needle"); + assert.equal(parses(), 4); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ThreadFind.ts b/apps/server/src/orchestration-v2/ThreadFind.ts index 431a27034250..c6a2c012f758 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.ts @@ -39,6 +39,14 @@ interface FindDocument extends FindRow { readonly runId: OrchestrationV2TurnItem["runId"]; readonly count: number; } +type CompletedScans = Map< + string, + { + sequence: number; + documents: readonly FindDocument[]; + sourceThreadIds: readonly ThreadId[]; + } +>; class TextKey extends Data.Class<{ text: string; role: "user" | "assistant" | "plan"; @@ -235,25 +243,19 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( }); const makeTextCache = (capacity: number) => Cache.make({ capacity, lookup: (key: TextKey) => Effect.sync(() => parseText(key)) }); - // Keys are message content, so entries never go stale. Cache TTL only expires on access and - // would keep the text of the last searched threads forever, so drop it all once Find is idle. - const textCaches = yield* RcRef.make({ + // Text keys are message content, so entries never go stale. Cache TTL only expires on access + // and would keep the last searched threads forever. Each search borrows this for its whole + // stream, and all of it is dropped once Find is idle. + const findCaches = yield* RcRef.make({ acquire: Effect.all({ // Scans visit every message; a cache smaller than the thread churns without hits. text: makeTextCache(8192), // Bounded separately because each entry can hold a large body; typing reuses the parse. largeText: makeTextCache(32), + scans: Effect.sync((): CompletedScans => new Map()), }), idleTimeToLive: "1 minute", }); - const scans = new Map< - string, - { - sequence: number; - documents: readonly FindDocument[]; - sourceThreadIds: readonly ThreadId[]; - } - >(); const readRevision = Effect.fn("ThreadFind.readRevision")(function* ( threadIds: readonly ThreadId[], ) { @@ -277,7 +279,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( return revision[0]?.sequence ?? 0; }); const snapshot = Effect.fn("ThreadFind.snapshot")( - function* (input: OrchestrationV2SearchThreadInput) { + function* (input: OrchestrationV2SearchThreadInput, scans: CompletedScans) { const { threadId } = input; const active = yield* sql<{ cwd: string | null }>` SELECT COALESCE(json_extract(t.payload_json, '$.worktreePath'), p.workspace_root) AS cwd @@ -328,11 +330,11 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( const searchThreadStream = (input: OrchestrationV2SearchThreadInput) => Stream.unwrap( Effect.gen(function* () { + const { scans, ...textCache } = yield* RcRef.get(findCaches); // Release SQLite before decoding and parsing. Both frames describe one snapshot. - const data = yield* snapshot(input); + const data = yield* snapshot(input, scans); if (data.result) return Stream.succeed(data.result); const { payloads, cwd, sequence, cacheKey, sourceThreadIds } = data; - const textCache = yield* RcRef.get(textCaches); const decode = (row: FindPayload) => decodeItem(row.payload).pipe( Effect.map((item) => ({ ...row, item })), From 559c34da73a18dff0972aac7f4b8b61ceae93d01 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 10 Nov 2026 18:53:12 -0800 Subject: [PATCH 3/3] test(server): Find test fails when a repeated query scans again Co-Authored-By: Claude Fable 5.2 --- apps/server/src/orchestration-v2/ThreadFind.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index a825cd109339..9366fdf193db 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -474,7 +474,7 @@ describe("V2 thread find", () => { .mocked(searchableMessageSegments) .mock.calls.filter(([message]) => message.text.startsWith("idle needle")).length; const search = (query: string) => projection.searchThread({ threadId, query }); - // A search still in progress holds the caches past the idle time. + // One idle minute passes inside this search. The later counts fail if it let the caches go. yield* projection.searchThreadStream({ threadId, query: "needle" }).pipe( Stream.tap((frame) => frame.complete === false ? TestClock.adjust("1 minute") : Effect.void, @@ -486,7 +486,10 @@ describe("V2 thread find", () => { yield* search("idle"); yield* TestClock.adjust("40 seconds"); // A repeated query is answered from the completed scan and still counts as use. - yield* search("needle"); + const repeated = yield* projection + .searchThreadStream({ threadId, query: "needle" }) + .pipe(Stream.runCollect); + assert.equal(repeated.length, 1); yield* TestClock.adjust("40 seconds"); yield* search("one"); assert.equal(parses(), 2);