1import { assert, it } from "@effect/vitest";
2import {
3 EventId,
4 MessageId,
5 NodeId,
6 ProjectId,
7 ProviderDriverKind,
8 ProviderInstanceId,
9 ProviderSessionId,
10 ProviderThreadId,
11 ProviderTurnId,
12 RunAttemptId,
13 RunId,
14 RuntimeRequestId,
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 SqlitePersistence from "../persistence/Sqlite.ts";
23import * as ProjectionStore from "./ProjectionStore.ts";
24import * as ProviderSessionManager from "./ProviderSessionManager.ts";
25import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
26import * as RuntimeRequestService from "./RuntimeRequestService.ts";
27
28const threadId = ThreadId.make("thread:control-reads");
29const providerThreadId = ProviderThreadId.make("provider-thread:control-reads");
30const providerTurnId = ProviderTurnId.make("provider-turn:control-reads");
31const providerSessionId = ProviderSessionId.make("session:control-reads");
32const attemptId = RunAttemptId.make("attempt:control-reads");
33const runId = RunId.make("run:control-reads");
34const messageId = MessageId.make("message:control-reads");
35const requestId = RuntimeRequestId.make("request:control-reads");
36const nodeId = NodeId.make("node:control-reads");
37const providerInstanceId = ProviderInstanceId.make("codex");
38const driver = ProviderDriverKind.make("codex");
39const modelSelection = { instanceId: providerInstanceId, model: "gpt-6" };
40
41function fixtureEvents(now: DateTime.Utc): ReadonlyArray<OrchestrationV2DomainEvent> {
42 const common = { threadId, occurredAt: now };
43 return [
44 {
45 ...common,
46 id: EventId.make("control:thread"),
47 type: "thread.created",
48 payload: {
49 id: threadId,
50 projectId: ProjectId.make("project:control-reads"),
51 title: "Control reads",
52 providerInstanceId,
53 modelSelection,
54 createdBy: "user",
55 creationSource: "web",
56 runtimeMode: "full-access",
57 interactionMode: "default",
58 branch: null,
59 worktreePath: "/workspace",
60 activeProviderThreadId: providerThreadId,
61 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
62 forkedFrom: null,
63 createdAt: now,
64 updatedAt: now,
65 archivedAt: null,
66 settledOverride: null,
67 settledAt: null,
68 lastVisitedAt: null,
69 deletedAt: null,
70 },
71 },
72 {
73 ...common,
74 id: EventId.make("control:provider-thread"),
75 type: "provider-thread.updated",
76 payload: {
77 id: providerThreadId,
78 driver,
79 providerInstanceId,
80 providerSessionId,
81 appThreadId: threadId,
82 ownerNodeId: null,
83 nativeThreadRef: { driver, nativeId: "native-control-thread", strength: "strong" },
84 nativeConversationHeadRef: null,
85 status: "active",
86 firstRunOrdinal: 1,
87 lastRunOrdinal: 1,
88 handoffIds: [],
89 forkedFrom: null,
90 createdAt: now,
91 updatedAt: now,
92 },
93 },
94 {
95 ...common,
96 id: EventId.make("control:run"),
97 type: "run.created",
98 payload: {
99 id: runId,
100 threadId,
101 ordinal: 1,
102 providerInstanceId,
103 modelSelection,
104 providerThreadId,
105 userMessageId: messageId,
106 rootNodeId: nodeId,
107 activeAttemptId: attemptId,
108 status: "running",
109 requestedAt: now,
110 startedAt: now,
111 completedAt: null,
112 checkpointId: null,
113 contextHandoffId: null,
114 },
115 },
116 {
117 ...common,
118 id: EventId.make("control:attempt"),
119 type: "run-attempt.created",
120 payload: {
121 id: attemptId,
122 runId,
123 attemptOrdinal: 1,
124 rootNodeId: nodeId,
125 providerInstanceId,
126 providerThreadId,
127 providerTurnId,
128 reason: "initial",
129 status: "running",
130 startedAt: now,
131 completedAt: null,
132 },
133 },
134 {
135 ...common,
136 id: EventId.make("control:turn"),
137 type: "provider-turn.updated",
138 payload: {
139 id: providerTurnId,
140 providerThreadId,
141 nodeId,
142 runAttemptId: attemptId,
143 nativeTurnRef: { driver, nativeId: "native-control-turn", strength: "strong" },
144 ordinal: 1,
145 status: "running",
146 startedAt: now,
147 completedAt: null,
148 },
149 },
150 {
151 ...common,
152 id: EventId.make("control:message"),
153 type: "message.updated",
154 payload: {
155 id: messageId,
156 threadId,
157 runId,
158 nodeId,
159 role: "user",
160 text: "Use the smaller fix.",
161 attachments: [],
162 streaming: false,
163 createdBy: "user",
164 creationSource: "web",
165 createdAt: now,
166 updatedAt: now,
167 },
168 },
169 {
170 ...common,
171 id: EventId.make("control:request"),
172 type: "runtime-request.updated",
173 payload: {
174 id: requestId,
175 nodeId,
176 providerTurnId,
177 nativeRequestRef: null,
178 kind: "command",
179 status: "resolved",
180 responseCapability: { type: "live", providerSessionId },
181 createdAt: now,
182 resolvedAt: now,
183 },
184 },
185 ];
186}
187
188const storageCases = [
189 {
190 storage: "sqlite" as const,
191 storeLayer: ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
192 },
193];
194
195it.effect.each(storageCases)(
196 "$storage: finds the active root turn without an attempt reverse link",
197 ({ storeLayer }) =>
198 Effect.gen(function* () {
199 const store = yield* ProjectionStore.ProjectionStoreV2;
200 const now = yield* DateTime.now;
201 const events = fixtureEvents(now).map((event) =>
202 event.type === "run-attempt.created"
203 ? { ...event, payload: { ...event.payload, providerTurnId: null } }
204 : event,
205 );
206 yield* Effect.forEach(events, (event) => store.apply(event), { discard: true });
207 const running = yield* store.getRunningTurnContext(threadId);
208 assert.equal(running.providerTurn?.id, providerTurnId);
209 const turnEvent = events.find((event) => event.type === "provider-turn.updated")!;
210 yield* store.apply({
211 ...turnEvent,
212 payload: { ...turnEvent.payload, status: "completed", completedAt: now },
213 });
214 assert.isUndefined((yield* store.getRunningTurnContext(threadId)).providerTurn);
215 }).pipe(Effect.provide(storeLayer)),
216);
217it.effect.each(storageCases)(
218 "$storage: controls and replies read only their exact durable targets",
219 ({ storage, storeLayer }) =>
220 Effect.gen(function* () {
221 const store = yield* ProjectionStore.ProjectionStoreV2;
222 const now = yield* DateTime.now;
223 const events = fixtureEvents(now);
224 yield* Effect.forEach(events, (event) => store.apply(event), { discard: true });
225 if (storage === "sqlite") {
226 const sql = yield* SqlClient.SqlClient;
227 yield* sql`INSERT INTO orchestration_v2_projection_messages
228 (message_id, thread_id, run_id, node_id, role, streaming, created_at, updated_at, payload_json)
229 VALUES ('message:unrelated-obsolete', ${threadId}, ${runId}, ${nodeId}, 'assistant', 0,
230 ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)}, '{"obsolete":"transcript"}')`;
231 assert.equal((yield* Effect.exit(store.getThreadProjection(threadId)))._tag, "Failure");
232 const queryPlan = yield* sql<{ detail: string }>`EXPLAIN QUERY PLAN
233 SELECT payload_json FROM orchestration_v2_projection_turn_items
234 WHERE thread_id = ${threadId} AND node_id = ${nodeId}
235 AND type IN ('approval_request', 'user_input_request')
236 AND json_extract(payload_json, '$.requestId') = ${requestId}
237 ORDER BY ordinal ASC LIMIT 1`;
238 assert.isTrue(queryPlan.some((row) => row.detail.includes("turn_items_node_ordinal_idx")));
239 }
240 const running = yield* store.getRunningTurnContext(threadId);
241 assert.equal(running.run?.id, runId);
242 assert.equal(running.providerThread?.id, providerThreadId);
243 assert.equal(running.providerTurn?.id, providerTurnId);
244 const providerContext = yield* store.getThreadProviderContext(threadId, providerInstanceId);
245 assert.equal(providerContext.thread.id, threadId);
246 assert.deepEqual(
247 providerContext.providerThreads.map((thread) => thread.id),
248 [providerThreadId],
249 );
250 const responseContext = yield* store.getRuntimeResponseContext(threadId, requestId);
251 assert.equal(responseContext.request?.id, requestId);
252 const target = { providerThreadId, providerTurnId, attemptId, messageId };
253 const context = yield* store.getProviderControlContext(threadId, target);
254 assert.equal(context.providerThread?.id, providerThreadId);
255 assert.equal(context.providerTurn?.id, providerTurnId);
256 assert.equal(context.attempt?.id, attemptId);
257 assert.equal(context.message?.text, "Use the smaller fix.");
258 assert.equal(context.run?.id, runId);
259 assert.equal((yield* store.getRuntimeRequest(threadId, requestId))?.status, "resolved");
260 assert.isUndefined(
261 yield* store.getRuntimeRequest(threadId, RuntimeRequestId.make("request:missing")),
262 );
263 const absent = yield* store.getProviderControlContext(threadId, {
264 providerThreadId: ProviderThreadId.make("provider-thread:missing"),
265 providerTurnId: ProviderTurnId.make("provider-turn:missing"),
266 });
267 assert.isUndefined(absent.providerThread);
268 assert.isUndefined(absent.providerTurn);
269 const missingThread = yield* store
270 .getRuntimeRequest(ThreadId.make("thread:missing"), requestId)
271 .pipe(Effect.flip);
272 assert.instanceOf(missingThread, ProjectionStore.ProjectionStoreThreadNotFoundError);
273
274 const calls: string[] = [];
275 const layerSessions = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
276 get: () =>
277 Effect.succeedSome({
278 interruptTurn: () =>
279 Effect.sync(() => {
280 calls.push("interrupt");
281 }),
282 steerTurn: (input: {
283 message: {
284 text: string;
285 };
286 runId: RunId;
287 }) =>
288 Effect.sync(() => {
289 assert.equal(input.runId, runId);
290 calls.push(input.message.text);
291 }),
292 respondToRuntimeRequest: () =>
293 Effect.sync(() => {
294 calls.push("reply");
295 }),
296 } as never),
297 });
298 yield* Effect.gen(function* () {
299 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
300 const reply = yield* RuntimeRequestService.RuntimeRequestServiceV2;
301 yield* control.interrupt({ threadId, providerThreadId, providerTurnId, providerSessionId });
302 yield* control.steer({
303 threadId,
304 providerThreadId,
305 providerTurnId,
306 providerSessionId,
307 messageId,
308 });
309 yield* reply.respond({ threadId, providerSessionId, requestId, decision: "accept" });
310 assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply"]);
311 const wrongSession = ProviderSessionId.make("session:wrong");
312 assert.equal(
313 (yield* Effect.exit(
314 control.interrupt({
315 threadId,
316 providerThreadId,
317 providerTurnId,
318 providerSessionId: wrongSession,
319 }),
320 ))._tag,
321 "Failure",
322 );
323 assert.equal(
324 (yield* Effect.exit(
325 reply.respond({
326 threadId,
327 providerSessionId: wrongSession,
328 requestId,
329 decision: "accept",
330 }),
331 ))._tag,
332 "Failure",
333 );
334 assert.lengthOf(calls, 3);
335 const turn = context.providerTurn!;
336 yield* store.apply({
337 id: EventId.make("control:turn-completed"),
338 type: "provider-turn.updated",
339 threadId,
340 occurredAt: now,
341 payload: { ...turn, status: "completed", completedAt: now },
342 });
343 yield* control.interrupt({ threadId, providerThreadId, providerTurnId, providerSessionId });
344 assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply", "interrupt"]);
345 }).pipe(
346 Effect.provide(
347 Layer.merge(ProviderTurnControlService.layer, RuntimeRequestService.layer).pipe(
348 Layer.provide(layerSessions),
349 ),
350 ),
351 );
352 }).pipe(Effect.provide(Layer.merge(storeLayer, SqlitePersistence.layerMemory))),
353);