fix(server): Find frees searched thread text after one idle minute

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;

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));