1import * as Cloudflare from "@/Cloudflare";
2import * as Alchemy from "@/index.ts";
3import * as Test from "@/Test/Alchemy";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import * as RpcClient from "effect/rpc/RpcClient";
10import ChatPersistenceRpcWorker from "./fixtures/ChatPersistenceRpcWorker.ts";
11import { ChatRpcs } from "./fixtures/ChatRpcs.ts";
12import { Gateway } from "./fixtures/Gateway.ts";
13
14const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
15 providers: Cloudflare.providers(),
16});
17
18// `Test.rpcClientLayer` speaks `RpcSerialization.ndjson` (streaming
19// procedures require newline framing on the wire, matching the worker) and
20// guards the transport against edge-generated HTML bodies (workers.dev
21// placeholder, error pages) that the RPC protocol would otherwise surface as
22// an opaque `RpcClientDefect`; see Test/Http.ts.
23const clientLayer = Test.rpcClientLayer;
24
25const logLevel = Effect.provideService(
26 MinimumLogLevel,
27 process.env.DEBUG ? "Debug" : "Info",
28);
29
30const Stack = Alchemy.Stack(
31 "AiGatewayChatPersistenceRpcStack",
32 {
33 providers: Cloudflare.providers(),
34 state: Cloudflare.state(),
35 },
36 Effect.gen(function* () {
37 const worker = yield* ChatPersistenceRpcWorker;
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
49// Cap exponential backoff at 3s so retries stay bounded when the CF edge is
50// slow (otherwise the geometric blow-up dominates wall time).
51const readinessSchedule = Schedule.min([
52 Schedule.exponential("500 millis"),
53 Schedule.spaced("3 seconds"),
54]);
55
56// A freshly deployed worker isn't live on every Cloudflare edge yet, so the
57// first RPC calls fail with `Handler does not export a fetch() function.` —
58// which arrives at the client as a DEFECT, and `Effect.retry` does not retry
59// defects. Promote defects to failures so the readiness retry absorbs the
60// transient cold-start error (a genuine bug just keeps failing until the
61// retry budget is exhausted).
62const retryReady = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
63 eff.pipe(
64 Effect.catchDefect((defect) => Effect.fail(defect)),
65 Effect.retry({ schedule: readinessSchedule, times: 15 }),
66 );
67
68test(
69 "send round-trips text and turns through the RpcWorker → DO",
70 Effect.gen(function* () {
71 const { url } = yield* stack;
72 const id = `rpc-send-${Date.now()}`;
73
74 yield* Effect.gen(function* () {
75 const client = yield* RpcClient.make(ChatRpcs);
76 const result = yield* client
77 .send({ id, prompt: "Say the single word 'pong'." })
78 .pipe(retryReady);
79
80 expect(typeof result.text).toBe("string");
81 expect(result.text.length).toBeGreaterThan(0);
82 // One turn appends the user prompt and the assistant reply.
83 expect(result.turns).toBe(2);
84 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
85 }).pipe(logLevel),
86 {
87 tags: ["provider:cloudflare", "provider:cloudflare:ai", "live"],
88 timeout: 180_000,
89 },
90);
91
92test(
93 "streamMessage yields effect/ai Response parts that decode to real classes",
94 Effect.gen(function* () {
95 const { url } = yield* stack;
96 const id = `rpc-stream-${Date.now()}`;
97
98 yield* Effect.gen(function* () {
99 const client = yield* RpcClient.make(ChatRpcs);
100
101 const parts = yield* client
102 .streamMessage({
103 id,
104 prompt: "Write a short sentence about Effect TS.",
105 })
106 .pipe(Stream.runCollect, retryReady);
107
108 // The stream parts arrived as decoded `effect/ai` instances (a
109 // discriminated union by `type`), not opaque JSON.
110 expect(parts.length).toBeGreaterThan(0);
111
112 const text = parts
113 .filter((p) => p.type === "text-delta")
114 .map((p) => p.delta)
115 .join("");
116 expect(text.length).toBeGreaterThan(0);
117
118 const finish = parts.find((p) => p.type === "finish");
119 expect(finish).toBeDefined();
120 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
121 }).pipe(logLevel),
122 {
123 tags: ["provider:cloudflare", "provider:cloudflare:ai", "live"],
124 timeout: 180_000,
125 },
126);
127
128test(
129 "streamed turn is persisted: a follow-up send recalls it",
130 Effect.gen(function* () {
131 const { url } = yield* stack;
132 const id = `rpc-memory-${Date.now()}`;
133
134 yield* Effect.gen(function* () {
135 const client = yield* RpcClient.make(ChatRpcs);
136
137 // First turn over the streaming procedure — persistence saves the
138 // appended turn when the stream finalizes.
139 yield* client
140 .streamMessage({ id, prompt: "My name is Sam. Remember it." })
141 .pipe(Stream.runDrain, retryReady);
142
143 // Second turn hits the same DO instance; history is reloaded from
144 // `state.storage`, so the model can recall the name and the turn
145 // count reflects the restored history (2 from the stream + 2).
146 const result = yield* client
147 .send({ id, prompt: "What is my name? Answer with just the name." })
148 .pipe(retryReady);
149
150 expect(result.text.toLowerCase()).toContain("sam");
151 expect(result.turns).toBe(4);
152 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
153 }).pipe(logLevel),
154 {
155 tags: ["provider:cloudflare", "provider:cloudflare:ai", "live"],
156 timeout: 240_000,
157 },
158);