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