diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 6793421a6433..9366fdf193db 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,40 @@ describe("V2 thread find", () => { }).pipe(Effect.provide(layerTest)), ); + 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("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 }); + // 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, + ), + Stream.runDrain, + ); + assert.equal(parses(), 2); + yield* TestClock.adjust("40 seconds"); + yield* search("idle"); + yield* TestClock.adjust("40 seconds"); + // A repeated query is answered from the completed scan and still counts as use. + 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); + yield* TestClock.adjust("1 minute"); + yield* search("needle"); + assert.equal(parses(), 4); + }).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..c6a2c012f758 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"; @@ -38,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"; @@ -232,26 +241,21 @@ 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)) }); + // 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[], ) { @@ -275,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 @@ -326,8 +330,9 @@ 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 decode = (row: FindPayload) => @@ -357,7 +362,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));