1import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
2import * as Cloudflare from "@/Cloudflare/index.ts";
3import * as Test from "@/Test/Alchemy";
4import * as kv from "@distilled.cloud/cloudflare/kv";
5import { expect } from "alchemy-test";
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 pathe from "pathe";
11import { expectUrlContains } from "../Utils/Http.ts";
12import { waitForWorkerToBeDeleted } from "../Utils/Worker.ts";
13
14const { test } = Test.make({ providers: Cloudflare.providers() });
15
16const logLevel = Effect.provideService(
17 MinimumLogLevel,
18 process.env.DEBUG ? "Debug" : "Info",
19);
20
21/**
22 * Producer: logs on every request so each invocation produces a streaming
23 * tail session (onset → log → outcome) for its streaming tail consumers.
24 */
25const producerScript = `export default {
26 fetch() {
27 console.log("alchemy-streaming-tail-marker");
28 return new Response("streaming-producer-ok");
29 },
30};`;
31
32/** The recorded shape of one streamed `TailEvent` in a consumer session. */
33interface RecordedTailEvent {
34 invocationId?: string;
35 event?: { type?: string; outcome?: string; message?: unknown };
36}
37
38/**
39 * Cloud-side `tailStream()` delivery is not yet available: Cloudflare's
40 * production API refuses the `streaming_tail_worker` compatibility flag —
41 * the flag that enables workerd's streaming tail model — with the typed
42 * `ScriptStartupError` (code 10021) "The compatibility flag
43 * streaming_tail_worker is experimental and cannot yet be used in Workers
44 * deployed to Cloudflare." A deployed consumer therefore cannot opt into
45 * the `tailStream()` handler, so no events are ever delivered to it. The
46 * ungated probe test below pins that exact rejection.
47 *
48 * Probed 2026-08-04 on the testing account: the `streaming_tail_consumers`
49 * metadata PUT succeeds and the producer serves, but producer invocations
50 * delivered ZERO streaming sessions across 5 minutes of polling (60 × 5s,
51 * each poll re-invoking the producer) — with and without
52 * `observability.traces.enabled` on the producer — while a plain `tail()`
53 * consumer on the same account delivers within seconds
54 * (TailConsumers.test.ts, green). Local workerd delivery is real and
55 * covered by StreamingTailConsumers.local.test.ts. Set this env var once
56 * Cloudflare ships production delivery (the probe test will fail then) to
57 * assert delivery end-to-end.
58 */
59const STREAMING_TAIL_DELIVERY =
60 !!process.env.CLOUDFLARE_TEST_STREAMING_TAIL_DELIVERY;
61
62/**
63 * Ungated platform probe: production refuses the experimental
64 * `streaming_tail_worker` compatibility flag, which is the root cause of
65 * zero cloud-side `tailStream()` delivery (a deployed consumer cannot
66 * enable the streaming tail model). The day this test FAILS — the flag
67 * deploys, or the rejection changes shape — streaming tail workers have
68 * shipped (or changed): re-verify delivery with
69 * CLOUDFLARE_TEST_STREAMING_TAIL_DELIVERY=1 and drop the gate above.
70 */
71test.provider(
72 "production refuses the streaming_tail_worker compatibility flag (probe)",
73 (stack) =>
74 Effect.gen(function* () {
75 yield* stack.destroy();
76
77 const error = yield* stack
78 .deploy(
79 Effect.gen(function* () {
80 return yield* Cloudflare.Worker("StreamingTailFlagProbe", {
81 script: producerScript,
82 compatibility: { flags: ["streaming_tail_worker"] },
83 });
84 }),
85 )
86 .pipe(Effect.flip);
87
88 expect(error._tag).toEqual("ScriptStartupError");
89 if (error._tag === "ScriptStartupError") {
90 expect(error.message).toContain(
91 "streaming_tail_worker is experimental and cannot yet be used",
92 );
93 }
94
95 yield* stack.destroy();
96 }).pipe(logLevel),
97 {
98 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
99 timeout: 240_000,
100 },
101);
102
103/**
104 * `streamingTailConsumers` uploads `streaming_tail_consumers` in the script
105 * metadata. Unlike plain `tail_consumers`, the script-settings read endpoint
106 * does not expose the field, so the attachment is asserted through the
107 * recorded attribute plus (when `CLOUDFLARE_TEST_STREAMING_TAIL_DELIVERY` is
108 * set) actual delivery: the consumer's `tailStream()` persists each completed
109 * session into KV, observed out-of-band via distilled KV reads.
110 */
111test.provider(
112 "worker with streamingTailConsumers attaches and detaches the consumer (delivery gated)",
113 (stack) =>
114 Effect.gen(function* () {
115 const { accountId } = yield* yield* CloudflareEnvironment;
116
117 yield* stack.destroy();
118
119 const deployStack = (tails: "attached" | "detached") =>
120 Effect.gen(function* () {
121 const events = yield* Cloudflare.KV.Namespace("StreamTailEvents");
122 const consumer = yield* Cloudflare.Worker("StreamingTailConsumer", {
123 main: pathe.resolve(
124 import.meta.dirname,
125 "fixtures/tail/streaming-tail-consumer.ts",
126 ),
127 env: { EVENTS: events },
128 });
129 const producer = yield* Cloudflare.Worker("StreamingTailProducer", {
130 script: producerScript,
131 streamingTailConsumers: tails === "attached" ? [consumer] : [],
132 });
133 return { events, consumer, producer };
134 });
135
136 const v1 = yield* stack.deploy(deployStack("attached"));
137
138 // The upload carried `streaming_tail_consumers` (the PUT would have
139 // failed otherwise) and the attribute records the consumer by deployed
140 // script name. GET script-settings has no field for streaming
141 // consumers, so there is no out-of-band settings read to assert
142 // against — the gated delivery assertion below is the only cloud-side
143 // proof of attachment available.
144 expect(v1.producer.streamingTailConsumers).toEqual([
145 { service: v1.consumer.workerName },
146 ]);
147 expect(v1.producer.tailConsumers ?? []).toEqual([]);
148
149 // The producer serves with the streaming consumer attached — the
150 // attachment never breaks the producer's own deploy or execution.
151 yield* expectUrlContains(v1.producer.url!, "streaming-producer-ok", {
152 label: "streaming tail producer",
153 });
154
155 // Delivery assertion — gated (see STREAMING_TAIL_DELIVERY above).
156 if (STREAMING_TAIL_DELIVERY) {
157 // Poll KV until a completed streaming session lands. Each poll
158 // re-invokes the producer so delivery that begins slightly after the
159 // first request still surfaces.
160 const eventKeys = yield* Effect.gen(function* () {
161 yield* expectUrlContains(v1.producer.url!, "streaming-producer-ok", {
162 timeout: "15 seconds",
163 label: "streaming tail producer (poll)",
164 });
165 const keys = yield* kv.listNamespaceKeys({
166 accountId,
167 namespaceId: v1.events.namespaceId,
168 prefix: "evt:",
169 });
170 return keys.result.map((k) => k.name);
171 }).pipe(
172 Effect.repeat({
173 schedule: Schedule.spaced("5 seconds"),
174 until: (keys): boolean => keys.length > 0,
175 times: 18,
176 }),
177 );
178 expect(eventKeys.length).toBeGreaterThan(0);
179
180 // The recorded session is a full onset → log → outcome stream from
181 // one producer invocation, carrying the producer's log marker.
182 const session = yield* kv
183 .getNamespaceValue({
184 accountId,
185 namespaceId: v1.events.namespaceId,
186 keyName: eventKeys[0],
187 })
188 .pipe(
189 Effect.flatMap((res) =>
190 Effect.tryPromise(() =>
191 new Response(
192 Stream.toReadableStream(res.body) as BodyInit,
193 ).text(),
194 ),
195 ),
196 );
197 expect(session).toContain("alchemy-streaming-tail-marker");
198 const events = JSON.parse(session) as RecordedTailEvent[];
199 const types = events.map((tailEvent) => tailEvent.event?.type);
200 expect(types).toContain("onset");
201 expect(types).toContain("outcome");
202 const log = events.find((tailEvent) => tailEvent.event?.type === "log");
203 expect(log?.event?.message).toEqual(["alchemy-streaming-tail-marker"]);
204 }
205
206 // Detaching every consumer is an in-place update, never a replace.
207 const v2 = yield* stack.deploy(deployStack("detached"));
208 expect(v2.producer.workerId).toEqual(v1.producer.workerId);
209 expect(v2.producer.workerName).toEqual(v1.producer.workerName);
210 expect(v2.producer.streamingTailConsumers ?? []).toEqual([]);
211
212 yield* stack.destroy();
213 yield* waitForWorkerToBeDeleted(v1.producer.workerName, accountId);
214 yield* waitForWorkerToBeDeleted(v1.consumer.workerName, accountId);
215 }).pipe(logLevel),
216 {
217 tags: [
218 "provider:cloudflare",
219 "provider:cloudflare:kv",
220 "provider:cloudflare:worker",
221 "live",
222 ],
223 timeout: 240_000,
224 },
225);