diff --git a/apps/mobile/modules/t3-markdown-text/src/markdownLinks.ts b/apps/mobile/modules/t3-markdown-text/src/markdownLinks.ts index 8b491c0c77c15fe578a9a684e592ad66621cae1d..913f7cb37167a8652fe29459ecdfc26c8c7fc9aa 100644 --- a/apps/mobile/modules/t3-markdown-text/src/markdownLinks.ts +++ b/apps/mobile/modules/t3-markdown-text/src/markdownLinks.ts @@ -1,10 +1,10 @@ +import { fileBasename } from "@t3tools/shared/path"; +import { formatFilePathPosition } from "@t3tools/shared/fileLinks"; import { - fileBasename, - formatFilePathPosition, inlineCodeFilePathCandidate, normalizeMarkdownLinkDestination, parseMarkdownFileLink, -} from "@t3tools/client-runtime/markdown-links"; +} from "@t3tools/shared/markdownLinks"; import { parseThreadLinkHref } from "@t3tools/shared/threadLinks"; import { videoMimeType } from "@t3tools/shared/video"; diff --git a/apps/mobile/modules/t3-markdown-text/src/nativeMarkdownText.ts b/apps/mobile/modules/t3-markdown-text/src/nativeMarkdownText.ts index 6ae50075093f715c6c580e8d9c3603588e0e22d1..58df022faf1d459cce35a58667f769a501859dbd 100644 --- a/apps/mobile/modules/t3-markdown-text/src/nativeMarkdownText.ts +++ b/apps/mobile/modules/t3-markdown-text/src/nativeMarkdownText.ts @@ -1,5 +1,5 @@ import type { MarkdownNode } from "react-native-nitro-markdown/headless"; -import { isMarkdownFileLinkLabel } from "@t3tools/client-runtime/markdown-links"; +import { isMarkdownFileLinkLabel } from "@t3tools/shared/markdownLinks"; import { collectComposerInlineTokens } from "@t3tools/shared/composerInlineTokens"; import { imageMimeType } from "@t3tools/shared/image"; import { isWindowsAbsolutePath } from "@t3tools/shared/path"; diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 79990884085d1b22fb31c03d0bd37ff9e583ff48..b6930deb5e8643ab7b0bb8a4e0f504ce30e6d02f 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -29,7 +29,7 @@ import type { import { appendCodexArtifactTemplateUsePrompt, type CodexArtifactTemplate, -} from "@t3tools/client-runtime/codex-artifact-templates"; +} from "@t3tools/shared/codexArtifactTemplates"; import type { ThreadUserInputQuestion } from "@t3tools/client-runtime/state/thread-requests"; import { presentPendingBackgroundWork, diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 4cfa8f2676b7c9e5ebe9876b77a3acaa5719f8da..44673dcf2759dffddbf8639979151c5461eb5173 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -36,9 +36,9 @@ import { writeComposerContextClipboard } from "../../lib/composerContextClipboar import { codexArtifactTemplatePresentationLabel, type CodexArtifactTemplate, -} from "@t3tools/client-runtime/codex-artifact-templates"; +} from "@t3tools/shared/codexArtifactTemplates"; import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets"; -import { isMarkdownFileLinkLabel } from "@t3tools/client-runtime/markdown-links"; +import { isMarkdownFileLinkLabel } from "@t3tools/shared/markdownLinks"; import { getTextContent, type MarkdownNode } from "react-native-nitro-markdown/headless"; import { formatAttachmentSize } from "@t3tools/client-runtime/state/attachments"; import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; @@ -50,7 +50,7 @@ import { resolveViewedImageAsset } from "@t3tools/client-runtime/work-log/presen import { renderCodexFileCitationsAsMarkdown, splitCodexArtifactTemplateMarkdown, -} from "@t3tools/client-runtime/codex-markdown-directives"; +} from "@t3tools/shared/codexMarkdownDirectives"; import { CHAT_LIST_ANCHOR_OFFSET, resolveChatListAnchoredEndSpace } from "@t3tools/shared/chatList"; import { imageMimeType } from "@t3tools/shared/image"; import { videoMimeType } from "@t3tools/shared/video"; diff --git a/apps/mobile/src/features/threads/fileChipMenu.ts b/apps/mobile/src/features/threads/fileChipMenu.ts index 9f82e089444dcbeb9dc4ae03b2f223604e367a19..9835c0c37cc17485975c83fb7464b49df010368f 100644 --- a/apps/mobile/src/features/threads/fileChipMenu.ts +++ b/apps/mobile/src/features/threads/fileChipMenu.ts @@ -1,4 +1,4 @@ -import { fileBasename } from "@t3tools/client-runtime/markdown-links"; +import { fileBasename } from "@t3tools/shared/path"; import type { ThreadId } from "@t3tools/contracts"; import { resolveMarkdownLinkPresentation } from "@t3tools/mobile-markdown-text/links"; import type { MarkdownFileContextMenu } from "@t3tools/mobile-markdown-text/types"; diff --git a/apps/mobile/src/features/threads/thread-search-match.tsx b/apps/mobile/src/features/threads/thread-search-match.tsx index 553160dbdd8c666a713be6fda97117db3b3e0edc..1f03e2c2adf03cf512fd50f2a7b5e69a716148bc 100644 --- a/apps/mobile/src/features/threads/thread-search-match.tsx +++ b/apps/mobile/src/features/threads/thread-search-match.tsx @@ -1,48 +1,9 @@ +import { splitThreadSearchText } from "@t3tools/shared/threadSearch"; import type { EnvironmentThreadSearchMatch } from "@t3tools/client-runtime/state/thread-search"; import { AppText as Text } from "../../components/AppText"; import { cn } from "../../lib/cn"; -function foldAsciiCase(value: string): string { - return value.replace(/[A-Z]/g, (character) => character.toLowerCase()); -} - -function splitHighlightParts(text: string, query: string) { - const normalizedText = foldAsciiCase(text); - const normalizedQuery = foldAsciiCase(query.trim()); - if (normalizedQuery.length === 0) { - return [{ text, highlighted: false, start: 0 }]; - } - - const parts: Array<{ - readonly text: string; - readonly highlighted: boolean; - readonly start: number; - }> = []; - let cursor = 0; - while (cursor < text.length) { - const matchIndex = normalizedText.indexOf(normalizedQuery, cursor); - if (matchIndex === -1) { - parts.push({ text: text.slice(cursor), highlighted: false, start: cursor }); - break; - } - if (matchIndex > cursor) { - parts.push({ - text: text.slice(cursor, matchIndex), - highlighted: false, - start: cursor, - }); - } - parts.push({ - text: text.slice(matchIndex, matchIndex + normalizedQuery.length), - highlighted: true, - start: matchIndex, - }); - cursor = matchIndex + normalizedQuery.length; - } - return parts; -} - export function ThreadSearchMatchExcerpt(props: { readonly match: EnvironmentThreadSearchMatch; readonly query: string; @@ -51,7 +12,7 @@ export function ThreadSearchMatchExcerpt(props: { readonly sidebar?: boolean; }) { const isUser = props.match.source === "user"; - const parts = splitHighlightParts(props.match.snippet, props.query); + const parts = splitThreadSearchText(props.match.snippet, props.query); return ( Effect.Effect; + readonly searchThreadStream: ( + input: OrchestrationV2SearchThreadInput, + ) => Stream.Stream; + readonly searchThread: ( + input: OrchestrationV2SearchThreadInput, + ) => Effect.Effect; + readonly getThreadHistoryPage: ( + threadId: ThreadId, + cursor: string, + throughEntryId?: string, + conversationOnly?: boolean, + ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, options: ProjectionTimelinePageOptions, @@ -10817,6 +10834,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio recoverDelegatedTask, delegatedTaskResultPending, dispatch: dispatchWithReceipt, + searchThreadStream: (input) => + projectionStore + .searchThreadStream(input) + .pipe( + Stream.mapError( + (cause) => new OrchestratorProjectionError({ threadId: input.threadId, cause }), + ), + ), + searchThread: (input) => + projectionStore + .searchThread(input) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: input.threadId, cause }), + ), + ), + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + projectionStore + .getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly) + .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))), getTimelinePage: (threadId, options) => projectionStore .getTimelinePage(threadId, options) @@ -10951,6 +10988,11 @@ const layerUnavailable: Layer.Layer = Layer.succeed( cause: "Orchestration V2 live runtime is not configured.", }), ), + searchThreadStream: (input) => + Stream.fail(new OrchestratorProjectionError({ threadId: input.threadId })), + searchThread: (input) => + Effect.fail(new OrchestratorProjectionError({ threadId: input.threadId })), + getThreadHistoryPage: (threadId) => Effect.fail(new OrchestratorProjectionError({ threadId })), getTimelinePage: (threadId) => Effect.fail(new OrchestratorProjectionError({ threadId })), getMessageCount: (threadId) => Effect.fail(new OrchestratorProjectionError({ threadId })), getTurnItem: ({ threadId }) => Effect.fail(new OrchestratorProjectionError({ threadId })), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index df123755cfc42abe6ce9cbf505bb737d37a3d413..86b2d503c6b237edd609ebf4ea05f9768a85a3af 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,3 +1,11 @@ +import { projectTurnItemForWire } from "./WireProjection.ts"; +import * as Stream from "effect/Stream"; +import { makeThreadFind, findProjectedThreadItems } from "./ThreadFind.ts"; +import type { + OrchestrationV2SearchThreadInput, + OrchestrationV2SearchThreadResult, + OrchestrationV2ThreadHistoryPage, +} from "@t3tools/contracts"; import { latestRootProviderFailure, latestUnheldRun, @@ -28,7 +36,6 @@ import type { ProviderThreadId, ProviderTurnId, RunAttemptId, - MessageId, } from "@t3tools/contracts"; import { OrchestrationV2AppThreadJson as OrchestrationV2AppThreadJsonSchema, @@ -49,6 +56,7 @@ import { OrchestrationV2TurnItemJson as OrchestrationV2TurnItemJsonSchema, orchestrationV2RunWorkStartedAt, RunId, + MessageId, RuntimeRequestId, CheckpointScopeId, ThreadId, @@ -74,8 +82,13 @@ import { MCP_APP_OUTPUT_KEY } from "@t3tools/shared/mcpApp"; import { threadHtmlRenderAttachmentIds } from "../attachmentStore.ts"; import { isThreadHistoryUserTurn, + isConversationHistoryItem, isThreadHistoryTurnStart, THREAD_HISTORY_MAX_RAW_TURNS, + selectHistoryPageFromCursor, + selectRecentTimelineWindow, + THREAD_HISTORY_PAGE_POLICY, + OLDER_THREAD_USER_TURN_LIMIT, } from "./threadHistoryPaging.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( @@ -318,9 +331,21 @@ export interface ShellSnapshotOptions { } export interface ProjectionStoreV2Shape { + readonly searchThreadStream: ( + input: OrchestrationV2SearchThreadInput, + ) => Stream.Stream; + readonly searchThread: ( + input: OrchestrationV2SearchThreadInput, + ) => Effect.Effect; readonly getThreadAttachmentIds: ( threadId: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; + readonly getThreadHistoryPage: ( + threadId: ThreadId, + cursor: string, + throughEntryId?: string, + conversationOnly?: boolean, + ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, options: ProjectionTimelinePageOptions, @@ -969,6 +994,11 @@ type ShellRunItemCountRow = { readonly item_count: number; }; +const encodeTimelineSources = Schema.encodeEffect( + Schema.fromJsonString( + Schema.Array(Schema.Struct({ threadId: Schema.String, id: Schema.String })), + ), +); const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema.String))); const encodeThreadPayload = Schema.encodeEffect( @@ -4706,12 +4736,17 @@ export const layer: Layer.Layer = ); // Only the timeline index is assembled across fork ancestry. Payloads are read - // after page selection, so a small MCP page never hydrates all tool output. + // after page selection, so history and MCP pages hydrate only the requested rows. type TimelineIndexItem = Pick< OrchestrationV2TurnItem, "id" | "threadId" | "runId" | "nodeId" | "type" > & { - readonly inputIntent?: "queued_turn"; + readonly inputIntent?: Extract< + OrchestrationV2TurnItem, + { type: "user_message" } + >["inputIntent"]; + readonly createdBy?: "user" | "agent" | "system"; + readonly messageId?: MessageId; }; type TimelineIndexRow = { readonly sourceThreadId: ThreadId; @@ -4738,9 +4773,13 @@ export const layer: Layer.Layer = run_id: string | null; node_id: string | null; type: OrchestrationV2TurnItem["type"]; - input_intent: string | null; + input_intent: NonNullable | null; + created_by: "user" | "agent" | "system" | null; + message_id: string | null; }>`SELECT turn_item_id, run_id, node_id, type, - CASE WHEN type = 'user_message' THEN json_extract(payload_json, '$.inputIntent') END AS input_intent + CASE WHEN type = 'user_message' THEN json_extract(payload_json, '$.inputIntent') END AS input_intent, + CASE WHEN type = 'user_message' THEN json_extract(payload_json, '$.createdBy') END AS created_by, + CASE WHEN type IN ('user_message', 'assistant_message') THEN json_extract(payload_json, '$.messageId') END AS message_id FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} ORDER BY ordinal ASC, turn_item_id ASC`; const items: Array = rows.map((row) => ({ @@ -4749,7 +4788,9 @@ export const layer: Layer.Layer = runId: row.run_id === null ? null : RunId.make(row.run_id), nodeId: row.node_id === null ? null : NodeId.make(row.node_id), type: row.type, - ...(row.input_intent === "queued_turn" ? { inputIntent: "queued_turn" as const } : {}), + ...(row.input_intent === null ? {} : { inputIntent: row.input_intent }), + ...(row.created_by === null ? {} : { createdBy: row.created_by }), + ...(row.message_id === null ? {} : { messageId: MessageId.make(row.message_id) }), })); const local: Array = items.map((item) => ({ sourceThreadId: threadId, @@ -4819,6 +4860,116 @@ export const layer: Layer.Layer = ), ); + const { searchThread, searchThreadStream } = yield* makeThreadFind((threadId) => + readTimelineIndex(threadId, new Set()).pipe(Effect.map((index) => index.visible)), + ); + + const getThreadHistoryPage: ProjectionStoreV2Shape["getThreadHistoryPage"] = ( + threadId, + cursor, + throughEntryId, + conversationOnly, + ) => + sql + .withTransaction( + Effect.gen(function* () { + const index = yield* readTimelineIndex(threadId, new Set()); + if (index.records.thread.deletedAt !== null) + return yield* new ProjectionStoreThreadNotFoundError({ threadId }); + const revisions = yield* sql<{ + sequence: number; + }>`SELECT COALESCE(MAX(sequence), 0) AS sequence + FROM orchestration_events WHERE application_event_version = 2 + AND aggregate_kind = 'thread' AND stream_id = ${threadId}`; + const snapshotSequence = revisions[0]?.sequence ?? 0; + // Choose complete turns from the light index before loading any bodies. + // A history page does not need nodes, provider turns, or other control state. + const page = yield* Effect.try({ + try: () => + selectHistoryPageFromCursor({ + items: index.visible.map((row, position) => ({ + ...row, + position, + canonicalPosition: position, + })), + cursor, + snapshotSequence, + rowEncodedBytes: () => 0, + throughEntryId, + }), + catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), + }); + const end = (page.items.at(-1)?.canonicalPosition ?? -1) + 1; + const hasUserTurns = index.visible + .slice(0, end) + .some((row) => isThreadHistoryUserTurn(row.item)); + const materialized = + conversationOnly && hasUserTurns + ? page.items.filter((row) => isConversationHistoryItem(row.item)) + : page.items; + const sources = yield* encodeTimelineSources( + materialized + .filter((row) => !row.synthetic) + .map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })), + ); + const payloads = yield* sql<{ + thread_id: string; + turn_item_id: string; + payload_json: string; + }>` + SELECT item.thread_id, item.turn_item_id, item.payload_json + FROM orchestration_v2_projection_turn_items item JOIN json_each(${sources}) wanted + ON item.thread_id = json_extract(wanted.value, '$.threadId') + AND item.turn_item_id = json_extract(wanted.value, '$.id')`; + const byIdentity = new Map( + payloads.map((row) => [`${row.thread_id}:${row.turn_item_id}`, row.payload_json]), + ); + const items = yield* Effect.forEach(materialized, (row) => + Effect.gen(function* () { + const payload = byIdentity.get(`${row.sourceThreadId}:${row.sourceItemId}`); + const item = + row.synthetic ?? + (payload === undefined + ? yield* new ProjectionStoreReadError({ threadId }) + : yield* decodeTurnItemPayload(payload)); + return { + position: row.canonicalPosition, + visibility: row.visibility, + sourceThreadId: row.sourceThreadId, + sourceItemId: row.sourceItemId, + item: projectTurnItemForWire(item), + }; + }), + ); + if (hasUserTurns) + return { + ...page, + snapshotSequence, + items: items.map((row, position) => ({ ...row, position })), + }; + // The no-turn fallback also respects the encoded byte budget after projection. + const bounded = selectRecentTimelineWindow({ + items, + snapshotSequence, + policy: { + ...THREAD_HISTORY_PAGE_POLICY, + maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, + maxItems: + THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2), + }, + }); + return { + snapshotSequence, + items: conversationOnly + ? bounded.items.filter((row) => isConversationHistoryItem(row.item)) + : bounded.items, + nextCursor: bounded.hasMoreHistory ? bounded.nextCursor : page.nextCursor, + hasMoreHistory: bounded.hasMoreHistory || page.hasMoreHistory, + }; + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const getTimelinePage: ProjectionStoreV2Shape["getTimelinePage"] = (threadId, options) => sql .withTransaction( @@ -5756,6 +5907,9 @@ export const layer: Layer.Layer = getThreadSnapshot, getThreadSnapshotWindow, getTimelinePage, + getThreadHistoryPage, + searchThread, + searchThreadStream, getThreadAttachmentIds, } satisfies ProjectionStoreV2Shape; }), @@ -6314,6 +6468,53 @@ export const layerMemory: Layer.Layer = Layer.effect( ), ), ), + searchThreadStream: (input) => Stream.fromEffect(service.searchThread(input)), + searchThread: (input) => + service.getThreadSnapshot(input.threadId).pipe( + Effect.flatMap(({ projection, snapshotSequence }) => + projection.thread.deletedAt !== null + ? Effect.fail(new ProjectionStoreThreadNotFoundError({ threadId: input.threadId })) + : Effect.succeed( + // This store has no project records, so only worktree paths resolve labels. + findProjectedThreadItems( + projection.visibleTurnItems, + input, + snapshotSequence, + projection.thread.worktreePath ?? undefined, + ), + ), + ), + ), + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + service.getThreadSnapshot(threadId).pipe( + Effect.flatMap(({ projection, snapshotSequence }) => + Effect.gen(function* () { + if (projection.thread.deletedAt !== null) + return yield* new ProjectionStoreThreadNotFoundError({ threadId }); + return yield* Effect.try({ + try: () => { + const page = selectHistoryPageFromCursor({ + items: projection.visibleTurnItems.map((row) => ({ + ...row, + item: projectTurnItemForWire(row.item), + })), + cursor, + throughEntryId, + snapshotSequence, + }); + return { + ...page, + snapshotSequence, + items: conversationOnly + ? page.items.filter((row) => isConversationHistoryItem(row.item)) + : page.items, + }; + }, + catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), + }); + }), + ), + ), getTimelinePage: (threadId, options) => service.getThreadProjection(threadId).pipe( Effect.map((projection) => { diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 09c80f66fa2cf857567e5fe9efa46316340cf15e..0808386da538865a3ea92d505ac3384372266a1d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -225,6 +225,9 @@ it.effect( getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), + searchThread: () => Effect.die("unused"), + searchThreadStream: () => Stream.empty, + getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..d5ec86dad6c44989ecaa2906711c44fc73a163c9 --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -0,0 +1,1012 @@ +import { assert, it, describe } from "@effect/vitest"; +import { + EventId, + MessageId, + PlanId, + RunId, + RuntimeRequestId, + TurnItemId, + type OrchestrationV2TurnItem, + type OrchestrationV2Run, + type OrchestrationV2SearchThreadResult, + type OrchestrationV2ThreadHistoryPage, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +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 { vi } from "vite-plus/test"; +import * as ThreadFindText from "@t3tools/shared/threadFindText"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProjectStore from "./ProjectStore.ts"; +import * as EventStore from "./EventStore.ts"; +import { encodeThreadHistoryCursor, selectHistoryPageFromCursor } from "./threadHistoryPaging.ts"; +import { projectTurnItemForWire } from "./WireProjection.ts"; + +const layerTest = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +const providerInstanceId = ProviderInstanceId.make("codex"); +const at = (minute: number) => DateTime.makeUnsafe(Date.UTC(2026, 8, 27, 0, minute)); + +const createProject = (projectId: ProjectId) => + Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) => + projects.apply({ + sequence: 0, + eventId: EventId.make(`created:${projectId}`), + aggregateKind: "project", + aggregateId: projectId, + occurredAt: DateTime.formatIso(at(0)), + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId, + title: projectId, + workspaceRoot: `/work/${projectId}`, + defaultModelSelection: null, + scripts: [], + createdAt: DateTime.formatIso(at(0)), + updatedAt: DateTime.formatIso(at(0)), + }, + }), + ); + +const thread = ( + threadId: ThreadId, + projectId: ProjectId, + overrides: { readonly archivedAt?: DateTime.Utc; readonly deletedAt?: DateTime.Utc } = {}, +): OrchestrationV2DomainEvent => ({ + id: EventId.make(`created:${threadId}`), + type: "thread.created", + threadId, + providerInstanceId, + occurredAt: at(0), + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: threadId, + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: at(0), + updatedAt: at(0), + archivedAt: overrides.archivedAt ?? null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: overrides.deletedAt ?? null, + }, +}); + +const threadId = ThreadId.make("thread:find"); +const projectId = ProjectId.make("project:find"); +const runId = RunId.make("run:find"); + +function item( + id: string, + ordinal: number, + text: string, + kind: "user_message" | "assistant_message" | "proposed_plan" = "assistant_message", + owner = threadId, +): Extract< + OrchestrationV2TurnItem, + { type: "user_message" | "assistant_message" | "proposed_plan" } +> { + const base = { + id: TurnItemId.make(id), + threadId: owner, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed" as const, + title: null, + // Creation time deliberately opposes the canonical item order. + startedAt: at(100 - ordinal), + completedAt: at(100 - ordinal), + updatedAt: at(1), + }; + if (kind === "proposed_plan") + return { ...base, type: kind, planId: PlanId.make(id), markdown: text, streaming: false }; + if (kind === "user_message") + return { + ...base, + type: kind, + messageId: MessageId.make(id), + text, + inputIntent: "turn_start", + attachments: [], + createdBy: "user", + creationSource: "web", + }; + return { ...base, type: kind, messageId: MessageId.make(id), text, streaming: false }; +} + +function run( + owner = threadId, + id = runId, + ordinal = 1, + status: OrchestrationV2Run["status"] = "completed", +): Extract { + return { + id: EventId.make(`run:${owner}:${id}:${status}`), + threadId: owner, + occurredAt: at(1), + type: "run.created", + payload: { + id, + threadId: owner, + ordinal, + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-6" }, + providerThreadId: null, + userMessageId: MessageId.make("prompt"), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: at(1), + startedAt: at(1), + completedAt: at(2), + checkpointId: null, + contextHandoffId: null, + }, + }; +} + +const commit = Effect.fn("ThreadFindTest.commit")(function* ( + events: readonly OrchestrationV2DomainEvent[], +) { + const store = yield* EventStore.EventStoreV2; + const projection = yield* ProjectionStore.ProjectionStoreV2; + yield* store.append({ events }); + yield* Effect.forEach(events, projection.apply, { discard: true }); +}); +const putItems = (items: readonly OrchestrationV2TurnItem[], version = "initial") => + commit( + items.map((payload) => ({ + id: EventId.make(`${payload.threadId}:${payload.id}:${version}`), + threadId: payload.threadId, + occurredAt: at(1), + type: "turn-item.updated" as const, + payload, + })), + ); +const setup = Effect.gen(function* () { + yield* createProject(projectId); + yield* commit([thread(threadId, projectId), run()]); + return yield* ProjectionStore.ProjectionStoreV2; +}); + +describe("V2 thread find", () => { + it.effect("delivers the reading-position match before parsing the remaining history", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("before", 0, "before progressive needle needle"), + item("reading", 1, "reading progressive needle needle"), + item("after", 2, "after progressive needle"), + ]); + const parse = vi.spyOn(ThreadFindText, "searchableMessageSegments"); + const frames: OrchestrationV2SearchThreadResult[] = []; + try { + yield* projection + .searchThreadStream({ + threadId, + query: "needle", + start: { entryId: "reading", occurrence: 1 }, + }) + .pipe( + Stream.runForEach((frame) => + Effect.sync(() => { + frames.push(frame); + if (frame.complete === false) { + assert.deepEqual(frame.match, { entryId: "reading", runId, occurrence: 1 }); + assert.equal(parse.mock.calls.length, 1); + assert.equal(parse.mock.calls[0]![0].text, "reading progressive needle needle"); + } + }), + ), + ); + assert.equal(frames.length, 2); + assert.notEqual(frames[1]!.complete, false); + assert.deepEqual(frames[1]!.match, frames[0]!.match); + assert.equal(frames[1]!.totalMatches, 5); + assert.equal(frames[1]!.activeIndex, 3); + const cached = yield* projection + .searchThreadStream({ threadId, query: "needle" }) + .pipe(Stream.runCollect); + assert.equal(cached.length, 1); + assert.notEqual(cached[0]!.complete, false); + } finally { + parse.mockRestore(); + } + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("wraps from an exhausted bottom anchor and cancels without counting the rest", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("first", 0, "cancel progressive needle"), + item("later", 1, "cancel later needle needle"), + item("bottom", 2, "cancel bottom needle"), + ]); + const parse = vi.spyOn(ThreadFindText, "searchableMessageSegments"); + try { + const frames = yield* projection + .searchThreadStream({ + threadId, + query: "needle", + start: { entryId: "bottom", occurrence: 1 }, + }) + .pipe(Stream.take(1), Stream.runCollect); + assert.equal(frames[0]!.complete, false); + assert.equal(frames[0]!.match?.entryId, "first"); + assert.deepEqual( + parse.mock.calls.map(([message]) => message.text), + ["cancel bottom needle", "cancel progressive needle"], + ); + const complete = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(complete.totalMatches, 4); + assert.equal(parse.mock.calls.length, 3); + } finally { + parse.mockRestore(); + } + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("finishes one snapshot while later searches include newly inserted matches", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("selected", 1, "snapshot needle"), + item("last", 2, "snapshot last needle"), + ]); + const frames = yield* projection.searchThreadStream({ threadId, query: "needle" }).pipe( + Stream.tap((frame) => + frame.complete === false + ? putItems([item("inserted", 0, "snapshot inserted needle")], "during-search") + : Effect.void, + ), + Stream.runCollect, + ); + assert.equal(frames.length, 2); + assert.equal(frames[1]!.totalMatches, 2); + assert.deepEqual(frames[1]!.match, frames[0]!.match); + const updated = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "selected", occurrence: 0 }, + }); + assert.equal(updated.totalMatches, 3); + assert.equal(updated.activeIndex, 1); + assert.equal(updated.match?.entryId, "selected"); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("returns only a final frame for empty searches and relative navigation", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([item("only", 0, "needle needle")]); + const empty = yield* projection + .searchThreadStream({ threadId, query: "missing" }) + .pipe(Stream.runCollect); + assert.equal(empty.length, 1); + assert.equal(empty[0]!.totalMatches, 0); + assert.notEqual(empty[0]!.complete, false); + const relative = yield* projection + .searchThreadStream({ threadId, query: "needle", offset: 1 }) + .pipe(Stream.runCollect); + assert.equal(relative.length, 1); + assert.equal(relative[0]!.match?.occurrence, 1); + assert.notEqual(relative[0]!.complete, false); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("returns bounded navigation counts around the selection, including wraparound", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems( + Array.from({ length: 30 }, (_, i) => + item(`message:${i}`, i, "needle ".repeat((i % 3) + 1)), + ), + ); + const current = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "message:14", occurrence: 1 }, + }); + assert.equal(current.totalMatches, 60); + assert.equal(current.activeIndex, 28); + assert.equal(current.navigation?.length, 17); + assert.deepEqual( + current.navigation?.map((entry) => entry.entryId), + Array.from({ length: 17 }, (_, i) => `message:${i + 6}`), + ); + assert.deepEqual( + current.navigation?.find((entry) => entry.entryId === "message:14"), + { + entryId: "message:14", + runId, + startIndex: 27, + count: 3, + }, + ); + const first = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(first.navigation?.length, 17); + assert.include(first.navigation?.map((entry) => entry.entryId) ?? [], "message:29"); + const empty = yield* projection.searchThread({ threadId, query: "absent" }); + assert.deepEqual(empty.navigation, []); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("keeps navigation compact for a message with thousands of occurrences", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([item("many", 1, "needle ".repeat(5000))]); + const result = yield* projection.searchThread({ threadId, query: "needle", index: 2000 }); + assert.equal(result.totalMatches, 5000); + assert.equal(result.match?.occurrence, 2000); + assert.deepEqual(result.navigation, [{ entryId: "many", runId, startIndex: 0, count: 5000 }]); + assert.isBelow(JSON.stringify(result).length, 500); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("cycles relative to the current match when new matches are inserted", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("first", 10, "needle"), + item("current", 20, "needle"), + item("last", 30, "needle"), + ]); + const initial = yield* projection.searchThread({ threadId, query: "needle", index: 1 }); + assert.equal(initial.match?.entryId, "current"); + yield* putItems([ + item("new-before", 15, "needle"), + item("new-after", 25, "needle"), + item("new-last", 40, "needle"), + ]); + const start = { entryId: "current", occurrence: 0 }; + const refreshed = yield* projection.searchThread({ threadId, query: "needle", start }); + assert.equal(refreshed.totalMatches, 6); + assert.equal(refreshed.activeIndex, 2); + assert.equal(refreshed.match?.entryId, "current"); + const next = yield* projection.searchThread({ threadId, query: "needle", start, offset: 1 }); + assert.equal(next.match?.entryId, "new-after"); + const previous = yield* projection.searchThread({ + threadId, + query: "needle", + start, + offset: -1, + }); + assert.equal(previous.match?.entryId, "new-before"); + const end = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "last", occurrence: 0 }, + offset: 1, + }); + assert.equal(end.match?.entryId, "new-last"); + const wrapped = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "new-last", occurrence: 0 }, + offset: 1, + }); + assert.equal(wrapped.match?.entryId, "first"); + const backwards = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "first", occurrence: 0 }, + offset: -1, + }); + assert.equal(backwards.match?.entryId, "new-last"); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("reuses parses of large messages across navigation, edits, and new queries", () => + Effect.gen(function* () { + const projection = yield* setup; + const text = `${"large message ".repeat(2600)}needle`; + yield* putItems(Array.from({ length: 8 }, (_, i) => item(`large:${i}`, i + 1, text))); + const parse = vi.spyOn(ThreadFindText, "searchableMessageSegments"); + try { + const first = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(first.totalMatches, 8); + // Identical bodies share one parse. + assert.equal(parse.mock.calls.length, 1); + parse.mockClear(); + const next = yield* projection.searchThread({ threadId, query: "needle", index: 4 }); + assert.equal(next.match?.entryId, "large:4"); + assert.equal(next.snapshotSequence, first.snapshotSequence); + assert.equal(parse.mock.calls.length, 0); + yield* putItems([item("large:4", 5, `${text} needle`)], "changed"); + const changed = yield* projection.searchThread({ threadId, query: "needle", index: 5 }); + assert.equal(changed.totalMatches, 9); + assert.deepEqual(changed.match, { entryId: "large:4", runId, occurrence: 1 }); + // Only the edited message is parsed again. + assert.equal(parse.mock.calls.length, 1); + parse.mockClear(); + const retyped = yield* projection.searchThread({ threadId, query: "large message" }); + assert.equal(retyped.totalMatches, 8 * 2600); + assert.equal(parse.mock.calls.length, 0); + } finally { + parse.mockRestore(); + } + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("searches visible rendered messages and plans in canonical item order", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("user", 1, "Check **the build**", "user_message"), + item("assistant", 2, "check the build twice: check the build"), + item("plan-item", 3, "# Check the build\nShip it", "proposed_plan"), + item("url", 4, "[documentation](https://hidden.example/needle)"), + ]); + const matches = []; + for (let index = 0; index < 4; index++) { + const result = yield* projection.searchThread({ + threadId, + query: "CHECK THE BUILD", + index, + }); + assert.equal(result.totalMatches, 4); + matches.push(result.match); + } + assert.deepEqual(matches, [ + { entryId: "user", runId, occurrence: 0 }, + { entryId: "assistant", runId, occurrence: 0 }, + { entryId: "assistant", runId, occurrence: 1 }, + { entryId: "plan-item", runId, occurrence: 0 }, + ]); + assert.equal( + (yield* projection.searchThread({ threadId, query: "hidden.example" })).totalMatches, + 0, + ); + assert.equal( + (yield* projection.searchThread({ threadId, query: "check", index: 99 })).activeIndex, + 3, + ); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("keeps a search snapshot consistent while another thread commits", () => + Effect.gen(function* () { + const projection = yield* setup; + const other = ThreadId.make("thread:busy"); + yield* commit([thread(other, projectId), run(other, RunId.make("run:busy"))]); + yield* putItems(Array.from({ length: 260 }, (_, i) => item(`message:${i}`, i, "needle"))); + const before = yield* projection.searchThread({ threadId, query: "needle" }); + const [during] = yield* Effect.all( + [ + projection.searchThread({ threadId, query: "needle", index: 259 }), + putItems([item("busy", 1, "busy", "assistant_message", other)]), + ], + { concurrency: "unbounded" }, + ); + assert.equal(during.totalMatches, 260); + assert.equal(during.match?.entryId, "message:259"); + assert.equal(during.snapshotSequence, before.snapshotSequence); + assert.equal( + (yield* projection.searchThread({ threadId, query: "needle" })).snapshotSequence, + before.snapshotSequence, + ); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("searches displayed skill labels and invalidates parsed text when labels change", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([item("skill", 1, "Use $test-t3-app now")]); + const input = { + threadId, + query: "T3 App Testing", + skills: [{ name: "test-t3-app", displayName: "T3 App Testing" }], + }; + assert.equal((yield* projection.searchThread(input)).totalMatches, 1); + assert.equal((yield* projection.searchThread({ ...input, skills: [] })).totalMatches, 0); + // Plans render `$skill` tokens literally, so they match as written, not by label. + yield* putItems( + [item("skill-plan", 2, "# Verify\nUse $test-t3-app now", "proposed_plan")], + "plan", + ); + assert.equal((yield* projection.searchThread(input)).totalMatches, 1); + assert.equal( + (yield* projection.searchThread({ ...input, query: "$test-t3-app" })).totalMatches, + 1, + ); + // The message label "T3 App Testing" plus the plan's literal "$test-t3-app". + assert.equal((yield* projection.searchThread({ ...input, query: "App" })).totalMatches, 2); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("never mixes message versions when the searched thread changes during a scan", () => + Effect.gen(function* () { + const projection = yield* setup; + const sql = yield* SqlClient.SqlClient; + const items = Array.from({ length: 260 }, (_, i) => item(`message:${i}`, i, "needle")); + yield* putItems(items); + const updated = items.map((item) => + item.type === "assistant_message" ? { ...item, text: "needle needle" } : item, + ); + const [during] = yield* Effect.all( + [ + projection.searchThread({ threadId, query: "needle" }), + sql.withTransaction(putItems(updated, "concurrent")), + ], + { concurrency: "unbounded" }, + ); + assert.include([260, 520], during.totalMatches); + assert.equal( + (yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, + 520, + ); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("starts at the reading position and wraps when no later match exists", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems([ + item("first", 1, "needle"), + item("reading", 2, "needle above, needle below"), + item("between", 3, "nothing"), + item("last", 4, "needle"), + item("end", 5, "nothing"), + ]); + const from = (entryId: string, occurrence = 0) => + projection.searchThread({ threadId, query: "needle", start: { entryId, occurrence } }); + assert.equal((yield* from("reading")).activeIndex, 1); + const below = yield* from("reading", 1); + assert.equal(below.activeIndex, 2); + assert.equal(below.match?.occurrence, 1); + assert.equal((yield* from("between")).match?.entryId, "last"); + assert.equal((yield* from("reading", 2)).activeIndex, 3); + assert.equal((yield* from("end")).match?.entryId, "first"); + assert.equal((yield* from("missing")).activeIndex, 0); + const explicit = yield* projection.searchThread({ + threadId, + query: "needle", + start: { entryId: "last", occurrence: 0 }, + index: 1, + }); + assert.equal(explicit.match?.entryId, "reading"); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("finds matches beyond the recent history window", () => + Effect.gen(function* () { + const projection = yield* setup; + const items = Array.from({ length: 520 }, (_, i) => + item(`message:${i}`, i + 1, i === 8 || i === 510 ? "sentinel" : "other message"), + ); + yield* putItems(items); + const first = yield* projection.searchThread({ threadId, query: "sentinel" }); + const last = yield* projection.searchThread({ threadId, query: "sentinel", index: 1 }); + assert.equal(first.totalMatches, 2); + assert.equal(first.match?.entryId, "message:8"); + assert.equal(last.match?.entryId, "message:510"); + }).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; + const original = item("streaming", 1, "needle"); + yield* putItems([original]); + assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1); + if (original.type !== "assistant_message") return; + yield* putItems([{ ...original, text: "needle needle", streaming: true }], "streaming"); + assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 2); + const updated = run(threadId, runId, 1, "rolled_back"); + yield* commit([{ ...updated, type: "run.updated" }]); + assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 0); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect( + "does not search undispatched inputs, cancelled queue rows, or folded question answers", + () => + Effect.gen(function* () { + const projection = yield* setup; + const queuedRunId = RunId.make("queued:cancelled"); + yield* commit([run(threadId, queuedRunId, 2, "cancelled")]); + const queued = item("cancelled", 1, "needle", "user_message"); + if (queued.type !== "user_message") return; + const answer = item("async-answer:question", 2, "needle", "user_message"); + const question: OrchestrationV2TurnItem = { + ...item("question", 3, ""), + type: "user_input_request", + requestId: RuntimeRequestId.make("question"), + questions: [], + questionAnswer: { requestId: "question", answers: {}, attachmentsByQuestionId: {} }, + }; + yield* putItems([ + { ...queued, runId: queuedRunId, inputIntent: "queued_turn" }, + answer, + question, + ]); + const event: OrchestrationV2DomainEvent = { + id: EventId.make("undispatched"), + threadId, + occurredAt: at(2), + type: "message.updated", + payload: { + id: MessageId.make("undispatched"), + threadId, + runId: null, + nodeId: null, + role: "user", + text: "needle", + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: at(2), + updatedAt: at(2), + }, + }; + yield* commit([event]); + const result = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(result.totalMatches, 0); + assert.isNull(result.match); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect( + "searches inherited fork history through the fork boundary, using source item identities", + () => + Effect.gen(function* () { + const projection = yield* setup; + const parent = ThreadId.make("thread:parent"); + const laterRun = RunId.make("run:later"); + const parentRun = RunId.make("run:parent"); + yield* commit([ + thread(parent, projectId), + run(parent, parentRun), + run(parent, laterRun, 2), + ]); + yield* putItems([ + { ...item("inherited", 1, "needle", "assistant_message", parent), runId: parentRun }, + { ...item("after-fork", 2, "needle", "assistant_message", parent), runId: laterRun }, + ]); + const child = thread(threadId, projectId); + if (child.type !== "thread.created") return; + yield* commit([ + { + ...child, + id: EventId.make("child:fork"), + type: "thread.metadata-updated", + payload: { + ...child.payload, + forkedFrom: { type: "run", threadId: parent, runId: parentRun }, + }, + }, + ]); + yield* putItems([item("local", 1, "needle")]); + const inherited = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(inherited.totalMatches, 2); + assert.equal(inherited.match?.entryId, "inherited"); + assert.equal( + (yield* projection.searchThread({ threadId, query: "needle", index: 1 })).match?.entryId, + "local", + ); + const snapshot = yield* projection.getThreadSnapshot(threadId); + const history = yield* projection.getThreadHistoryPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "local", + position: 2, + }), + ); + assert.deepEqual( + history.items, + snapshot.projection.visibleTurnItems + .slice(0, -1) + .map((row) => ({ ...row, item: projectTurnItemForWire(row.item) })), + ); + assert.equal(history.items[0]?.sourceThreadId, parent); + assert.equal(history.items.at(-1)?.item.type, "fork"); + // Parent updates invalidate a child's cached inherited result too. + yield* putItems( + [{ ...item("inherited", 1, "gone", "assistant_message", parent), runId: parentRun }], + "parent-update", + ); + assert.equal( + (yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, + 1, + ); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("rejects deleted threads and keeps results scoped to their thread", () => + Effect.gen(function* () { + const projection = yield* setup; + const other = ThreadId.make("thread:other"); + yield* commit([thread(other, projectId), run(other, RunId.make("run:other"))]); + yield* putItems([ + item("shared-id", 1, "needle"), + item("other-id", 1, "other text", "assistant_message", other), + ]); + assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1); + assert.equal( + (yield* projection.searchThread({ threadId: other, query: "needle" })).totalMatches, + 0, + ); + const deleted = thread(threadId, projectId, { deletedAt: at(3) }); + if (deleted.type !== "thread.created") return; + yield* commit([{ ...deleted, id: EventId.make("deleted"), type: "thread.deleted" }]); + const error = yield* Effect.flip(projection.searchThread({ threadId, query: "needle" })); + assert.equal(error._tag, "ProjectionStoreThreadNotFoundError"); + const historyError = yield* Effect.flip( + projection.getThreadHistoryPage( + threadId, + encodeThreadHistoryCursor({ + snapshotSequence: 0, + sourceThreadId: threadId, + sourceItemId: "shared-id", + position: 0, + }), + ), + ); + assert.equal(historyError._tag, "ProjectionStoreThreadNotFoundError"); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("invalidates a cached fork search when its empty source gains its first message", () => + Effect.gen(function* () { + const projection = yield* setup; + const parent = ThreadId.make("thread:empty-parent"); + const parentRun = RunId.make("run:empty-parent"); + yield* commit([thread(parent, projectId), run(parent, parentRun)]); + const child = thread(threadId, projectId); + if (child.type !== "thread.created") return; + yield* commit([ + { + ...child, + id: EventId.make("child:empty-fork"), + type: "thread.metadata-updated", + payload: { + ...child.payload, + forkedFrom: { type: "run", threadId: parent, runId: parentRun }, + }, + }, + ]); + yield* putItems([item("local", 1, "needle")]); + assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1); + yield* putItems([ + { ...item("inherited", 1, "needle", "assistant_message", parent), runId: parentRun }, + ]); + const updated = yield* projection.searchThread({ threadId, query: "needle" }); + assert.equal(updated.totalMatches, 2); + assert.equal(updated.match?.entryId, "inherited"); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("memory store resolves worktree file labels and rejects deleted threads", () => + Effect.gen(function* () { + const projection = yield* ProjectionStore.ProjectionStoreV2; + const created = thread(threadId, projectId); + if (created.type !== "thread.created") return; + const withWorktree = { + ...created, + payload: { ...created.payload, worktreePath: "/work/tree" }, + }; + const message = item("file", 1, "See `docs/notes.md:12`."); + yield* Effect.forEach( + [ + withWorktree, + run(), + { + id: EventId.make("file"), + threadId, + occurredAt: at(1), + type: "turn-item.updated", + payload: message, + }, + ] satisfies OrchestrationV2DomainEvent[], + projection.apply, + { discard: true }, + ); + assert.equal( + (yield* projection.searchThread({ threadId, query: "notes.md · L12" })).totalMatches, + 1, + ); + const deleted = thread(threadId, projectId, { deletedAt: at(3) }); + if (deleted.type !== "thread.created") return; + yield* projection.apply({ ...deleted, id: EventId.make("deleted"), type: "thread.deleted" }); + const error = yield* Effect.flip(projection.searchThread({ threadId, query: "notes" })); + assert.equal(error._tag, "ProjectionStoreThreadNotFoundError"); + }).pipe(Effect.provide(ProjectionStore.layerMemory)), + ); +}); + +describe("V2 history materialization", () => { + it.effect( + "combines history pages through a match's message identity without skipping intervening turns", + () => + Effect.gen(function* () { + const projection = yield* setup; + const items = Array.from({ length: 45 }, (_, turn) => [ + item(`user:${turn}`, turn * 3, `turn ${turn}`, "user_message"), + { + ...item(`answer:${turn}`, turn * 3 + 1, `answer ${turn}`), + messageId: MessageId.make(`message:${turn}`), + }, + { + ...item(`tool:${turn}`, turn * 3 + 2, ""), + type: "command_execution" as const, + input: "pwd", + output: "huge output".repeat(1000), + exitCode: 0, + }, + ]).flat(); + yield* putItems(items); + const snapshot = yield* projection.getThreadSnapshot(threadId); + const cursor = encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "absent", + position: items.length, + }); + const ordinary = yield* projection.getThreadHistoryPage(threadId, cursor); + assert.equal(ordinary.items[0]?.sourceItemId, "user:25"); + const targeted = yield* projection.getThreadHistoryPage(threadId, cursor, "message:10"); + assert.equal(targeted.items[0]?.sourceItemId, "user:10"); + assert.deepEqual( + targeted.items.map((row) => row.sourceItemId), + items.slice(30).map((entry) => entry.id), + ); + const conversation = yield* projection.getThreadHistoryPage( + threadId, + cursor, + "message:10", + true, + ); + assert.equal(conversation.nextCursor, targeted.nextCursor); + assert.equal(conversation.hasMoreHistory, targeted.hasMoreHistory); + assert.deepEqual( + conversation.items.map((row) => row.sourceItemId), + targeted.items + .filter((row) => row.item.type !== "command_execution") + .map((row) => row.sourceItemId), + ); + assert.equal(conversation.items.length, 70); + assert.equal(targeted.items.length, 105); + assert.isTrue(targeted.hasMoreHistory); + const earlier = yield* projection.getThreadHistoryPage(threadId, targeted.nextCursor!); + assert.deepEqual( + earlier.items.map((row) => row.sourceItemId), + items.slice(0, 30).map((entry) => entry.id), + ); + assert.isFalse(earlier.hasMoreHistory); + const capped = yield* projection.getThreadHistoryPage(threadId, cursor, "message:0"); + assert.equal(capped.items[0]?.sourceItemId, "user:5"); + const absent = yield* projection.getThreadHistoryPage( + threadId, + cursor, + "another-thread-message", + ); + assert.deepEqual(absent, ordinary); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("pages complete turns without duplicating or skipping messages", () => + Effect.gen(function* () { + const projection = yield* setup; + const items = Array.from({ length: 45 }, (_, turn) => [ + item(`user:${turn}`, turn * 3, `turn ${turn}`, "user_message"), + { + ...item(`steer:${turn}`, turn * 3 + 1, "steer", "user_message"), + inputIntent: "steer" as const, + }, + item( + `answer:${turn}`, + turn * 3 + 2, + turn === 40 ? "large ".repeat(200_000) : `answer ${turn}`, + ), + ]).flat(); + yield* putItems(items); + const snapshot = yield* projection.getThreadSnapshot(threadId); + const rows = snapshot.projection.visibleTurnItems.map((row) => ({ + ...row, + item: projectTurnItemForWire(row.item), + })); + let cursor: string | null = encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "absent", + position: rows.length, + }); + const collected: string[] = []; + while (cursor !== null) { + const page: OrchestrationV2ThreadHistoryPage = yield* projection.getThreadHistoryPage( + threadId, + cursor, + ); + assert.deepEqual(page, { + snapshotSequence: snapshot.snapshotSequence, + ...selectHistoryPageFromCursor({ + items: rows, + cursor, + snapshotSequence: snapshot.snapshotSequence, + }), + }); + assert.equal(page.items[0]?.item.type, "user_message"); + collected.unshift(...page.items.map((row) => String(row.sourceItemId))); + cursor = page.nextCursor; + } + assert.deepEqual( + collected, + items.map((entry) => String(entry.id)), + ); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("enforces fallback item and byte limits and always admits an oversized row", () => + Effect.gen(function* () { + const projection = yield* setup; + yield* putItems( + Array.from({ length: 80 }, (_, ordinal) => + item(`row:${ordinal}`, ordinal, ordinal < 78 ? "small" : "large ".repeat(200_000)), + ), + ); + const snapshot = yield* projection.getThreadSnapshot(threadId); + const rows = snapshot.projection.visibleTurnItems.map((row) => ({ + ...row, + item: projectTurnItemForWire(row.item), + })); + let cursor: string | null = encodeThreadHistoryCursor({ + snapshotSequence: snapshot.snapshotSequence, + sourceThreadId: threadId, + sourceItemId: "absent", + position: rows.length, + }); + const sizes: number[] = []; + while (cursor !== null) { + const page: OrchestrationV2ThreadHistoryPage = yield* projection.getThreadHistoryPage( + threadId, + cursor, + ); + assert.deepEqual(page, { + snapshotSequence: snapshot.snapshotSequence, + ...selectHistoryPageFromCursor({ + items: rows, + cursor, + snapshotSequence: snapshot.snapshotSequence, + }), + }); + sizes.push(page.items.length); + cursor = page.nextCursor; + } + assert.deepEqual(sizes, [1, 1, 75, 3]); + }).pipe(Effect.provide(layerTest)), + ); +}); diff --git a/apps/server/src/orchestration-v2/ThreadFind.ts b/apps/server/src/orchestration-v2/ThreadFind.ts new file mode 100644 index 0000000000000000000000000000000000000000..a764da8dd5f2dd3229448950e78ba5466c9aa0ae --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadFind.ts @@ -0,0 +1,449 @@ +import { + OrchestrationV2TurnItemJson, + type OrchestrationV2ProjectedTurnItem, + OrchestrationV2SearchThreadInput, + type OrchestrationV2SearchThreadResult, + type OrchestrationV2TurnItem, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import { searchableMessageSegments, searchablePlanSegments } from "@t3tools/shared/threadFindText"; +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 Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; +import { + ProjectionStoreReadError, + ProjectionStoreThreadNotFoundError, + type ProjectionStoreV2Error, +} from "./ProjectionStore.ts"; + +interface FindRow { + readonly position: number; + readonly visibility: OrchestrationV2ProjectedTurnItem["visibility"]; + readonly sourceThreadId: ThreadId; + readonly sourceItemId: TurnItemId; +} +interface FindPayload extends FindRow { + readonly payload: string; + readonly entryId: string; + readonly type: OrchestrationV2TurnItem["type"]; +} +interface FindDocument extends FindRow { + readonly entryId: string; + readonly runId: OrchestrationV2TurnItem["runId"]; + readonly count: number; +} +class TextKey extends Data.Class<{ + text: string; + role: "user" | "assistant" | "plan"; + streaming: boolean; + hasContext: boolean; + cwd: string | undefined; + skills: NonNullable; +}> {} + +function textKey( + item: OrchestrationV2TurnItem, + cwd: string | undefined, + skills: TextKey["skills"], +) { + return new TextKey({ + text: + item.type === "proposed_plan" + ? item.markdown + : item.type === "user_message" || item.type === "assistant_message" + ? item.text + : "", + role: + item.type === "proposed_plan" ? "plan" : item.type === "user_message" ? "user" : "assistant", + streaming: item.type === "assistant_message" && item.streaming, + hasContext: item.type === "user_message" && item.context !== undefined, + cwd, + skills, + }); +} + +function parseText(key: TextKey): readonly string[] { + return key.role === "plan" + ? searchablePlanSegments(key.text, key.cwd) + : (searchableMessageSegments( + { + role: key.role, + text: key.text, + streaming: key.streaming, + context: key.hasContext ? { version: 1, records: [] } : undefined, + }, + key.cwd, + key.skills, + ) ?? []); +} + +function searchableRows(items: readonly OrchestrationV2ProjectedTurnItem[]) { + const foldedAnswers = new Set( + items.flatMap(({ item }) => + item.type === "user_input_request" && item.questionAnswer + ? [`async-answer:${item.questionAnswer.requestId}`] + : [], + ), + ); + return items.filter( + ({ item }) => + ["user_message", "assistant_message", "proposed_plan"].includes(item.type) && + !(item.type === "user_message" && foldedAnswers.has(item.messageId)), + ); +} + +function documentFor(row: OrchestrationV2ProjectedTurnItem, count: number): FindDocument { + return { + position: row.position, + visibility: row.visibility, + sourceThreadId: row.sourceThreadId, + sourceItemId: row.sourceItemId, + entryId: + row.item.type === "user_message" || row.item.type === "assistant_message" + ? row.item.messageId + : row.item.id, + runId: row.item.runId, + count, + }; +} + +function countSegments(segments: readonly string[], query: string) { + return segments.reduce((sum, text) => sum + countThreadSearchOccurrences(text, query), 0); +} + +function selectMatch( + documents: readonly FindDocument[], + input: Pick, +) { + const totalMatches = documents.reduce((sum, doc) => sum + doc.count, 0); + let requestedIndex = input.index ?? 0; + if (input.index === undefined && input.start) { + const startIndex = documents.findIndex((doc) => doc.entryId === input.start?.entryId); + if (startIndex >= 0) { + requestedIndex = + documents.slice(0, startIndex).reduce((sum, doc) => sum + doc.count, 0) + + Math.min(input.start.occurrence, documents[startIndex]!.count); + if (requestedIndex >= totalMatches) requestedIndex = 0; + } + } + const activeIndex = + input.index === undefined && totalMatches > 0 + ? (((requestedIndex + (input.offset ?? 0)) % totalMatches) + totalMatches) % totalMatches + : Math.min(requestedIndex, Math.max(0, totalMatches - 1)); + let occurrence = activeIndex; + for (const document of documents) { + if (occurrence < document.count) return { totalMatches, activeIndex, document, occurrence }; + occurrence -= document.count; + } + return { totalMatches, activeIndex, document: null, occurrence: 0 }; +} + +/** The memory projection uses the same match ordering as SQLite. */ +export function findProjectedThreadItems( + items: readonly OrchestrationV2ProjectedTurnItem[], + input: OrchestrationV2SearchThreadInput, + snapshotSequence: number, + cwd?: string, +): OrchestrationV2SearchThreadResult { + const rows = searchableRows(items); + const docs = rows.map((row) => + documentFor( + row, + countSegments(parseText(textKey(row.item, cwd, input.skills ?? [])), input.query), + ), + ); + return resultForSelection(selectMatch(docs, input), docs, snapshotSequence); +} + +function resultForSelection( + selected: ReturnType, + documents: readonly FindDocument[], + snapshotSequence: number, +): OrchestrationV2SearchThreadResult { + let startIndex = 0; + const entries = documents.flatMap(({ entryId, runId, count }) => { + const entry = { entryId, runId, count, startIndex }; + startIndex += count; + return count > 0 ? [entry] : []; + }); + const selectedEntry = entries.findIndex((entry) => entry.entryId === selected.document?.entryId); + const navigation = + entries.length <= 17 + ? entries + : Array.from( + { length: 17 }, + (_, i) => entries[(selectedEntry + i - 8 + entries.length) % entries.length]!, + ); + return { + snapshotSequence, + totalMatches: selected.totalMatches, + activeIndex: selected.activeIndex, + navigation, + match: + selected.document === null + ? null + : { + entryId: selected.document.entryId, + runId: selected.document.runId, + occurrence: selected.occurrence, + }, + }; +} + +/** Read the canonical index once per scan, including inherited fork rows, without tool bodies. */ +export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( + readIndex: (threadId: ThreadId) => Effect.Effect< + readonly (Omit & { + readonly item: Pick; + })[], + ProjectionStoreV2Error + >, +) { + const sql = yield* SqlClient.SqlClient; + const encodeCacheKey = Schema.encodeEffect( + Schema.fromJsonString( + Schema.Struct({ + threadId: ThreadId, + cwd: Schema.optional(Schema.String), + query: Schema.String, + skills: OrchestrationV2SearchThreadInput.fields.skills, + }), + ), + ); + const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId))); + const encodeSources = Schema.encodeEffect( + Schema.fromJsonString(Schema.Array(Schema.Struct({ threadId: ThreadId, id: TurnItemId }))), + ); + const decodeItem = Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2TurnItemJson)); + const isSnapshotError = Schema.is( + Schema.Union([ProjectionStoreReadError, ProjectionStoreThreadNotFoundError]), + ); + const load = Effect.fn("ThreadFind.load")(function* ( + threadId: ThreadId, + rows: readonly FindRow[], + ) { + if (rows.length === 0) return []; + const sources = yield* encodeSources( + rows.map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })), + ).pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause }))); + const payloads = yield* sql<{ + thread_id: string; + turn_item_id: string; + payload_json: string; + entry_id: string; + item_type: OrchestrationV2TurnItem["type"]; + }>` + SELECT item.thread_id, item.turn_item_id, item.payload_json, + COALESCE(json_extract(item.payload_json, '$.messageId'), item.turn_item_id) AS entry_id, + json_extract(item.payload_json, '$.type') AS item_type + FROM orchestration_v2_projection_turn_items AS item + JOIN json_each(${sources}) AS wanted + ON item.thread_id = json_extract(wanted.value, '$.threadId') + AND item.turn_item_id = json_extract(wanted.value, '$.id') + `.pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause }))); + const byThread = new Map>(); + for (const row of payloads) { + let items = byThread.get(row.thread_id); + if (!items) byThread.set(row.thread_id, (items = new Map())); + items.set(row.turn_item_id, row); + } + return yield* Effect.forEach(rows, (row) => { + const payload = byThread.get(row.sourceThreadId)?.get(row.sourceItemId); + return payload === undefined + ? Effect.fail(new ProjectionStoreReadError({ threadId })) + : Effect.succeed({ + ...row, + payload: payload.payload_json, + entryId: payload.entry_id, + type: payload.item_type, + }); + }); + }); + // Cache rendered segments, not queries; typing and navigation reuse Markdown parsing. + const textCache = yield* Cache.make({ + capacity: 512, + 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)), + }); + // Scans retain counts and item references, never message bodies. Navigation checks + // the current revision in its transaction without reloading message payloads. + const scans = new Map< + string, + { + sequence: number; + documents: readonly FindDocument[]; + sourceThreadIds: readonly ThreadId[]; + } + >(); + const readRevision = Effect.fn("ThreadFind.readRevision")(function* ( + threadIds: readonly ThreadId[], + ) { + const sources = yield* encodeThreadIds(threadIds); + const revision = yield* sql<{ sequence: number }>` + WITH RECURSIVE source_threads(thread_id) AS ( + SELECT value FROM json_each(${sources}) + UNION + SELECT json_extract(t.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads t + JOIN source_threads s ON s.thread_id = t.thread_id + WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run' + ) + SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events + WHERE application_event_version = 2 AND aggregate_kind = 'thread' + AND stream_id IN (SELECT thread_id FROM source_threads) + `; + return revision[0]?.sequence ?? 0; + }); + const snapshot = Effect.fn("ThreadFind.snapshot")( + function* (input: OrchestrationV2SearchThreadInput) { + const { threadId } = input; + const active = yield* sql<{ cwd: string | null }>` + SELECT COALESCE(json_extract(t.payload_json, '$.worktreePath'), p.workspace_root) AS cwd + FROM orchestration_v2_projection_threads t JOIN projection_projects p ON p.project_id = t.project_id + WHERE t.thread_id = ${threadId} AND t.deleted_at IS NULL AND p.deleted_at IS NULL + `; + if (!active[0]) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); + const cwd = active[0].cwd ?? undefined; + const cacheKey = yield* encodeCacheKey({ + threadId, + cwd, + query: input.query, + skills: input.skills ?? [], + }); + const cached = scans.get(cacheKey); + if (cached && cached.sequence === (yield* readRevision(cached.sourceThreadIds))) { + const selected = selectMatch(cached.documents, input); + scans.delete(cacheKey); + scans.set(cacheKey, cached); + return { result: resultForSelection(selected, cached.documents, cached.sequence) }; + } + const index = yield* readIndex(threadId); + const sourceThreadIds = [...new Set([threadId, ...index.map((row) => row.sourceThreadId)])]; + const sequence = yield* readRevision(sourceThreadIds); + const candidates = index + .map((row, position) => ({ ...row, position })) + .filter((row) => + ["user_message", "assistant_message", "proposed_plan", "user_input_request"].includes( + row.item.type, + ), + ); + const payloads: FindPayload[] = []; + for (let start = 0; start < candidates.length; start += 128) { + payloads.push(...(yield* load(threadId, candidates.slice(start, start + 128)))); + } + return { payloads, cwd, sequence, cacheKey, sourceThreadIds }; + }, + sql.withTransaction, + (effect, input) => + effect.pipe( + Effect.mapError((cause) => + isSnapshotError(cause) + ? cause + : new ProjectionStoreReadError({ threadId: input.threadId, cause }), + ), + ), + ); + const searchThreadStream = (input: OrchestrationV2SearchThreadInput) => + Stream.unwrap( + Effect.gen(function* () { + // Release SQLite before decoding and parsing. Both frames describe one snapshot. + const data = yield* snapshot(input); + if (data.result) return Stream.succeed(data.result); + const { payloads, cwd, sequence, cacheKey, sourceThreadIds } = data; + const decode = (row: FindPayload) => + decodeItem(row.payload).pipe( + Effect.map((item) => ({ ...row, item })), + Effect.mapError( + (cause) => new ProjectionStoreReadError({ threadId: input.threadId, cause }), + ), + ); + // Answers rendered inside question cards must not appear as separate matches. + const foldedAnswers = new Set(); + for (const payload of payloads) { + if (payload.type !== "user_input_request") continue; + const { item } = yield* decode(payload); + if (item.type === "user_input_request" && item.questionAnswer) + foldedAnswers.add(`async-answer:${item.questionAnswer.requestId}`); + } + const rows = payloads.filter( + (row) => + row.type !== "user_input_request" && + !(row.type === "user_message" && foldedAnswers.has(row.entryId)), + ); + const documents = new Map(); + const count = Effect.fnUntraced(function* (row: FindPayload) { + const cached = documents.get(row.position); + if (cached) return cached; + 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, + ); + const document = documentFor(decoded, countSegments(segments, input.query)); + documents.set(row.position, document); + yield* Effect.yieldNow; + return document; + }); + const finish = Effect.gen(function* () { + for (const row of rows) yield* count(row); + const ordered = rows.map((row) => documents.get(row.position)!); + scans.delete(cacheKey); + scans.set(cacheKey, { sequence, documents: ordered, sourceThreadIds }); + if (scans.size > 8) { + const oldest = scans.keys().next().value; + if (oldest !== undefined) scans.delete(oldest); + } + return resultForSelection(selectMatch(ordered, input), ordered, sequence); + }); + // Relative navigation needs the final counts. Initial searches can reveal an + // identity immediately, scanning at/below the reading anchor before wrapping. + if (input.index !== undefined || (input.offset ?? 0) !== 0) + return Stream.fromEffect(finish); + const anchor = Math.max( + 0, + rows.findIndex((row) => row.entryId === input.start?.entryId), + ); + for (let step = 0; step <= rows.length; step++) { + const index = (anchor + step) % rows.length; + const row = rows[index]; + if (!row) break; + const document = yield* count(row); + const occurrence = + step === 0 && row.entryId === input.start?.entryId + ? Math.min(input.start.occurrence, document.count) + : 0; + if (occurrence >= document.count) continue; + const first: OrchestrationV2SearchThreadResult = { + complete: false, + snapshotSequence: sequence, + totalMatches: 1, + activeIndex: 0, + match: { entryId: document.entryId, runId: document.runId, occurrence }, + }; + // The remainder is pulled only after the first frame has been delivered. + return Stream.concat(Stream.succeed(first), Stream.fromEffect(finish)); + } + return Stream.fromEffect(finish); + }), + ); + return { + searchThreadStream, + searchThread: (input: OrchestrationV2SearchThreadInput) => + searchThreadStream(input).pipe( + Stream.runCollect, + Effect.map((results) => results[results.length - 1]!), + ), + }; +}); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index e59fce35bcbdce87d6f78a1709bc1c14013452b7..a12e6680ca7e2bf036f9d10e513b3a883d4ac407 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1,3 +1,7 @@ +import type { + OrchestrationV2SearchThreadInput, + OrchestrationV2SearchThreadResult, +} from "@t3tools/contracts"; import type { ProjectionRecordField, ProjectionRecordFilter, @@ -271,12 +275,19 @@ export type ThreadManagementError = typeof ThreadManagementError.Type; type ThreadManagementFailure = ThreadManagementError | Orchestrator.OrchestratorV2Error; export interface ThreadManagementServiceShape { + readonly searchThreadStream: ( + input: OrchestrationV2SearchThreadInput, + ) => Stream.Stream; + readonly searchThread: ( + input: OrchestrationV2SearchThreadInput, + ) => Effect.Effect; readonly ensureLegacyTranscript: ( threadId: ThreadId, ) => Effect.Effect; readonly dispatch: ( command: OrchestrationV2ServerCommand, ) => Effect.Effect; + readonly getThreadHistoryPage: Orchestrator.OrchestratorV2["Service"]["getThreadHistoryPage"]; readonly getTimelinePage: Orchestrator.OrchestratorV2["Service"]["getTimelinePage"]; readonly getMessageCount: Orchestrator.OrchestratorV2["Service"]["getMessageCount"]; /** @@ -859,7 +870,23 @@ const make = Effect.gen(function* () { return ThreadManagementService.of({ ensureLegacyTranscript, + searchThreadStream: (input) => + Stream.unwrap( + ensureProjectionTranscript(input.threadId).pipe( + Effect.as(orchestrator.searchThreadStream(input)), + ), + ), + searchThread: (input) => + ensureProjectionTranscript(input.threadId).pipe( + Effect.andThen(orchestrator.searchThread(input)), + ), dispatch, + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + ensureProjectionTranscript(threadId).pipe( + Effect.andThen( + orchestrator.getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly), + ), + ), getTimelinePage: (threadId, options) => ensureProjectionTranscript(threadId).pipe( Effect.andThen(orchestrator.getTimelinePage(threadId, options)), diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 8f8c30034c44cc9d97d5a223bd52058a91c1d9c2..8f67002333ba9bdeb6331d130f24b3d260a30e7f 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -1,8 +1,6 @@ import { AuthOrchestrationReadScope, EnvironmentHttpApi, - ThreadId, - TurnItemId, type OrchestrationProjectShell, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -24,7 +22,6 @@ import { buildBoundedThreadProjection, decodeThreadHistoryCursor, InvalidThreadHistoryCursorError, - selectHistoryPageFromCursor, THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, THREAD_HISTORY_PAGE_POLICY, OLDER_THREAD_USER_TURN_LIMIT, @@ -43,22 +40,6 @@ function isThreadNotFound(error: unknown): boolean { ); } -function selectHistoryPageFromCursorOrError( - input: Parameters[0], -): - | { readonly _tag: "ok"; readonly page: ReturnType } - | { readonly _tag: "invalid_cursor" } - | { readonly _tag: "error"; readonly cause: unknown } { - try { - return { _tag: "ok", page: selectHistoryPageFromCursor(input) }; - } catch (cause) { - if (cause instanceof InvalidThreadHistoryCursorError) { - return { _tag: "invalid_cursor" }; - } - return { _tag: "error", cause }; - } -} - /** * Serves orchestration V2 snapshots over HTTP so clients can load the * (potentially large) shell and thread projections off the socket; gzip @@ -222,41 +203,33 @@ export const layer = HttpApiBuilder.group( Effect.fn("environment.orchestration.threadHistoryPage")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - let anchorItemId; try { - anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(args.query.cursor).si); + decodeThreadHistoryCursor(args.query.cursor); } catch (cause) { - if (cause instanceof InvalidThreadHistoryCursorError) { + if (cause instanceof InvalidThreadHistoryCursorError) return yield* failEnvironmentInvalidRequest("invalid_history_cursor"); - } return yield* failEnvironmentInternal("orchestration_thread_history_failed", cause); } - const decodedCursor = decodeThreadHistoryCursor(args.query.cursor); - const snapshot = yield* loadThreadSnapshotWindow( - args.params.threadId, - anchorItemId, - ThreadId.make(decodedCursor.st), - ).pipe(traceLocalHandlerWork); - const pageOrError = selectHistoryPageFromCursorOrError({ - items: snapshot.projection.visibleTurnItems, - cursor: args.query.cursor, - snapshotSequence: snapshot.snapshotSequence, - }); - if (pageOrError._tag === "invalid_cursor") { - return yield* failEnvironmentInvalidRequest("invalid_history_cursor"); - } - if (pageOrError._tag === "error") { - return yield* failEnvironmentInternal( - "orchestration_thread_history_failed", - pageOrError.cause, + return yield* threadManagement + .getThreadHistoryPage( + args.params.threadId, + args.query.cursor, + args.query.throughEntryId, + args.query.view === "conversation", + ) + .pipe( + traceLocalHandlerWork, + Effect.catch( + Effect.fnUntraced(function* (cause) { + if (isThreadNotFound(cause)) + return yield* failEnvironmentNotFound("thread_not_found"); + return yield* failEnvironmentInternal( + "orchestration_thread_history_failed", + cause, + ); + }), + ), ); - } - return { - snapshotSequence: snapshot.snapshotSequence, - items: pageOrError.page.items, - nextCursor: pageOrError.page.nextCursor, - hasMoreHistory: pageOrError.page.hasMoreHistory, - }; }), ); }), diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index a2e6c87e20df79d09fa27095b52731d5feef01a1..8ee7c2641bd3028fbd96bb4b0c2f426cb5be0d0a 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -17,6 +17,11 @@ export const THREAD_HISTORY_PAGE_POLICY = { } as const; export const OLDER_THREAD_USER_TURN_LIMIT = 20; + +/** Load conversation rows before the bulk of a turn's folded tool activity. */ +export function isConversationHistoryItem(item: Pick): boolean { + return !["command_execution", "reasoning", "file_change", "dynamic_tool"].includes(item.type); +} export const THREAD_HISTORY_MAX_RAW_TURNS = 150; /** Extra rows let the projection query retain an inclusive cursor and prove another page exists. */ @@ -39,8 +44,19 @@ export type ThreadHistoryCursorPayload = { readonly p: number; }; -export type SelectTimelinePageResult = { - readonly items: ReadonlyArray; +type HistoryRow = Pick< + OrchestrationV2ProjectedTurnItem, + "position" | "visibility" | "sourceThreadId" | "sourceItemId" +> & { + readonly item: Pick & { + readonly id?: string; + readonly messageId?: string; + readonly inputIntent?: string; + readonly createdBy?: string; + }; +}; +export type SelectTimelinePageResult = { + readonly items: ReadonlyArray; readonly nextCursor: string | null; readonly hasMoreHistory: boolean; }; @@ -67,7 +83,7 @@ function bytesOfJson(value: unknown): number { } /** Ordinary history-page cost: one projected row (item is nested once). */ -export function projectedRowEncodedBytes(row: OrchestrationV2ProjectedTurnItem): number { +export function projectedRowEncodedBytes(row: HistoryRow): number { return bytesOfJson(row); } @@ -87,9 +103,7 @@ export function projectedRowBoundedSnapshotEncodedBytes( return rowBytes; } -function renumberPositions( - items: ReadonlyArray, -): OrchestrationV2ProjectedTurnItem[] { +function renumberPositions(items: ReadonlyArray): Row[] { return items.map((row, position) => (row.position === position ? row : { ...row, position })); } @@ -148,7 +162,7 @@ export function decodeThreadHistoryCursor(cursor: string): ThreadHistoryCursorPa return parsed as ThreadHistoryCursorPayload; } -export function isThreadHistoryTurnStart(item: OrchestrationV2TurnItem): boolean { +export function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean { return ( item.type === "user_message" && (item.inputIntent === "turn_start" || item.inputIntent === "queued_turn") @@ -156,7 +170,7 @@ export function isThreadHistoryTurnStart(item: OrchestrationV2TurnItem): boolean } /** Steering belongs to its existing turn and must not consume another page slot. */ -export function isThreadHistoryUserTurn(item: OrchestrationV2TurnItem): boolean { +export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean { return ( isThreadHistoryTurnStart(item) && item.type === "user_message" && item.createdBy === "user" ); @@ -167,13 +181,13 @@ export function isThreadHistoryUserTurn(item: OrchestrationV2TurnItem): boolean * through complete user turns. Histories without turn starts use row budgets * and always admit at least one row so oversized items cannot deadlock paging. */ -function selectOlderTimelinePage(input: { - readonly items: ReadonlyArray; +function selectOlderTimelinePage(input: { + readonly items: ReadonlyArray; readonly exclusiveEndIndex: number; readonly snapshotSequence: number; readonly policy?: ThreadHistoryPagePolicy | undefined; - readonly rowEncodedBytes?: ((row: OrchestrationV2ProjectedTurnItem) => number) | undefined; -}): SelectTimelinePageResult { + readonly rowEncodedBytes?: ((row: Row) => number) | undefined; +}): SelectTimelinePageResult { const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY; const rowCost = input.rowEncodedBytes ?? projectedRowEncodedBytes; const end = Math.min(Math.max(input.exclusiveEndIndex, 0), input.items.length); @@ -181,7 +195,7 @@ function selectOlderTimelinePage(input: { return { items: [], nextCursor: null, hasMoreHistory: false }; } - const selected: OrchestrationV2ProjectedTurnItem[] = []; + const selected: Row[] = []; let encodedBytes = 0; let userTurns = 0; let rawTurns = 0; @@ -247,7 +261,7 @@ export function selectRecentTimelineWindow(input: { } function findCursorIndex( - items: ReadonlyArray, + items: ReadonlyArray>, cursor: ThreadHistoryCursorPayload, ): number { const byIdentity = items.findIndex( @@ -264,22 +278,48 @@ function findCursorIndex( return Math.min(cursor.p, items.length); } -export function selectHistoryPageFromCursor(input: { - readonly items: ReadonlyArray; +export function selectHistoryPageFromCursor(input: { + readonly items: ReadonlyArray; readonly cursor: string; readonly snapshotSequence: number; readonly policy?: ThreadHistoryPagePolicy | undefined; -}): SelectTimelinePageResult { + readonly rowEncodedBytes?: ((row: Row) => number) | undefined; + readonly throughEntryId?: string | undefined; +}): SelectTimelinePageResult { const decoded = decodeThreadHistoryCursor(input.cursor); const anchorIndex = findCursorIndex(input.items, decoded); + let policy = input.policy ?? { + ...THREAD_HISTORY_PAGE_POLICY, + maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, + }; + if (input.throughEntryId !== undefined) { + const targetIndex = input.items.findIndex( + (row) => (row.item.messageId ?? row.item.id) === input.throughEntryId, + ); + if (targetIndex >= 0 && targetIndex < anchorIndex) { + // A search jump can combine two pages, but never grow into an unbounded read. + const turns = input.items + .slice(targetIndex, anchorIndex) + .filter((row) => isThreadHistoryUserTurn(row.item)).length; + policy = { + ...policy, + maxUserTurns: Math.min( + OLDER_THREAD_USER_TURN_LIMIT * 2, + Math.max( + OLDER_THREAD_USER_TURN_LIMIT, + turns + (isThreadHistoryUserTurn(input.items[targetIndex]!.item) ? 0 : 1), + ), + ), + maxItems: policy.maxItems * 2, + }; + } + } return selectOlderTimelinePage({ items: input.items, exclusiveEndIndex: anchorIndex, snapshotSequence: input.snapshotSequence, - policy: input.policy ?? { - ...THREAD_HISTORY_PAGE_POLICY, - maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT, - }, + ...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }), + policy, }); } diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index aa64f048ad807043eed3241b7532c0a55f4d2c75..f10af26391c678bde457aa6a81344082a17dda6c 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -193,6 +193,9 @@ const makeTestRelay = Effect.fnUntraced(function* ( readShellSnapshot: unused, ensureLegacyTranscript: unused, dispatch: unused, + searchThread: () => Effect.die("unused"), + searchThreadStream: () => Stream.empty, + getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), getTurnItem: () => Effect.die("unused turn item read"), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 70238f10a882662355f6135c88d7b15f6b5db6c8..2a11852ecbc50fa884fa8dffde2c25f60e095dfd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -52,6 +52,7 @@ import { type AcpRegistrySetProviderInput, OrchestrationGetFullThreadDiffError, OrchestrationSearchThreadsError, + OrchestrationV2SearchThreadError, OrchestrationGetTurnDiffError, ORCHESTRATION_V2_WS_METHODS, ORCHESTRATION_PROTOCOL_QUERY_PARAM, @@ -1727,6 +1728,8 @@ const layerWsRpc = ( shellResumeCompletionMarker: true, threadResumeCompletionMarker: true, threadSnapshotPagination: true, + threadFind: true, + threadFindProgressive: true, ...Option.match(scratchWorkspaceRoot, { onNone: () => ({}), onSome: (root) => ({ scratchWorkspaceRoot: root }), @@ -1894,6 +1897,14 @@ const layerWsRpc = ( }), ), ), + [ORCHESTRATION_V2_WS_METHODS.searchThread]: (input) => + threadManagement + .searchThread(input) + .pipe(Effect.mapError((cause) => new OrchestrationV2SearchThreadError({ cause }))), + [ORCHESTRATION_V2_WS_METHODS.searchThreadStream]: (input) => + threadManagement + .searchThreadStream(input) + .pipe(Stream.mapError((cause) => new OrchestrationV2SearchThreadError({ cause }))), [ORCHESTRATION_V2_WS_METHODS.searchThreads]: (input) => threadSearch.search(input).pipe( Effect.mapError( diff --git a/apps/web/package.json b/apps/web/package.json index 915466100bcff14f04f39c95ce3982aa5c99e2a7..20e180aad46b50d3a95e796f52cb7bab890d9a80 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -60,10 +60,7 @@ "react": "19.2.6", "react-dom": "19.2.6", "react-markdown": "^10.1.0", - "rehype-raw": "^7.0.0", - "rehype-sanitize": "^6.0.0", - "remark-breaks": "^4.0.0", - "remark-gfm": "^4.0.1", + "remark-gfm": "catalog:", "tailwind-merge": "^3.4.0", "unbash": "^5.0.0", "zustand": "^5.0.11" @@ -88,7 +85,7 @@ "jsdom": "^30.0.1", "react-test-renderer": "19.2.6", "tailwindcss": "^4.0.0", - "unified": "^11.0.5", + "unified": "catalog:", "vite": "catalog:", "vite-plus": "catalog:" } diff --git a/apps/web/src/components/ChatMarkdown.test.tsx b/apps/web/src/components/ChatMarkdown.test.tsx index b1ceb33062c8ba226dabe2cfa3f772e1c6c3c785..08fb61eac0bb0f81d9aa07429f69a797ba0f580c 100644 --- a/apps/web/src/components/ChatMarkdown.test.tsx +++ b/apps/web/src/components/ChatMarkdown.test.tsx @@ -1,6 +1,12 @@ // @vitest-environment jsdom import { EnvironmentId, type AuthEnvironmentScope } from "@t3tools/contracts"; +import { createRoot } from "react-dom/client"; +import { useThreadFindHighlights } from "./chat/threadFindHighlights"; +import { searchableMessageSegments } from "@t3tools/shared/threadFindText"; +import { countThreadSearchOccurrences } from "@t3tools/shared/threadSearch"; + +import { MarkdownFindContext } from "./chat/markdownFindContext"; import { act, type ComponentProps, type ReactNode } from "react"; import { renderToStaticMarkup } from "react-dom/server"; import { create, type ReactTestRenderer } from "react-test-renderer"; @@ -12,6 +18,10 @@ import { Button } from "./ui/button"; import { setMarkdownTaskChecked } from "./files/filePreviewMode"; vi.mock("@effect/atom-react", () => ({ useAtomValue: () => null })); +vi.mock("./chat/MermaidDiagram", () => ({ + // Real Mermaid needs layout APIs jsdom lacks; a rendered diagram is an SVG. + MermaidDiagram: () => , +})); vi.mock("../hooks/useTheme", () => ({ useTheme: () => ({ resolvedTheme: "dark" }) })); vi.mock("../hooks/useSettings", async (importOriginal) => { const actual = await importOriginal(); @@ -1034,3 +1044,231 @@ describe("ChatMarkdown Windows file links", () => { expect(html).not.toContain("chat-markdown-file-link"); }); }); + +it("opens a disclosure only when find selects a match inside it", async () => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.stubGlobal( + "Highlight", + class extends Set { + constructor(...ranges: Range[]) { + super(ranges); + } + }, + ); + const highlights = new Map>(); + vi.stubGlobal("CSS", { highlights, escape: (value: string) => value }); + const container = document.createElement("div"); + document.body.append(container); + const root = createRoot(container); + const openStates = () => + [...container.querySelectorAll("[data-markdown-details-open]")].map((node) => + node.getAttribute("data-markdown-details-open"), + ); + function Probe({ + activeOccurrence, + searching = true, + }: { + activeOccurrence: number; + searching?: boolean; + }) { + useThreadFindHighlights({ + container, + query: searching ? "needle" : "", + activeRowId: "row", + activeOccurrence, + onActiveRange: () => {}, + }); + return ( +
+
+ + Unrelated

nothing here

", + "
Outer
Inner

needle

", + ].join("\n\n")} + /> + +
+
+ ); + } + const frame = () => + act(() => new Promise((resolve) => requestAnimationFrame(() => resolve()))); + try { + // Closed panels are unmounted until find starts, as in the app. + await act(() => root.render()); + expect(container.textContent).not.toContain("nothing here"); + await act(() => root.render()); + await frame(); + // Selecting the visible match opens nothing; the folded one is counted but not painted. + expect(openStates()).toEqual(["false", "false", "false"]); + expect(container.textContent).toContain("nothing here"); + expect( + [...(highlights.get("t3-thread-find-active") ?? [])].map((range) => range.toString()), + ).toEqual(["needle"]); + expect(highlights.get("t3-thread-find")?.size).toBe(0); + + await act(() => root.render()); + await frame(); + await frame(); + // Stepping to the folded match opens its two ancestors, not the unrelated one. + expect(openStates()).toEqual(["false", "true", "true"]); + expect(highlights.get("t3-thread-find-active")?.size).toBe(1); + } finally { + await act(() => root.unmount()); + container.remove(); + vi.unstubAllGlobals(); + } +}); + +it("keeps Mermaid diagrams rendered until find selects a match in their source", async () => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.stubGlobal( + "Highlight", + class extends Set { + constructor(...ranges: Range[]) { + super(ranges); + } + }, + ); + const highlights = new Map>(); + vi.stubGlobal("CSS", { highlights, escape: (value: string) => value }); + const container = document.createElement("div"); + document.body.append(container); + const root = createRoot(container); + const diagram = "```mermaid\ngraph TD; Alpha-->Beta\n```"; + function Probe({ query }: { query: string }) { + useThreadFindHighlights({ + container, + query, + activeRowId: "row", + activeOccurrence: 0, + onActiveRange: () => {}, + }); + return ( +
+
+ + + +
+
+ ); + } + const frame = () => + act(() => new Promise((resolve) => requestAnimationFrame(() => resolve()))); + const diagrams = () => container.querySelectorAll('svg[aria-label="Diagram"]').length; + try { + await act(() => root.render()); + await frame(); + expect(diagrams()).toBe(2); + await act(() => root.render()); + await frame(); + await frame(); + // Only the diagram holding the selected match switches to source. + expect(diagrams()).toBe(1); + expect( + [...(highlights.get("t3-thread-find-active") ?? [])].map((range) => range.toString()), + ).toEqual(["Alpha"]); + } finally { + await act(() => root.unmount()); + container.remove(); + vi.unstubAllGlobals(); + } +}); + +it.each([ + { + text: "```mermaid\ngraph TD; SearchSourceAlpha-->B\n```", + query: "SearchSourceAlpha", + count: 1, + }, + { + text: "★ Insight ─────\nfirst line\nsecond line", + query: "first line second", + count: 0, + lineBreaks: true, + }, + { text: '```ts title="src/needle.ts"\nconst a = 1;\n```', query: "needle", count: 0 }, + { text: "```weirdlang\nconst a = 1;\n```", query: "weirdlang", count: 0 }, + { text: "Use $test-t3-app now", query: "T3 App Testing", count: 1 }, + { text: "`/tmp/file.ts:42`", query: "file.ts · L42", count: 1, user: true, lineBreaks: true }, + { text: "> [!NOTE]\n> Searchable alert", query: "Searchable alert", count: 1 }, + { + text: "
Folded

Hidden needle

", + query: "Hidden needle", + count: 1, + }, + { text: ARTIFACT_TEMPLATE_DIRECTIVE, query: "Hello World", count: 1, useTemplate: true }, + { text: ARTIFACT_TEMPLATE_DIRECTIVE, query: "Document template", count: 1, useTemplate: true }, + { text: ARTIFACT_TEMPLATE_DIRECTIVE, query: "World Document", count: 0, useTemplate: true }, + { text: ARTIFACT_TEMPLATE_DIRECTIVE, query: "Use template", count: 0, useTemplate: true }, +])( + "highlights the indexed occurrences of $query in $text", + async ({ text, query, count, lineBreaks, user, useTemplate }) => { + const skills = [{ name: "test-t3-app", displayName: "T3 App Testing" }]; + const highlights = new Map>(); + vi.stubGlobal( + "Highlight", + class extends Set { + constructor(...ranges: Range[]) { + super(ranges); + } + }, + ); + vi.stubGlobal("CSS", { highlights, escape: (value: string) => value }); + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + const container = document.createElement("div"); + document.body.append(container); + const root = createRoot(container); + function Probe() { + useThreadFindHighlights({ + container, + query, + activeRowId: "row", + activeOccurrence: 0, + onActiveRange: () => {}, + }); + return ( +
+
+ + undefined : undefined} + /> + +
+
+ ); + } + try { + await act(() => root.render()); + await act(() => new Promise((resolve) => requestAnimationFrame(() => resolve()))); + const ranges = [...highlights.values()].flatMap((value) => [...value]); + expect(ranges.map((range) => range.toString())).toEqual( + Array.from({ length: count }, () => query), + ); + const segments = + searchableMessageSegments( + { role: user ? "user" : "assistant", text, streaming: false }, + undefined, + skills, + ) ?? []; + expect( + segments.reduce((sum, segment) => sum + countThreadSearchOccurrences(segment, query), 0), + ).toBe(count); + } finally { + await act(() => root.unmount()); + container.remove(); + vi.unstubAllGlobals(); + } + }, +); diff --git a/apps/web/src/components/ChatMarkdown.tsx b/apps/web/src/components/ChatMarkdown.tsx index d926f5aaf412d472c7444967a235e1ee3a67f5bd..42b4a0618b6b3abe63bffb2f8a191c83e5cc3af3 100644 --- a/apps/web/src/components/ChatMarkdown.tsx +++ b/apps/web/src/components/ChatMarkdown.tsx @@ -1,5 +1,24 @@ +import { MarkdownFindContext, useFindRevealRef } from "./chat/markdownFindContext"; +import { + buildFileLinkParentSuffixByPath, + fileLinkLabel, + resolvePathLinkTarget, +} from "@t3tools/shared/fileLinks"; +import { + isWindowsDrivePathHref, + normalizeMarkdownLinkDestination, + extractInlineCodeSpans, + extractMarkdownLinkHrefs, + inlineCodeFilePathCandidate, +} from "@t3tools/shared/markdownLinks"; +import { isAbsolutePath } from "@t3tools/shared/path"; import { usePullRequestLinking } from "~/hooks/usePullRequestLinking"; import { AuthFilesystemReadScope, AuthOrchestrationOperateScope } from "@t3tools/contracts"; +import { + CHAT_MARKDOWN_REMARK_PLUGINS, + CHAT_MARKDOWN_REMARK_PLUGINS_WITH_BREAKS, + CHAT_MARKDOWN_REHYPE_PLUGINS, +} from "@t3tools/shared/markdownPipeline"; import { useAtomValue } from "@effect/atom-react"; import { COMPOSER_CONTEXT_CLIPBOARD_MIME, @@ -46,12 +65,11 @@ import { codexArtifactTemplatePresentationLabel, type CodexArtifactTemplate, type CodexArtifactTemplateKind, -} from "@t3tools/client-runtime/codex-artifact-templates"; +} from "@t3tools/shared/codexArtifactTemplates"; import { classifyMarkdownImageSource, markdownImageSourceFragment, } from "@t3tools/client-runtime/markdown-images"; -import { inlineCodeFilePathCandidate } from "@t3tools/client-runtime/markdown-links"; import { mediaFileReference, mediaUrlReference } from "@t3tools/client-runtime/media-reference"; import { mediaKindFromPath, mediaMimeTypeFromExtension } from "@t3tools/shared/filePreview"; import * as Cause from "effect/Cause"; @@ -83,24 +101,16 @@ import ReactMarkdown from "react-markdown"; import { toHtml } from "hast-util-to-html"; import { createIncrementalMarkdownPlugin } from "../markdown-incremental"; import { defaultUrlTransform } from "react-markdown"; -import rehypeRaw from "rehype-raw"; -import rehypeSanitize, { defaultSchema } from "rehype-sanitize"; -import remarkBreaks from "remark-breaks"; import { parseAssistantCitationHref } from "@t3tools/shared/assistantCitations"; import { parseComposerContextHref } from "@t3tools/shared/composerContextReferences"; -import { parseThreadLinkHref, THREAD_LINK_PROTOCOL } from "@t3tools/shared/threadLinks"; +import { parseThreadLinkHref } from "@t3tools/shared/threadLinks"; import { AssistantCitationChip } from "./chat/AssistantCitationChip"; import { MarkdownThreadLink } from "./chat/MarkdownThreadLink"; -import remarkGfm from "remark-gfm"; -import type { Processor } from "unified"; import { isWindowsAbsolutePath } from "@t3tools/shared/path"; -import { remarkGithubAlerts } from "../markdown-github-alerts"; import { artifactTemplateFromHastProperties, - CODEX_ARTIFACT_TEMPLATE_HAST_PROPERTIES, - remarkCodexDirectives, renderCodexFileCitationsAsMarkdown, -} from "@t3tools/client-runtime/codex-markdown-directives"; +} from "@t3tools/shared/codexMarkdownDirectives"; import { renderSkillInlineMarkdownChildren } from "./chat/SkillInlineText"; import { resolveMarkdownMediaPreview, @@ -153,11 +163,7 @@ import { serializeTableElementToCsv, serializeTableElementToMarkdown, } from "../markdown-clipboard"; -import { remarkNormalizeListItemIndentation } from "../markdown-list-indentation"; import { - extractMarkdownLinkHrefs, - isWindowsDrivePathHref, - normalizeMarkdownLinkDestination, resolveInlineCodeFileLinkMeta, resolveMarkdownFileLinkMeta, rewriteMarkdownFileUriHref, @@ -165,7 +171,7 @@ import { shouldOpenMarkdownFileLinkInEditor, type MarkdownFileLinkMeta, } from "../markdown-links"; -import { isMarkdownFileLinkLabel } from "@t3tools/client-runtime/markdown-links"; +import { isMarkdownFileLinkLabel } from "@t3tools/shared/markdownLinks"; import { readLocalApi } from "../localApi"; import { useAssetUrlRefresh, useAssetUrlState } from "../assets/assetUrls"; import { cn } from "../lib/utils"; @@ -195,7 +201,6 @@ import { import { useOpenLink } from "../browser/useOpenLink"; import { writeTextToClipboard } from "../hooks/useCopyToClipboard"; import { isPreviewAvailableFor } from "../browser/previewRuntime"; -import { isAbsolutePath, resolvePathLinkTarget } from "../terminal-links"; import { isBrowserPreviewFile, openFileInPreview, @@ -322,15 +327,17 @@ function CodexArtifactTemplateCard(props: { - - + {/* Block elements keep the name and label separate thread-find segments. */} +
+
{props.template.displayName} - - {presentationLabel} - +
+
{presentationLabel}
+
{props.onUse ? (