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