Open History to see the last commit
1import * as DocumentText from "../provider/DocumentText.ts";
2// @effect-diagnostics nodeBuiltinImport:off
3import * as NodeFS from "node:fs";
4import * as NodePath from "node:path";
5import * as NodeServices from "@effect/platform-node/NodeServices";
6import { expect, it } from "@effect/vitest";
7import {
8 ChatAttachmentId,
9 CommandId,
10 EventId,
11 GuestMessageId,
12 MessageId,
13 RuntimeRequestId,
14 ThreadId,
15 TurnItemId,
16 type OrchestrationV2Command,
17 type OrchestrationV2ServerCommand,
18 type OrchestrationV2StoredEvent,
19 type UserInputAttachments,
20} from "@t3tools/contracts";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import {
25 createAttachmentId,
26 createPendingAttachmentId,
27 resolveAttachmentPath,
28} from "../attachmentStore.ts";
29import * as ServerConfig from "../config.ts";
30import * as ServerSettings from "../serverSettings.ts";
31import {
32 OrchestratorCommandPreviouslyRejectedError,
33 OrchestratorDispatchError,
34} from "./Orchestrator.ts";
35import * as ThreadManagementService from "./ThreadManagementService.ts";
36import { dispatchCommand } from "./ThreadMessageIntake.ts";
37
38const layerIntakeTest = ServerConfig.layerTest(process.cwd(), {
39 prefix: "t3-question-intake-",
40}).pipe(
41 Layer.provideMerge(DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer))),
42 Layer.provideMerge(ServerSettings.layerTest()),
43);
44
45const layerFailingDispatchHandler =
46 (captured: OrchestrationV2ServerCommand[]) => (command: OrchestrationV2ServerCommand) => {
47 captured.push(command);
48 return Effect.fail(
49 new OrchestratorDispatchError({
50 commandId: command.commandId,
51 commandType: command.type,
52 cause: "Stop after intake",
53 }),
54 );
55 };
56
57const layerFailingDispatch = (captured: OrchestrationV2ServerCommand[]) =>
58 Layer.mock(ThreadManagementService.ThreadManagementService)({
59 dispatch: layerFailingDispatchHandler(captured),
60 });
61
62const NOW = DateTime.makeUnsafe("2026-09-13T00:00:00.000Z");
63
64const answeredRespondEvents = (input: {
65 readonly threadId: ThreadId;
66 readonly requestId: string;
67 readonly attachmentsByQuestionId: UserInputAttachments;
68}): OrchestrationV2StoredEvent[] => [
69 {
70 sequence: 1,
71 commandId: null,
72 event: {
73 id: EventId.make("evt-recorded-answer"),
74 type: "turn-item.updated",
75 threadId: input.threadId,
76 occurredAt: NOW,
77 payload: {
78 id: TurnItemId.make("item-recorded-answer"),
79 type: "user_input_request",
80 threadId: input.threadId,
81 runId: null,
82 nodeId: null,
83 providerThreadId: null,
84 providerTurnId: null,
85 nativeItemRef: null,
86 parentItemId: null,
87 ordinal: 1,
88 status: "completed",
89 title: null,
90 startedAt: null,
91 completedAt: NOW,
92 updatedAt: NOW,
93 requestId: RuntimeRequestId.make(input.requestId),
94 questions: [],
95 questionAnswer: {
96 requestId: input.requestId,
97 answers: {},
98 attachmentsByQuestionId: input.attachmentsByQuestionId,
99 },
100 },
101 },
102 },
103];
104
105it.effect("claims question uploads and passes readable paths through the V2 request command", () =>
106 Effect.gen(function* () {
107 const config = yield* ServerConfig.ServerConfig;
108 const id = ChatAttachmentId.make(createPendingAttachmentId()!);
109 NodeFS.writeFileSync(
110 NodePath.join(config.attachmentsDir, `${id}.png`),
111 new Uint8Array([1, 2, 3]),
112 );
113 const captured: OrchestrationV2ServerCommand[] = [];
114 yield* dispatchCommand({
115 type: "runtime-request.respond",
116 commandId: CommandId.make("answer-1"),
117 threadId: ThreadId.make("thread-answer"),
118 requestId: RuntimeRequestId.make("request-1"),
119 answers: { q: ["Selected option"] },
120 attachmentsByQuestionId: {
121 q: [{ type: "image", id, name: "screen.png", mimeType: "image/png", sizeBytes: 3 }],
122 },
123 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
124 expect(captured).toHaveLength(1);
125 const command = captured[0]!;
126 expect(command.type).toBe("runtime-request.respond");
127 if (command.type !== "runtime-request.respond") return;
128 const attachment = command.attachmentsByQuestionId?.q?.[0];
129 expect(attachment).toBeDefined();
130 const path = resolveAttachmentPath({
131 attachmentsDir: config.attachmentsDir,
132 attachment: attachment!,
133 });
134 expect(path).not.toBeNull();
135 expect(NodeFS.readFileSync(path!)).toEqual(Buffer.from([1, 2, 3]));
136 expect(command.answers?.q).toEqual([
137 "Selected option",
138 `Attached image "screen.png": ${JSON.stringify(path)}`,
139 ]);
140 }).pipe(Effect.provide(layerIntakeTest)),
141);
142
143it.effect("retains claimed copies when dispatch failure may have been accepted", () =>
144 Effect.gen(function* () {
145 const config = yield* ServerConfig.ServerConfig;
146 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
147 NodeFS.writeFileSync(
148 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
149 new Uint8Array([1, 2, 3]),
150 );
151 const captured: OrchestrationV2ServerCommand[] = [];
152 const result = yield* dispatchCommand({
153 type: "runtime-request.respond",
154 commandId: CommandId.make("answer-ambiguous"),
155 threadId: ThreadId.make("thread-ambiguous"),
156 requestId: RuntimeRequestId.make("request-ambiguous"),
157 answers: { q: ["one"] },
158 attachmentsByQuestionId: {
159 q: [
160 {
161 type: "image",
162 id: pendingId,
163 name: "screen.png",
164 mimeType: "image/png",
165 sizeBytes: 3,
166 },
167 ],
168 },
169 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
170 expect(result._tag).toBe("Failure");
171 expect(captured).toHaveLength(1);
172 const threadFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
173 entry.startsWith("thread-ambiguous-"),
174 );
175 expect(threadFiles).toHaveLength(1);
176 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
177 }).pipe(Effect.provide(layerIntakeTest)),
178);
179
180it.effect("releases claimed copies when the command was already rejected", () =>
181 Effect.gen(function* () {
182 const config = yield* ServerConfig.ServerConfig;
183 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
184 NodeFS.writeFileSync(
185 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
186 new Uint8Array([1, 2, 3]),
187 );
188 const result = yield* dispatchCommand({
189 type: "runtime-request.respond",
190 commandId: CommandId.make("answer-rejected"),
191 threadId: ThreadId.make("thread-rejected"),
192 requestId: RuntimeRequestId.make("request-rejected"),
193 answers: { q: ["one"] },
194 attachmentsByQuestionId: {
195 q: [
196 {
197 type: "image",
198 id: pendingId,
199 name: "screen.png",
200 mimeType: "image/png",
201 sizeBytes: 3,
202 },
203 ],
204 },
205 }).pipe(
206 Effect.provide(
207 Layer.mock(ThreadManagementService.ThreadManagementService)({
208 dispatch: (command) =>
209 Effect.fail(
210 new OrchestratorCommandPreviouslyRejectedError({
211 commandId: command.commandId,
212 commandType: command.type,
213 detail: "rejected earlier",
214 }),
215 ),
216 }),
217 ),
218 Effect.result,
219 );
220 expect(result._tag).toBe("Failure");
221 expect(
222 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
223 entry.startsWith("thread-rejected-"),
224 ),
225 ).toEqual([]);
226 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
227 }).pipe(Effect.provide(layerIntakeTest)),
228);
229
230it.effect("releases claimed copies when dispatch replays an earlier accepted response", () =>
231 Effect.gen(function* () {
232 const config = yield* ServerConfig.ServerConfig;
233 const threadId = ThreadId.make("thread-replay");
234 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
235 NodeFS.writeFileSync(
236 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
237 new Uint8Array([1, 2, 3]),
238 );
239 const earlierAttachment = {
240 type: "image" as const,
241 id: ChatAttachmentId.make("thread-replay-00000000-0000-4000-8000-0000000000aa"),
242 name: "screen.png",
243 mimeType: "image/png",
244 sizeBytes: 3,
245 };
246 NodeFS.writeFileSync(
247 NodePath.join(config.attachmentsDir, `${earlierAttachment.id}.png`),
248 new Uint8Array([4, 4, 4]),
249 );
250 const storedEvents = answeredRespondEvents({
251 threadId,
252 requestId: "request-replay",
253 attachmentsByQuestionId: { q: [earlierAttachment] },
254 });
255 const result = yield* dispatchCommand({
256 type: "runtime-request.respond",
257 commandId: CommandId.make("answer-replay"),
258 threadId,
259 requestId: RuntimeRequestId.make("request-replay"),
260 answers: { q: ["one"] },
261 attachmentsByQuestionId: {
262 q: [
263 {
264 type: "image",
265 id: pendingId,
266 name: "screen.png",
267 mimeType: "image/png",
268 sizeBytes: 3,
269 },
270 ],
271 },
272 }).pipe(
273 Effect.provide(
274 Layer.mock(ThreadManagementService.ThreadManagementService)({
275 dispatch: () => Effect.succeed({ sequence: 1, storedEvents }),
276 }),
277 ),
278 Effect.result,
279 );
280 expect(result._tag).toBe("Success");
281 expect(
282 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
283 entry.startsWith("thread-replay-"),
284 ),
285 ).toEqual([`${earlierAttachment.id}.png`]);
286 expect(
287 NodeFS.readFileSync(NodePath.join(config.attachmentsDir, `${earlierAttachment.id}.png`)),
288 ).toEqual(Buffer.from([4, 4, 4]));
289 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
290 }).pipe(Effect.provide(layerIntakeTest)),
291);
292
293it.effect("a retried response re-claims the preserved pending uploads", () =>
294 Effect.gen(function* () {
295 const config = yield* ServerConfig.ServerConfig;
296 const threadId = ThreadId.make("thread-retry");
297 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
298 const lateId = ChatAttachmentId.make(createPendingAttachmentId()!);
299 NodeFS.writeFileSync(
300 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
301 new Uint8Array([1, 2, 3]),
302 );
303 const command: OrchestrationV2Command = {
304 type: "runtime-request.respond",
305 commandId: CommandId.make("answer-retry"),
306 threadId,
307 requestId: RuntimeRequestId.make("request-retry"),
308 answers: { q1: ["one"], q2: ["two"] },
309 attachmentsByQuestionId: {
310 q1: [
311 {
312 type: "image",
313 id: pendingId,
314 name: "screen.png",
315 mimeType: "image/png",
316 sizeBytes: 3,
317 },
318 ],
319 q2: [
320 {
321 type: "image",
322 id: lateId,
323 name: "late.png",
324 mimeType: "image/png",
325 sizeBytes: 2,
326 },
327 ],
328 },
329 };
330 const captured: OrchestrationV2ServerCommand[] = [];
331 const first = yield* dispatchCommand(command).pipe(
332 Effect.provide(layerFailingDispatch(captured)),
333 Effect.result,
334 );
335 expect(first._tag).toBe("Failure");
336 expect(captured).toHaveLength(0);
337
338 NodeFS.writeFileSync(
339 NodePath.join(config.attachmentsDir, `${lateId}.png`),
340 new Uint8Array([7, 7]),
341 );
342 const retry = yield* dispatchCommand(command).pipe(
343 Effect.provide(
344 Layer.mock(ThreadManagementService.ThreadManagementService)({
345 dispatch: (dispatched) => {
346 captured.push(dispatched);
347 return Effect.succeed({ sequence: 1, storedEvents: [] });
348 },
349 }),
350 ),
351 Effect.result,
352 );
353 expect(retry._tag).toBe("Success");
354 const dispatched = captured[0]!;
355 expect(dispatched.type).toBe("runtime-request.respond");
356 if (dispatched.type !== "runtime-request.respond") return;
357 for (const attachment of [
358 ...(dispatched.attachmentsByQuestionId?.q1 ?? []),
359 ...(dispatched.attachmentsByQuestionId?.q2 ?? []),
360 ]) {
361 const path = resolveAttachmentPath({
362 attachmentsDir: config.attachmentsDir,
363 attachment,
364 });
365 expect(path).not.toBeNull();
366 expect(NodeFS.existsSync(path!)).toBe(true);
367 }
368 expect(
369 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
370 entry.startsWith("thread-retry-"),
371 ),
372 ).toHaveLength(2);
373 }).pipe(Effect.provide(layerIntakeTest)),
374);
375
376it.effect("sends a guest message with the guest's files, never ones the client names", () =>
377 Effect.gen(function* () {
378 const config = yield* ServerConfig.ServerConfig;
379 const guestFile = {
380 type: "image" as const,
381 id: ChatAttachmentId.make(createAttachmentId("thread-guest")!),
382 name: "screen.png",
383 mimeType: "image/png",
384 sizeBytes: 3,
385 };
386 NodeFS.writeFileSync(
387 NodePath.join(config.attachmentsDir, `${guestFile.id}.png`),
388 new Uint8Array([1, 2, 3]),
389 );
390 const guestMessageId = GuestMessageId.make("guest-message:files");
391 const captured: OrchestrationV2ServerCommand[] = [];
392 const layerGuestShare = Layer.mock(ThreadManagementService.ThreadManagementService)({
393 dispatch: layerFailingDispatchHandler(captured),
394 // Only the share is read.
395 getThreadSnapshotWindow: () =>
396 Effect.succeed({
397 projection: {
398 thread: {
399 guestShare: {
400 messages: [{ id: guestMessageId, status: "pending", attachments: [guestFile] }],
401 },
402 },
403 },
404 } as unknown as Effect.Success<
405 ReturnType<
406 ThreadManagementService.ThreadManagementServiceShape["getThreadSnapshotWindow"]
407 >
408 >),
409 });
410 yield* dispatchCommand({
411 type: "message.dispatch",
412 commandId: CommandId.make("send-guest-message"),
413 threadId: ThreadId.make("thread-guest"),
414 messageId: MessageId.make("message-guest"),
415 guestMessageId,
416 text: "",
417 attachments: [{ ...guestFile, id: ChatAttachmentId.make("thread-guest-owner-file") }],
418 dispatchMode: { type: "queue_after_active" },
419 createdBy: "user",
420 creationSource: "web",
421 }).pipe(Effect.provide(layerGuestShare), Effect.result);
422 const command = captured[0];
423 if (command?.type !== "message.dispatch") throw new Error("Expected a message dispatch.");
424 expect(command.attachments).toHaveLength(1);
425 // The guest's file, which is already the thread's.
426 expect(command.attachments).toEqual([guestFile]);
427 }).pipe(Effect.provide(layerIntakeTest)),
428);