1import * as Cloudflare from "@/Cloudflare";
2import * as Alchemy from "@/index.ts";
3import * as Test from "@/Test/Alchemy";
4import { expect } from "alchemy-test";
5import * as ConsoleService from "effect/Console";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import * as Stream from "effect/Stream";
10import * as Sse from "effect/encoding/Sse";
11import * as HttpClient from "effect/http/HttpClient";
12import { Gateway } from "./fixtures/Gateway.ts";
13import LanguageModelTestWorker from "./fixtures/LanguageModelWorker.ts";
14
15// Fresh `workers.dev` URLs return non-200 (404 / 500 "Script not
16// found") for a few seconds while the edge propagates. Each test uses
17// `HttpClient.filterStatusOk(yield* HttpClient.HttpClient)` so the
18// existing `Effect.retry` rides through these by converting the
19// bad-status response into a retryable Effect failure.
20
21const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
22 providers: Cloudflare.providers(),
23});
24
25const logLevel = Effect.provideService(
26 MinimumLogLevel,
27 process.env.DEBUG ? "Debug" : "Info",
28);
29
30const Stack = Alchemy.Stack(
31 "AiGatewayLanguageModelStack",
32 {
33 providers: Cloudflare.providers(),
34 state: Cloudflare.state(),
35 },
36 Effect.gen(function* () {
37 const worker = yield* LanguageModelTestWorker;
38 const gateway = yield* Gateway;
39 return {
40 gatewayId: gateway.gatewayId,
41 url: worker.url.as<string>(),
42 };
43 }),
44);
45
46const stack = beforeAll(deploy(Stack));
47afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
48
49type StreamPart = {
50 type: string;
51 id?: string;
52 name?: string;
53 delta?: string;
54 reason?: string;
55 usage?: {
56 inputTokens?: { total?: number };
57 outputTokens?: { total?: number };
58 };
59};
60
61const parseSse = (sse: string): ReadonlyArray<StreamPart> =>
62 sse
63 .split("\n\n")
64 .map((frame) => frame.replace(/^data:\s*/, "").trim())
65 .filter((line) => line.length > 0)
66 .map((line) => JSON.parse(line) as StreamPart);
67
68test(
69 "deployed worker generates text via AiGateway-backed LanguageModel",
70 Effect.gen(function* () {
71 const out = yield* stack;
72 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
73
74 const res = yield* client
75 .get(`${out.url}/generate?prompt=${encodeURIComponent("Say pong.")}`)
76 .pipe(
77 Effect.retry({
78 schedule: Schedule.exponential("500 millis"),
79 times: 10,
80 }),
81 );
82 expect(res.status).toBe(200);
83
84 const body = (yield* res.json) as {
85 text: string;
86 finishReason: string;
87 usage: {
88 inputTokens: number | undefined;
89 outputTokens: number | undefined;
90 };
91 };
92
93 expect(typeof body.text).toBe("string");
94 expect(body.text.length).toBeGreaterThan(0);
95 // Refactor invariant: `mapUsage` must populate both input and output
96 // tokens from Workers AI's `prompt_tokens` / `completion_tokens` fields.
97 expect(body.usage.inputTokens).toBeGreaterThan(0);
98 expect(body.usage.outputTokens).toBeGreaterThan(0);
99 // Refactor invariant: a normal completion maps to `stop`, not `unknown` /
100 // `other` / `error`.
101 expect(body.finishReason).toBe("stop");
102 }).pipe(logLevel),
103 {
104 tags: [
105 "provider:cloudflare",
106 "provider:cloudflare:ai",
107 "provider:cloudflare:worker",
108 "live",
109 ],
110 timeout: 180_000,
111 },
112);
113
114test(
115 "deployed worker streams text via AiGateway-backed LanguageModel",
116 Effect.gen(function* () {
117 const out = yield* stack;
118 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
119
120 // The AI-Gateway-backed model can answer 200 with an empty SSE on a cold
121 // or transient call (no tokens streamed), which `filterStatusOk` does not
122 // catch. Fold the parse into the retried effect and fail on an empty /
123 // unfinished stream so the same backoff rides out the blip.
124 const parts = yield* client
125 .get(`${out.url}/stream?prompt=${encodeURIComponent("Say pong.")}`)
126 .pipe(
127 Effect.flatMap((res) => res.text),
128 Effect.map(parseSse),
129 Effect.flatMap((parts) =>
130 parts.some((p) => p.type === "text-delta") &&
131 parts.some((p) => p.type === "finish")
132 ? Effect.succeed(parts)
133 : Effect.fail(new Error("AI stream not ready: empty/unfinished")),
134 ),
135 Effect.retry({
136 schedule: Schedule.exponential("500 millis"),
137 times: 10,
138 }),
139 );
140
141 const text = parts
142 .filter((p) => p.type === "text-delta")
143 .map((p) => p.delta ?? "")
144 .join("");
145 const finish = parts.find((p) => p.type === "finish");
146
147 expect(parts.length).toBeGreaterThan(0);
148 expect(text.length).toBeGreaterThan(0);
149 expect(finish).toBeDefined();
150 }).pipe(logLevel),
151 {
152 tags: [
153 "provider:cloudflare",
154 "provider:cloudflare:ai",
155 "provider:cloudflare:worker",
156 "live",
157 ],
158 timeout: 180_000,
159 },
160);
161
162test(
163 "streamed parts respect ordering: text-start → text-delta+ → text-end → finish",
164 Effect.gen(function* () {
165 const out = yield* stack;
166 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
167
168 // Same cold/transient empty-SSE hazard as above — retry until the stream
169 // actually carries the structural parts this test asserts ordering over.
170 const parts = yield* client
171 .get(`${out.url}/stream?prompt=${encodeURIComponent("Say pong.")}`)
172 .pipe(
173 Effect.flatMap((res) => res.text),
174 Effect.map(parseSse),
175 Effect.flatMap((parts) =>
176 parts.some((p) => p.type === "text-delta") &&
177 parts.some((p) => p.type === "finish")
178 ? Effect.succeed(parts)
179 : Effect.fail(new Error("AI stream not ready: empty/unfinished")),
180 ),
181 Effect.retry({
182 schedule: Schedule.exponential("500 millis"),
183 times: 10,
184 }),
185 );
186
187 const indexOfType = (type: string) =>
188 parts.findIndex((p) => p.type === type);
189 const lastIndexOfType = (type: string) => {
190 for (let i = parts.length - 1; i >= 0; i--) {
191 if (parts[i]!.type === type) return i;
192 }
193 return -1;
194 };
195
196 // Refactor invariant: `emitTextDelta` opens a text-start before the first
197 // text-delta, `finalizeStream` emits text-end before the final finish.
198 const startIdx = indexOfType("text-start");
199 const firstDeltaIdx = indexOfType("text-delta");
200 const lastDeltaIdx = lastIndexOfType("text-delta");
201 const endIdx = indexOfType("text-end");
202 const finishIdx = indexOfType("finish");
203
204 expect(startIdx).toBeGreaterThanOrEqual(0);
205 expect(firstDeltaIdx).toBeGreaterThan(startIdx);
206 expect(endIdx).toBeGreaterThan(lastDeltaIdx);
207 expect(finishIdx).toBe(parts.length - 1);
208 expect(finishIdx).toBeGreaterThan(endIdx);
209
210 // Exactly one text segment was opened and closed.
211 expect(parts.filter((p) => p.type === "text-start")).toHaveLength(1);
212 expect(parts.filter((p) => p.type === "text-end")).toHaveLength(1);
213 }).pipe(logLevel),
214 {
215 tags: [
216 "provider:cloudflare",
217 "provider:cloudflare:ai",
218 "provider:cloudflare:worker",
219 "live",
220 ],
221 timeout: 180_000,
222 },
223);
224
225test.skipIf(!process.env.DEBUG_RAW_STREAM)(
226 "DEBUG: dump raw Workers AI SSE stream",
227 Effect.gen(function* () {
228 const out = yield* stack;
229 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
230 // Through the Console SERVICE so the runner captures it per-test (raw
231 // fd writes bypass capture and corrupt the reporter/TUI).
232 const console = yield* ConsoleService.Console;
233 const print = (s: string) => console.log(s);
234
235 for (const includeUsage of ["0", "1"] as const) {
236 const res = yield* client
237 .get(
238 `${out.url}/raw-stream?include_usage=${includeUsage}&prompt=${encodeURIComponent("Say pong.")}`,
239 )
240 .pipe(
241 Effect.retry({
242 schedule: Schedule.exponential("500 millis"),
243 times: 10,
244 }),
245 );
246 print(`\n=== include_usage=${includeUsage} ===\n`);
247 print(yield* res.text);
248 print("\n=== end ===\n");
249 }
250
251 const toolRes = yield* client
252 .get(
253 `${out.url}/raw-tool-stream?prompt=${encodeURIComponent("What's the weather in Seattle?")}`,
254 )
255 .pipe(
256 Effect.retry({
257 schedule: Schedule.exponential("500 millis"),
258 times: 10,
259 }),
260 );
261 print(`\n=== raw tool stream ===\n`);
262 print(yield* toolRes.text);
263 print("\n=== end tool stream ===\n");
264 }).pipe(logLevel),
265 {
266 tags: [
267 "provider:cloudflare",
268 "provider:cloudflare:ai",
269 "provider:cloudflare:worker",
270 "live",
271 ],
272 timeout: 180_000,
273 },
274);
275
276test(
277 "stream finish part reports real token counts and a terminal reason",
278 Effect.gen(function* () {
279 const out = yield* stack;
280 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
281
282 const res = yield* client
283 .get(
284 `${out.url}/stream?prompt=${encodeURIComponent("Say the word 'pong' and nothing else.")}`,
285 )
286 .pipe(
287 Effect.retry({
288 schedule: Schedule.exponential("500 millis"),
289 times: 10,
290 }),
291 );
292 expect(res.status).toBe(200);
293
294 const parts = parseSse(yield* res.text);
295 const finish = parts.find((p) => p.type === "finish");
296 expect(finish).toBeDefined();
297 // Workers AI emits the real usage chunk and then a zero-valued
298 // "terminator" usage chunk; the refactor must keep the real one (see
299 // `hasNonZeroUsage` in updateChunkMeta).
300 expect(finish?.usage?.inputTokens?.total).toBeGreaterThan(0);
301 expect(finish?.usage?.outputTokens?.total).toBeGreaterThan(0);
302 // A clean `[DONE]` falls back to "stop"; some Workers AI responses now
303 // report the model's real "length" reason even for this short prompt.
304 expect(["stop", "length"]).toContain(finish?.reason);
305 }).pipe(logLevel),
306 {
307 tags: [
308 "provider:cloudflare",
309 "provider:cloudflare:ai",
310 "provider:cloudflare:worker",
311 "live",
312 ],
313 timeout: 180_000,
314 },
315);
316
317test(
318 "stream emits multiple text-delta chunks for a long-form response",
319 Effect.gen(function* () {
320 const out = yield* stack;
321 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
322
323 const res = yield* client
324 .get(
325 `${out.url}/stream?prompt=${encodeURIComponent(
326 "Write a short paragraph (around 80 words) about why TypeScript developers might enjoy Effect TS.",
327 )}`,
328 )
329 .pipe(
330 Effect.retry({
331 schedule: Schedule.exponential("500 millis"),
332 times: 10,
333 }),
334 );
335 expect(res.status).toBe(200);
336
337 const parts = parseSse(yield* res.text);
338 const deltas = parts.filter((p) => p.type === "text-delta");
339 // Refactor invariant: each model SSE chunk produces a text-delta — so a
340 // long response should produce many. A single fused delta would mean we
341 // accidentally buffered the stream.
342 expect(deltas.length).toBeGreaterThan(3);
343 const total = deltas.map((p) => p.delta ?? "").join("").length;
344 expect(total).toBeGreaterThan(20);
345 }).pipe(logLevel),
346 {
347 tags: [
348 "provider:cloudflare",
349 "provider:cloudflare:ai",
350 "provider:cloudflare:worker",
351 "live",
352 ],
353 timeout: 180_000,
354 },
355);
356
357// `persisted chat survives across DO invocations` exercises the
358// /chat?id=... route which is wired by `ChatAgent` from the Agent
359// slice. The fixture in this slice (`LanguageModelWorker`) does not
360// include that route — the test re-activates in the Agent PR.
361test.skip(
362 "persisted chat survives across DO invocations",
363 Effect.gen(function* () {
364 const out = yield* stack;
365 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
366 const id = `test-${Date.now()}`;
367
368 const r1 = yield* client
369 .get(
370 `${out.url}/chat?id=${id}&prompt=${encodeURIComponent("My name is Sam. Remember it.")}`,
371 )
372 .pipe(
373 Effect.retry({
374 schedule: Schedule.exponential("500 millis"),
375 times: 10,
376 }),
377 );
378 if (r1.status !== 200) {
379 console.error("turn1 error body:", yield* r1.text);
380 }
381 expect(r1.status).toBe(200);
382 const b1 = (yield* r1.json) as { text: string; turns: number };
383 expect(b1.turns).toBeGreaterThanOrEqual(2);
384
385 const r2 = yield* client
386 .get(
387 `${out.url}/chat?id=${id}&prompt=${encodeURIComponent("What is my name? Answer with just the name.")}`,
388 )
389 .pipe(
390 Effect.retry({
391 schedule: Schedule.exponential("500 millis"),
392 times: 10,
393 }),
394 );
395 expect(r2.status).toBe(200);
396 const b2 = (yield* r2.json) as { text: string; turns: number };
397
398 expect(b2.text.toLowerCase()).toContain("sam");
399 expect(b2.turns).toBeGreaterThanOrEqual(4);
400 }).pipe(logLevel),
401 {
402 tags: [
403 "provider:cloudflare",
404 "provider:cloudflare:ai",
405 "provider:cloudflare:worker",
406 "live",
407 ],
408 timeout: 240_000,
409 },
410);
411
412test(
413 "deployed worker invokes a tool via AiGateway-backed LanguageModel",
414 Effect.gen(function* () {
415 const out = yield* stack;
416 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
417
418 const res = yield* client
419 .get(
420 `${out.url}/tool?prompt=${encodeURIComponent(
421 "What's the weather in San Francisco?",
422 )}`,
423 )
424 .pipe(
425 Effect.retry({
426 schedule: Schedule.exponential("500 millis"),
427 times: 10,
428 }),
429 );
430 if (res.status !== 200) {
431 console.error("tool error body:", yield* res.text);
432 }
433 expect(res.status).toBe(200);
434
435 const body = (yield* res.json) as {
436 text: string;
437 finishReason: string;
438 toolCalls: Array<{
439 id: string;
440 name: string;
441 params: { city: string };
442 }>;
443 toolResults: Array<{
444 id: string;
445 name: string;
446 result: { city: string; temperatureF: number; condition: string };
447 isFailure: boolean;
448 }>;
449 };
450
451 expect(body.toolCalls.length).toBeGreaterThan(0);
452 const call = body.toolCalls[0]!;
453 expect(call.name).toBe("get_weather");
454 expect(typeof call.params.city).toBe("string");
455 expect(call.params.city.toLowerCase()).toContain("san francisco");
456
457 expect(body.toolResults.length).toBeGreaterThan(0);
458 const result = body.toolResults[0]!;
459 expect(result.name).toBe("get_weather");
460 expect(result.isFailure).toBe(false);
461 expect(result.result.temperatureF).toBe(72);
462 expect(result.result.condition).toBe("sunny");
463 }).pipe(logLevel),
464 {
465 tags: [
466 "provider:cloudflare",
467 "provider:cloudflare:ai",
468 "provider:cloudflare:worker",
469 "live",
470 ],
471 timeout: 180_000,
472 },
473);
474
475test(
476 "streams tool-call parts via AiGateway-backed LanguageModel",
477 Effect.gen(function* () {
478 const out = yield* stack;
479 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
480
481 const res = yield* client
482 .get(
483 `${out.url}/tool-stream?prompt=${encodeURIComponent(
484 "What's the weather in Seattle?",
485 )}`,
486 )
487 .pipe(
488 Effect.retry({
489 schedule: Schedule.exponential("500 millis"),
490 times: 10,
491 }),
492 );
493 expect(res.status).toBe(200);
494
495 const parts = parseSse(yield* res.text);
496 const toolParamsStart = parts.filter((p) => p.type === "tool-params-start");
497 const toolParamsDeltas = parts.filter(
498 (p) => p.type === "tool-params-delta",
499 );
500 const toolParamsEnd = parts.filter((p) => p.type === "tool-params-end");
501
502 expect(toolParamsStart.length).toBeGreaterThan(0);
503 expect(toolParamsStart[0]?.name).toBe("get_weather");
504 expect(toolParamsDeltas.length).toBeGreaterThan(0);
505
506 // Refactor invariant: `finalizeStream` closes any tool calls still open
507 // when the stream ends, emitting one `tool-params-end` per opened call.
508 expect(toolParamsEnd.length).toBe(toolParamsStart.length);
509 // Each start is matched to an end by id.
510 const startIds = new Set(toolParamsStart.map((p) => p.id));
511 const endIds = new Set(toolParamsEnd.map((p) => p.id));
512 expect(endIds).toEqual(startIds);
513
514 // Ordering: every tool-params-end comes after its corresponding start,
515 // every delta with the same id falls between start and end, and finish
516 // is last.
517 const indexOf = (type: string, id: string | undefined) =>
518 parts.findIndex((p) => p.type === type && p.id === id);
519 for (const start of toolParamsStart) {
520 const startIdx = indexOf("tool-params-start", start.id);
521 const endIdx = indexOf("tool-params-end", start.id);
522 expect(endIdx).toBeGreaterThan(startIdx);
523 for (const delta of toolParamsDeltas.filter((p) => p.id === start.id)) {
524 const dIdx = parts.indexOf(delta);
525 expect(dIdx).toBeGreaterThan(startIdx);
526 expect(dIdx).toBeLessThan(endIdx);
527 }
528 }
529 expect(parts[parts.length - 1]?.type).toBe("finish");
530 }).pipe(logLevel),
531 {
532 tags: [
533 "provider:cloudflare",
534 "provider:cloudflare:ai",
535 "provider:cloudflare:worker",
536 "live",
537 ],
538 timeout: 180_000,
539 },
540);
541
542test(
543 "concatenated tool-params-delta payloads parse back into the requested arguments",
544 Effect.gen(function* () {
545 const out = yield* stack;
546 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
547
548 const res = yield* client
549 .get(
550 `${out.url}/tool-stream?prompt=${encodeURIComponent(
551 "What's the weather in Portland?",
552 )}`,
553 )
554 .pipe(
555 Effect.retry({
556 schedule: Schedule.exponential("500 millis"),
557 times: 10,
558 }),
559 );
560 expect(res.status).toBe(200);
561
562 const parts = parseSse(yield* res.text);
563 const toolParamsStart = parts.filter((p) => p.type === "tool-params-start");
564 expect(toolParamsStart.length).toBeGreaterThan(0);
565 const firstId = toolParamsStart[0]!.id!;
566
567 // Refactor invariant: `handleToolDeltas` emits each `arguments` fragment
568 // as a `tool-params-delta` in order; concatenating them yields a valid
569 // JSON document with the originally-requested fields.
570 const joined = parts
571 .filter((p) => p.type === "tool-params-delta" && p.id === firstId)
572 .map((p) => p.delta ?? "")
573 .join("");
574 const args = yield* Effect.try({
575 try: () => JSON.parse(joined) as { city?: string },
576 catch: (cause) =>
577 new Error(
578 `Invalid concatenated tool arguments ${JSON.stringify(joined)}: ${cause}`,
579 ),
580 });
581 expect(typeof args.city).toBe("string");
582 expect(args.city!.toLowerCase()).toContain("portland");
583 }).pipe(logLevel),
584 {
585 tags: [
586 "provider:cloudflare",
587 "provider:cloudflare:ai",
588 "provider:cloudflare:worker",
589 "live",
590 ],
591 timeout: 180_000,
592 },
593);
594
595test(
596 "streams Effect-native parts and prints them live",
597 Effect.gen(function* () {
598 const out = yield* stack;
599 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
600
601 const res = yield* client
602 .get(
603 `${out.url}/stream?prompt=${encodeURIComponent("Write a short haiku about Effect TS.")}`,
604 )
605 .pipe(
606 Effect.retry({
607 schedule: Schedule.exponential("500 millis"),
608 times: 10,
609 }),
610 );
611 expect(res.status).toBe(200);
612
613 // Through the Console SERVICE: the runner buffers it per-test (the TUI
614 // detail pane live-tails it), and nothing leaks into the reporter
615 // output. Raw fd writes (the old fs.writeSync(2, ...)) bypass capture.
616 const console = yield* ConsoleService.Console;
617 const print = (s: string) => console.log(s);
618
619 print("--- live stream begin ---");
620 let collected = "";
621
622 yield* res.stream.pipe(
623 Stream.orDie,
624 Stream.decodeText(),
625 Stream.pipeThroughChannel(Sse.decode<never, unknown>()),
626 Stream.runForEach((event) =>
627 Effect.sync(() => {
628 const part = JSON.parse(event.data) as {
629 type: string;
630 delta?: string;
631 id?: string;
632 reason?: string;
633 };
634 switch (part.type) {
635 case "text-start":
636 print(`[text-start id=${part.id}]\n`);
637 break;
638 case "text-delta":
639 print(part.delta ?? "");
640 collected += part.delta ?? "";
641 break;
642 case "text-end":
643 print(`\n[text-end id=${part.id}]\n`);
644 break;
645 case "finish":
646 print(`[finish reason=${part.reason}]\n`);
647 break;
648 default:
649 print(`[${part.type}]\n`);
650 }
651 }),
652 ),
653 );
654
655 print("--- live stream end ---\n");
656 expect(collected.length).toBeGreaterThan(0);
657 }).pipe(logLevel),
658 {
659 tags: [
660 "provider:cloudflare",
661 "provider:cloudflare:ai",
662 "provider:cloudflare:worker",
663 "live",
664 ],
665 timeout: 180_000,
666 },
667);