1import { assert, it, describe } from "@effect/vitest";
2import { searchableMessageSegments } from "@t3tools/shared/threadFindText";
3import {
4 EventId,
5 MessageId,
6 PlanId,
7 RunId,
8 RuntimeRequestId,
9 TurnItemId,
10 type OrchestrationV2TurnItem,
11 type OrchestrationV2Run,
12 type OrchestrationV2ThreadHistoryPage,
13 ProjectId,
14 ProviderInstanceId,
15 ThreadId,
16 type OrchestrationV2DomainEvent,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as SqlClient from "effect/sql/SqlClient";
22import * as Stream from "effect/Stream";
23import * as TestClock from "effect/testing/TestClock";
24import { vi } from "vite-plus/test";
25
26import * as SqlitePersistence from "../persistence/Sqlite.ts";
27import * as ProjectionStore from "./ProjectionStore.ts";
28import * as ProjectStore from "./ProjectStore.ts";
29import * as EventStore from "./EventStore.ts";
30import {
31 encodeThreadHistoryCursor,
32 selectHistoryPageFromCursor,
33 THREAD_HISTORY_PAGE_POLICY,
34} from "./threadHistoryPaging.ts";
35import { projectTurnItemForWire } from "@t3tools/shared/wireProjection";
36
37vi.mock("@t3tools/shared/threadFindText", { spy: true });
38
39const layerTest = Layer.mergeAll(EventStore.layer, ProjectionStore.layer, ProjectStore.layer).pipe(
40 Layer.provideMerge(SqlitePersistence.layerMemory),
41);
42
43const providerInstanceId = ProviderInstanceId.make("codex");
44const at = (minute: number) => DateTime.makeUnsafe(Date.UTC(2026, 8, 27, 0, minute));
45
46const createProject = (projectId: ProjectId) =>
47 Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) =>
48 projects.apply({
49 sequence: 0,
50 eventId: EventId.make(`created:${projectId}`),
51 aggregateKind: "project",
52 aggregateId: projectId,
53 occurredAt: DateTime.formatIso(at(0)),
54 commandId: null,
55 causationEventId: null,
56 correlationId: null,
57 metadata: {},
58 type: "project.created",
59 payload: {
60 projectId,
61 title: projectId,
62 workspaceRoot: `/work/${projectId}`,
63 defaultModelSelection: null,
64 scripts: [],
65 createdAt: DateTime.formatIso(at(0)),
66 updatedAt: DateTime.formatIso(at(0)),
67 },
68 }),
69 );
70
71const thread = (
72 threadId: ThreadId,
73 projectId: ProjectId,
74 overrides: { readonly archivedAt?: DateTime.Utc; readonly deletedAt?: DateTime.Utc } = {},
75): OrchestrationV2DomainEvent => ({
76 id: EventId.make(`created:${threadId}`),
77 type: "thread.created",
78 threadId,
79 providerInstanceId,
80 occurredAt: at(0),
81 payload: {
82 createdBy: "user",
83 creationSource: "web",
84 id: threadId,
85 projectId,
86 title: threadId,
87 providerInstanceId,
88 modelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
89 runtimeMode: "full-access",
90 interactionMode: "default",
91 branch: null,
92 worktreePath: null,
93 activeProviderThreadId: null,
94 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
95 forkedFrom: null,
96 createdAt: at(0),
97 updatedAt: at(0),
98 archivedAt: overrides.archivedAt ?? null,
99 settledOverride: null,
100 settledAt: null,
101 lastVisitedAt: null,
102 deletedAt: overrides.deletedAt ?? null,
103 },
104});
105
106const threadId = ThreadId.make("thread:find");
107const projectId = ProjectId.make("project:find");
108const runId = RunId.make("run:find");
109
110function item(
111 id: string,
112 ordinal: number,
113 text: string,
114 kind: "user_message" | "assistant_message" | "proposed_plan" = "assistant_message",
115 owner = threadId,
116): Extract<
117 OrchestrationV2TurnItem,
118 { type: "user_message" | "assistant_message" | "proposed_plan" }
119> {
120 const base = {
121 id: TurnItemId.make(id),
122 threadId: owner,
123 runId,
124 nodeId: null,
125 providerThreadId: null,
126 providerTurnId: null,
127 nativeItemRef: null,
128 parentItemId: null,
129 ordinal,
130 status: "completed" as const,
131 title: null,
132 startedAt: at(100 - ordinal),
133 completedAt: at(100 - ordinal),
134 updatedAt: at(1),
135 };
136 if (kind === "proposed_plan")
137 return { ...base, type: kind, planId: PlanId.make(id), markdown: text, streaming: false };
138 if (kind === "user_message")
139 return {
140 ...base,
141 type: kind,
142 messageId: MessageId.make(id),
143 text,
144 inputIntent: "turn_start",
145 attachments: [],
146 createdBy: "user",
147 creationSource: "web",
148 };
149 return { ...base, type: kind, messageId: MessageId.make(id), text, streaming: false };
150}
151
152function run(
153 owner = threadId,
154 id = runId,
155 ordinal = 1,
156 status: OrchestrationV2Run["status"] = "completed",
157): Extract<OrchestrationV2DomainEvent, { type: "run.created" }> {
158 return {
159 id: EventId.make(`run:${owner}:${id}:${status}`),
160 threadId: owner,
161 occurredAt: at(1),
162 type: "run.created",
163 payload: {
164 id,
165 threadId: owner,
166 ordinal,
167 providerInstanceId,
168 modelSelection: { instanceId: providerInstanceId, model: "gpt-6" },
169 providerThreadId: null,
170 userMessageId: MessageId.make("prompt"),
171 rootNodeId: null,
172 activeAttemptId: null,
173 status,
174 requestedAt: at(1),
175 startedAt: at(1),
176 completedAt: at(2),
177 checkpointId: null,
178 contextHandoffId: null,
179 },
180 };
181}
182
183const commit = Effect.fn("ThreadFindTest.commit")(function* (
184 events: readonly OrchestrationV2DomainEvent[],
185) {
186 const store = yield* EventStore.EventStoreV2;
187 const projection = yield* ProjectionStore.ProjectionStoreV2;
188 yield* store.append({ events });
189 yield* Effect.forEach(events, projection.apply, { discard: true });
190});
191const putItems = (items: readonly OrchestrationV2TurnItem[], version = "initial") =>
192 commit(
193 items.map((payload) => ({
194 id: EventId.make(`${payload.threadId}:${payload.id}:${version}`),
195 threadId: payload.threadId,
196 occurredAt: at(1),
197 type: "turn-item.updated" as const,
198 payload,
199 })),
200 );
201const setup = Effect.gen(function* () {
202 yield* createProject(projectId);
203 yield* commit([thread(threadId, projectId), run()]);
204 return yield* ProjectionStore.ProjectionStoreV2;
205});
206
207describe("V2 thread find", () => {
208 it.effect("finishes one snapshot while later searches include newly inserted matches", () =>
209 Effect.gen(function* () {
210 const projection = yield* setup;
211 yield* putItems([
212 item("selected", 1, "snapshot needle"),
213 item("last", 2, "snapshot last needle"),
214 ]);
215 const frames = yield* projection.searchThreadStream({ threadId, query: "needle" }).pipe(
216 Stream.tap((frame) =>
217 frame.complete === false
218 ? putItems([item("inserted", 0, "snapshot inserted needle")], "during-search")
219 : Effect.void,
220 ),
221 Stream.runCollect,
222 );
223 assert.equal(frames.length, 2);
224 assert.equal(frames[1]!.totalMatches, 2);
225 assert.deepEqual(frames[1]!.match, frames[0]!.match);
226 const updated = yield* projection.searchThread({
227 threadId,
228 query: "needle",
229 start: { entryId: "selected", occurrence: 0 },
230 });
231 assert.equal(updated.totalMatches, 3);
232 assert.equal(updated.activeIndex, 1);
233 assert.equal(updated.match?.entryId, "selected");
234 }).pipe(Effect.provide(layerTest)),
235 );
236
237 it.effect("returns bounded navigation counts around the selection, including wraparound", () =>
238 Effect.gen(function* () {
239 const projection = yield* setup;
240 yield* putItems(
241 Array.from({ length: 30 }, (_, i) =>
242 item(`message:${i}`, i, "needle ".repeat((i % 3) + 1)),
243 ),
244 );
245 const current = yield* projection.searchThread({
246 threadId,
247 query: "needle",
248 start: { entryId: "message:14", occurrence: 1 },
249 });
250 assert.equal(current.totalMatches, 60);
251 assert.equal(current.activeIndex, 28);
252 assert.equal(current.navigation?.length, 17);
253 assert.deepEqual(
254 current.navigation?.map((entry) => entry.entryId),
255 Array.from({ length: 17 }, (_, i) => `message:${i + 6}`),
256 );
257 assert.deepEqual(
258 current.navigation?.find((entry) => entry.entryId === "message:14"),
259 {
260 entryId: "message:14",
261 runId,
262 startIndex: 27,
263 count: 3,
264 },
265 );
266 const first = yield* projection.searchThread({ threadId, query: "needle" });
267 assert.equal(first.navigation?.length, 17);
268 assert.include(first.navigation?.map((entry) => entry.entryId) ?? [], "message:29");
269 const empty = yield* projection.searchThread({ threadId, query: "absent" });
270 assert.deepEqual(empty.navigation, []);
271 }).pipe(Effect.provide(layerTest)),
272 );
273
274 it.effect("keeps navigation compact for a message with thousands of occurrences", () =>
275 Effect.gen(function* () {
276 const projection = yield* setup;
277 yield* putItems([item("many", 1, "needle ".repeat(5000))]);
278 const result = yield* projection.searchThread({ threadId, query: "needle", index: 2000 });
279 assert.equal(result.totalMatches, 5000);
280 assert.equal(result.match?.occurrence, 2000);
281 assert.deepEqual(result.navigation, [{ entryId: "many", runId, startIndex: 0, count: 5000 }]);
282 assert.isBelow(JSON.stringify(result).length, 500);
283 }).pipe(Effect.provide(layerTest)),
284 );
285
286 it.effect("cycles relative to the current match when new matches are inserted", () =>
287 Effect.gen(function* () {
288 const projection = yield* setup;
289 yield* putItems([
290 item("first", 10, "needle"),
291 item("current", 20, "needle"),
292 item("last", 30, "needle"),
293 ]);
294 const initial = yield* projection.searchThread({ threadId, query: "needle", index: 1 });
295 assert.equal(initial.match?.entryId, "current");
296 yield* putItems([
297 item("new-before", 15, "needle"),
298 item("new-after", 25, "needle"),
299 item("new-last", 40, "needle"),
300 ]);
301 const start = { entryId: "current", occurrence: 0 };
302 const refreshed = yield* projection.searchThread({ threadId, query: "needle", start });
303 assert.equal(refreshed.totalMatches, 6);
304 assert.equal(refreshed.activeIndex, 2);
305 assert.equal(refreshed.match?.entryId, "current");
306 const next = yield* projection.searchThread({ threadId, query: "needle", start, offset: 1 });
307 assert.equal(next.match?.entryId, "new-after");
308 const previous = yield* projection.searchThread({
309 threadId,
310 query: "needle",
311 start,
312 offset: -1,
313 });
314 assert.equal(previous.match?.entryId, "new-before");
315 const end = yield* projection.searchThread({
316 threadId,
317 query: "needle",
318 start: { entryId: "last", occurrence: 0 },
319 offset: 1,
320 });
321 assert.equal(end.match?.entryId, "new-last");
322 const wrapped = yield* projection.searchThread({
323 threadId,
324 query: "needle",
325 start: { entryId: "new-last", occurrence: 0 },
326 offset: 1,
327 });
328 assert.equal(wrapped.match?.entryId, "first");
329 const backwards = yield* projection.searchThread({
330 threadId,
331 query: "needle",
332 start: { entryId: "first", occurrence: 0 },
333 offset: -1,
334 });
335 assert.equal(backwards.match?.entryId, "new-last");
336 }).pipe(Effect.provide(layerTest)),
337 );
338
339 it.effect("searches visible rendered messages and plans in canonical item order", () =>
340 Effect.gen(function* () {
341 const projection = yield* setup;
342 yield* putItems([
343 item("user", 1, "Check **the build**", "user_message"),
344 item("assistant", 2, "check the build twice: check the build"),
345 item("plan-item", 3, "# Check the build\nShip it", "proposed_plan"),
346 item("url", 4, "[documentation](https://hidden.example/needle)"),
347 ]);
348 const matches = [];
349 for (let index = 0; index < 4; index++) {
350 const result = yield* projection.searchThread({
351 threadId,
352 query: "CHECK THE BUILD",
353 index,
354 });
355 assert.equal(result.totalMatches, 4);
356 matches.push(result.match);
357 }
358 assert.deepEqual(matches, [
359 { entryId: "user", runId, occurrence: 0 },
360 { entryId: "assistant", runId, occurrence: 0 },
361 { entryId: "assistant", runId, occurrence: 1 },
362 { entryId: "plan-item", runId, occurrence: 0 },
363 ]);
364 assert.equal(
365 (yield* projection.searchThread({ threadId, query: "hidden.example" })).totalMatches,
366 0,
367 );
368 assert.equal(
369 (yield* projection.searchThread({ threadId, query: "check", index: 99 })).activeIndex,
370 3,
371 );
372 }).pipe(Effect.provide(layerTest)),
373 );
374
375 it.effect("skips the empty-response fallback the timeline hides beside an HTML render", () =>
376 Effect.gen(function* () {
377 const projection = yield* setup;
378 const shownRunId = RunId.make("run:shown");
379 yield* commit([run(threadId, shownRunId, 2)]);
380 yield* putItems([
381 {
382 ...item("render", 0, ""),
383 type: "dynamic_tool",
384 toolName: "html_render",
385 input: {},
386 output: { htmlRender: { attachmentId: "visual", title: "Result", height: 120 } },
387 },
388 item("hidden", 1, ""),
389 { ...item("shown", 2, ""), runId: shownRunId },
390 ]);
391 const result = yield* projection.searchThread({ threadId, query: "empty" });
392 assert.equal(result.totalMatches, 1);
393 assert.equal(result.match?.entryId, "shown");
394 }).pipe(Effect.provide(layerTest)),
395 );
396
397 it.effect("searches displayed skill labels and invalidates parsed text when labels change", () =>
398 Effect.gen(function* () {
399 const projection = yield* setup;
400 yield* putItems([item("skill", 1, "Use $test-t3-app now")]);
401 const input = {
402 threadId,
403 query: "T3 App Testing",
404 skills: [{ name: "test-t3-app", displayName: "T3 App Testing" }],
405 };
406 assert.equal((yield* projection.searchThread(input)).totalMatches, 1);
407 assert.equal((yield* projection.searchThread({ ...input, skills: [] })).totalMatches, 0);
408 yield* putItems(
409 [item("skill-plan", 2, "# Verify\nUse $test-t3-app now", "proposed_plan")],
410 "plan",
411 );
412 assert.equal((yield* projection.searchThread(input)).totalMatches, 1);
413 assert.equal(
414 (yield* projection.searchThread({ ...input, query: "$test-t3-app" })).totalMatches,
415 1,
416 );
417 assert.equal((yield* projection.searchThread({ ...input, query: "App" })).totalMatches, 2);
418 }).pipe(Effect.provide(layerTest)),
419 );
420
421 it.effect("never mixes message versions when the searched thread changes during a scan", () =>
422 Effect.gen(function* () {
423 const projection = yield* setup;
424 const sql = yield* SqlClient.SqlClient;
425 const items = Array.from({ length: 260 }, (_, i) => item(`message:${i}`, i, "needle"));
426 yield* putItems(items);
427 const updated = items.map((item) =>
428 item.type === "assistant_message" ? { ...item, text: "needle needle" } : item,
429 );
430 const [during] = yield* Effect.all(
431 [
432 projection.searchThread({ threadId, query: "needle" }),
433 sql.withTransaction(putItems(updated, "concurrent")),
434 ],
435 { concurrency: "unbounded" },
436 );
437 assert.include([260, 520], during.totalMatches);
438 assert.equal(
439 (yield* projection.searchThread({ threadId, query: "needle" })).totalMatches,
440 520,
441 );
442 }).pipe(Effect.provide(layerTest)),
443 );
444
445 it.effect("starts at the reading position and wraps when no later match exists", () =>
446 Effect.gen(function* () {
447 const projection = yield* setup;
448 yield* putItems([
449 item("first", 1, "needle"),
450 item("reading", 2, "needle above, needle below"),
451 item("between", 3, "nothing"),
452 item("last", 4, "needle"),
453 item("end", 5, "nothing"),
454 ]);
455 const from = (entryId: string, occurrence = 0) =>
456 projection.searchThread({ threadId, query: "needle", start: { entryId, occurrence } });
457 assert.equal((yield* from("reading")).activeIndex, 1);
458 const below = yield* from("reading", 1);
459 assert.equal(below.activeIndex, 2);
460 assert.equal(below.match?.occurrence, 1);
461 assert.equal((yield* from("between")).match?.entryId, "last");
462 assert.equal((yield* from("reading", 2)).activeIndex, 3);
463 assert.equal((yield* from("end")).match?.entryId, "first");
464 assert.equal((yield* from("missing")).activeIndex, 0);
465 const explicit = yield* projection.searchThread({
466 threadId,
467 query: "needle",
468 start: { entryId: "last", occurrence: 0 },
469 index: 1,
470 });
471 assert.equal(explicit.match?.entryId, "reading");
472 }).pipe(Effect.provide(layerTest)),
473 );
474
475 it.effect("finds matches beyond the recent history window", () =>
476 Effect.gen(function* () {
477 const projection = yield* setup;
478 const items = Array.from({ length: 520 }, (_, i) =>
479 item(`message:${i}`, i + 1, i === 8 || i === 510 ? "sentinel" : "other message"),
480 );
481 yield* putItems(items);
482 const first = yield* projection.searchThread({ threadId, query: "sentinel" });
483 const last = yield* projection.searchThread({ threadId, query: "sentinel", index: 1 });
484 assert.equal(first.totalMatches, 2);
485 assert.equal(first.match?.entryId, "message:8");
486 assert.equal(last.match?.entryId, "message:510");
487 }).pipe(Effect.provide(layerTest)),
488 );
489
490 it.effect("keeps its caches while Find is in use and drops them after one idle minute", () =>
491 Effect.gen(function* () {
492 const projection = yield* setup;
493 yield* putItems([item("one", 1, "idle needle one"), item("two", 2, "idle needle two")]);
494 const parses = () =>
495 vi
496 .mocked(searchableMessageSegments)
497 .mock.calls.filter(([message]) => message.text.startsWith("idle needle")).length;
498 const search = (query: string) => projection.searchThread({ threadId, query });
499 // One idle minute passes inside this search. The later counts fail if it let the caches go.
500 yield* projection.searchThreadStream({ threadId, query: "needle" }).pipe(
501 Stream.tap((frame) =>
502 frame.complete === false ? TestClock.adjust("1 minute") : Effect.void,
503 ),
504 Stream.runDrain,
505 );
506 assert.equal(parses(), 2);
507 yield* TestClock.adjust("40 seconds");
508 yield* search("idle");
509 yield* TestClock.adjust("40 seconds");
510 // A repeated query is answered from the completed scan and still counts as use.
511 const repeated = yield* projection
512 .searchThreadStream({ threadId, query: "needle" })
513 .pipe(Stream.runCollect);
514 assert.equal(repeated.length, 1);
515 yield* TestClock.adjust("40 seconds");
516 yield* search("one");
517 assert.equal(parses(), 2);
518 yield* TestClock.adjust("1 minute");
519 yield* search("needle");
520 assert.equal(parses(), 4);
521 }).pipe(Effect.provide(layerTest)),
522 );
523
524 it.effect("invalidates cached counts when streamed text changes or a run rolls back", () =>
525 Effect.gen(function* () {
526 const projection = yield* setup;
527 const original = item("streaming", 1, "needle");
528 yield* putItems([original]);
529 assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1);
530 if (original.type !== "assistant_message") return;
531 yield* putItems([{ ...original, text: "needle needle", streaming: true }], "streaming");
532 assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 2);
533 const updated = run(threadId, runId, 1, "rolled_back");
534 yield* commit([{ ...updated, type: "run.updated" }]);
535 assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 0);
536 }).pipe(Effect.provide(layerTest)),
537 );
538
539 it.effect(
540 "does not search undispatched inputs, cancelled queue rows, or folded question answers",
541 () =>
542 Effect.gen(function* () {
543 const projection = yield* setup;
544 const queuedRunId = RunId.make("queued:cancelled");
545 yield* commit([run(threadId, queuedRunId, 2, "cancelled")]);
546 const queued = item("cancelled", 1, "needle", "user_message");
547 if (queued.type !== "user_message") return;
548 const answer = item("async-answer:question", 2, "needle", "user_message");
549 const question: OrchestrationV2TurnItem = {
550 ...item("question", 3, ""),
551 type: "user_input_request",
552 requestId: RuntimeRequestId.make("question"),
553 questions: [],
554 questionAnswer: { requestId: "question", answers: {}, attachmentsByQuestionId: {} },
555 };
556 yield* putItems([
557 { ...queued, runId: queuedRunId, inputIntent: "queued_turn" },
558 answer,
559 question,
560 ]);
561 const event: OrchestrationV2DomainEvent = {
562 id: EventId.make("undispatched"),
563 threadId,
564 occurredAt: at(2),
565 type: "message.updated",
566 payload: {
567 id: MessageId.make("undispatched"),
568 threadId,
569 runId: null,
570 nodeId: null,
571 role: "user",
572 text: "needle",
573 attachments: [],
574 streaming: false,
575 createdBy: "user",
576 creationSource: "web",
577 createdAt: at(2),
578 updatedAt: at(2),
579 },
580 };
581 yield* commit([event]);
582 const result = yield* projection.searchThread({ threadId, query: "needle" });
583 assert.equal(result.totalMatches, 0);
584 assert.isNull(result.match);
585 }).pipe(Effect.provide(layerTest)),
586 );
587
588 it.effect("finds an async answer that started a later run", () =>
589 Effect.gen(function* () {
590 yield* createProject(projectId);
591 const projection = yield* ProjectionStore.ProjectionStoreV2;
592 const queuedRun = RunId.make("run:queued");
593 const answerRun = RunId.make("run:answer");
594 yield* commit([
595 thread(threadId, projectId),
596 run(),
597 run(threadId, queuedRun, 2),
598 run(threadId, answerRun, 3),
599 ]);
600 const requestId = RuntimeRequestId.make("late");
601 const question: OrchestrationV2TurnItem = {
602 ...item("question", 1, ""),
603 type: "user_input_request",
604 requestId,
605 questions: [],
606 questionAnswer: { requestId, answers: {}, attachmentsByQuestionId: {} },
607 };
608 const items = [
609 question,
610 { ...item("queued", 2, "queued follow-up", "user_message"), runId: queuedRun },
611 { ...item(`async-answer:${requestId}`, 3, "needle", "user_message"), runId: answerRun },
612 ];
613 yield* commit(
614 items.map((payload) => ({
615 id: EventId.make(`${payload.id}:initial`),
616 threadId,
617 occurredAt: at(1),
618 type: "turn-item.updated" as const,
619 payload,
620 })),
621 );
622 const result = yield* projection.searchThread({ threadId, query: "needle" });
623 assert.equal(result.totalMatches, 1);
624 assert.deepEqual(result.match, {
625 entryId: `async-answer:${requestId}`,
626 runId: answerRun,
627 occurrence: 0,
628 });
629 }).pipe(Effect.provide(layerTest)),
630 );
631
632 it.effect(
633 "searches inherited fork history through the fork boundary, using source item identities",
634 () =>
635 Effect.gen(function* () {
636 const projection = yield* setup;
637 const parent = ThreadId.make("thread:parent");
638 const laterRun = RunId.make("run:later");
639 const parentRun = RunId.make("run:parent");
640 yield* commit([
641 thread(parent, projectId),
642 run(parent, parentRun),
643 run(parent, laterRun, 2),
644 ]);
645 yield* putItems([
646 { ...item("inherited", 1, "needle", "assistant_message", parent), runId: parentRun },
647 { ...item("after-fork", 2, "needle", "assistant_message", parent), runId: laterRun },
648 ]);
649 const child = thread(threadId, projectId);
650 if (child.type !== "thread.created") return;
651 yield* commit([
652 {
653 ...child,
654 id: EventId.make("child:fork"),
655 type: "thread.metadata-updated",
656 payload: {
657 ...child.payload,
658 forkedFrom: { type: "run", threadId: parent, runId: parentRun },
659 },
660 },
661 ]);
662 yield* putItems([item("local", 1, "needle")]);
663 const inherited = yield* projection.searchThread({ threadId, query: "needle" });
664 assert.equal(inherited.totalMatches, 2);
665 assert.equal(inherited.match?.entryId, "inherited");
666 assert.equal(
667 (yield* projection.searchThread({ threadId, query: "needle", index: 1 })).match?.entryId,
668 "local",
669 );
670 const snapshot = yield* projection.getThreadSnapshot(threadId);
671 const history = yield* projection.getThreadHistoryPage(
672 threadId,
673 encodeThreadHistoryCursor({
674 snapshotSequence: snapshot.snapshotSequence,
675 sourceThreadId: threadId,
676 sourceItemId: "local",
677 position: 2,
678 }),
679 );
680 assert.deepEqual(
681 history.items,
682 snapshot.projection.visibleTurnItems
683 .slice(0, -1)
684 .map((row) => ({ ...row, item: projectTurnItemForWire(row.item) })),
685 );
686 assert.equal(history.items[0]?.sourceThreadId, parent);
687 assert.equal(history.items.at(-1)?.item.type, "fork");
688 yield* putItems(
689 [{ ...item("inherited", 1, "gone", "assistant_message", parent), runId: parentRun }],
690 "parent-update",
691 );
692 assert.equal(
693 (yield* projection.searchThread({ threadId, query: "needle" })).totalMatches,
694 1,
695 );
696 }).pipe(Effect.provide(layerTest)),
697 );
698
699 it.effect("rejects deleted threads and keeps results scoped to their thread", () =>
700 Effect.gen(function* () {
701 const projection = yield* setup;
702 const other = ThreadId.make("thread:other");
703 yield* commit([thread(other, projectId), run(other, RunId.make("run:other"))]);
704 yield* putItems([
705 item("shared-id", 1, "needle"),
706 item("other-id", 1, "other text", "assistant_message", other),
707 ]);
708 assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1);
709 assert.equal(
710 (yield* projection.searchThread({ threadId: other, query: "needle" })).totalMatches,
711 0,
712 );
713 const deleted = thread(threadId, projectId, { deletedAt: at(3) });
714 if (deleted.type !== "thread.created") return;
715 yield* commit([{ ...deleted, id: EventId.make("deleted"), type: "thread.deleted" }]);
716 const error = yield* Effect.flip(projection.searchThread({ threadId, query: "needle" }));
717 assert.equal(error._tag, "ProjectionStoreThreadNotFoundError");
718 const historyError = yield* Effect.flip(
719 projection.getThreadHistoryPage(
720 threadId,
721 encodeThreadHistoryCursor({
722 snapshotSequence: 0,
723 sourceThreadId: threadId,
724 sourceItemId: "shared-id",
725 position: 0,
726 }),
727 ),
728 );
729 assert.equal(historyError._tag, "ProjectionStoreThreadNotFoundError");
730 }).pipe(Effect.provide(layerTest)),
731 );
732
733 it.effect("invalidates a cached fork search when its empty source gains its first message", () =>
734 Effect.gen(function* () {
735 const projection = yield* setup;
736 const parent = ThreadId.make("thread:empty-parent");
737 const parentRun = RunId.make("run:empty-parent");
738 yield* commit([thread(parent, projectId), run(parent, parentRun)]);
739 const child = thread(threadId, projectId);
740 if (child.type !== "thread.created") return;
741 yield* commit([
742 {
743 ...child,
744 id: EventId.make("child:empty-fork"),
745 type: "thread.metadata-updated",
746 payload: {
747 ...child.payload,
748 forkedFrom: { type: "run", threadId: parent, runId: parentRun },
749 },
750 },
751 ]);
752 yield* putItems([item("local", 1, "needle")]);
753 assert.equal((yield* projection.searchThread({ threadId, query: "needle" })).totalMatches, 1);
754 yield* putItems([
755 { ...item("inherited", 1, "needle", "assistant_message", parent), runId: parentRun },
756 ]);
757 const updated = yield* projection.searchThread({ threadId, query: "needle" });
758 assert.equal(updated.totalMatches, 2);
759 assert.equal(updated.match?.entryId, "inherited");
760 }).pipe(Effect.provide(layerTest)),
761 );
762});
763
764describe("V2 history materialization", () => {
765 it.effect(
766 "combines history pages through a match's message identity without skipping intervening turns",
767 () =>
768 Effect.gen(function* () {
769 const projection = yield* setup;
770 const items = Array.from({ length: 45 }, (_, turn) => [
771 item(`user:${turn}`, turn * 3, `turn ${turn}`, "user_message"),
772 {
773 ...item(`answer:${turn}`, turn * 3 + 1, `answer ${turn}`),
774 messageId: MessageId.make(`message:${turn}`),
775 },
776 {
777 ...item(`tool:${turn}`, turn * 3 + 2, ""),
778 type: "command_execution" as const,
779 input: "pwd",
780 output: "huge output".repeat(1000),
781 exitCode: 0,
782 },
783 ]).flat();
784 yield* putItems(items);
785 const snapshot = yield* projection.getThreadSnapshot(threadId);
786 const cursor = encodeThreadHistoryCursor({
787 snapshotSequence: snapshot.snapshotSequence,
788 sourceThreadId: threadId,
789 sourceItemId: "absent",
790 position: items.length,
791 });
792 const ordinary = yield* projection.getThreadHistoryPage(threadId, cursor);
793 assert.equal(ordinary.items[0]?.sourceItemId, "user:25");
794 const targeted = yield* projection.getThreadHistoryPage(threadId, cursor, "message:10");
795 assert.equal(targeted.items[0]?.sourceItemId, "user:10");
796 assert.deepEqual(
797 targeted.items.map((row) => row.sourceItemId),
798 items.slice(30).map((entry) => entry.id),
799 );
800 const conversation = yield* projection.getThreadHistoryPage(
801 threadId,
802 cursor,
803 "message:10",
804 true,
805 );
806 assert.equal(conversation.nextCursor, targeted.nextCursor);
807 assert.equal(conversation.hasMoreHistory, targeted.hasMoreHistory);
808 assert.deepEqual(
809 conversation.items.map((row) => row.sourceItemId),
810 targeted.items
811 .filter((row) => row.item.type !== "command_execution")
812 .map((row) => row.sourceItemId),
813 );
814 assert.equal(conversation.items.length, 70);
815 assert.equal(targeted.items.length, 105);
816 assert.isTrue(targeted.hasMoreHistory);
817 const earlier = yield* projection.getThreadHistoryPage(threadId, targeted.nextCursor!);
818 assert.deepEqual(
819 earlier.items.map((row) => row.sourceItemId),
820 items.slice(0, 30).map((entry) => entry.id),
821 );
822 assert.isFalse(earlier.hasMoreHistory);
823 const capped = yield* projection.getThreadHistoryPage(threadId, cursor, "message:0");
824 assert.equal(capped.items[0]?.sourceItemId, "user:5");
825 const absent = yield* projection.getThreadHistoryPage(
826 threadId,
827 cursor,
828 "another-thread-message",
829 );
830 assert.deepEqual(absent, ordinary);
831 }).pipe(Effect.provide(layerTest)),
832 );
833
834 it.effect("pages long turns without duplicating or skipping messages", () =>
835 Effect.gen(function* () {
836 const projection = yield* setup;
837 const items = Array.from({ length: 45 }, (_, turn) => [
838 item(`user:${turn}`, turn * 3, `turn ${turn}`, "user_message"),
839 {
840 ...item(`steer:${turn}`, turn * 3 + 1, "steer", "user_message"),
841 inputIntent: "steer" as const,
842 },
843 item(
844 `answer:${turn}`,
845 turn * 3 + 2,
846 turn === 40 ? "large ".repeat(200_000) : `answer ${turn}`,
847 ),
848 ]).flat();
849 yield* putItems(items);
850 const snapshot = yield* projection.getThreadSnapshot(threadId);
851 const rows = snapshot.projection.visibleTurnItems.map((row) => ({
852 ...row,
853 item: projectTurnItemForWire(row.item),
854 }));
855 let cursor: string | null = encodeThreadHistoryCursor({
856 snapshotSequence: snapshot.snapshotSequence,
857 sourceThreadId: threadId,
858 sourceItemId: "absent",
859 position: rows.length,
860 });
861 const collected: string[] = [];
862 while (cursor !== null) {
863 const page: OrchestrationV2ThreadHistoryPage = yield* projection.getThreadHistoryPage(
864 threadId,
865 cursor,
866 );
867 assert.deepEqual(page, {
868 snapshotSequence: snapshot.snapshotSequence,
869 ...selectHistoryPageFromCursor({
870 items: rows,
871 cursor,
872 snapshotSequence: snapshot.snapshotSequence,
873 }),
874 });
875 // A turn larger than the byte budget is split, but only a lone oversized item may exceed it.
876 assert.isTrue(
877 page.items.length === 1 ||
878 Buffer.byteLength(JSON.stringify(page.items), "utf8") <=
879 THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes,
880 );
881 collected.unshift(...page.items.map((row) => String(row.sourceItemId)));
882 cursor = page.nextCursor;
883 }
884 assert.deepEqual(
885 collected,
886 items.map((entry) => String(entry.id)),
887 );
888 }).pipe(Effect.provide(layerTest)),
889 );
890
891 it.effect("enforces fallback item and byte limits and always admits an oversized row", () =>
892 Effect.gen(function* () {
893 const projection = yield* setup;
894 yield* putItems(
895 Array.from({ length: 80 }, (_, ordinal) =>
896 item(`row:${ordinal}`, ordinal, ordinal < 78 ? "small" : "large ".repeat(200_000)),
897 ),
898 );
899 const snapshot = yield* projection.getThreadSnapshot(threadId);
900 const rows = snapshot.projection.visibleTurnItems.map((row) => ({
901 ...row,
902 item: projectTurnItemForWire(row.item),
903 }));
904 let cursor: string | null = encodeThreadHistoryCursor({
905 snapshotSequence: snapshot.snapshotSequence,
906 sourceThreadId: threadId,
907 sourceItemId: "absent",
908 position: rows.length,
909 });
910 const sizes: number[] = [];
911 while (cursor !== null) {
912 const page: OrchestrationV2ThreadHistoryPage = yield* projection.getThreadHistoryPage(
913 threadId,
914 cursor,
915 );
916 assert.deepEqual(page, {
917 snapshotSequence: snapshot.snapshotSequence,
918 ...selectHistoryPageFromCursor({
919 items: rows,
920 cursor,
921 snapshotSequence: snapshot.snapshotSequence,
922 }),
923 });
924 sizes.push(page.items.length);
925 cursor = page.nextCursor;
926 }
927 assert.deepEqual(sizes, [1, 1, 75, 3]);
928 }).pipe(Effect.provide(layerTest)),
929 );
930});