1import {
2 OrchestrationV2TurnItemJson,
3 type OrchestrationV2ProjectedTurnItem,
4 OrchestrationV2SearchThreadInput,
5 type OrchestrationV2SearchThreadResult,
6 type OrchestrationV2TurnItem,
7 ThreadId,
8 TurnItemId,
9} from "@t3tools/contracts";
10import { searchableMessageSegments, searchablePlanSegments } from "@t3tools/shared/threadFindText";
11import { countThreadSearchOccurrences } from "@t3tools/shared/threadSearch";
12import * as Cache from "effect/Cache";
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Schema from "effect/Schema";
16import * as Stream from "effect/Stream";
17import * as SqlClient from "effect/sql/SqlClient";
18import {
19 ProjectionStoreReadError,
20 ProjectionStoreThreadNotFoundError,
21 type ProjectionStoreV2Error,
22} from "./ProjectionStore.ts";
23
24interface FindRow {
25 readonly position: number;
26 readonly visibility: OrchestrationV2ProjectedTurnItem["visibility"];
27 readonly sourceThreadId: ThreadId;
28 readonly sourceItemId: TurnItemId;
29}
30interface FindPayload extends FindRow {
31 readonly payload: string;
32 readonly entryId: string;
33 readonly type: OrchestrationV2TurnItem["type"];
34}
35interface FindDocument extends FindRow {
36 readonly entryId: string;
37 readonly runId: OrchestrationV2TurnItem["runId"];
38 readonly count: number;
39}
40class TextKey extends Data.Class<{
41 text: string;
42 role: "user" | "assistant" | "plan";
43 streaming: boolean;
44 hasContext: boolean;
45 cwd: string | undefined;
46 skills: NonNullable<OrchestrationV2SearchThreadInput["skills"]>;
47}> {}
48
49function textKey(
50 item: OrchestrationV2TurnItem,
51 cwd: string | undefined,
52 skills: TextKey["skills"],
53) {
54 return new TextKey({
55 text:
56 item.type === "proposed_plan"
57 ? item.markdown
58 : item.type === "user_message" || item.type === "assistant_message"
59 ? item.text
60 : "",
61 role:
62 item.type === "proposed_plan" ? "plan" : item.type === "user_message" ? "user" : "assistant",
63 streaming: item.type === "assistant_message" && item.streaming,
64 hasContext: item.type === "user_message" && item.context !== undefined,
65 cwd,
66 skills,
67 });
68}
69
70function parseText(key: TextKey): readonly string[] {
71 return key.role === "plan"
72 ? searchablePlanSegments(key.text, key.cwd)
73 : (searchableMessageSegments(
74 {
75 role: key.role,
76 text: key.text,
77 streaming: key.streaming,
78 context: key.hasContext ? { version: 1, records: [] } : undefined,
79 },
80 key.cwd,
81 key.skills,
82 ) ?? []);
83}
84
85function searchableRows(items: readonly OrchestrationV2ProjectedTurnItem[]) {
86 const foldedAnswers = new Set(
87 items.flatMap(({ item }) =>
88 item.type === "user_input_request" && item.questionAnswer
89 ? [`async-answer:${item.questionAnswer.requestId}`]
90 : [],
91 ),
92 );
93 return items.filter(
94 ({ item }) =>
95 ["user_message", "assistant_message", "proposed_plan"].includes(item.type) &&
96 !(item.type === "user_message" && foldedAnswers.has(item.messageId)),
97 );
98}
99
100function documentFor(row: OrchestrationV2ProjectedTurnItem, count: number): FindDocument {
101 return {
102 position: row.position,
103 visibility: row.visibility,
104 sourceThreadId: row.sourceThreadId,
105 sourceItemId: row.sourceItemId,
106 entryId:
107 row.item.type === "user_message" || row.item.type === "assistant_message"
108 ? row.item.messageId
109 : row.item.id,
110 runId: row.item.runId,
111 count,
112 };
113}
114
115function countSegments(segments: readonly string[], query: string) {
116 return segments.reduce((sum, text) => sum + countThreadSearchOccurrences(text, query), 0);
117}
118
119function selectMatch(
120 documents: readonly FindDocument[],
121 input: Pick<OrchestrationV2SearchThreadInput, "index" | "start" | "offset">,
122) {
123 const totalMatches = documents.reduce((sum, doc) => sum + doc.count, 0);
124 let requestedIndex = input.index ?? 0;
125 if (input.index === undefined && input.start) {
126 const startIndex = documents.findIndex((doc) => doc.entryId === input.start?.entryId);
127 if (startIndex >= 0) {
128 requestedIndex =
129 documents.slice(0, startIndex).reduce((sum, doc) => sum + doc.count, 0) +
130 Math.min(input.start.occurrence, documents[startIndex]!.count);
131 if (requestedIndex >= totalMatches) requestedIndex = 0;
132 }
133 }
134 const activeIndex =
135 input.index === undefined && totalMatches > 0
136 ? (((requestedIndex + (input.offset ?? 0)) % totalMatches) + totalMatches) % totalMatches
137 : Math.min(requestedIndex, Math.max(0, totalMatches - 1));
138 let occurrence = activeIndex;
139 for (const document of documents) {
140 if (occurrence < document.count) return { totalMatches, activeIndex, document, occurrence };
141 occurrence -= document.count;
142 }
143 return { totalMatches, activeIndex, document: null, occurrence: 0 };
144}
145
146/** The memory projection uses the same match ordering as SQLite. */
147export function findProjectedThreadItems(
148 items: readonly OrchestrationV2ProjectedTurnItem[],
149 input: OrchestrationV2SearchThreadInput,
150 snapshotSequence: number,
151 cwd?: string,
152): OrchestrationV2SearchThreadResult {
153 const rows = searchableRows(items);
154 const docs = rows.map((row) =>
155 documentFor(
156 row,
157 countSegments(parseText(textKey(row.item, cwd, input.skills ?? [])), input.query),
158 ),
159 );
160 return resultForSelection(selectMatch(docs, input), docs, snapshotSequence);
161}
162
163function resultForSelection(
164 selected: ReturnType<typeof selectMatch>,
165 documents: readonly FindDocument[],
166 snapshotSequence: number,
167): OrchestrationV2SearchThreadResult {
168 let startIndex = 0;
169 const entries = documents.flatMap(({ entryId, runId, count }) => {
170 const entry = { entryId, runId, count, startIndex };
171 startIndex += count;
172 return count > 0 ? [entry] : [];
173 });
174 const selectedEntry = entries.findIndex((entry) => entry.entryId === selected.document?.entryId);
175 const navigation =
176 entries.length <= 17
177 ? entries
178 : Array.from(
179 { length: 17 },
180 (_, i) => entries[(selectedEntry + i - 8 + entries.length) % entries.length]!,
181 );
182 return {
183 snapshotSequence,
184 totalMatches: selected.totalMatches,
185 activeIndex: selected.activeIndex,
186 navigation,
187 match:
188 selected.document === null
189 ? null
190 : {
191 entryId: selected.document.entryId,
192 runId: selected.document.runId,
193 occurrence: selected.occurrence,
194 },
195 };
196}
197
198/** Read the canonical index once per scan, including inherited fork rows, without tool bodies. */
199export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
200 readIndex: (threadId: ThreadId) => Effect.Effect<
201 readonly (Omit<FindRow, "position"> & {
202 readonly item: Pick<OrchestrationV2TurnItem, "type">;
203 })[],
204 ProjectionStoreV2Error
205 >,
206) {
207 const sql = yield* SqlClient.SqlClient;
208 const encodeCacheKey = Schema.encodeEffect(
209 Schema.fromJsonString(
210 Schema.Struct({
211 threadId: ThreadId,
212 cwd: Schema.optional(Schema.String),
213 query: Schema.String,
214 skills: OrchestrationV2SearchThreadInput.fields.skills,
215 }),
216 ),
217 );
218 const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId)));
219 const encodeSources = Schema.encodeEffect(
220 Schema.fromJsonString(Schema.Array(Schema.Struct({ threadId: ThreadId, id: TurnItemId }))),
221 );
222 const decodeItem = Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2TurnItemJson));
223 const isSnapshotError = Schema.is(
224 Schema.Union([ProjectionStoreReadError, ProjectionStoreThreadNotFoundError]),
225 );
226 const load = Effect.fn("ThreadFind.load")(function* (
227 threadId: ThreadId,
228 rows: readonly FindRow[],
229 ) {
230 if (rows.length === 0) return [];
231 const sources = yield* encodeSources(
232 rows.map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })),
233 ).pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));
234 const payloads = yield* sql<{
235 thread_id: string;
236 turn_item_id: string;
237 payload_json: string;
238 entry_id: string;
239 item_type: OrchestrationV2TurnItem["type"];
240 }>`
241 SELECT item.thread_id, item.turn_item_id, item.payload_json,
242 COALESCE(json_extract(item.payload_json, '$.messageId'), item.turn_item_id) AS entry_id,
243 json_extract(item.payload_json, '$.type') AS item_type
244 FROM orchestration_v2_projection_turn_items AS item
245 JOIN json_each(${sources}) AS wanted
246 ON item.thread_id = json_extract(wanted.value, '$.threadId')
247 AND item.turn_item_id = json_extract(wanted.value, '$.id')
248 `.pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));
249 const byThread = new Map<string, Map<string, (typeof payloads)[number]>>();
250 for (const row of payloads) {
251 let items = byThread.get(row.thread_id);
252 if (!items) byThread.set(row.thread_id, (items = new Map()));
253 items.set(row.turn_item_id, row);
254 }
255 return yield* Effect.forEach(rows, (row) => {
256 const payload = byThread.get(row.sourceThreadId)?.get(row.sourceItemId);
257 return payload === undefined
258 ? Effect.fail(new ProjectionStoreReadError({ threadId }))
259 : Effect.succeed({
260 ...row,
261 payload: payload.payload_json,
262 entryId: payload.entry_id,
263 type: payload.item_type,
264 });
265 });
266 });
267 // Cache rendered segments, not queries; typing and navigation reuse Markdown parsing.
268 const textCache = yield* Cache.make({
269 capacity: 512,
270 timeToLive: "1 minute",
271 lookup: (key: TextKey) => Effect.sync(() => parseText(key)),
272 });
273 // Bounded separately because each entry can hold a large body; typing reuses the parse.
274 const largeTextCache = yield* Cache.make({
275 capacity: 32,
276 timeToLive: "1 minute",
277 lookup: (key: TextKey) => Effect.sync(() => parseText(key)),
278 });
279 // Scans retain counts and item references, never message bodies. Navigation checks
280 // the current revision in its transaction without reloading message payloads.
281 const scans = new Map<
282 string,
283 {
284 sequence: number;
285 documents: readonly FindDocument[];
286 sourceThreadIds: readonly ThreadId[];
287 }
288 >();
289 const readRevision = Effect.fn("ThreadFind.readRevision")(function* (
290 threadIds: readonly ThreadId[],
291 ) {
292 const sources = yield* encodeThreadIds(threadIds);
293 const revision = yield* sql<{ sequence: number }>`
294 WITH RECURSIVE source_threads(thread_id) AS (
295 SELECT value FROM json_each(${sources})
296 UNION
297 SELECT json_extract(t.payload_json, '$.forkedFrom.threadId')
298 FROM orchestration_v2_projection_threads t
299 JOIN source_threads s ON s.thread_id = t.thread_id
300 WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
301 )
302 SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events
303 WHERE application_event_version = 2 AND aggregate_kind = 'thread'
304 AND stream_id IN (SELECT thread_id FROM source_threads)
305 `;
306 return revision[0]?.sequence ?? 0;
307 });
308 const snapshot = Effect.fn("ThreadFind.snapshot")(
309 function* (input: OrchestrationV2SearchThreadInput) {
310 const { threadId } = input;
311 const active = yield* sql<{ cwd: string | null }>`
312 SELECT COALESCE(json_extract(t.payload_json, '$.worktreePath'), p.workspace_root) AS cwd
313 FROM orchestration_v2_projection_threads t JOIN projection_projects p ON p.project_id = t.project_id
314 WHERE t.thread_id = ${threadId} AND t.deleted_at IS NULL AND p.deleted_at IS NULL
315 `;
316 if (!active[0]) return yield* new ProjectionStoreThreadNotFoundError({ threadId });
317 const cwd = active[0].cwd ?? undefined;
318 const cacheKey = yield* encodeCacheKey({
319 threadId,
320 cwd,
321 query: input.query,
322 skills: input.skills ?? [],
323 });
324 const cached = scans.get(cacheKey);
325 if (cached && cached.sequence === (yield* readRevision(cached.sourceThreadIds))) {
326 const selected = selectMatch(cached.documents, input);
327 scans.delete(cacheKey);
328 scans.set(cacheKey, cached);
329 return { result: resultForSelection(selected, cached.documents, cached.sequence) };
330 }
331 const index = yield* readIndex(threadId);
332 const sourceThreadIds = [...new Set([threadId, ...index.map((row) => row.sourceThreadId)])];
333 const sequence = yield* readRevision(sourceThreadIds);
334 const candidates = index
335 .map((row, position) => ({ ...row, position }))
336 .filter((row) =>
337 ["user_message", "assistant_message", "proposed_plan", "user_input_request"].includes(
338 row.item.type,
339 ),
340 );
341 const payloads: FindPayload[] = [];
342 for (let start = 0; start < candidates.length; start += 128) {
343 payloads.push(...(yield* load(threadId, candidates.slice(start, start + 128))));
344 }
345 return { payloads, cwd, sequence, cacheKey, sourceThreadIds };
346 },
347 sql.withTransaction,
348 (effect, input) =>
349 effect.pipe(
350 Effect.mapError((cause) =>
351 isSnapshotError(cause)
352 ? cause
353 : new ProjectionStoreReadError({ threadId: input.threadId, cause }),
354 ),
355 ),
356 );
357 const searchThreadStream = (input: OrchestrationV2SearchThreadInput) =>
358 Stream.unwrap(
359 Effect.gen(function* () {
360 // Release SQLite before decoding and parsing. Both frames describe one snapshot.
361 const data = yield* snapshot(input);
362 if (data.result) return Stream.succeed(data.result);
363 const { payloads, cwd, sequence, cacheKey, sourceThreadIds } = data;
364 const decode = (row: FindPayload) =>
365 decodeItem(row.payload).pipe(
366 Effect.map((item) => ({ ...row, item })),
367 Effect.mapError(
368 (cause) => new ProjectionStoreReadError({ threadId: input.threadId, cause }),
369 ),
370 );
371 // Answers rendered inside question cards must not appear as separate matches.
372 const foldedAnswers = new Set<string>();
373 for (const payload of payloads) {
374 if (payload.type !== "user_input_request") continue;
375 const { item } = yield* decode(payload);
376 if (item.type === "user_input_request" && item.questionAnswer)
377 foldedAnswers.add(`async-answer:${item.questionAnswer.requestId}`);
378 }
379 const rows = payloads.filter(
380 (row) =>
381 row.type !== "user_input_request" &&
382 !(row.type === "user_message" && foldedAnswers.has(row.entryId)),
383 );
384 const documents = new Map<number, FindDocument>();
385 const count = Effect.fnUntraced(function* (row: FindPayload) {
386 const cached = documents.get(row.position);
387 if (cached) return cached;
388 const decoded = yield* decode(row);
389 const key = textKey(decoded.item, cwd, input.skills ?? []);
390 const segments = yield* Cache.get(
391 key.text.length <= 32_768 ? textCache : largeTextCache,
392 key,
393 );
394 const document = documentFor(decoded, countSegments(segments, input.query));
395 documents.set(row.position, document);
396 yield* Effect.yieldNow;
397 return document;
398 });
399 const finish = Effect.gen(function* () {
400 for (const row of rows) yield* count(row);
401 const ordered = rows.map((row) => documents.get(row.position)!);
402 scans.delete(cacheKey);
403 scans.set(cacheKey, { sequence, documents: ordered, sourceThreadIds });
404 if (scans.size > 8) {
405 const oldest = scans.keys().next().value;
406 if (oldest !== undefined) scans.delete(oldest);
407 }
408 return resultForSelection(selectMatch(ordered, input), ordered, sequence);
409 });
410 // Relative navigation needs the final counts. Initial searches can reveal an
411 // identity immediately, scanning at/below the reading anchor before wrapping.
412 if (input.index !== undefined || (input.offset ?? 0) !== 0)
413 return Stream.fromEffect(finish);
414 const anchor = Math.max(
415 0,
416 rows.findIndex((row) => row.entryId === input.start?.entryId),
417 );
418 for (let step = 0; step <= rows.length; step++) {
419 const index = (anchor + step) % rows.length;
420 const row = rows[index];
421 if (!row) break;
422 const document = yield* count(row);
423 const occurrence =
424 step === 0 && row.entryId === input.start?.entryId
425 ? Math.min(input.start.occurrence, document.count)
426 : 0;
427 if (occurrence >= document.count) continue;
428 const first: OrchestrationV2SearchThreadResult = {
429 complete: false,
430 snapshotSequence: sequence,
431 totalMatches: 1,
432 activeIndex: 0,
433 match: { entryId: document.entryId, runId: document.runId, occurrence },
434 };
435 // The remainder is pulled only after the first frame has been delivered.
436 return Stream.concat(Stream.succeed(first), Stream.fromEffect(finish));
437 }
438 return Stream.fromEffect(finish);
439 }),
440 );
441 return {
442 searchThreadStream,
443 searchThread: (input: OrchestrationV2SearchThreadInput) =>
444 searchThreadStream(input).pipe(
445 Stream.runCollect,
446 Effect.map((results) => results[results.length - 1]!),
447 ),
448 };
449});