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