1import {
2 IsoDateTime,
3 MessageId,
4 OrchestrationThreadSearchSource,
5 OrchestrationV2AppThreadJson,
6 OrchestrationV2ThreadOpens,
7 ThreadLinkedPullRequest,
8 ThreadPullRequestLink,
9 type OrchestrationSearchThreadsInput,
10 type OrchestrationSearchThreadsResult,
11 type OrchestrationV2SearchThreadsInput,
12 type OrchestrationV2SearchThreadsResult,
13 type OrchestrationV2ThreadShellSnapshot,
14 ProjectId,
15 ThreadId,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Context from "effect/Context";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Schema from "effect/Schema";
23import * as SqlClient from "effect/sql/SqlClient";
24import * as SqlSchema from "effect/sql/SqlSchema";
25import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
26import {
27 messageSearchTokens,
28 foldSearchQuery,
29 rankMessageMatches,
30 rankSearchItems,
31 messageMatcher,
32 scoreThreadSearch,
33 type ThreadSearchSubject,
34} from "@t3tools/shared/paletteSearch";
35import { threadSearchPullRequests } from "@t3tools/shared/threadPullRequests";
36import {
37 indexedMessageCondition,
38 messageSearchText,
39 nonPlaceholderMessageCondition,
40} from "./messageSearchIndex.ts";
41import { loadOlderSettledThreadsPage } from "./ShellStream.ts";
42
43// When the user last wrote in the thread aliased `threads`, as the clients
44// rank recency; agent notices in the user role do not count.
45const latestUserMessageAt = (sql: SqlClient.SqlClient) => sql`(
46 SELECT messages.updated_at
47 FROM orchestration_v2_projection_messages AS messages
48 WHERE messages.thread_id = threads.thread_id
49 AND messages.role = 'user'
50 AND COALESCE(json_extract(messages.payload_json, '$.createdBy'), 'user') = 'user'
51 ORDER BY messages.updated_at DESC, messages.message_id DESC
52 LIMIT 1
53)`;
54
55export class ThreadSearchError extends Schema.TaggedError<ThreadSearchError>()(
56 "ThreadSearchError",
57 {
58 operation: Schema.Literals(["query", "decode"]),
59 cause: Schema.Defect(),
60 },
61) {
62 override get message(): string {
63 return `Thread search ${this.operation} failed.`;
64 }
65}
66
67// Token lists bind as JSON arrays, so any number of them is one parameter.
68const TokenList = Schema.fromJsonString(Schema.Array(Schema.String));
69const SearchRequest = Schema.Struct({
70 /** The FTS5 query, or null to scan messages instead. */
71 match: Schema.NullOr(Schema.String),
72 /** Tokens every hit must contain, checked on the text. */
73 substrings: TokenList,
74 /** For each short word outside ASCII, the spellings a hit must contain one of. */
75 shortSpellings: Schema.fromJsonString(Schema.Array(Schema.Array(Schema.String))),
76 projectId: Schema.NullOr(ProjectId),
77 /** Threads the caller already lists, left out before the hit limit. */
78 excludedThreadIds: TokenList,
79});
80const SearchRow = Schema.Struct({
81 messageRowid: Schema.Number,
82 messageId: MessageId,
83 threadId: ThreadId,
84 projectId: ProjectId,
85 source: OrchestrationThreadSearchSource,
86 text: Schema.String,
87 messageCreatedAt: Schema.NullOr(IsoDateTime),
88});
89
90const ThreadOpensRow = Schema.Struct({
91 threadId: ThreadId,
92 opens: Schema.NullOr(Schema.String),
93 latestUserMessageAt: Schema.NullOr(IsoDateTime),
94});
95const decodeThreadOpens = Schema.decodeUnknownOption(
96 Schema.fromJsonString(OrchestrationV2ThreadOpens),
97);
98
99const SNIPPET_LENGTH = 240;
100// The newest hits read per query. A common word stops here instead of reading
101// every message, so time stays flat as history grows.
102const MESSAGE_HIT_LIMIT = 2000;
103// How many of the newest index hits the filters may pass over to fill that
104// limit: a rare short word or hidden threads would otherwise read every hit.
105const MESSAGE_CANDIDATE_LIMIT = 10_000;
106// Without the index, how many of the newest messages a search scans.
107const MESSAGE_SCAN_LIMIT = 20_000;
108const NON_ASCII = /[\u0080-\uffff]/;
109
110/** SQLite's lower() folds ASCII only, so a short word is checked as each spelling it can match. */
111function shortTokenSpellings(token: string): ReadonlyArray<string> {
112 const forms = (char: string) => {
113 const plain = char.normalize("NFD").replace(/\p{M}/gu, "");
114 return [char, char.toUpperCase(), plain, plain.toUpperCase()];
115 };
116 const spellings = Array.from(token).reduce<ReadonlyArray<string>>(
117 (prefixes, char) => prefixes.flatMap((prefix) => forms(char).map((form) => prefix + form)),
118 [""],
119 );
120 return [...new Set(spellings)];
121}
122
123/**
124 * The SQL inputs for a message search, or null when the query has no token
125 * long enough to search. Tokens of three or more characters go to the
126 * trigram index as quoted phrases; shorter ones, and every token when the
127 * index cannot serve the query, are checked as substrings of the hits.
128 */
129function buildMessageSearch(
130 query: string,
131 hasIndex: boolean,
132): Omit<typeof SearchRequest.Type, "projectId" | "excludedThreadIds"> | null {
133 const { long, short } = messageSearchTokens(query);
134 if (long.length === 0) return null;
135 // Two-character CJK words are too short for trigrams.
136 const indexed = hasIndex && long.every((token) => Array.from(token).length >= 3);
137 const narrowing = short.filter((token) => !NON_ASCII.test(token));
138 return {
139 match: indexed ? long.map((token) => `"${token.replaceAll('"', '""')}"`).join(" ") : null,
140 substrings: indexed ? narrowing : [...long, ...narrowing],
141 shortSpellings: short.filter((token) => NON_ASCII.test(token)).map(shortTokenSpellings),
142 };
143}
144
145// A thread shows a link as its text, so in-thread find cannot reach a match
146// in the address; messages are matched as they read.
147const MARKDOWN_LINK = /!?\[([^\]\n]*)\]\([^)\s]*\)/g;
148
149// Result rows show one line, so the match must sit near the start to be seen.
150const SNIPPET_LEAD = 24;
151
152const isLowSurrogate = (code: number) => code >= 0xdc00 && code <= 0xdfff;
153
154/**
155 * Up to 240 characters starting a few words before the match at `matchStart`,
156 * whitespace collapsed, with ellipses where the text was cut. The client finds
157 * the highlights again in the result.
158 */
159export function buildSearchSnippet(text: string, matchStart: number): string {
160 let start = Math.max(0, matchStart - SNIPPET_LEAD);
161 const wordStart = text.slice(start, matchStart).search(/\s/);
162 if (start > 0 && wordStart !== -1) start += wordStart + 1;
163 // Cut between characters, not inside an emoji's surrogate pair.
164 if (isLowSurrogate(text.charCodeAt(start))) start++;
165 // Collapsing whitespace shortens the text, so read more of it.
166 const rawEnd = Math.min(text.length, start + SNIPPET_LENGTH * 2);
167 // Emphasis and code marks do not show in a thread either.
168 const shown = text
169 .slice(start, rawEnd)
170 .replace(/\*\*|`+/g, "")
171 .replace(/\s+/g, " ")
172 .trim();
173 let end = Math.min(shown.length, SNIPPET_LENGTH - 2);
174 if (isLowSurrogate(shown.charCodeAt(end))) end--;
175 const cut = end < shown.length || rawEnd < text.length;
176 return `${start > 0 ? "…" : ""}${shown.slice(0, end)}${cut ? "…" : ""}`;
177}
178
179// Filter hidden subagents and side chats before the hit limit so visible matches aren't lost.
180// Word starts and ranking follow the title rules in JS.
181// Without FTS5 trigrams, or for two-character CJK words, scan the same eligible messages.
182// The scan folds ASCII case only, so accent variants and non-ASCII case variants may not match.
183export const messageHitsQuery = (
184 sql: SqlClient.SqlClient,
185 { match, substrings, shortSpellings, projectId, excludedThreadIds }: typeof SearchRequest.Encoded,
186) => sql`
187 WITH eligible AS MATERIALIZED (
188 SELECT threads.thread_id, threads.project_id
189 FROM orchestration_v2_projection_threads AS threads
190 INNER JOIN projection_projects AS projects
191 ON projects.project_id = threads.project_id
192 WHERE threads.deleted_at IS NULL
193 AND threads.archived_at IS NULL
194 AND projects.deleted_at IS NULL
195 AND (${projectId} IS NULL OR threads.project_id = ${projectId})
196 AND threads.thread_id NOT IN (SELECT value FROM json_each(${excludedThreadIds}))
197 AND json_extract(threads.payload_json, '$.lineage.relationshipToParent')
198 IS NOT 'subagent'
199 AND json_extract(threads.payload_json, '$.lineage.sideChat') IS NOT 1
200 )
201 ${
202 match === null
203 ? sql`
204 SELECT
205 messages.rowid AS "messageRowid",
206 messages.message_id AS "messageId",
207 messages.thread_id AS "threadId",
208 eligible.project_id AS "projectId",
209 messages.role AS source,
210 ${messageSearchText(sql)} AS text,
211 messages.created_at AS "messageCreatedAt"
212 FROM orchestration_v2_projection_messages AS messages
213 INNER JOIN eligible ON eligible.thread_id = messages.thread_id
214 WHERE ${indexedMessageCondition(sql)}
215 -- A scan reads every message it passes, so it stops at the newest ones.
216 AND messages.rowid > (
217 SELECT coalesce(max(rowid), 0) FROM orchestration_v2_projection_messages
218 ) - ${MESSAGE_SCAN_LIMIT}
219 `
220 : sql`
221 SELECT
222 orchestration_v2_message_search.rowid AS "messageRowid",
223 messages.message_id AS "messageId",
224 messages.thread_id AS "threadId",
225 eligible.project_id AS "projectId",
226 messages.role AS source,
227 ${messageSearchText(sql)} AS text,
228 messages.created_at AS "messageCreatedAt"
229 FROM orchestration_v2_message_search
230 INNER JOIN orchestration_v2_projection_messages AS messages
231 ON messages.rowid = orchestration_v2_message_search.rowid
232 INNER JOIN eligible ON eligible.thread_id = messages.thread_id
233 WHERE orchestration_v2_message_search MATCH ${match}
234 AND orchestration_v2_message_search.rowid >= coalesce((
235 SELECT orchestration_v2_message_search.rowid FROM orchestration_v2_message_search
236 WHERE orchestration_v2_message_search MATCH ${match}
237 ORDER BY orchestration_v2_message_search.rowid DESC
238 LIMIT 1 OFFSET ${MESSAGE_CANDIDATE_LIMIT - 1}
239 ), 0)
240 AND ${nonPlaceholderMessageCondition(sql)}
241 `
242 }
243 -- Answers to an agent's question show inside its question card, where
244 -- neither find nor a jump can reach them.
245 AND messages.message_id NOT LIKE 'async-answer:%'
246 AND NOT EXISTS (
247 SELECT 1 FROM json_each(${substrings}) AS token
248 WHERE instr(lower(${messageSearchText(sql)}), token.value) = 0
249 )
250 AND NOT EXISTS (
251 SELECT 1 FROM json_each(${shortSpellings}) AS token
252 WHERE NOT EXISTS (
253 SELECT 1 FROM json_each(token.value) AS spelling
254 WHERE instr(${messageSearchText(sql)}, spelling.value) > 0
255 )
256 )
257 -- The index's own rowid, so FTS5 returns hits newest first and stops at
258 -- the limit; ordering by the messages table's rowid sorts every hit.
259 ORDER BY "messageRowid" DESC
260 LIMIT ${MESSAGE_HIT_LIMIT}
261 `;
262
263const TitleCandidateRow = Schema.Struct({
264 thread: Schema.fromJsonString(OrchestrationV2AppThreadJson),
265 projectTitle: Schema.String,
266 latestUserMessageAt: Schema.NullOr(IsoDateTime),
267});
268
269export interface ThreadSearchHit {
270 readonly threadId: ThreadId;
271 readonly projectId: ProjectId;
272 readonly title: string;
273 readonly source: "title" | OrchestrationThreadSearchSource;
274 readonly snippet?: string;
275 readonly messageCreatedAt?: string | null;
276}
277
278export class ThreadSearch extends Context.Service<
279 ThreadSearch,
280 {
281 /** Message text matches, for clients that rank their own thread list. */
282 readonly search: (
283 input: OrchestrationSearchThreadsInput,
284 ) => Effect.Effect<OrchestrationSearchThreadsResult, ThreadSearchError>;
285 /**
286 * Title, branch, pull request, and project matches ranked as the command
287 * palette ranks them, then message matches for the remaining threads.
288 */
289 readonly searchThreads: (
290 input: OrchestrationSearchThreadsInput & { readonly projectId?: ProjectId | undefined },
291 ) => Effect.Effect<{ readonly matches: ReadonlyArray<ThreadSearchHit> }, ThreadSearchError>;
292 /**
293 * The settled threads before `before` that clients omitting older settled threads should
294 * receive for a query: the best title, branch, pull request and project matches, ranked as
295 * the palette ranks them, then the older ones among `messageThreadIds`, in their order.
296 */
297 readonly searchOlderSettled: (input: {
298 readonly query: string;
299 readonly before: string;
300 readonly projectIds?: ReadonlyArray<ProjectId> | undefined;
301 /** The caller's name for this environment, which its palette also matches. */
302 readonly environmentLabel?: string | undefined;
303 readonly messageThreadIds: ReadonlyArray<ThreadId>;
304 }) => Effect.Effect<ReadonlyArray<ThreadId>, ThreadSearchError>;
305 }
306>()("t3/orchestration-v2/ThreadSearch") {}
307
308// Rows a client gets per search: enough for the shown results without sending every match.
309const OLDER_SETTLED_TITLE_ROWS = 15;
310const OLDER_SETTLED_MESSAGE_ROWS = 5;
311
312// Only the fields the palette scorer reads, so an interactive search does not decode payloads.
313const OlderSettledTitleRow = Schema.Struct({
314 id: ThreadId,
315 title: Schema.String,
316 branch: Schema.NullOr(Schema.String),
317 pullRequests: Schema.NullOr(Schema.fromJsonString(Schema.Array(ThreadPullRequestLink))),
318 linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)),
319 branchPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)),
320 opens: Schema.NullOr(Schema.String),
321 createdAt: IsoDateTime,
322 projectTitle: Schema.String,
323 latestUserMessageAt: Schema.NullOr(IsoDateTime),
324});
325
326/** @public */
327export const make = Effect.gen(function* () {
328 const sql = yield* SqlClient.SqlClient;
329
330 const searchRows = SqlSchema.findAll({
331 Request: SearchRequest,
332 Result: SearchRow,
333 execute: (request) => messageHitsQuery(sql, request),
334 });
335
336 const threadOpens = SqlSchema.findAll({
337 Request: Schema.Struct({ threadIds: TokenList }),
338 Result: ThreadOpensRow,
339 execute: ({ threadIds }) => sql`
340 SELECT
341 threads.thread_id AS "threadId",
342 json_extract(threads.payload_json, '$.opens') AS opens,
343 ${latestUserMessageAt(sql)} AS "latestUserMessageAt"
344 FROM orchestration_v2_projection_threads AS threads
345 WHERE threads.thread_id IN (SELECT value FROM json_each(${threadIds}))
346 `,
347 });
348
349 const titleCandidates = SqlSchema.findAll({
350 Request: Schema.Struct({ projectId: Schema.NullOr(ProjectId) }),
351 Result: TitleCandidateRow,
352 execute: ({ projectId }) => sql`
353 SELECT
354 threads.payload_json AS thread,
355 projects.title AS "projectTitle",
356 ${latestUserMessageAt(sql)} AS "latestUserMessageAt"
357 FROM orchestration_v2_projection_threads AS threads
358 INNER JOIN projection_projects AS projects
359 ON projects.project_id = threads.project_id
360 WHERE threads.deleted_at IS NULL
361 AND threads.archived_at IS NULL
362 AND projects.deleted_at IS NULL
363 AND json_extract(threads.payload_json, '$.lineage.relationshipToParent')
364 IS NOT 'subagent'
365 AND json_extract(threads.payload_json, '$.lineage.sideChat') IS NOT 1
366 AND (${projectId} IS NULL OR threads.project_id = ${projectId})
367 `,
368 });
369
370 const olderSettledTitleCandidates = SqlSchema.findAll({
371 Request: Schema.Struct({
372 before: Schema.String,
373 projectIds: Schema.NullOr(TokenList),
374 }),
375 Result: OlderSettledTitleRow,
376 execute: ({ before, projectIds }) => sql`
377 SELECT
378 threads.thread_id AS id,
379 json_extract(threads.payload_json, '$.title') AS title,
380 json_extract(threads.payload_json, '$.branch') AS branch,
381 json_extract(threads.payload_json, '$.pullRequests') AS "pullRequests",
382 json_extract(threads.payload_json, '$.linkedPullRequest') AS "linkedPullRequest",
383 json_extract(threads.payload_json, '$.branchPullRequest') AS "branchPullRequest",
384 json_extract(threads.payload_json, '$.opens') AS opens,
385 json_extract(threads.payload_json, '$.createdAt') AS "createdAt",
386 projects.title AS "projectTitle",
387 ${latestUserMessageAt(sql)} AS "latestUserMessageAt"
388 FROM orchestration_v2_projection_threads AS threads
389 INNER JOIN projection_projects AS projects
390 ON projects.project_id = threads.project_id
391 WHERE threads.deleted_at IS NULL
392 AND json_extract(threads.payload_json, '$.archivedAt') IS NULL
393 AND projects.deleted_at IS NULL
394 AND json_extract(threads.payload_json, '$.lineage.relationshipToParent')
395 IS NOT 'subagent'
396 AND json_extract(threads.payload_json, '$.lineage.sideChat') IS NOT 1
397 AND json_extract(threads.payload_json, '$.settledOverride') = 'settled'
398 AND json_extract(threads.payload_json, '$.settledAt') < ${before}
399 AND (${projectIds} IS NULL
400 OR threads.project_id IN (SELECT value FROM json_each(${projectIds})))
401 `,
402 });
403
404 const olderSettledIds = SqlSchema.findAll({
405 Request: Schema.Struct({
406 before: Schema.String,
407 threadIds: TokenList,
408 projectIds: Schema.NullOr(TokenList),
409 }),
410 Result: Schema.Struct({ id: ThreadId }),
411 execute: ({ before, threadIds, projectIds }) => sql`
412 SELECT threads.thread_id AS id
413 FROM orchestration_v2_projection_threads AS threads
414 WHERE threads.thread_id IN (SELECT value FROM json_each(${threadIds}))
415 AND threads.deleted_at IS NULL
416 AND json_extract(threads.payload_json, '$.archivedAt') IS NULL
417 AND json_extract(threads.payload_json, '$.lineage.relationshipToParent')
418 IS NOT 'subagent'
419 AND json_extract(threads.payload_json, '$.lineage.sideChat') IS NOT 1
420 AND json_extract(threads.payload_json, '$.settledOverride') = 'settled'
421 AND json_extract(threads.payload_json, '$.settledAt') < ${before}
422 AND (${projectIds} IS NULL
423 OR threads.project_id IN (SELECT value FROM json_each(${projectIds})))
424 `,
425 });
426
427 const read = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
428 sql.withTransaction(effect).pipe(
429 NodeSqliteClient.readOnly,
430 Effect.mapError(
431 (cause) =>
432 new ThreadSearchError({
433 operation: Schema.isSchemaError(cause) ? "decode" : "query",
434 cause,
435 }),
436 ),
437 );
438
439 // A project scope applies before ranking, so other projects' stronger hits
440 // cannot use up the limit.
441 const searchMessages = Effect.fn("ThreadSearch.search")(function* (
442 input: OrchestrationSearchThreadsInput & {
443 readonly projectId: ProjectId | null;
444 /** Threads the caller already lists, which must not use up the limit. */
445 readonly excludedThreadIds?: ReadonlySet<ThreadId>;
446 /** Off for an agent's search, which has no use for the user's picks. */
447 readonly picks?: boolean;
448 },
449 ) {
450 // One message per thread: the best word match, then what the user wrote,
451 // since that is what they remember asking, then the newest by when it was
452 // written. Row order is not age: imported histories arrive in one batch.
453 // Text relevance (bm25) would cost a pass over every hit in history.
454 type Hit = {
455 readonly row: typeof SearchRow.Type;
456 /** The message as it reads, links as their text. */
457 readonly text: string;
458 readonly quality: number;
459 readonly offset: number;
460 };
461 const newer = (left: Hit, right: Hit) =>
462 (right.row.messageCreatedAt ?? "").localeCompare(left.row.messageCreatedAt ?? "");
463 const compareHits = (left: Hit, right: Hit) =>
464 right.quality - left.quality ||
465 Number(left.row.source !== "user") - Number(right.row.source !== "user") ||
466 newer(left, right);
467 const rows = yield* read(
468 Effect.gen(function* () {
469 const index = yield* sql`
470 SELECT 1 FROM sqlite_master WHERE name = 'orchestration_v2_message_search'
471 `;
472 const messageSearch = buildMessageSearch(input.query, index.length > 0);
473 return messageSearch === null
474 ? []
475 : yield* searchRows({
476 ...messageSearch,
477 projectId: input.projectId,
478 excludedThreadIds: [...(input.excludedThreadIds ?? [])],
479 });
480 }),
481 );
482 const best = new Map<ThreadId, Hit>();
483 // Matching messages per thread at its best quality: a thread that keeps
484 // using the word ranks above a passing mention, and weaker hits ("Trust"
485 // for "rust") never add to a whole word's count.
486 const hitCounts = new Map<ThreadId, number>();
487 const hitCount = (threadId: ThreadId) => hitCounts.get(threadId) ?? 1;
488 const matchMessage = messageMatcher(input.query);
489 for (const row of rows) {
490 const text = row.text.replace(MARKDOWN_LINK, "$1");
491 const match = matchMessage(text);
492 if (match === null) continue;
493 const hit = { row, text, ...match };
494 const current = best.get(row.threadId);
495 if (current === undefined || hit.quality > current.quality) {
496 hitCounts.set(row.threadId, 1);
497 } else if (hit.quality === current.quality) {
498 hitCounts.set(row.threadId, hitCount(row.threadId) + 1);
499 }
500 if (current === undefined || compareHits(hit, current) < 0) best.set(row.threadId, hit);
501 }
502 // Ranked as clients rank them, with picks and use, before the limit cuts:
503 // a thread picked for this query must come back even when it only
504 // matches in a few old messages.
505 const opens = best.size === 0 ? [] : yield* read(threadOpens({ threadIds: [...best.keys()] }));
506 const subjectById = new Map(
507 opens.map((row) => [
508 row.threadId,
509 {
510 title: "",
511 activityAt: row.latestUserMessageAt === null ? null : Date.parse(row.latestUserMessageAt),
512 opens: Option.getOrNull(decodeThreadOpens(row.opens)),
513 },
514 ]),
515 );
516 const now = DateTime.toEpochMillis(yield* DateTime.now);
517 const ranked = rankMessageMatches(
518 [...best.values()].map((hit) => ({
519 ...hit,
520 hits: hitCount(hit.row.threadId),
521 messageCreatedAt: hit.row.messageCreatedAt,
522 })),
523 (hit) => subjectById.get(hit.row.threadId),
524 foldSearchQuery(input.query),
525 now,
526 { picks: input.picks ?? true },
527 );
528 return {
529 matches: ranked.slice(0, input.limit ?? 50).map(({ row, text, quality, offset }) => ({
530 hits: hitCount(row.threadId),
531 threadId: row.threadId,
532 messageId: row.messageId,
533 projectId: row.projectId,
534 source: row.source,
535 snippet: buildSearchSnippet(text, offset),
536 messageCreatedAt: row.messageCreatedAt,
537 quality,
538 })),
539 };
540 });
541
542 const search: ThreadSearch["Service"]["search"] = (input) =>
543 searchMessages({ ...input, projectId: null });
544
545 const searchThreads: ThreadSearch["Service"]["searchThreads"] = Effect.fn(
546 "ThreadSearch.searchThreads",
547 )(function* (input) {
548 const limit = input.limit ?? 50;
549 const normalizedQuery = foldSearchQuery(input.query);
550 const now = DateTime.toEpochMillis(yield* DateTime.now);
551 // Agents call this rarely, and a project has hundreds of threads, not
552 // millions, so ranking every active thread in memory stays cheap.
553 const candidates = yield* read(titleCandidates({ projectId: input.projectId ?? null }));
554 const subjectById = new Map(
555 candidates.map((row): [ThreadId, ThreadSearchSubject] => [
556 row.thread.id,
557 {
558 id: row.thread.id,
559 title: row.thread.title,
560 branch: row.thread.branch,
561 projectTitle: row.projectTitle,
562 pullRequests: threadSearchPullRequests(row.thread),
563 // Like the clients: what the user last wrote, else when the thread began.
564 activityAt: Date.parse(
565 row.latestUserMessageAt ?? DateTime.formatIso(row.thread.createdAt),
566 ),
567 opens: row.thread.opens ?? null,
568 },
569 ]),
570 );
571 const titleHits: ThreadSearchHit[] = rankSearchItems(candidates, (row, options) =>
572 // An agent's search has no use for the user's palette picks.
573 scoreThreadSearch(normalizedQuery, subjectById.get(row.thread.id)!, now, {
574 ...options,
575 picks: false,
576 }),
577 )
578 .slice(0, limit)
579 .map(({ item }) => ({
580 threadId: item.thread.id,
581 projectId: item.thread.projectId,
582 title: item.thread.title,
583 source: "title",
584 }));
585 const shown = new Set(titleHits.map((hit) => hit.threadId));
586 const { matches } = yield* searchMessages({
587 query: input.query,
588 limit: 50,
589 projectId: input.projectId ?? null,
590 excludedThreadIds: shown,
591 picks: false,
592 });
593 const messageHits = matches
594 .filter((match) => subjectById.has(match.threadId))
595 .map(({ quality: _quality, hits: _hits, messageId: _messageId, ...match }) => ({
596 ...match,
597 title: subjectById.get(match.threadId)!.title,
598 }));
599 // Titles come first, but a fifth of the slots stay for message matches,
600 // as the palette keeps its "In messages" section.
601 const messageSlots = Math.min(messageHits.length, Math.floor(limit / 5));
602 return {
603 matches: [...titleHits.slice(0, limit - messageSlots), ...messageHits].slice(0, limit),
604 };
605 });
606
607 const searchOlderSettled: ThreadSearch["Service"]["searchOlderSettled"] = Effect.fn(
608 "ThreadSearch.searchOlderSettled",
609 )(function* ({ query, before, projectIds, environmentLabel, messageThreadIds }) {
610 const normalizedQuery = foldSearchQuery(query);
611 const now = DateTime.toEpochMillis(yield* DateTime.now);
612 const candidates = yield* read(
613 olderSettledTitleCandidates({ before, projectIds: projectIds ?? null }),
614 );
615 const subjectOf = (row: typeof OlderSettledTitleRow.Type): ThreadSearchSubject => ({
616 id: row.id,
617 title: row.title,
618 branch: row.branch,
619 projectTitle: row.projectTitle,
620 environmentLabel: environmentLabel ?? null,
621 pullRequests: threadSearchPullRequests({
622 pullRequests: row.pullRequests ?? undefined,
623 linkedPullRequest: row.linkedPullRequest,
624 branchPullRequest: row.branchPullRequest,
625 }),
626 // Like the clients: what the user last wrote, else when the thread began.
627 activityAt: Date.parse(row.latestUserMessageAt ?? row.createdAt),
628 opens: Option.getOrNull(decodeThreadOpens(row.opens)),
629 });
630 // Picks on, as in the command palette: a thread picked for this query must make the cut.
631 const titleIds = rankSearchItems(candidates, (row, options) =>
632 scoreThreadSearch(normalizedQuery, subjectOf(row), now, { ...options, picks: true }),
633 )
634 .slice(0, OLDER_SETTLED_TITLE_ROWS)
635 .map(({ item }) => item.id);
636 const shown = new Set(titleIds);
637 const messageCandidates = [...new Set(messageThreadIds)].filter((id) => !shown.has(id));
638 const older =
639 messageCandidates.length === 0
640 ? new Set<ThreadId>()
641 : new Set(
642 (yield* read(
643 olderSettledIds({
644 before,
645 threadIds: messageCandidates,
646 projectIds: projectIds ?? null,
647 }),
648 )).map((row) => row.id),
649 );
650 return [
651 ...titleIds,
652 ...messageCandidates.filter((id) => older.has(id)).slice(0, OLDER_SETTLED_MESSAGE_ROWS),
653 ];
654 });
655
656 return ThreadSearch.of({ search, searchThreads, searchOlderSettled });
657});
658
659export const layer = Layer.effect(ThreadSearch, make);
660
661/**
662 * The client search: message hits, and with `olderSettledBefore`, the rows of the best older
663 * settled matches, which shells that omit them do not hold.
664 */
665export const searchThreadsWithOlderSettledRows = Effect.fn(
666 "ThreadSearch.searchThreadsWithOlderSettledRows",
667)(function* <E1, E2, E3, E4, E5>(input: {
668 readonly request: OrchestrationV2SearchThreadsInput;
669 readonly search: (
670 input: OrchestrationSearchThreadsInput,
671 ) => Effect.Effect<OrchestrationSearchThreadsResult, E1>;
672 readonly searchOlderSettled: (
673 input: Parameters<ThreadSearch["Service"]["searchOlderSettled"]>[0],
674 ) => Effect.Effect<ReadonlyArray<ThreadId>, E2>;
675 readonly sql: SqlClient.SqlClient;
676 readonly readThreads: (
677 threadIds: ReadonlyArray<ThreadId>,
678 ) => Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E4>, E3>;
679 readonly latestSequence: Effect.Effect<number, E5>;
680}) {
681 const { olderSettledBefore, projectIds, environmentLabel, titlesOnly, ...request } =
682 input.request;
683 const messages: OrchestrationV2SearchThreadsResult =
684 titlesOnly === true ? { matches: [] } : yield* input.search(request);
685 if (olderSettledBefore === undefined) return messages;
686 // Ranked before the rows' transaction, so scoring never holds the shared reader.
687 const threadIds = yield* input.searchOlderSettled({
688 query: request.query,
689 before: olderSettledBefore,
690 projectIds,
691 environmentLabel,
692 messageThreadIds: messages.matches.map((match) => match.threadId),
693 });
694 const older = yield* loadOlderSettledThreadsPage({
695 sql: input.sql,
696 listThreads: Effect.succeed({ threadIds, counts: [] }),
697 readThreads: input.readThreads,
698 latestSequence: input.latestSequence,
699 });
700 return { ...messages, snapshotSequence: older.snapshotSequence, threads: older.threads };
701});