1import { assert, it } from "@effect/vitest";
2import {
3 CheckpointScopeId,
4 NodeId,
5 EventId,
6 MessageId,
7 ProjectId,
8 ProviderInstanceId,
9 RunId,
10 ThreadId,
11 TurnItemId,
12 type OrchestrationV2AppThread,
13 type OrchestrationV2Run,
14} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as SqlClient from "effect/sql/SqlClient";
19import * as SqlitePersistence from "../persistence/Sqlite.ts";
20import * as ProjectionStore from "./ProjectionStore.ts";
21
22const layerDatabase = Layer.mergeAll(
23 SqlitePersistence.layerMemory,
24 ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
25);
26
27it.effect.each(["sqlite", "memory"] as const)(
28 "loads startup state without obsolete transcript payloads in %s",
29 (storage) =>
30 Effect.gen(function* () {
31 const store = yield* ProjectionStore.ProjectionStoreV2;
32 const now = yield* DateTime.now;
33 const threadId = ThreadId.make("thread:completion-reads");
34 const runId = RunId.make("run:completion-reads");
35 const messageId = MessageId.make("message:completion-reads");
36 const instanceId = ProviderInstanceId.make("codex");
37 const thread: OrchestrationV2AppThread = {
38 id: threadId,
39 projectId: ProjectId.make("project:completion-reads"),
40 title: "Completion controls",
41 providerInstanceId: instanceId,
42 modelSelection: { instanceId, model: "test" },
43 runtimeMode: "full-access",
44 interactionMode: "default",
45 branch: null,
46 worktreePath: null,
47 activeProviderThreadId: null,
48 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
49 forkedFrom: null,
50 createdBy: "user",
51 creationSource: "web",
52 createdAt: now,
53 updatedAt: now,
54 archivedAt: null,
55 settledOverride: null,
56 settledAt: null,
57 lastVisitedAt: null,
58 deletedAt: null,
59 };
60 const run: OrchestrationV2Run = {
61 id: runId,
62 threadId,
63 ordinal: 2,
64 providerInstanceId: instanceId,
65 modelSelection: thread.modelSelection,
66 providerThreadId: null,
67 userMessageId: messageId,
68 rootNodeId: NodeId.make("startup:root"),
69 activeAttemptId: null,
70 status: "queued",
71 requestedAt: now,
72 startedAt: null,
73 completedAt: null,
74 checkpointId: null,
75 contextHandoffId: null,
76 };
77 const putThread = (value: OrchestrationV2AppThread) =>
78 store.apply({
79 id: EventId.make(`event:thread:${value.id}`),
80 type: "thread.created",
81 threadId: value.id,
82 occurredAt: now,
83 payload: value,
84 });
85 const putRun = (value: OrchestrationV2Run) =>
86 store.apply({
87 id: EventId.make(`event:run:${value.id}:${value.status}`),
88 type: "run.updated",
89 threadId,
90 runId: value.id,
91 occurredAt: now,
92 payload: value,
93 });
94 yield* putThread(thread);
95 yield* putRun(run);
96 const scopeId = CheckpointScopeId.make("startup:shared-scope");
97 yield* store.apply({
98 id: EventId.make("startup:root-event"),
99 type: "node.updated",
100 threadId,
101 occurredAt: now,
102 payload: {
103 id: run.rootNodeId!,
104 threadId,
105 runId,
106 parentNodeId: null,
107 rootNodeId: run.rootNodeId!,
108 kind: "root_turn",
109 status: "pending",
110 countsForRun: true,
111 providerThreadId: null,
112 providerTurnId: null,
113 nativeItemRef: null,
114 runtimeRequestId: null,
115 checkpointScopeId: scopeId,
116 startedAt: null,
117 completedAt: null,
118 },
119 });
120 yield* store.apply({
121 id: EventId.make("startup:scope-event"),
122 type: "checkpoint-scope.created",
123 threadId,
124 occurredAt: now,
125 payload: {
126 id: scopeId,
127 threadId,
128 runId: RunId.make("run:old"),
129 nodeId: NodeId.make("startup:earlier-root"),
130 parentScopeId: null,
131 providerThreadId: null,
132 kind: "root_run",
133 ordinalWithinParent: 0,
134 advancesAppRunCount: true,
135 cwd: "/repo/worktree",
136 createdAt: now,
137 },
138 });
139 const message = {
140 id: messageId,
141 threadId,
142 runId,
143 nodeId: null,
144 role: "user" as const,
145 text: "Continue",
146 attachments: [],
147 streaming: false,
148 createdBy: "agent" as const,
149 creationSource: "server" as const,
150 createdAt: now,
151 updatedAt: now,
152 delegatedCompletion: { parentRunId: RunId.make("parent-run"), generation: 1, taskIds: [] },
153 };
154 yield* store.apply({
155 id: EventId.make("event:input"),
156 type: "message.updated",
157 threadId,
158 runId,
159 occurredAt: now,
160 payload: message,
161 });
162 const oldRunId = RunId.make("run:old");
163 yield* putRun({ ...run, id: oldRunId, ordinal: 1, status: "completed" });
164 const item = {
165 id: TurnItemId.make("item:history"),
166 threadId,
167 runId: oldRunId,
168 nodeId: null,
169 providerThreadId: null,
170 providerTurnId: null,
171 nativeItemRef: null,
172 parentItemId: null,
173 ordinal: 1,
174 status: "completed" as const,
175 title: "old command",
176 startedAt: now,
177 completedAt: now,
178 updatedAt: now,
179 type: "command_execution" as const,
180 input: "echo history",
181 output: "historical result",
182 };
183 yield* store.apply({
184 id: EventId.make("item-event"),
185 type: "turn-item.updated",
186 threadId,
187 runId: oldRunId,
188 occurredAt: now,
189 payload: item,
190 });
191 if (storage === "sqlite") {
192 const sql = yield* SqlClient.SqlClient;
193 yield* sql`INSERT INTO orchestration_v2_projection_messages
194 (message_id, thread_id, run_id, node_id, role, streaming, created_at, updated_at, payload_json)
195 VALUES ('obsolete-history', ${threadId}, NULL, NULL, 'assistant', 0,
196 ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
197 yield* sql`INSERT INTO orchestration_v2_projection_turn_items
198 (turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
199 type, status, ordinal, updated_at, payload_json)
200 VALUES ('obsolete-tool', ${threadId}, ${oldRunId}, NULL, NULL, NULL,
201 'tool_call', 'completed', 2, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
202 assert.equal((yield* Effect.exit(store.getThreadProjection(threadId)))._tag, "Failure");
203 }
204 const selected = yield* store.getThreadRecords(threadId, ["runs", "messages"], {
205 runIds: [runId],
206 messageIds: [messageId],
207 });
208 assert.deepEqual(Object.keys(selected).sort(), ["messages", "runs", "thread"]);
209 assert.deepEqual(
210 selected.runs.map((r) => r.id),
211 [runId],
212 );
213 assert.deepEqual(
214 selected.messages.map((m) => m.id),
215 [messageId],
216 );
217 assert.deepEqual(
218 (yield* store.getThreadRecords(threadId, ["messages"], { messageIds: [] })).messages,
219 [],
220 );
221 assert.deepEqual(
222 (yield* store.getThreadRecords(threadId, ["turnItems"], {
223 turnItemTypes: ["command_execution"],
224 turnItemRunId: oldRunId,
225 })).turnItems,
226 [item],
227 );
228 assert.deepEqual(
229 (yield* store.getThreadRecords(threadId, ["turnItems"], {
230 turnItemTypes: ["command_execution"],
231 turnItemRunId: runId,
232 })).turnItems,
233 [],
234 );
235 assert.equal(yield* store.getNextTurnItemOrdinal(threadId), storage === "sqlite" ? 3 : 2);
236 assert.equal(yield* store.getMessageCount(threadId), storage === "sqlite" ? 2 : 1);
237 assert.deepEqual(
238 (yield* store.getThreadRecords(threadId, ["messages"], {
239 messageRunIds: [runId],
240 })).messages.map((message) => message.id),
241 [messageId],
242 );
243 assert.deepEqual(
244 (yield* store.getThreadRecords(threadId, ["messages"], { messageRunIds: [] })).messages,
245 [],
246 );
247 assert.deepEqual(
248 (yield* store.getThreadRecords(threadId, ["turnItems"], {
249 turnItemTypes: ["command_execution"],
250 turnItemRunIds: [oldRunId],
251 })).turnItems,
252 [item],
253 );
254 assert.deepEqual(
255 (yield* store.getThreadRecords(threadId, ["turnItems"], { turnItemRunIds: [] })).turnItems,
256 [],
257 );
258 assert.deepEqual(yield* store.getThreadAttachmentIds(threadId), []);
259 if (storage === "sqlite") {
260 const sql = yield* SqlClient.SqlClient;
261 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'`;
262 assert.deepEqual(yield* store.getThreadAttachmentIds(threadId), ["attachment:old"]);
263 }
264 const page = yield* store.getTimelinePage(threadId, { view: "activity", limit: 1 });
265 assert.deepEqual(
266 page.items.map((row) => row.item),
267 [item],
268 );
269 assert.equal(page.hasMore, storage === "sqlite");
270 assert.equal(page.totalItems, storage === "sqlite" ? 2 : 1);
271 assert.deepEqual(
272 (yield* store.getTimelinePage(threadId, { view: "messages", limit: 1 })).items,
273 [],
274 );
275 const context = yield* store.getTurnStartContext(threadId, runId);
276 assert.equal(context.thread.id, threadId);
277 assert.deepEqual(
278 context.checkpointScopes.map((scope) => scope.id),
279 [scopeId],
280 );
281 assert.equal(context.messages.find((m) => m.id === messageId)?.text, "Continue");
282 assert.isTrue(context.hasConversation);
283 assert.deepEqual(context.turnItems, []);
284 assert.deepEqual(yield* store.getTurnStartHistory(threadId), [item]);
285 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [oldRunId]), [item]);
286 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [runId]), []);
287 assert.deepEqual(yield* store.getTurnStartHistory(threadId, []), []);
288 // Pairing depends on IDs, not payload decoding or item status.
289 const resultId = TurnItemId.make("interrupt-result");
290 assert.isTrue(yield* store.hasUnpairedRunInterruptRequest(threadId, item.id, resultId));
291 assert.isFalse(yield* store.hasUnpairedRunInterruptRequest(threadId, resultId, item.id));
292 assert.isFalse(
293 yield* store.hasUnpairedRunInterruptRequest(
294 ThreadId.make("other-thread"),
295 item.id,
296 resultId,
297 ),
298 );
299 yield* store.apply({
300 id: EventId.make("interrupt-result-event"),
301 type: "turn-item.updated",
302 threadId,
303 runId: oldRunId,
304 occurredAt: now,
305 payload: { ...item, id: resultId },
306 });
307 assert.isFalse(yield* store.hasUnpairedRunInterruptRequest(threadId, item.id, resultId));
308 if (storage === "sqlite") {
309 const sql = yield* SqlClient.SqlClient;
310 const original = yield* sql<{ payload_json: string }>`SELECT payload_json
311 FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${item.id}`;
312 yield* sql`UPDATE orchestration_v2_projection_turn_items
313 SET payload_json = 'invalid JSON' WHERE turn_item_id = ${item.id}`;
314 assert.isTrue(
315 yield* store.hasUnpairedRunInterruptRequest(
316 threadId,
317 item.id,
318 TurnItemId.make("absent-result"),
319 ),
320 );
321 yield* sql`UPDATE orchestration_v2_projection_turn_items
322 SET payload_json = ${original[0]!.payload_json} WHERE turn_item_id = ${item.id}`;
323 }
324 for (const text of ["/compact", " \t/COMPACT\n", "\u00a0/compact\u3000"]) {
325 yield* store.apply({
326 id: EventId.make("compact-input"),
327 type: "message.updated",
328 threadId,
329 runId,
330 occurredAt: now,
331 payload: { ...message, text },
332 });
333 assert.isFalse((yield* store.getTurnStartContext(threadId, runId)).hasConversation);
334 }
335 const marker = {
336 ...message,
337 id: MessageId.make("old-compact"),
338 runId: oldRunId,
339 text: "\t/COMPACT\n",
340 };
341 yield* store.apply({
342 id: EventId.make("old-compact-input"),
343 type: "message.updated",
344 threadId,
345 runId: oldRunId,
346 occurredAt: now,
347 payload: marker,
348 });
349 assert.include(
350 (yield* store.getTurnStartContext(threadId, runId)).messages.map((m) => m.id),
351 marker.id,
352 );
353 const other = ThreadId.make("other");
354 yield* putThread({ ...thread, id: other });
355 assert.deepEqual((yield* store.getTurnStartContext(other, runId)).messages, []);
356 assert.deepEqual(yield* store.getTurnStartHistory(other, [oldRunId]), []);
357 assert.isFalse((yield* store.getTurnStartContext(other, runId)).hasConversation);
358 assert.instanceOf(
359 yield* store.getTurnStartContext(ThreadId.make("missing"), runId).pipe(Effect.flip),
360 ProjectionStore.ProjectionStoreThreadNotFoundError,
361 );
362 }).pipe(
363 Effect.provide(
364 storage === "sqlite"
365 ? layerDatabase
366 : Layer.merge(SqlitePersistence.layerMemory, ProjectionStore.layerMemory),
367 ),
368 ),
369);