1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3// Alchemy modifications: uses Array<T> syntax for non-tuple array types to match the repository convention.
4import { runInDurableObject } from "cloudflare:test";
5import { env } from "cloudflare:workers";
6import { NonRetryableError } from "cloudflare:workflows";
7import { afterEach, describe, it, vi } from "vitest";
8import workerdUnsafe from "workerd:unsafe";
9import { InstanceEvent } from "../index.ts";
10import { REDACTED_STEP_OUTPUT } from "../context.ts";
11import { computeHash } from "../lib/cache.ts";
12import {
13 InvalidStepReadableStreamError,
14 OversizedStreamChunkError,
15 UnsupportedStreamChunkError,
16 WorkflowTimeoutError,
17 isAbortError,
18} from "../lib/errors.ts";
19import {
20 STREAMING_STEP_CHUNKS_TABLE,
21 getStreamOutputMetaKey,
22 StreamOutputState,
23 rollbackStreamOutput,
24 writeStreamOutput,
25} from "../lib/streams.ts";
26import { MODIFIER_KEYS } from "../modifier.ts";
27import { runWorkflow, runWorkflowAndAwait } from "./utils.ts";
28import type { Engine, EngineLogs } from "../engine.ts";
29import type { StreamOutputMeta } from "../lib/streams.ts";
30
31afterEach(async () => {
32 await workerdUnsafe.abortAllDurableObjects();
33});
34
35describe("Context", () => {
36 it("should provide attempt count 1 on first successful attempt", async ({
37 expect,
38 }) => {
39 let receivedAttempt: number | undefined;
40
41 const engineStub = await runWorkflow(
42 "MOCK-INSTANCE-ID",
43 async (_event, step) => {
44 const result = await step.do("a successful step", async (ctx) => {
45 receivedAttempt = ctx.attempt;
46 return "success";
47 });
48 return result;
49 },
50 );
51
52 await vi.waitUntil(
53 async () => {
54 const logs = (await engineStub.readLogs()) as EngineLogs;
55 return logs.logs.some(
56 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
57 );
58 },
59 { timeout: 5000 },
60 );
61
62 expect(receivedAttempt).toBe(1);
63 });
64
65 it("should provide attempt count to callback", async ({ expect }) => {
66 const receivedAttempts: Array<number> = [];
67
68 const engineStub = await runWorkflow(
69 "MOCK-INSTANCE-ID-RETRY",
70 async (_event, step) => {
71 const result = await step.do(
72 "retrying step",
73 {
74 retries: {
75 limit: 2,
76 delay: 0,
77 },
78 },
79 async (ctx) => {
80 receivedAttempts.push(ctx.attempt);
81 throw new Error(`Throwing`);
82 },
83 );
84 return result;
85 },
86 );
87
88 await vi.waitUntil(
89 async () => {
90 const logs = (await engineStub.readLogs()) as EngineLogs;
91 return logs.logs.some(
92 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
93 );
94 },
95 { timeout: 5000 },
96 );
97
98 // Should have received attempts 1, 2, and 3
99 expect(receivedAttempts).toEqual([1, 2, 3]);
100 });
101
102 it("should wait for retry delays by default", async ({ expect }) => {
103 const start = Date.now();
104 const engineStub = await runWorkflowAndAwait(
105 "MOCK-INSTANCE-RETRY-DELAYS",
106 async (_event, step) => {
107 const result = await step.do(
108 "retrying step with delay",
109 {
110 retries: {
111 limit: 1,
112 delay: "1 second",
113 backoff: "constant",
114 },
115 },
116 async () => {
117 throw new Error("Always fails");
118 },
119 );
120 return result;
121 },
122 );
123 const elapsed = Date.now() - start;
124
125 const logs = (await engineStub.readLogs()) as EngineLogs;
126 expect(
127 logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START),
128 ).toHaveLength(2);
129 // Should have waited at least ~1 second for the retry delay
130 expect(elapsed).toBeGreaterThanOrEqual(900);
131 });
132
133 it("should skip retry delays when disableRetryDelays is set", async ({
134 expect,
135 }) => {
136 const engineId = env.ENGINE.idFromName("MOCK-INSTANCE-DISABLE-RETRY");
137 const engineStub = env.ENGINE.get(engineId);
138
139 // Set the disableRetryDelays flag before running the workflow
140 await runInDurableObject(engineStub, async (_engine, state) => {
141 await state.storage.put(MODIFIER_KEYS.DISABLE_ALL_RETRY_DELAYS, true);
142 });
143
144 const start = Date.now();
145 const stub = await runWorkflowAndAwait(
146 "MOCK-INSTANCE-DISABLE-RETRY",
147 async (_event, step) => {
148 const result = await step.do(
149 "retrying step with delay",
150 {
151 retries: {
152 limit: 2,
153 delay: "10 seconds",
154 backoff: "constant",
155 },
156 },
157 async () => {
158 throw new Error("Always fails");
159 },
160 );
161 return result;
162 },
163 );
164 const elapsed = Date.now() - start;
165
166 const logs = (await stub.readLogs()) as EngineLogs;
167 expect(
168 logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START),
169 ).toHaveLength(3);
170 // Without disableRetryDelays, this would take 20+ seconds (10s + 10s)
171 expect(elapsed).toBeLessThan(5000);
172 });
173
174 // NOTE: The `step` and `config` fields on WorkflowStepContext are new additions
175 // that haven't been published to @cloudflare/workers-types yet. The runtime
176 // provides them, but TypeScript doesn't know about them, so we capture ctx as
177 // `unknown` and assert on the shape.
178 // TODO: Remove these workarounds once https://github.com/cloudflare/workerd/pull/6523 lands.
179
180 it("should provide step name and count in context", async ({ expect }) => {
181 let receivedCtx: unknown;
182
183 const engineStub = await runWorkflow(
184 "MOCK-INSTANCE-STEP-CTX",
185 async (_event, step) => {
186 const result = await step.do("my step", async (ctx) => {
187 receivedCtx = ctx;
188 return "done";
189 });
190 return result;
191 },
192 );
193
194 await vi.waitUntil(
195 async () => {
196 const logs = (await engineStub.readLogs()) as EngineLogs;
197 return logs.logs.some(
198 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
199 );
200 },
201 { timeout: 5000 },
202 );
203
204 expect(receivedCtx).toMatchObject({
205 step: { name: "my step", count: 1 },
206 });
207 });
208
209 it("should increment step count for steps with the same name", async ({
210 expect,
211 }) => {
212 const receivedContexts: Array<unknown> = [];
213
214 const engineStub = await runWorkflow(
215 "MOCK-INSTANCE-STEP-COUNT",
216 async (_event, step) => {
217 await step.do("repeated step", async (ctx) => {
218 receivedContexts.push(ctx);
219 return "first";
220 });
221 await step.do("repeated step", async (ctx) => {
222 receivedContexts.push(ctx);
223 return "second";
224 });
225 await step.do("different step", async (ctx) => {
226 receivedContexts.push(ctx);
227 return "third";
228 });
229 return "done";
230 },
231 );
232
233 // Needs extra headroom: 3 sequential step.do calls + cold DO startup on Windows
234 await vi.waitUntil(
235 async () => {
236 const logs = (await engineStub.readLogs()) as EngineLogs;
237 return logs.logs.some(
238 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
239 );
240 },
241 { timeout: 10000 },
242 );
243
244 expect(receivedContexts[0]).toMatchObject({
245 step: { name: "repeated step", count: 1 },
246 });
247 expect(receivedContexts[1]).toMatchObject({
248 step: { name: "repeated step", count: 2 },
249 });
250 expect(receivedContexts[2]).toMatchObject({
251 step: { name: "different step", count: 1 },
252 });
253 });
254
255 it("should provide resolved config with defaults in context", async ({
256 expect,
257 }) => {
258 let receivedCtx: unknown;
259
260 const engineStub = await runWorkflow(
261 "MOCK-INSTANCE-CTX-CONFIG-DEFAULTS",
262 async (_event, step) => {
263 // No config provided — ctx.config should have all defaults
264 const result = await step.do("default config step", async (ctx) => {
265 receivedCtx = ctx;
266 return "done";
267 });
268 return result;
269 },
270 );
271
272 await vi.waitUntil(
273 async () => {
274 const logs = (await engineStub.readLogs()) as EngineLogs;
275 return logs.logs.some(
276 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
277 );
278 },
279 { timeout: 5000 },
280 );
281
282 expect(receivedCtx).toMatchObject({
283 config: {
284 retries: {
285 limit: 5,
286 delay: 1000,
287 backoff: "exponential",
288 },
289 timeout: "10 minutes",
290 },
291 });
292 });
293
294 it("should provide resolved config with user overrides merged in context", async ({
295 expect,
296 }) => {
297 let receivedCtx: unknown;
298
299 const engineStub = await runWorkflow(
300 "MOCK-INSTANCE-CTX-CONFIG-OVERRIDES",
301 async (_event, step) => {
302 const result = await step.do(
303 "custom config step",
304 {
305 retries: {
306 limit: 3,
307 delay: 500,
308 },
309 timeout: "5 minutes",
310 },
311 async (ctx) => {
312 receivedCtx = ctx;
313 return "done";
314 },
315 );
316 return result;
317 },
318 );
319
320 await vi.waitUntil(
321 async () => {
322 const logs = (await engineStub.readLogs()) as EngineLogs;
323 return logs.logs.some(
324 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
325 );
326 },
327 { timeout: 5000 },
328 );
329
330 // User overrides should be merged with defaults
331 expect(receivedCtx).toMatchObject({
332 config: {
333 retries: {
334 limit: 3,
335 delay: 500,
336 backoff: "exponential",
337 },
338 timeout: "5 minutes",
339 },
340 });
341 });
342
343 it("should not allow user callback to mutate engine retry config", async ({
344 expect,
345 }) => {
346 const receivedAttempts: Array<number> = [];
347
348 const engineStub = await runWorkflow(
349 "MOCK-INSTANCE-CTX-CONFIG-MUTATION",
350 async (_event, step) => {
351 const result = await step.do(
352 "mutating step",
353 {
354 retries: {
355 limit: 1,
356 delay: 0,
357 },
358 },
359 async (ctx) => {
360 receivedAttempts.push(ctx.attempt);
361 // Attempt to escalate retries from 1 to 100
362 (
363 ctx as unknown as { config: { retries: { limit: number } } }
364 ).config.retries.limit = 100;
365 throw new Error("retry me");
366 },
367 );
368 return result;
369 },
370 );
371
372 await vi.waitUntil(
373 async () => {
374 const logs = (await engineStub.readLogs()) as EngineLogs;
375 return logs.logs.some(
376 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
377 );
378 },
379 { timeout: 5000 },
380 );
381
382 // With limit: 1, the engine should execute attempt 1 + 1 retry = 2 attempts total.
383 // If the mutation leaked, this would keep going far beyond 2.
384 expect(receivedAttempts).toEqual([1, 2]);
385 });
386});
387
388// ── Helpers for stream tests ────────────────────────────────────────────────
389
390function encodeUtf8(str: string): Uint8Array {
391 return new TextEncoder().encode(str);
392}
393
394function decodeUtf8(bytes: Uint8Array): string {
395 return new TextDecoder().decode(bytes);
396}
397
398async function readStreamBytes(
399 stream: ReadableStream<Uint8Array>,
400): Promise<Uint8Array> {
401 const chunks: Array<Uint8Array> = [];
402 const reader = stream.getReader();
403 while (true) {
404 const { done, value } = await reader.read();
405 if (done) {
406 break;
407 }
408 chunks.push(value);
409 }
410 let totalLength = 0;
411 for (const chunk of chunks) {
412 totalLength += chunk.byteLength;
413 }
414 const result = new Uint8Array(totalLength);
415 let offset = 0;
416 for (const chunk of chunks) {
417 result.set(chunk, offset);
418 offset += chunk.byteLength;
419 }
420 return result;
421}
422
423function countStreamOutputChunks(
424 state: DurableObjectState,
425 cacheKey: string,
426): number {
427 const row = state.storage.sql
428 .exec<{ cnt: number }>(
429 `SELECT COUNT(*) AS cnt FROM streaming_step_chunks WHERE cache_key = ?`,
430 cacheKey,
431 )
432 .one();
433 return row?.cnt ?? 0;
434}
435
436describe("Context - ReadableStream step outputs", () => {
437 it("should persist a readable stream output and replay from cache", async ({
438 expect,
439 }) => {
440 const payload = "hello from a readable stream ".repeat(500); // ~14KB
441 const payloadBytes = encodeUtf8(payload);
442 let callCount = 0;
443
444 const engineStub = await runWorkflow(
445 "STREAM-BASIC",
446 async (_event, step) => {
447 const stream = await step.do("stream step", async () => {
448 callCount++;
449 return new ReadableStream<Uint8Array>({
450 start(controller) {
451 // Enqueue in two chunks
452 const mid = Math.floor(payloadBytes.length / 2);
453 controller.enqueue(payloadBytes.slice(0, mid));
454 controller.enqueue(payloadBytes.slice(mid));
455 controller.close();
456 },
457 });
458 });
459
460 // The result should be a ReadableStream we can read
461 const bytes = await readStreamBytes(
462 stream as ReadableStream<Uint8Array>,
463 );
464 return decodeUtf8(bytes);
465 },
466 );
467
468 await vi.waitUntil(
469 async () => {
470 const logs = (await engineStub.readLogs()) as EngineLogs;
471 return logs.logs.some(
472 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
473 );
474 },
475 { timeout: 5000 },
476 );
477
478 const logs = (await engineStub.readLogs()) as EngineLogs;
479 const successLog = logs.logs.find(
480 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
481 );
482 expect(successLog?.metadata.result).toBe(payload);
483
484 // The closure should have been called exactly once (cache hit on replay)
485 expect(callCount).toBe(1);
486
487 // Verify the stream metadata is stored and marked complete
488 const hash = await computeHash("stream step");
489 const cacheKey = `${hash}-1`;
490 const metaKey = getStreamOutputMetaKey(cacheKey);
491
492 await runInDurableObject(engineStub, async (_engine, state) => {
493 const meta = (await state.storage.get(metaKey)) as StreamOutputMeta;
494 expect(meta).toBeDefined();
495 expect(meta.state).toBe(StreamOutputState.Complete);
496 expect(meta.totalBytes).toBe(payloadBytes.byteLength);
497 expect(meta.chunkCount).toBeGreaterThanOrEqual(1);
498 });
499 });
500
501 it("should persist an empty readable stream", async ({ expect }) => {
502 const engineStub = await runWorkflow(
503 "STREAM-EMPTY",
504 async (_event, step) => {
505 const stream = await step.do("empty stream step", async () => {
506 return new ReadableStream<Uint8Array>({
507 start(controller) {
508 controller.close();
509 },
510 });
511 });
512
513 // Should be a readable stream that yields no data
514 const bytes = await readStreamBytes(
515 stream as ReadableStream<Uint8Array>,
516 );
517 return bytes.byteLength;
518 },
519 );
520
521 await vi.waitUntil(
522 async () => {
523 const logs = (await engineStub.readLogs()) as EngineLogs;
524 return logs.logs.some(
525 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
526 );
527 },
528 { timeout: 5000 },
529 );
530
531 const logs = (await engineStub.readLogs()) as EngineLogs;
532 const successLog = logs.logs.find(
533 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
534 );
535 expect(successLog?.metadata.result).toBe(0);
536
537 // Verify stream metadata
538 const hash = await computeHash("empty stream step");
539 const cacheKey = `${hash}-1`;
540 const metaKey = getStreamOutputMetaKey(cacheKey);
541
542 await runInDurableObject(engineStub, async (_engine, state) => {
543 const meta = (await state.storage.get(metaKey)) as StreamOutputMeta;
544 expect(meta).toBeDefined();
545 expect(meta.state).toBe(StreamOutputState.Complete);
546 expect(meta.totalBytes).toBe(0);
547 expect(meta.chunkCount).toBe(0);
548 });
549 });
550
551 // ── Tests that exercise stream error paths directly inside the DO ────
552 //
553 // The tests below call writeStreamOutput() inside runInDurableObject()
554 // rather than running a full workflow through step.do().
555 //
556 // Why: In the test harness the user callback runs in a separate
557 // WorkerEntrypoint (TestWorkflow) and communicates with the Engine
558 // Durable Object over RPC. Returning a problematic ReadableStream from
559 // that callback fails at the *RPC transfer* layer with native workerd
560 // errors that are different from the custom errors the engine raises:
561 //
562 // • Locked stream → TypeError: "The ReadableStream has been locked
563 // to a reader."
564 // • Non-byte chunks → TypeError: "This ReadableStream did not return
565 // bytes."
566 // • Very large chunks → Error: "Network connection lost." (the RPC
567 // pipe disconnects under load)
568 //
569 // In production the callback executes in the *same* isolate as the
570 // engine, so the stream reaches writeStreamOutput() directly and the
571 // engine's own validation (iterateStreamChunks, normalizeChunkToUint8Array,
572 // chunk-size check) surfaces the correct custom error types.
573 //
574 // By calling writeStreamOutput() inside the DO we replicate the
575 // production code-path and avoid the RPC transfer artefacts.
576
577 it("should surface a locked readable stream as a fatal error", async ({
578 expect,
579 }) => {
580 const engineId = env.ENGINE.idFromName("STREAM-LOCKED");
581 const engineStub = env.ENGINE.get(engineId);
582
583 await runInDurableObject(engineStub, async (_engine, state) => {
584 // Create the streaming table (normally done during engine init)
585 // eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
586 state.storage.sql.exec(`
587 CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
588 cache_key TEXT NOT NULL,
589 attempt INTEGER NOT NULL,
590 chunk_index INTEGER NOT NULL,
591 chunk BLOB NOT NULL,
592 PRIMARY KEY (cache_key, attempt, chunk_index)
593 ) WITHOUT ROWID
594 `);
595
596 const stream = new ReadableStream<Uint8Array>({
597 start(controller) {
598 controller.enqueue(encodeUtf8("data"));
599 controller.close();
600 },
601 });
602 // Lock the stream by acquiring a reader
603 stream.getReader();
604
605 try {
606 await writeStreamOutput({
607 storage: state.storage,
608 cacheKey: "locked-stream-test",
609 attempt: 1,
610 stream,
611 });
612 expect.unreachable("writeStreamOutput should have thrown");
613 } catch (e) {
614 expect(e).toBeInstanceOf(InvalidStepReadableStreamError);
615 }
616 });
617 });
618
619 it("should surface an unsupported chunk type as a fatal error", async ({
620 expect,
621 }) => {
622 const engineId = env.ENGINE.idFromName("STREAM-UNSUPPORTED-CHUNK");
623 const engineStub = env.ENGINE.get(engineId);
624
625 await runInDurableObject(engineStub, async (_engine, state) => {
626 // eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
627 state.storage.sql.exec(`
628 CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
629 cache_key TEXT NOT NULL,
630 attempt INTEGER NOT NULL,
631 chunk_index INTEGER NOT NULL,
632 chunk BLOB NOT NULL,
633 PRIMARY KEY (cache_key, attempt, chunk_index)
634 ) WITHOUT ROWID
635 `);
636
637 const stream = new ReadableStream({
638 start(controller) {
639 // Enqueue a string -- not supported (only ArrayBuffer / TypedArray)
640 controller.enqueue("this is a string, not bytes");
641 controller.close();
642 },
643 });
644
645 try {
646 await writeStreamOutput({
647 storage: state.storage,
648 cacheKey: "unsupported-chunk-test",
649 attempt: 1,
650 stream,
651 });
652 expect.unreachable("writeStreamOutput should have thrown");
653 } catch (e) {
654 expect(e).toBeInstanceOf(UnsupportedStreamChunkError);
655 }
656 });
657 });
658
659 it("should surface an oversized stream chunk as a fatal error", async ({
660 expect,
661 }) => {
662 const engineId = env.ENGINE.idFromName("STREAM-OVERSIZED-CHUNK");
663 const engineStub = env.ENGINE.get(engineId);
664
665 await runInDurableObject(engineStub, async (_engine, state) => {
666 // eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
667 state.storage.sql.exec(`
668 CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
669 cache_key TEXT NOT NULL,
670 attempt INTEGER NOT NULL,
671 chunk_index INTEGER NOT NULL,
672 chunk BLOB NOT NULL,
673 PRIMARY KEY (cache_key, attempt, chunk_index)
674 ) WITHOUT ROWID
675 `);
676
677 const stream = new ReadableStream<Uint8Array>({
678 start(controller) {
679 // 17 MiB chunk -- exceeds the 16 MiB per-chunk limit
680 controller.enqueue(new Uint8Array(17 * 1024 * 1024));
681 controller.close();
682 },
683 });
684
685 try {
686 await writeStreamOutput({
687 storage: state.storage,
688 cacheKey: "oversized-chunk-test",
689 attempt: 1,
690 stream,
691 });
692 expect.unreachable("writeStreamOutput should have thrown");
693 } catch (e) {
694 expect(e).toBeInstanceOf(OversizedStreamChunkError);
695 }
696 });
697 });
698
699 it("should clean up streaming_step_chunks on restart", async ({ expect }) => {
700 const instanceId = "STREAM-RESTART-CLEANUP";
701 const engineId = env.ENGINE.idFromName(instanceId);
702
703 const engineStub = await runWorkflowAndAwait(
704 instanceId,
705 async (_event, step) => {
706 const stream = await step.do("stream before restart", async () => {
707 return new ReadableStream<Uint8Array>({
708 start(controller) {
709 controller.enqueue(encodeUtf8("data for restart test"));
710 controller.close();
711 },
712 });
713 });
714 const bytes = await readStreamBytes(
715 stream as ReadableStream<Uint8Array>,
716 );
717 return decodeUtf8(bytes);
718 },
719 );
720
721 // Verify chunks exist before restart
722 const hash = await computeHash("stream before restart");
723 const cacheKey = `${hash}-1`;
724
725 await runInDurableObject(engineStub, async (_engine, state) => {
726 const chunkCount = countStreamOutputChunks(state, cacheKey);
727 expect(chunkCount).toBeGreaterThanOrEqual(1);
728 });
729
730 // Trigger restart
731 try {
732 await runInDurableObject(engineStub, async (engine) => {
733 await engine.changeInstanceStatus("restart");
734 });
735 } catch (e) {
736 if (!isAbortError(e)) {
737 throw e;
738 }
739 }
740
741 const restartedStub: DurableObjectStub<Engine> = env.ENGINE.get(engineId);
742 await runInDurableObject(restartedStub, async (engine, state) => {
743 await engine.attemptRestart();
744
745 // Check immediately after attemptRestart returns, before the
746 // fire-and-forget init() call (engine.ts:892) has a chance to
747 // re-execute the workflow and recreate the chunks.
748 const chunkCount = countStreamOutputChunks(state, cacheKey);
749 expect(chunkCount).toBe(0);
750 });
751 });
752
753 it("should preserve mock stream chunks across restart", async ({
754 expect,
755 }) => {
756 const mockPayload = "mock stream survives restart";
757 const mockPayloadBytes = encodeUtf8(mockPayload);
758
759 const instanceId = "STREAM-MOCK-SURVIVES-RESTART";
760 const engineId = env.ENGINE.idFromName(instanceId);
761 const engineStub = env.ENGINE.get(engineId);
762
763 // Set up a mocked stream step via the modifier
764 await runInDurableObject(engineStub, async (engine) => {
765 const modifier = engine.getInstanceModifier();
766 await modifier.mockStepResult(
767 { name: "mocked restart stream" },
768 new ReadableStream<Uint8Array>({
769 start(controller) {
770 controller.enqueue(mockPayloadBytes);
771 controller.close();
772 },
773 }),
774 );
775 });
776
777 const hash = await computeHash("mocked restart stream");
778 const baseCacheKey = `${hash}-1`;
779
780 // Run the workflow once so the engine is initialised and can be restarted
781 const stub = await runWorkflow(instanceId, async (_event, step) => {
782 const stream = await step.do("mocked restart stream", async () => {
783 return new ReadableStream<Uint8Array>({
784 start(controller) {
785 controller.enqueue(encodeUtf8("WRONG - real step ran"));
786 controller.close();
787 },
788 });
789 });
790 const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
791 return decodeUtf8(bytes);
792 });
793
794 await vi.waitUntil(
795 async () => {
796 const logs = (await stub.readLogs()) as EngineLogs;
797 return logs.logs.some(
798 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
799 );
800 },
801 { timeout: 5000 },
802 );
803
804 // Trigger restart
805 try {
806 await runInDurableObject(stub, async (engine) => {
807 await engine.changeInstanceStatus("restart");
808 });
809 } catch (e) {
810 if (!isAbortError(e)) {
811 throw e;
812 }
813 }
814
815 // After restart, mock stream chunks (attempt=0) must still exist
816 const restartedStub: DurableObjectStub<Engine> = env.ENGINE.get(engineId);
817 await runInDurableObject(restartedStub, async (engine, state) => {
818 await engine.attemptRestart();
819
820 const chunkCount = countStreamOutputChunks(state, baseCacheKey);
821 expect(chunkCount).toBeGreaterThanOrEqual(1);
822 });
823
824 // Run the workflow again — the mock sentinel (KV) and its SQL chunks
825 // should both be present, allowing the mocked stream to replay correctly.
826 const stub2 = await runWorkflow(instanceId, async (_event, step) => {
827 const stream = await step.do("mocked restart stream", async () => {
828 return new ReadableStream<Uint8Array>({
829 start(controller) {
830 controller.enqueue(
831 encodeUtf8("WRONG - real step ran after restart"),
832 );
833 controller.close();
834 },
835 });
836 });
837 const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
838 return decodeUtf8(bytes);
839 });
840
841 await vi.waitUntil(
842 async () => {
843 const logs = (await stub2.readLogs()) as EngineLogs;
844 return logs.logs.some(
845 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
846 );
847 },
848 { timeout: 5000 },
849 );
850
851 const logs = (await stub2.readLogs()) as EngineLogs;
852 const successLog = logs.logs.find(
853 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
854 );
855 expect(successLog?.metadata.result).toBe(mockPayload);
856 });
857
858 it("should normalize TypedArray chunks to Uint8Array on replay", async ({
859 expect,
860 }) => {
861 // Return Int16Array chunks -- they should be stored as raw bytes
862 // and replayed as Uint8Array
863 const int16Data = new Int16Array([1, 2, 3, 256, -1]);
864 const expectedBytes = new Uint8Array(
865 int16Data.buffer,
866 int16Data.byteOffset,
867 int16Data.byteLength,
868 );
869
870 const engineStub = await runWorkflow(
871 "STREAM-TYPED-ARRAY",
872 async (_event, step) => {
873 const stream = await step.do("typed array step", async () => {
874 return new ReadableStream({
875 start(controller) {
876 controller.enqueue(int16Data);
877 controller.close();
878 },
879 });
880 });
881
882 const bytes = await readStreamBytes(
883 stream as ReadableStream<Uint8Array>,
884 );
885 return Array.from(bytes);
886 },
887 );
888
889 await vi.waitUntil(
890 async () => {
891 const logs = (await engineStub.readLogs()) as EngineLogs;
892 return logs.logs.some(
893 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
894 );
895 },
896 { timeout: 5000 },
897 );
898
899 const logs = (await engineStub.readLogs()) as EngineLogs;
900 const successLog = logs.logs.find(
901 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
902 );
903 expect(successLog?.metadata.result).toEqual(Array.from(expectedBytes));
904 });
905
906 it("should return a replay ReadableStream from waitForStepResult", async ({
907 expect,
908 }) => {
909 const payload = "stream content for waitForStepResult";
910 const payloadBytes = encodeUtf8(payload);
911
912 const engineStub = await runWorkflow(
913 "STREAM-WAIT-FOR-STEP",
914 async (_event, step) => {
915 const stream = await step.do("stream step", async () => {
916 return new ReadableStream<Uint8Array>({
917 start(controller) {
918 controller.enqueue(payloadBytes);
919 controller.close();
920 },
921 });
922 });
923 const bytes = await readStreamBytes(
924 stream as ReadableStream<Uint8Array>,
925 );
926 return decodeUtf8(bytes);
927 },
928 );
929
930 // Use engine's waitForStepResult to get the stream
931 const stepResult = await engineStub.waitForStepResult("stream step");
932
933 // Should be a ReadableStream
934 expect(stepResult).toBeInstanceOf(ReadableStream);
935
936 const replayBytes = await readStreamBytes(
937 stepResult as ReadableStream<Uint8Array>,
938 );
939 expect(decodeUtf8(replayBytes)).toBe(payload);
940
941 // Wait for workflow to finish
942 await vi.waitUntil(
943 async () => {
944 const logs = (await engineStub.readLogs()) as EngineLogs;
945 return logs.logs.some(
946 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
947 );
948 },
949 { timeout: 5000 },
950 );
951 });
952
953 it("should mock a step result with a ReadableStream via modifier", async ({
954 expect,
955 }) => {
956 const mockPayload = "mocked stream content from modifier";
957 const mockPayloadBytes = encodeUtf8(mockPayload);
958
959 const instanceId = "STREAM-MOCK-STEP-RESULT";
960 const engineId = env.ENGINE.idFromName(instanceId);
961 const engineStub = env.ENGINE.get(engineId);
962
963 // Set up the mock stream result via the modifier before running the workflow
964 await runInDurableObject(engineStub, async (engine) => {
965 const modifier = engine.getInstanceModifier();
966 await modifier.mockStepResult(
967 { name: "mocked stream step" },
968 new ReadableStream<Uint8Array>({
969 start(controller) {
970 controller.enqueue(mockPayloadBytes);
971 controller.close();
972 },
973 }),
974 );
975 });
976
977 const hash = await computeHash("mocked stream step");
978 const baseCacheKey = `${hash}-1`;
979 const valueKey = `${baseCacheKey}-value`;
980
981 await runInDurableObject(engineStub, async (_engine, state) => {
982 // normal non-mocked key should not exist
983 expect(
984 await state.storage.get(getStreamOutputMetaKey(baseCacheKey)),
985 ).toBeUndefined();
986
987 const replaceResult = await state.storage.get<{
988 __mockStreamOutput: true;
989 cacheKey: string;
990 meta: StreamOutputMeta;
991 }>(`${MODIFIER_KEYS.REPLACE_RESULT}${valueKey}`);
992
993 expect(replaceResult?.__mockStreamOutput).toBe(true);
994 expect(replaceResult?.cacheKey).toBe(baseCacheKey);
995 expect(replaceResult?.meta.state).toBe(StreamOutputState.Complete);
996 expect(replaceResult?.meta.attempt).toBe(0);
997 expect(
998 countStreamOutputChunks(state, baseCacheKey),
999 ).toBeGreaterThanOrEqual(1);
1000 });
1001
1002 // Run a workflow that uses the mocked step
1003 const stub = await runWorkflow(instanceId, async (_event, step) => {
1004 const stream = await step.do("mocked stream step", async () => {
1005 // This should NOT be called -- the mock replaces it
1006 return new ReadableStream<Uint8Array>({
1007 start(controller) {
1008 controller.enqueue(encodeUtf8("WRONG - real step ran"));
1009 controller.close();
1010 },
1011 });
1012 });
1013 const bytes = await readStreamBytes(stream as ReadableStream<Uint8Array>);
1014 return decodeUtf8(bytes);
1015 });
1016
1017 await vi.waitUntil(
1018 async () => {
1019 const logs = (await stub.readLogs()) as EngineLogs;
1020 return logs.logs.some(
1021 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1022 );
1023 },
1024 { timeout: 5000 },
1025 );
1026
1027 const logs = (await stub.readLogs()) as EngineLogs;
1028 const successLog = logs.logs.find(
1029 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1030 );
1031 // The workflow should have received the mocked stream content
1032 expect(successLog?.metadata.result).toBe(mockPayload);
1033 });
1034
1035 it("should resolve stream output to a text preview in readLogs", async ({
1036 expect,
1037 }) => {
1038 const payload = "hello from stream preview test";
1039 const payloadBytes = encodeUtf8(payload);
1040
1041 const engineStub = await runWorkflow(
1042 "STREAM-PREVIEW-TEXT",
1043 async (_event, step) => {
1044 const stream = await step.do("preview step", async () => {
1045 return new ReadableStream<Uint8Array>({
1046 start(controller) {
1047 controller.enqueue(payloadBytes);
1048 controller.close();
1049 },
1050 });
1051 });
1052 const bytes = await readStreamBytes(
1053 stream as ReadableStream<Uint8Array>,
1054 );
1055 return decodeUtf8(bytes);
1056 },
1057 );
1058
1059 await vi.waitUntil(
1060 async () => {
1061 const logs = (await engineStub.readLogs()) as EngineLogs;
1062 return logs.logs.some(
1063 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1064 );
1065 },
1066 { timeout: 5000 },
1067 );
1068
1069 const logs = (await engineStub.readLogs()) as EngineLogs;
1070 const stepLog = logs.logs.find(
1071 (val) => val.event === InstanceEvent.STEP_SUCCESS,
1072 );
1073 // readLogs() should resolve the stream output metadata to the preview text
1074 expect(stepLog?.metadata.result).toBe(payload);
1075 });
1076
1077 it("should truncate a long stream preview in readLogs", async ({
1078 expect,
1079 }) => {
1080 // Generate a payload longer than the 1024-char preview limit
1081 const payload = "A".repeat(2048);
1082 const payloadBytes = encodeUtf8(payload);
1083
1084 const engineStub = await runWorkflow(
1085 "STREAM-PREVIEW-TRUNCATED",
1086 async (_event, step) => {
1087 const stream = await step.do("long preview step", async () => {
1088 return new ReadableStream<Uint8Array>({
1089 start(controller) {
1090 controller.enqueue(payloadBytes);
1091 controller.close();
1092 },
1093 });
1094 });
1095 const bytes = await readStreamBytes(
1096 stream as ReadableStream<Uint8Array>,
1097 );
1098 return decodeUtf8(bytes);
1099 },
1100 );
1101
1102 await vi.waitUntil(
1103 async () => {
1104 const logs = (await engineStub.readLogs()) as EngineLogs;
1105 return logs.logs.some(
1106 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1107 );
1108 },
1109 { timeout: 5000 },
1110 );
1111
1112 const logs = (await engineStub.readLogs()) as EngineLogs;
1113 const stepLog = logs.logs.find(
1114 (val) => val.event === InstanceEvent.STEP_SUCCESS,
1115 );
1116 // readLogs() should truncate the preview to 1024 chars
1117 expect(stepLog?.metadata.result).toBe(
1118 "A".repeat(1024) + "[truncated output]",
1119 );
1120 });
1121
1122 it("should resolve non-UTF-8 stream output to a binary summary in readLogs", async ({
1123 expect,
1124 }) => {
1125 // Write raw bytes that are not valid UTF-8
1126 const invalidUtf8 = new Uint8Array([0xff, 0xfe, 0x80, 0x81]);
1127
1128 const engineStub = await runWorkflow(
1129 "STREAM-PREVIEW-BINARY",
1130 async (_event, step) => {
1131 const stream = await step.do("binary step", async () => {
1132 return new ReadableStream<Uint8Array>({
1133 start(controller) {
1134 controller.enqueue(invalidUtf8);
1135 controller.close();
1136 },
1137 });
1138 });
1139 const bytes = await readStreamBytes(
1140 stream as ReadableStream<Uint8Array>,
1141 );
1142 return bytes.byteLength;
1143 },
1144 );
1145
1146 await vi.waitUntil(
1147 async () => {
1148 const logs = (await engineStub.readLogs()) as EngineLogs;
1149 return logs.logs.some(
1150 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1151 );
1152 },
1153 { timeout: 5000 },
1154 );
1155
1156 const logs = (await engineStub.readLogs()) as EngineLogs;
1157 const stepLog = logs.logs.find(
1158 (val) => val.event === InstanceEvent.STEP_SUCCESS,
1159 );
1160 // readLogs() should fall back to a binary size summary
1161 expect(stepLog?.metadata.result).toBe(
1162 `[ReadableStream (binary): ${invalidUtf8.byteLength} bytes]`,
1163 );
1164 });
1165
1166 it("should time out during stream write and roll back", async ({
1167 expect,
1168 }) => {
1169 // In the full workflow path a stream-write timeout triggers retries
1170 // (default limit: 5, exponential backoff from 1 s), so the test would
1171 // exceed its timeout before WORKFLOW_FAILURE is ever logged. Testing
1172 // writeStreamOutput() directly lets us verify timeout + rollback
1173 // behaviour without retry overhead.
1174
1175 const engineId = env.ENGINE.idFromName("STREAM-TIMEOUT");
1176 const engineStub = env.ENGINE.get(engineId);
1177
1178 const cacheKey = "timeout-stream-test";
1179
1180 await runInDurableObject(engineStub, async (_engine, state) => {
1181 // eslint-disable-next-line workers-sdk/no-unsafe-command-execution -- DO SQL exec, not child_process
1182 state.storage.sql.exec(`
1183 CREATE TABLE IF NOT EXISTS ${STREAMING_STEP_CHUNKS_TABLE} (
1184 cache_key TEXT NOT NULL,
1185 attempt INTEGER NOT NULL,
1186 chunk_index INTEGER NOT NULL,
1187 chunk BLOB NOT NULL,
1188 PRIMARY KEY (cache_key, attempt, chunk_index)
1189 ) WITHOUT ROWID
1190 `);
1191
1192 const abortController = new AbortController();
1193
1194 // Create a timeout promise that rejects after 1 second, mirroring
1195 // the timeoutPromise() used by Context.do() in production.
1196 const timeoutTask = new Promise<never>((_, reject) => {
1197 setTimeout(() => {
1198 const error = new WorkflowTimeoutError(
1199 "Execution timed out after 1000ms",
1200 );
1201 abortController.abort(error);
1202 reject(error);
1203 }, 1000);
1204 });
1205
1206 // Stream that never closes — each pull emits a small chunk then
1207 // waits longer than the timeout.
1208 const stream = new ReadableStream<Uint8Array>({
1209 async pull(controller) {
1210 controller.enqueue(encodeUtf8("chunk "));
1211 await new Promise((resolve) => setTimeout(resolve, 2000));
1212 },
1213 });
1214
1215 try {
1216 await writeStreamOutput({
1217 storage: state.storage,
1218 cacheKey,
1219 attempt: 1,
1220 stream,
1221 signal: abortController.signal,
1222 timeoutTask,
1223 });
1224 expect.unreachable("writeStreamOutput should have thrown");
1225 } catch (e) {
1226 expect(e).toBeInstanceOf(WorkflowTimeoutError);
1227 }
1228
1229 // Mimic the rollback that context.ts performs in the outer catch
1230 // (line ~704) when a stream write times out.
1231 await rollbackStreamOutput(state.storage, cacheKey, 1);
1232
1233 // Verify that all chunks and metadata were cleaned up
1234 const chunkCount = countStreamOutputChunks(state, cacheKey);
1235 expect(chunkCount).toBe(0);
1236
1237 const metaKey = getStreamOutputMetaKey(cacheKey);
1238 const meta = await state.storage.get(metaKey);
1239 expect(meta).toBeUndefined();
1240 });
1241 });
1242});
1243
1244describe("Context - typed-array step outputs (issue #14101)", () => {
1245 it("should persist a 200KB Uint8Array step output on a tight backing buffer", async ({
1246 expect,
1247 }) => {
1248 // Baseline: a 200KB view sized exactly to its backing buffer must
1249 // succeed. This case worked on `main` too — anchors the regression test
1250 // suite to "small enough to fit, regardless of normalisation".
1251 const engineStub = await runWorkflow(
1252 "UINT8-TIGHT-200K",
1253 async (_event, step) => {
1254 return await step.do("emit-tight-bytes", async () => {
1255 return new Uint8Array(200_000);
1256 });
1257 },
1258 );
1259
1260 await vi.waitUntil(
1261 async () => {
1262 const logs = (await engineStub.readLogs()) as EngineLogs;
1263 return logs.logs.some(
1264 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1265 );
1266 },
1267 { timeout: 10000 },
1268 );
1269
1270 const logs = (await engineStub.readLogs()) as EngineLogs;
1271 expect(
1272 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
1273 ).toBe(true);
1274 expect(
1275 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
1276 ).toBe(false);
1277 });
1278
1279 it("should persist a 200KB Uint8Array view sliced from an 800KB backing buffer (regression #14101)", async ({
1280 expect,
1281 }) => {
1282 // The exact reporter's repro: a Uint8Array view sized 200KB but
1283 // sliced from a much larger backing ArrayBuffer (800KB here). Prior
1284 // to the fix, workerd's v8::ValueSerializer would serialise the
1285 // entire 800KB backing buffer along with the view, blowing past the
1286 // 1MiB SQL blob limit and surfacing a raw `SQLITE_TOOBIG` error.
1287 const engineStub = await runWorkflow(
1288 "UINT8-SLICED-200K-OF-800K",
1289 async (_event, step) => {
1290 return await step.do("emit-sliced-bytes", async () => {
1291 const backing = new ArrayBuffer(800_000);
1292 return new Uint8Array(backing, 0, 200_000);
1293 });
1294 },
1295 );
1296
1297 await vi.waitUntil(
1298 async () => {
1299 const logs = (await engineStub.readLogs()) as EngineLogs;
1300 return logs.logs.some(
1301 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1302 );
1303 },
1304 { timeout: 10000 },
1305 );
1306
1307 const logs = (await engineStub.readLogs()) as EngineLogs;
1308 expect(
1309 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
1310 ).toBe(true);
1311 expect(
1312 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
1313 ).toBe(false);
1314 });
1315
1316 it("should return the original Uint8Array to the next step in the live execution path", async ({
1317 expect,
1318 }) => {
1319 // The persisted form preserves the view type (`Uint8Array` stays
1320 // `Uint8Array`) so the live execution path and cached replays observe
1321 // the same shape — downstream user code that branches on
1322 // `value instanceof Uint8Array` works either way.
1323 let observedInNext: unknown = "not-seen";
1324 const engineStub = await runWorkflow(
1325 "UINT8-ROUND-TRIP",
1326 async (_event, step) => {
1327 const bytes = await step.do("emit-bytes", async () => {
1328 return new Uint8Array([1, 2, 3, 4, 5]);
1329 });
1330 await step.do("read-bytes", async () => {
1331 observedInNext = bytes;
1332 return "ok";
1333 });
1334 },
1335 );
1336
1337 await vi.waitUntil(
1338 async () => {
1339 const logs = (await engineStub.readLogs()) as EngineLogs;
1340 return logs.logs.some(
1341 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1342 );
1343 },
1344 { timeout: 10000 },
1345 );
1346
1347 expect(observedInNext).toBeInstanceOf(Uint8Array);
1348 expect(Array.from(observedInNext as Uint8Array)).toEqual([1, 2, 3, 4, 5]);
1349 });
1350
1351 it("should preserve typed-array view types other than Uint8Array on the live execution path", async ({
1352 expect,
1353 }) => {
1354 // Persisting must preserve the constructor type (`Int16Array` stays
1355 // `Int16Array`, etc.) so downstream `instanceof` checks behave the
1356 // same whether the step ran live or via cached replay.
1357 let observedInNext: unknown = "not-seen";
1358 const engineStub = await runWorkflow(
1359 "INT16-ROUND-TRIP",
1360 async (_event, step) => {
1361 const bytes = await step.do("emit-int16", async () => {
1362 return new Int16Array([1, 2, 3, 4, 5]);
1363 });
1364 await step.do("read-int16", async () => {
1365 observedInNext = bytes;
1366 return "ok";
1367 });
1368 },
1369 );
1370
1371 await vi.waitUntil(
1372 async () => {
1373 const logs = (await engineStub.readLogs()) as EngineLogs;
1374 return logs.logs.some(
1375 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1376 );
1377 },
1378 { timeout: 10000 },
1379 );
1380
1381 expect(observedInNext).toBeInstanceOf(Int16Array);
1382 expect(Array.from(observedInNext as Int16Array)).toEqual([1, 2, 3, 4, 5]);
1383 });
1384
1385 it("should compact typed-array views nested inside an object (deep-walk regression)", async ({
1386 expect,
1387 }) => {
1388 // A view sliced from a much larger backing buffer would bloat past
1389 // SQLITE_TOOBIG if nested-level normalisation were missing. Recurses
1390 // into plain objects.
1391 const engineStub = await runWorkflow(
1392 "NESTED-UINT8-SLICED",
1393 async (_event, step) => {
1394 return await step.do("emit-nested-sliced-bytes", async () => {
1395 const backing = new ArrayBuffer(800_000);
1396 return {
1397 image: new Uint8Array(backing, 0, 200_000),
1398 meta: { width: 100, height: 100 },
1399 };
1400 });
1401 },
1402 );
1403
1404 await vi.waitUntil(
1405 async () => {
1406 const logs = (await engineStub.readLogs()) as EngineLogs;
1407 return logs.logs.some(
1408 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1409 );
1410 },
1411 { timeout: 10000 },
1412 );
1413
1414 const logs = (await engineStub.readLogs()) as EngineLogs;
1415 expect(
1416 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
1417 ).toBe(true);
1418 expect(
1419 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
1420 ).toBe(false);
1421 });
1422
1423 it("should compact typed-array views nested deep inside arrays of objects", async ({
1424 expect,
1425 }) => {
1426 // Two levels of nesting (array of object of view). Confirms the
1427 // walker descends through both arrays and plain objects.
1428 const engineStub = await runWorkflow(
1429 "ARRAY-OF-OBJECTS-WITH-SLICED-VIEWS",
1430 async (_event, step) => {
1431 return await step.do("emit-array-of-objects", async () => {
1432 const backing = new ArrayBuffer(800_000);
1433 return [
1434 { data: new Uint8Array(backing, 0, 100_000) },
1435 { data: new Uint8Array(backing, 100_000, 100_000) },
1436 ];
1437 });
1438 },
1439 );
1440
1441 await vi.waitUntil(
1442 async () => {
1443 const logs = (await engineStub.readLogs()) as EngineLogs;
1444 return logs.logs.some(
1445 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1446 );
1447 },
1448 { timeout: 10000 },
1449 );
1450
1451 const logs = (await engineStub.readLogs()) as EngineLogs;
1452 expect(
1453 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
1454 ).toBe(true);
1455 expect(
1456 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
1457 ).toBe(false);
1458 });
1459});
1460
1461describe("Sensitive step output", () => {
1462 it("should redact a sensitive step's output in logs while passing the real value downstream", async ({
1463 expect,
1464 }) => {
1465 let downstreamValue: unknown;
1466
1467 const engineStub = await runWorkflowAndAwait(
1468 "SENSITIVE-STEP-OUTPUT",
1469 async (_event, step) => {
1470 const secret = await step.do(
1471 "sensitive step",
1472 { sensitive: "output" },
1473 async () => {
1474 return { token: "super-secret" };
1475 },
1476 );
1477 await step.do("downstream step", async () => {
1478 downstreamValue = secret;
1479 return "ok";
1480 });
1481 },
1482 );
1483
1484 await vi.waitUntil(
1485 async () => {
1486 const logs = (await engineStub.readLogs()) as EngineLogs;
1487 return logs.logs.some(
1488 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1489 );
1490 },
1491 { timeout: 5000 },
1492 );
1493
1494 const logs = (await engineStub.readLogs()) as EngineLogs;
1495 const stepLog = logs.logs.find(
1496 (val) =>
1497 val.event === InstanceEvent.STEP_SUCCESS &&
1498 val.target === "sensitive step-1",
1499 );
1500 expect(stepLog?.metadata.result).toBe(REDACTED_STEP_OUTPUT);
1501
1502 // The real value is still cached and handed to the running workflow.
1503 expect(downstreamValue).toEqual({ token: "super-secret" });
1504 });
1505
1506 it("should redact a sensitive step's output from waitForStepResult", async ({
1507 expect,
1508 }) => {
1509 const engineStub = await runWorkflowAndAwait(
1510 "SENSITIVE-STEP-WAIT-RESULT",
1511 async (_event, step) => {
1512 await step.do(
1513 "sensitive step",
1514 { sensitive: "output" },
1515 async () => "super-secret",
1516 );
1517 },
1518 );
1519
1520 await vi.waitUntil(
1521 async () => {
1522 const logs = (await engineStub.readLogs()) as EngineLogs;
1523 return logs.logs.some(
1524 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1525 );
1526 },
1527 { timeout: 5000 },
1528 );
1529
1530 const result = await engineStub.waitForStepResult("sensitive step");
1531 expect(result).toBe(REDACTED_STEP_OUTPUT);
1532 });
1533
1534 it("should not redact a sensitive step's error", async ({ expect }) => {
1535 const engineStub = await runWorkflowAndAwait(
1536 "SENSITIVE-STEP-ERROR",
1537 async (_event, step) => {
1538 try {
1539 await step.do(
1540 "sensitive failing step",
1541 { sensitive: "output", retries: { limit: 0, delay: 0 } },
1542 async () => {
1543 throw new NonRetryableError("boom with secret context");
1544 },
1545 );
1546 } catch {}
1547 },
1548 );
1549
1550 await vi.waitUntil(
1551 async () => {
1552 const logs = (await engineStub.readLogs()) as EngineLogs;
1553 return logs.logs.some(
1554 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1555 );
1556 },
1557 { timeout: 5000 },
1558 );
1559
1560 const logs = (await engineStub.readLogs()) as EngineLogs;
1561 const attemptFailure = logs.logs.find(
1562 (val) => val.event === InstanceEvent.ATTEMPT_FAILURE,
1563 );
1564 const error = attemptFailure?.metadata.error as
1565 | { message: string }
1566 | undefined;
1567 expect(error?.message).toContain("boom with secret context");
1568 });
1569
1570 it("should redact a sensitive streaming step output in readLogs", async ({
1571 expect,
1572 }) => {
1573 const payload = "streamed secret";
1574 const payloadBytes = encodeUtf8(payload);
1575
1576 const engineStub = await runWorkflowAndAwait(
1577 "SENSITIVE-STREAM-OUTPUT",
1578 async (_event, step) => {
1579 const stream = await step.do(
1580 "sensitive stream step",
1581 { sensitive: "output" },
1582 async () => {
1583 return new ReadableStream<Uint8Array>({
1584 start(controller) {
1585 controller.enqueue(payloadBytes);
1586 controller.close();
1587 },
1588 });
1589 },
1590 );
1591 const bytes = await readStreamBytes(
1592 stream as ReadableStream<Uint8Array>,
1593 );
1594 return decodeUtf8(bytes);
1595 },
1596 );
1597
1598 await vi.waitUntil(
1599 async () => {
1600 const logs = (await engineStub.readLogs()) as EngineLogs;
1601 return logs.logs.some(
1602 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
1603 );
1604 },
1605 { timeout: 5000 },
1606 );
1607
1608 const logs = (await engineStub.readLogs()) as EngineLogs;
1609 const stepLog = logs.logs.find(
1610 (val) =>
1611 val.event === InstanceEvent.STEP_SUCCESS &&
1612 val.target === "sensitive stream step-1",
1613 );
1614 expect(stepLog?.metadata.result).toBe(REDACTED_STEP_OUTPUT);
1615 });
1616});