Open History to see the last commit
1import { assert, it } from "@effect/vitest";
2import {
3 CheckpointScopeId,
4 ContextTransferId,
5 NodeId,
6 EventId,
7 MessageId,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 RunId,
12 RuntimeRequestId,
13 ThreadId,
14 TurnItemId,
15 type OrchestrationV2AppThread,
16 type OrchestrationV2Run,
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 SqlitePersistence from "../persistence/Sqlite.ts";
23import * as ProjectionStore from "./ProjectionStore.ts";
24
25const layerDatabase = Layer.mergeAll(
26 SqlitePersistence.layerMemory,
27 ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
28);
29
30it.effect.each(["sqlite"] as const)(
31 "loads startup state without obsolete transcript payloads in %s",
32 (storage) =>
33 Effect.gen(function* () {
34 const store = yield* ProjectionStore.ProjectionStoreV2;
35 const now = yield* DateTime.now;
36 const threadId = ThreadId.make("thread:completion-reads");
37 const runId = RunId.make("run:completion-reads");
38 const messageId = MessageId.make("message:completion-reads");
39 const instanceId = ProviderInstanceId.make("codex");
40 const thread: OrchestrationV2AppThread = {
41 id: threadId,
42 projectId: ProjectId.make("project:completion-reads"),
43 title: "Completion controls",
44 providerInstanceId: instanceId,
45 modelSelection: { instanceId, model: "test" },
46 runtimeMode: "full-access",
47 interactionMode: "default",
48 branch: null,
49 worktreePath: null,
50 activeProviderThreadId: null,
51 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
52 forkedFrom: null,
53 createdBy: "user",
54 creationSource: "web",
55 createdAt: now,
56 updatedAt: now,
57 archivedAt: null,
58 settledOverride: null,
59 settledAt: null,
60 lastVisitedAt: null,
61 deletedAt: null,
62 };
63 const run: OrchestrationV2Run = {
64 id: runId,
65 threadId,
66 ordinal: 2,
67 providerInstanceId: instanceId,
68 modelSelection: thread.modelSelection,
69 providerThreadId: null,
70 userMessageId: messageId,
71 rootNodeId: NodeId.make("startup:root"),
72 activeAttemptId: null,
73 status: "queued",
74 requestedAt: now,
75 startedAt: null,
76 completedAt: null,
77 checkpointId: null,
78 contextHandoffId: null,
79 };
80 const putThread = (value: OrchestrationV2AppThread) =>
81 store.apply({
82 id: EventId.make(`event:thread:${value.id}`),
83 type: "thread.created",
84 threadId: value.id,
85 occurredAt: now,
86 payload: value,
87 });
88 const putRun = (value: OrchestrationV2Run) =>
89 store.apply({
90 id: EventId.make(`event:run:${value.id}:${value.status}`),
91 type: "run.updated",
92 threadId,
93 runId: value.id,
94 occurredAt: now,
95 payload: value,
96 });
97 yield* putThread(thread);
98 yield* putRun(run);
99 const scopeId = CheckpointScopeId.make("startup:shared-scope");
100 yield* store.apply({
101 id: EventId.make("startup:root-event"),
102 type: "node.updated",
103 threadId,
104 occurredAt: now,
105 payload: {
106 id: run.rootNodeId!,
107 threadId,
108 runId,
109 parentNodeId: null,
110 rootNodeId: run.rootNodeId!,
111 kind: "root_turn",
112 status: "pending",
113 countsForRun: true,
114 providerThreadId: null,
115 providerTurnId: null,
116 nativeItemRef: null,
117 runtimeRequestId: null,
118 checkpointScopeId: scopeId,
119 startedAt: null,
120 completedAt: null,
121 },
122 });
123 yield* store.apply({
124 id: EventId.make("startup:scope-event"),
125 type: "checkpoint-scope.created",
126 threadId,
127 occurredAt: now,
128 payload: {
129 id: scopeId,
130 threadId,
131 runId: RunId.make("run:old"),
132 nodeId: NodeId.make("startup:earlier-root"),
133 parentScopeId: null,
134 providerThreadId: null,
135 kind: "root_run",
136 ordinalWithinParent: 0,
137 advancesAppRunCount: true,
138 cwd: "/repo/worktree",
139 createdAt: now,
140 },
141 });
142 const message = {
143 id: messageId,
144 threadId,
145 runId,
146 nodeId: null,
147 role: "user" as const,
148 text: "Continue",
149 attachments: [],
150 streaming: false,
151 createdBy: "agent" as const,
152 creationSource: "server" as const,
153 createdAt: now,
154 updatedAt: now,
155 delegatedCompletion: { parentRunId: RunId.make("parent-run"), generation: 1, taskIds: [] },
156 };
157 yield* store.apply({
158 id: EventId.make("event:input"),
159 type: "message.updated",
160 threadId,
161 runId,
162 occurredAt: now,
163 payload: message,
164 });
165 const oldRunId = RunId.make("run:old");
166 yield* putRun({ ...run, id: oldRunId, ordinal: 1, status: "completed" });
167 const item = {
168 id: TurnItemId.make("item:history"),
169 threadId,
170 runId: oldRunId,
171 nodeId: null,
172 providerThreadId: null,
173 providerTurnId: null,
174 nativeItemRef: null,
175 parentItemId: null,
176 ordinal: 1,
177 status: "completed" as const,
178 title: "old command",
179 startedAt: now,
180 completedAt: now,
181 updatedAt: now,
182 type: "command_execution" as const,
183 input: "echo history",
184 output: "historical result",
185 };
186 yield* store.apply({
187 id: EventId.make("item-event"),
188 type: "turn-item.updated",
189 threadId,
190 runId: oldRunId,
191 occurredAt: now,
192 payload: item,
193 });
194 if (storage === "sqlite") {
195 const sql = yield* SqlClient.SqlClient;
196 yield* sql`INSERT INTO orchestration_v2_projection_messages
197 (message_id, thread_id, run_id, node_id, role, streaming, created_at, updated_at, payload_json)
198 VALUES ('obsolete-history', ${threadId}, NULL, NULL, 'assistant', 0,
199 ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
200 yield* sql`INSERT INTO orchestration_v2_projection_turn_items
201 (turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
202 type, status, ordinal, updated_at, payload_json)
203 VALUES ('obsolete-tool', ${threadId}, ${oldRunId}, NULL, NULL, NULL,
204 'tool_call', 'completed', 2, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
205 assert.equal((yield* Effect.exit(store.getThreadProjection(threadId)))._tag, "Failure");
206 }
207 const selected = yield* store.getThreadRecords(threadId, ["runs", "messages"], {
208 runIds: [runId],
209 messageIds: [messageId],
210 });
211 assert.deepEqual(Object.keys(selected).sort(), ["messages", "runs", "thread"]);
212 assert.deepEqual(
213 selected.runs.map((r) => r.id),
214 [runId],
215 );
216 assert.deepEqual(
217 selected.messages.map((m) => m.id),
218 [messageId],
219 );
220 assert.deepEqual(
221 (yield* store.getThreadRecords(threadId, ["messages"], { messageIds: [] })).messages,
222 [],
223 );
224 assert.deepEqual(
225 (yield* store.getThreadRecords(threadId, ["turnItems"], {
226 turnItemTypes: ["command_execution"],
227 turnItemRunId: oldRunId,
228 })).turnItems,
229 [item],
230 );
231 assert.deepEqual(
232 (yield* store.getThreadRecords(threadId, ["turnItems"], {
233 turnItemTypes: ["command_execution"],
234 turnItemRunId: runId,
235 })).turnItems,
236 [],
237 );
238 assert.equal(yield* store.getNextTurnItemOrdinal(threadId), storage === "sqlite" ? 3 : 2);
239 assert.equal(yield* store.getMessageCount(threadId), storage === "sqlite" ? 2 : 1);
240 assert.deepEqual(
241 (yield* store.getThreadRecords(threadId, ["messages"], {
242 messageRunIds: [runId],
243 })).messages.map((message) => message.id),
244 [messageId],
245 );
246 assert.deepEqual(
247 (yield* store.getThreadRecords(threadId, ["messages"], { messageRunIds: [] })).messages,
248 [],
249 );
250 assert.deepEqual(
251 (yield* store.getThreadRecords(threadId, ["turnItems"], {
252 turnItemTypes: ["command_execution"],
253 turnItemRunIds: [oldRunId],
254 })).turnItems,
255 [item],
256 );
257 assert.deepEqual(
258 (yield* store.getThreadRecords(threadId, ["turnItems"], { turnItemRunIds: [] })).turnItems,
259 [],
260 );
261 assert.deepEqual(yield* store.getThreadAttachmentIds(threadId), []);
262 if (storage === "sqlite") {
263 const sql = yield* SqlClient.SqlClient;
264 yield* sql`UPDATE orchestration_v2_projection_messages SET payload_json = json_object('text', ${"x".repeat(512_000)}, 'attachments', json_array(json_object('id', 'attachment:old'))) WHERE message_id = 'obsolete-history'`;
265 assert.deepEqual(yield* store.getThreadAttachmentIds(threadId), ["attachment:old"]);
266 }
267 const page = yield* store.getTimelinePage(threadId, { view: "activity", limit: 1 });
268 assert.deepEqual(
269 page.items.map((row) => row.item),
270 [item],
271 );
272 assert.equal(page.hasMore, storage === "sqlite");
273 assert.equal(page.totalItems, storage === "sqlite" ? 2 : 1);
274 assert.deepEqual(
275 (yield* store.getTimelinePage(threadId, { view: "messages", limit: 1 })).items,
276 [],
277 );
278 const context = yield* store.getTurnStartContext(threadId, runId);
279 assert.equal(context.thread.id, threadId);
280 assert.deepEqual(
281 context.checkpointScopes.map((scope) => scope.id),
282 [scopeId],
283 );
284 assert.equal(context.messages.find((m) => m.id === messageId)?.text, "Continue");
285 assert.isTrue(context.hasConversation);
286 assert.deepEqual(context.turnItems, []);
287 assert.deepEqual(yield* store.getTurnStartHistory(threadId), [item]);
288 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [oldRunId]), [item]);
289 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [runId]), []);
290 assert.deepEqual(yield* store.getTurnStartHistory(threadId, []), []);
291 const resultId = TurnItemId.make("interrupt-result");
292 assert.isTrue(yield* store.hasUnpairedRunInterruptRequest(threadId, item.id, resultId));
293 assert.isFalse(yield* store.hasUnpairedRunInterruptRequest(threadId, resultId, item.id));
294 assert.isFalse(
295 yield* store.hasUnpairedRunInterruptRequest(
296 ThreadId.make("other-thread"),
297 item.id,
298 resultId,
299 ),
300 );
301 yield* store.apply({
302 id: EventId.make("interrupt-result-event"),
303 type: "turn-item.updated",
304 threadId,
305 runId: oldRunId,
306 occurredAt: now,
307 payload: { ...item, id: resultId },
308 });
309 assert.isFalse(yield* store.hasUnpairedRunInterruptRequest(threadId, item.id, resultId));
310 if (storage === "sqlite") {
311 const sql = yield* SqlClient.SqlClient;
312 const original = yield* sql<{ payload_json: string }>`SELECT payload_json
313 FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${item.id}`;
314 yield* sql`UPDATE orchestration_v2_projection_turn_items
315 SET payload_json = 'invalid JSON' WHERE turn_item_id = ${item.id}`;
316 assert.isTrue(
317 yield* store.hasUnpairedRunInterruptRequest(
318 threadId,
319 item.id,
320 TurnItemId.make("absent-result"),
321 ),
322 );
323 yield* sql`UPDATE orchestration_v2_projection_turn_items
324 SET payload_json = ${original[0]!.payload_json} WHERE turn_item_id = ${item.id}`;
325 }
326 for (const text of ["/compact", " \t/COMPACT\n", "\u00a0/compact\u3000"]) {
327 yield* store.apply({
328 id: EventId.make("compact-input"),
329 type: "message.updated",
330 threadId,
331 runId,
332 occurredAt: now,
333 payload: { ...message, text },
334 });
335 assert.isFalse((yield* store.getTurnStartContext(threadId, runId)).hasConversation);
336 }
337 const marker = {
338 ...message,
339 id: MessageId.make("old-compact"),
340 runId: oldRunId,
341 text: "\t/COMPACT\n",
342 };
343 yield* store.apply({
344 id: EventId.make("old-compact-input"),
345 type: "message.updated",
346 threadId,
347 runId: oldRunId,
348 occurredAt: now,
349 payload: marker,
350 });
351 assert.include(
352 (yield* store.getTurnStartContext(threadId, runId)).messages.map((m) => m.id),
353 marker.id,
354 );
355 const other = ThreadId.make("other");
356 yield* putThread({ ...thread, id: other });
357 assert.deepEqual((yield* store.getTurnStartContext(other, runId)).messages, []);
358 assert.deepEqual(yield* store.getTurnStartHistory(other, [oldRunId]), []);
359 assert.isFalse((yield* store.getTurnStartContext(other, runId)).hasConversation);
360 assert.instanceOf(
361 yield* store.getTurnStartContext(ThreadId.make("missing"), runId).pipe(Effect.flip),
362 ProjectionStore.ProjectionStoreThreadNotFoundError,
363 );
364 const file = {
365 type: "file" as const,
366 id: "message-file",
367 name: "notes.pdf",
368 mimeType: "application/pdf",
369 sizeBytes: 12,
370 };
371 yield* store.apply({
372 id: EventId.make("attachment:message-event"),
373 type: "message.updated",
374 threadId,
375 runId,
376 occurredAt: now,
377 payload: { ...message, attachments: [file] },
378 });
379 yield* store.apply({
380 id: EventId.make("attachment:question-event"),
381 type: "turn-item.updated",
382 threadId,
383 runId: oldRunId,
384 occurredAt: now,
385 payload: {
386 ...item,
387 id: TurnItemId.make("attachment:question"),
388 type: "user_input_request",
389 requestId: RuntimeRequestId.make("attachment:request"),
390 questions: [],
391 questionAnswer: {
392 requestId: "attachment:request",
393 answers: { document: ["See document"], image: ["See image"] },
394 attachmentsByQuestionId: {
395 document: [{ ...file, id: "answer-document" }],
396 image: [
397 {
398 type: "image",
399 id: "answer-image",
400 name: "photo.png",
401 mimeType: "image/png",
402 sizeBytes: 8,
403 },
404 ],
405 },
406 },
407 },
408 });
409 const recordingId = "thread-completion-reads-0b9f6c1e-3c1d-4f7a-9d2e-5a6b7c8d9e0f-webm";
410 yield* store.apply({
411 id: EventId.make("attachment:recording-event"),
412 type: "turn-item.updated",
413 threadId,
414 runId: oldRunId,
415 occurredAt: now,
416 payload: {
417 ...item,
418 id: TurnItemId.make("attachment:recording"),
419 type: "dynamic_tool",
420 toolName: "mcp__t3-code__preview_recording_stop",
421 input: {},
422 output: [
423 { type: "text", text: JSON.stringify({ id: recordingId, mimeType: "video/webm" }) },
424 ],
425 },
426 });
427 assert.deepEqual((yield* store.getThreadAttachmentIds(threadId)).toSorted(), [
428 "answer-document",
429 "answer-image",
430 "attachment:old",
431 "message-file",
432 recordingId,
433 ]);
434 }).pipe(Effect.provide(layerDatabase)),
435);
436
437it.effect(
438 "counts a native fork's inherited history as conversation for its first /compact in sqlite",
439 () =>
440 Effect.gen(function* () {
441 const store = yield* ProjectionStore.ProjectionStoreV2;
442 const now = yield* DateTime.now;
443 const instanceId = ProviderInstanceId.make("claudeAgent");
444 const sourceId = ThreadId.make("thread:fork-source");
445 const forkId = ThreadId.make("thread:fork-target");
446 const sourceRunId = RunId.make("run:fork-source-third");
447 const forkRunId = RunId.make("run:fork-compact");
448 const source: OrchestrationV2AppThread = {
449 id: sourceId,
450 projectId: ProjectId.make("project:fork-compact"),
451 title: "Source",
452 providerInstanceId: instanceId,
453 modelSelection: { instanceId, model: "claude-opus-5-5" },
454 runtimeMode: "full-access",
455 interactionMode: "default",
456 branch: null,
457 worktreePath: null,
458 activeProviderThreadId: null,
459 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: sourceId },
460 forkedFrom: null,
461 createdBy: "user",
462 creationSource: "web",
463 createdAt: now,
464 updatedAt: now,
465 archivedAt: null,
466 settledOverride: null,
467 settledAt: null,
468 lastVisitedAt: null,
469 deletedAt: null,
470 };
471 const fork: OrchestrationV2AppThread = {
472 ...source,
473 id: forkId,
474 title: "Fork",
475 lineage: { parentThreadId: sourceId, relationshipToParent: "fork", rootThreadId: sourceId },
476 forkedFrom: { type: "run", threadId: sourceId, runId: sourceRunId },
477 };
478 const run = (threadId: ThreadId, id: RunId, status: OrchestrationV2Run["status"]) =>
479 ({
480 id,
481 threadId,
482 ordinal: 1,
483 providerInstanceId: instanceId,
484 modelSelection: source.modelSelection,
485 providerThreadId: null,
486 userMessageId: MessageId.make(`message:${id}`),
487 rootNodeId: NodeId.make(`node:${id}`),
488 activeAttemptId: null,
489 status,
490 requestedAt: now,
491 startedAt: null,
492 completedAt: null,
493 checkpointId: null,
494 contextHandoffId: null,
495 }) satisfies OrchestrationV2Run;
496 const userMessage = (threadId: ThreadId, runId: RunId, text: string) =>
497 store.apply({
498 id: EventId.make(`event:message:${runId}`),
499 type: "message.updated",
500 threadId,
501 runId,
502 occurredAt: now,
503 payload: {
504 id: MessageId.make(`message:${runId}`),
505 threadId,
506 runId,
507 nodeId: null,
508 role: "user",
509 text,
510 attachments: [],
511 streaming: false,
512 createdBy: "user",
513 creationSource: "web",
514 createdAt: now,
515 updatedAt: now,
516 },
517 });
518 for (const thread of [source, fork]) {
519 yield* store.apply({
520 id: EventId.make(`event:thread:${thread.id}`),
521 type: "thread.created",
522 threadId: thread.id,
523 occurredAt: now,
524 payload: thread,
525 });
526 }
527 for (const [threadId, value] of [
528 [sourceId, run(sourceId, sourceRunId, "completed")],
529 [forkId, run(forkId, forkRunId, "queued")],
530 ] as const) {
531 yield* store.apply({
532 id: EventId.make(`event:run:${value.id}`),
533 type: "run.updated",
534 threadId,
535 runId: value.id,
536 occurredAt: now,
537 payload: value,
538 });
539 }
540 yield* userMessage(sourceId, sourceRunId, "THIRD");
541 yield* store.apply({
542 id: EventId.make("event:fork-transfer"),
543 type: "context-transfer.created",
544 threadId: forkId,
545 occurredAt: now,
546 payload: {
547 id: ContextTransferId.make("transfer:fork-compact"),
548 type: "fork",
549 sourceThreadId: sourceId,
550 targetThreadId: forkId,
551 sourcePoint: { threadId: sourceId, runId: sourceRunId },
552 basePoint: null,
553 sourceProviderInstanceId: instanceId,
554 targetProviderInstanceId: instanceId,
555 targetRunId: forkRunId,
556 status: "pending",
557 resolution: {
558 strategy: "native_fork",
559 providerThreadRef: {
560 driver: ProviderDriverKind.make("claudeAgent"),
561 nativeId: "native-fork-source",
562 strength: "strong",
563 },
564 },
565 createdBy: "user",
566 error: null,
567 createdAt: now,
568 updatedAt: now,
569 consumedAt: null,
570 },
571 });
572 yield* userMessage(forkId, forkRunId, "/compact");
573
574 assert.isTrue((yield* store.getTurnStartContext(forkId, forkRunId)).hasConversation);
575 }).pipe(Effect.provide(layerDatabase)),
576);