1import { assert, it } from "@effect/vitest";
2import {
3 EventId,
4 MessageId,
5 ProjectId,
6 ProviderInstanceId,
7 RunId,
8 ThreadId,
9 type OrchestrationV2AppThread,
10 type OrchestrationV2Run,
11} from "@t3tools/contracts";
12import * as DateTime from "effect/DateTime";
13import * as Effect from "effect/Effect";
14import * as Layer from "effect/Layer";
15import * as SqlClient from "effect/sql/SqlClient";
16import * as SqlitePersistence from "../persistence/Sqlite.ts";
17import * as ProjectionStore from "./ProjectionStore.ts";
18
19const layerDatabase = Layer.mergeAll(
20 SqlitePersistence.layerMemory,
21 ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
22);
23
24it.effect.each(["sqlite", "memory"] as const)(
25 "checks queue eligibility and delivery ownership without unrelated history in %s",
26 (storage) =>
27 Effect.gen(function* () {
28 const store = yield* ProjectionStore.ProjectionStoreV2;
29 const now = yield* DateTime.now;
30 const threadId = ThreadId.make("thread:completion-reads");
31 const runId = RunId.make("run:completion-reads");
32 const messageId = MessageId.make("message:completion-reads");
33 const instanceId = ProviderInstanceId.make("codex");
34 const thread: OrchestrationV2AppThread = {
35 id: threadId,
36 projectId: ProjectId.make("project:completion-reads"),
37 title: "Completion controls",
38 providerInstanceId: instanceId,
39 modelSelection: { instanceId, model: "test" },
40 runtimeMode: "full-access",
41 interactionMode: "default",
42 branch: null,
43 worktreePath: null,
44 activeProviderThreadId: null,
45 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
46 forkedFrom: null,
47 createdBy: "user",
48 creationSource: "web",
49 createdAt: now,
50 updatedAt: now,
51 archivedAt: null,
52 settledOverride: null,
53 settledAt: null,
54 lastVisitedAt: null,
55 deletedAt: null,
56 };
57 const run: OrchestrationV2Run = {
58 id: runId,
59 threadId,
60 ordinal: 1,
61 providerInstanceId: instanceId,
62 modelSelection: thread.modelSelection,
63 providerThreadId: null,
64 userMessageId: messageId,
65 rootNodeId: null,
66 activeAttemptId: null,
67 status: "queued",
68 requestedAt: now,
69 startedAt: null,
70 completedAt: null,
71 checkpointId: null,
72 contextHandoffId: null,
73 };
74 const putThread = (value: OrchestrationV2AppThread) =>
75 store.apply({
76 id: EventId.make(`event:thread:${value.id}`),
77 type: "thread.created",
78 threadId: value.id,
79 occurredAt: now,
80 payload: value,
81 });
82 const putRun = (value: OrchestrationV2Run) =>
83 store.apply({
84 id: EventId.make(`event:run:${value.id}:${value.status}`),
85 type: "run.updated",
86 threadId,
87 runId: value.id,
88 occurredAt: now,
89 payload: value,
90 });
91 yield* putThread(thread);
92 yield* putRun(run);
93 const message = {
94 id: messageId,
95 threadId,
96 runId,
97 nodeId: null,
98 role: "user" as const,
99 text: "Continue",
100 attachments: [],
101 streaming: false,
102 createdBy: "agent" as const,
103 creationSource: "server" as const,
104 createdAt: now,
105 updatedAt: now,
106 delegatedCompletion: { parentRunId: RunId.make("parent-run"), generation: 1, taskIds: [] },
107 };
108 yield* store.apply({
109 id: EventId.make("event:input"),
110 type: "message.updated",
111 threadId,
112 runId,
113 occurredAt: now,
114 payload: message,
115 });
116 if (storage === "sqlite") {
117 const sql = yield* SqlClient.SqlClient;
118 yield* sql`INSERT INTO orchestration_v2_projection_messages
119 (message_id, thread_id, run_id, node_id, role, streaming, created_at, updated_at, payload_json)
120 VALUES ('obsolete-history', ${threadId}, NULL, NULL, 'assistant', 0, ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
121 assert.equal((yield* Effect.exit(store.getThreadProjection(threadId)))._tag, "Failure");
122 }
123 assert.deepEqual(yield* store.getRunMessage(threadId, runId), message);
124 assert.isTrue(yield* store.canStartQueuedRun(threadId));
125 yield* putRun({ ...run, queueHeld: true });
126 assert.isFalse(yield* store.canStartQueuedRun(threadId));
127 yield* putRun(run);
128 const blocker = { ...run, id: RunId.make("blocker"), ordinal: 2 };
129 for (const status of ["preparing", "starting", "running", "waiting"] as const) {
130 yield* putRun({ ...blocker, status });
131 assert.isFalse(yield* store.canStartQueuedRun(threadId));
132 }
133 yield* putRun({ ...blocker, status: "completed" });
134 assert.isTrue(yield* store.canStartQueuedRun(threadId));
135 for (const barrier of ["archivedAt", "deletedAt"] as const) {
136 yield* putThread({ ...thread, [barrier]: now });
137 assert.isFalse(yield* store.canStartQueuedRun(threadId));
138 }
139 yield* putThread(thread);
140 yield* putRun({ ...run, status: "completed" });
141 assert.isFalse(yield* store.canStartQueuedRun(threadId));
142 const otherId = ThreadId.make("thread:other");
143 yield* putThread({ ...thread, id: otherId });
144 assert.isUndefined(yield* store.getRunMessage(otherId, runId));
145 assert.isUndefined(yield* store.getRunMessage(threadId, RunId.make("missing-run")));
146 const missing = ThreadId.make("missing-thread");
147 assert.instanceOf(
148 yield* store.canStartQueuedRun(missing).pipe(Effect.flip),
149 ProjectionStore.ProjectionStoreThreadNotFoundError,
150 );
151 assert.instanceOf(
152 yield* store.getRunMessage(missing, runId).pipe(Effect.flip),
153 ProjectionStore.ProjectionStoreThreadNotFoundError,
154 );
155 }).pipe(
156 Effect.provide(
157 storage === "sqlite"
158 ? layerDatabase
159 : Layer.merge(SqlitePersistence.layerMemory, ProjectionStore.layerMemory),
160 ),
161 ),
162);