1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3// Alchemy modifications: tests RPC disposal with a typed stub because the current Vitest pool does not expose methods replaced on a live Durable Object.
4import { createExecutionContext, runInDurableObject } from "cloudflare:test";
5import { env } from "cloudflare:workers";
6import { describe, it, vi } from "vitest";
7import { InstanceEvent, InstanceStatus } from "../index.ts";
8import { WorkflowBinding, WorkflowHandle } from "../binding.ts";
9import { WorkflowSubscriptionTarget } from "../subscription.ts";
10import { setTestWorkflowCallback } from "./test-entry.ts";
11import type { Engine, EngineLogs } from "../engine.ts";
12import type {
13 WorkflowSubscription,
14 WorkflowSubscriptionEvent,
15} from "../subscription.ts";
16import type { WorkflowEvent } from "cloudflare:workers";
17
18let instanceCounter = 0;
19function uniqueId(prefix = "instance"): string {
20 return `${prefix}-${++instanceCounter}`;
21}
22
23function createBinding(): WorkflowBinding {
24 const ctx = createExecutionContext();
25 return new WorkflowBinding(ctx, {
26 ENGINE: env.ENGINE,
27 BINDING_NAME: "TEST_WORKFLOW",
28 WORKFLOW_NAME: "test-workflow",
29 });
30}
31
32async function waitUntilLogEvent(
33 engineStub: DurableObjectStub<Engine>,
34 event: InstanceEvent,
35 timeout = 5000,
36): Promise<void> {
37 await vi.waitUntil(
38 async () => {
39 const logs = (await engineStub.readLogs()) as EngineLogs;
40 const hasEvent = logs.logs.some((log) => log.event === event);
41 return hasEvent;
42 },
43 { timeout },
44 );
45}
46
47describe("WorkflowBinding", () => {
48 describe("create()", () => {
49 it("should create an instance with provided id and params", async ({
50 expect,
51 }) => {
52 const id = uniqueId();
53 const binding = createBinding();
54 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
55
56 setTestWorkflowCallback(async (event) => {
57 return (event as WorkflowEvent<{ key: string }>).payload;
58 });
59
60 const params = { key: "test-value" };
61 const result = await binding.create({ id, params });
62
63 expect(result.id).toBe(id);
64
65 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
66
67 const instance = await binding.get(id);
68 const status = await instance.status();
69 expect(status.output).toEqual(params);
70 });
71
72 it("should pass the workflow name in the event", async ({ expect }) => {
73 const id = uniqueId();
74 const binding = createBinding();
75 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
76
77 setTestWorkflowCallback(async (event) => {
78 return (event as WorkflowEvent<unknown>).workflowName;
79 });
80
81 await binding.create({ id });
82
83 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
84
85 const instance = await binding.get(id);
86 const status = await instance.status();
87 expect(status.output).toBe("test-workflow");
88 });
89
90 it("should auto-generate id when not provided", async ({ expect }) => {
91 const binding = createBinding();
92 setTestWorkflowCallback(async () => "done");
93 const result = await binding.create();
94
95 expect(result.id).toBeDefined();
96 expect(result.id.length).toBeGreaterThan(0);
97
98 // Wait for the workflow to complete before the test ends so
99 // the fire-and-forget init() RPC settles before teardown.
100 const instance = await binding.get(result.id);
101 await vi.waitUntil(
102 async () => {
103 const s = await instance.status();
104 return s.status === "complete";
105 },
106 { timeout: 5000 },
107 );
108 });
109
110 it("should throw WorkflowError for invalid instance id", async ({
111 expect,
112 }) => {
113 const binding = createBinding();
114 await expect(binding.create({ id: "#invalid!" })).rejects.toThrow(
115 "Workflow instance has invalid id",
116 );
117 });
118
119 it("should block creation when pending persistence deletion fails", async ({
120 expect,
121 }) => {
122 const binding = new WorkflowBinding(createExecutionContext(), {
123 ENGINE: env.ENGINE,
124 BINDING_NAME: "TEST_WORKFLOW",
125 WORKFLOW_NAME: "test-workflow",
126 MINIFLARE_LOOPBACK: {
127 fetch: () => Promise.resolve(new Response(null, { status: 500 })),
128 } as unknown as Fetcher,
129 });
130
131 await expect(binding.create({ id: "cleanup-failed" })).rejects.toThrow(
132 "Failed to wait for persisted workflow instance 'cleanup-failed' deletion",
133 );
134 });
135 });
136
137 describe("get()", () => {
138 it("should return a WorkflowHandle for an existing instance", async ({
139 expect,
140 }) => {
141 const id = uniqueId();
142 const binding = createBinding();
143 env.ENGINE.get(env.ENGINE.idFromName(id));
144
145 setTestWorkflowCallback(async () => "done");
146 await binding.create({ id });
147
148 const instance = await binding.get(id);
149
150 expect(instance).toMatchObject({
151 id,
152 status: expect.any(Function),
153 pause: expect.any(Function),
154 resume: expect.any(Function),
155 terminate: expect.any(Function),
156 restart: expect.any(Function),
157 delete: expect.any(Function),
158 });
159
160 // Wait for the workflow to complete before the test ends so
161 // the fire-and-forget init() RPC settles before teardown.
162 await vi.waitUntil(
163 async () => {
164 const s = await instance.status();
165 return s.status === "complete";
166 },
167 { timeout: 5000 },
168 );
169 });
170 });
171
172 describe("instance deletion", () => {
173 it("deleteInstance should delete an instance and wipe its stored state", async ({
174 expect,
175 }) => {
176 const id = uniqueId();
177 const binding = createBinding();
178
179 setTestWorkflowCallback(async () => "done");
180 await binding.create({ id });
181
182 const instance = await binding.get(id);
183 await vi.waitUntil(
184 async () => {
185 const status = await instance.status();
186 return status.status === "complete";
187 },
188 { timeout: 5000 },
189 );
190
191 await expect(binding.deleteInstance(id)).resolves.toBeUndefined();
192 await expect(binding.get(id)).rejects.toThrow("instance.not_found");
193 });
194
195 it("should reject an invalid instance ID", async ({ expect }) => {
196 await expect(createBinding().deleteInstance("")).rejects.toThrow(
197 "(instance.invalid_id) Instance ID is invalid",
198 );
199 });
200
201 it("should accept a cron-generated instance ID", async ({ expect }) => {
202 await expect(
203 createBinding().deleteInstance("*/30 * * * *-1786001400000"),
204 ).rejects.toThrow("instance.not_found");
205 });
206
207 it("should let a running instance delete itself and stop execution", async ({
208 expect,
209 }) => {
210 const id = uniqueId();
211 const binding = createBinding();
212 let deleteStarted = false;
213 let continuedAfterDelete = false;
214
215 setTestWorkflowCallback(async () => {
216 deleteStarted = true;
217 const instance = await binding.get(id);
218 await (instance as unknown as { delete(): Promise<void> }).delete();
219 continuedAfterDelete = true;
220 });
221 await binding.create({ id });
222 await vi.waitUntil(() => deleteStarted, { timeout: 5000 });
223
224 await vi.waitUntil(
225 async () => {
226 try {
227 await binding.get(id);
228 return false;
229 } catch {
230 return true;
231 }
232 },
233 { timeout: 5000 },
234 );
235
236 await scheduler.wait(50);
237 expect(continuedAfterDelete).toBe(false);
238 await expect(binding.get(id)).rejects.toThrow("instance.not_found");
239 });
240 });
241
242 describe("deleteBatch()", () => {
243 it("should delete instances and wipe their stored state", async ({
244 expect,
245 }) => {
246 const ids = [uniqueId(), uniqueId()];
247 const binding = createBinding();
248
249 setTestWorkflowCallback(async () => "done");
250 await binding.createBatch(ids.map((id) => ({ id })));
251
252 for (const id of ids) {
253 const instance = await binding.get(id);
254 await vi.waitUntil(
255 async () => {
256 const status = await instance.status();
257 return status.status === "complete";
258 },
259 { timeout: 5000 },
260 );
261 }
262
263 await expect(binding.deleteBatch({ instances: ids })).resolves.toEqual({
264 deleted: ids.map((id) => ({ id })),
265 errors: [],
266 });
267 for (const id of ids) {
268 await expect(binding.get(id)).rejects.toThrow("instance.not_found");
269 }
270 });
271
272 it("should report each duplicate missing cron-generated ID", async ({
273 expect,
274 }) => {
275 const binding = createBinding();
276 const cronId = "*/30 * * * *-1786001400000";
277 await expect(
278 binding.deleteBatch({
279 instances: [cronId, cronId],
280 }),
281 ).resolves.toEqual({
282 deleted: [],
283 errors: [
284 {
285 id: cronId,
286 code: 10400,
287 message: "workflows.api.error.instance.not_found",
288 },
289 {
290 id: cronId,
291 code: 10400,
292 message: "workflows.api.error.instance.not_found",
293 },
294 ],
295 });
296 });
297
298 it("should normalize unexpected deletion errors", async ({ expect }) => {
299 const deleteInstance = vi
300 .fn()
301 .mockRejectedValue(new Error("sensitive failure"));
302 const binding = new WorkflowBinding(createExecutionContext(), {
303 ENGINE: {
304 idFromName: (id: string) => id,
305 get: () => ({ deleteInstance }),
306 } as unknown as DurableObjectNamespace<Engine>,
307 BINDING_NAME: "TEST_WORKFLOW",
308 WORKFLOW_NAME: "test-workflow",
309 });
310
311 await expect(
312 binding.deleteBatch({ instances: ["broken-instance"] }),
313 ).resolves.toEqual({
314 deleted: [],
315 errors: [
316 {
317 id: "broken-instance",
318 code: 10001,
319 message: "workflows.api.error.internal_server",
320 },
321 ],
322 });
323 expect(deleteInstance).toHaveBeenCalledOnce();
324 });
325
326 it("should report persistence cleanup failures per instance", async ({
327 expect,
328 }) => {
329 const abort = vi.fn(() =>
330 Promise.reject(new Error("Durable Object aborted")),
331 );
332 const loopbackFetch = vi.fn((url: string) =>
333 Promise.resolve(
334 new Response(null, {
335 status: url.includes("/cleanup-failed") ? 500 : 204,
336 }),
337 ),
338 );
339 const binding = new WorkflowBinding(createExecutionContext(), {
340 ENGINE: {
341 idFromName: (id: string) => ({ toString: () => id }),
342 get: (id: { toString(): string }) => ({
343 id,
344 deleteInstance: () => {
345 if (id.toString() === "missing") {
346 return Promise.reject(
347 new Error("(instance.not_found) Instance does not exist"),
348 );
349 }
350 return Promise.resolve();
351 },
352 unsafeAbort: abort,
353 }),
354 } as unknown as DurableObjectNamespace<Engine>,
355 BINDING_NAME: "TEST_WORKFLOW",
356 WORKFLOW_NAME: "test-workflow",
357 MINIFLARE_LOOPBACK: { fetch: loopbackFetch } as unknown as Fetcher,
358 });
359
360 await expect(
361 binding.deleteBatch({
362 instances: ["cleanup-failed", "missing", "deleted", "cleanup-failed"],
363 }),
364 ).resolves.toEqual({
365 deleted: [{ id: "deleted" }],
366 errors: [
367 {
368 id: "cleanup-failed",
369 code: 10001,
370 message: "workflows.api.error.internal_server",
371 },
372 {
373 id: "missing",
374 code: 10400,
375 message: "workflows.api.error.instance.not_found",
376 },
377 {
378 id: "cleanup-failed",
379 code: 10001,
380 message: "workflows.api.error.internal_server",
381 },
382 ],
383 });
384 expect(abort).toHaveBeenCalledOnce();
385 expect(loopbackFetch.mock.calls.map(([url]) => url)).toEqual(
386 expect.arrayContaining([
387 expect.stringContaining("/missing?defer=1"),
388 expect.stringContaining("/missing?waitForPendingDelete=1"),
389 ]),
390 );
391 });
392
393 it("should reject invalid batches", async ({ expect }) => {
394 const binding = createBinding();
395 await expect(binding.deleteBatch({ instances: [] })).rejects.toThrow(
396 "(body) batchDeleteInstances should have at least 1 instance",
397 );
398 await expect(
399 binding.deleteBatch({
400 instances: Array.from({ length: 101 }, (_, i) => `instance-${i}`),
401 }),
402 ).rejects.toThrow(
403 "(body) batchDeleteInstances only supports 100 instances at a time",
404 );
405 await expect(binding.deleteBatch({ instances: [""] })).rejects.toThrow(
406 "(instance.invalid_id) Instance ID is invalid",
407 );
408 });
409 });
410
411 describe("createBatch()", () => {
412 it("should create multiple instances in a batch", async ({ expect }) => {
413 const binding = createBinding();
414 const ids = ["batch-1", "batch-2", "batch-3"];
415 setTestWorkflowCallback(async () => "done");
416
417 const results = await binding.createBatch(ids.map((id) => ({ id })));
418
419 expect(results).toHaveLength(3);
420 expect(results.map((r) => r.id)).toEqual(ids);
421
422 for (const id of ids) {
423 const instance = await binding.get(id);
424 expect(instance.id).toBe(id);
425 }
426
427 // Wait for all batch workflows to complete before the test ends
428 // so the fire-and-forget init() RPCs settle before teardown.
429 for (const id of ids) {
430 const instance = await binding.get(id);
431 await vi.waitUntil(
432 async () => {
433 const s = await instance.status();
434 return s.status === "complete";
435 },
436 { timeout: 5000 },
437 );
438 }
439 });
440
441 it("should throw error for empty batch", async ({ expect }) => {
442 const binding = createBinding();
443
444 await expect(binding.createBatch([])).rejects.toThrow(
445 "WorkflowError: batchCreate should have at least 1 instance",
446 );
447 });
448 });
449});
450
451describe("WorkflowBinding", () => {
452 it("should not call dispose when sending an event to an instance", async ({
453 expect,
454 }) => {
455 const id = uniqueId();
456 const disposeSpy = vi.fn();
457 const receiveEvent = vi.fn(() =>
458 Object.assign(Promise.resolve(), { [Symbol.dispose]: disposeSpy }),
459 );
460 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
461 Object.defineProperty(engineStub, "receiveEvent", { value: receiveEvent });
462 const instance = new WorkflowHandle(id, engineStub, () => engineStub);
463
464 using _ = (await instance.sendEvent({
465 type: "test-event",
466 payload: { test: "data" },
467 })) as unknown as Disposable;
468
469 expect(receiveEvent).toHaveBeenCalledOnce();
470 expect(disposeSpy).not.toHaveBeenCalled();
471 });
472});
473
474describe("WorkflowHandle", () => {
475 describe("status()", () => {
476 it("should return running status for a workflow waiting for an event", async ({
477 expect,
478 }) => {
479 const id = uniqueId();
480 const binding = createBinding();
481 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
482
483 setTestWorkflowCallback(async (_event, step) => {
484 await step.waitForEvent("wait-for-event", {
485 type: "some-event",
486 timeout: "2 seconds",
487 });
488 return "completed";
489 });
490
491 await binding.create({ id });
492 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
493
494 const instance = await binding.get(id);
495 const status = await instance.status();
496
497 expect(status.status).toBe("running");
498 expect(status.output).toBeNull();
499 expect(status.error).toBeUndefined();
500
501 // Terminate the waiting workflow so the init() RPC settles
502 // before teardown (the 2-second waitForEvent timeout would
503 // otherwise leave the workflow running past test end).
504 await instance.terminate();
505 });
506
507 it("should return complete status and output for a successful workflow", async ({
508 expect,
509 }) => {
510 const id = uniqueId();
511 const binding = createBinding();
512 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
513 const expectedOutput = { result: "success", value: 42 };
514
515 setTestWorkflowCallback(async () => expectedOutput);
516 await binding.create({ id });
517 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
518
519 const instance = await binding.get(id);
520 const status = await instance.status();
521
522 expect(status.status).toBe("complete");
523 expect(status.output).toEqual(expectedOutput);
524 expect(status.error).toBeUndefined();
525 });
526
527 it("should return errored status and error for a failed workflow", async ({
528 expect,
529 }) => {
530 const id = uniqueId();
531 const binding = createBinding();
532 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
533
534 setTestWorkflowCallback(async () => {
535 throw new Error("Workflow failed intentionally");
536 });
537
538 await binding.create({ id });
539 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_FAILURE);
540
541 const instance = await binding.get(id);
542 const status = await instance.status();
543
544 expect(status.status).toBe("errored");
545 expect(status.error).toBeDefined();
546 expect(status.error?.message).toBe("Workflow failed intentionally");
547 expect(status.output).toBeNull();
548 });
549
550 it("should return step outputs in __LOCAL_DEV_STEP_OUTPUTS", async ({
551 expect,
552 }) => {
553 const id = uniqueId();
554 const binding = createBinding();
555 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
556
557 setTestWorkflowCallback(async (_event, step) => {
558 const step1Result = await step.do(
559 "step-1",
560 async () => "result-from-step-1",
561 );
562 const step2Result = await step.do("step-2", async () => ({
563 data: "result-from-step-2",
564 }));
565 const step3Result = await step.do("step-3", async () => 123);
566 return { step1Result, step2Result, step3Result };
567 });
568
569 await binding.create({ id });
570 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
571
572 const instance = (await binding.get(id)) as WorkflowHandle;
573 const status = await instance.status();
574
575 expect(status.status).toBe("complete");
576 expect(status.__LOCAL_DEV_STEP_OUTPUTS).toHaveLength(3);
577 expect(status.__LOCAL_DEV_STEP_OUTPUTS[0]).toBe("result-from-step-1");
578 expect(status.__LOCAL_DEV_STEP_OUTPUTS[1]).toEqual({
579 data: "result-from-step-2",
580 });
581 expect(status.__LOCAL_DEV_STEP_OUTPUTS[2]).toBe(123);
582 });
583
584 it("should return terminated status for a terminated instance", async ({
585 expect,
586 }) => {
587 const id = uniqueId();
588 const binding = createBinding();
589 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
590
591 setTestWorkflowCallback(async (_event, step) => {
592 await step.waitForEvent("wait-for-event", {
593 type: "some-event",
594 timeout: "1 second",
595 });
596 return "completed";
597 });
598
599 await binding.create({ id });
600 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
601
602 const instance = await binding.get(id);
603 await instance.terminate();
604
605 const newInstance = await binding.get(id);
606 const status = await newInstance.status();
607
608 expect(status.status).toBe("terminated");
609 });
610 });
611
612 describe("subscribe()", () => {
613 it("finishes a pending next request when disposed", async ({ expect }) => {
614 const started = Promise.withResolvers<void>();
615 const pendingEvent =
616 Promise.withResolvers<
617 IteratorResult<WorkflowSubscriptionEvent, undefined>
618 >();
619 const onClose = vi.fn();
620 const subscription = new WorkflowSubscriptionTarget(async () => {
621 started.resolve();
622 return pendingEvent.promise;
623 }, onClose);
624
625 const result = subscription.next();
626 await started.promise;
627 subscription[Symbol.dispose]();
628 pendingEvent.resolve({
629 done: false,
630 value: {
631 type: "workflow_queued",
632 instanceId: "disposed-subscription",
633 eventId: 0,
634 timestamp: Date.now(),
635 },
636 });
637
638 expect(await result).toEqual({ done: true, value: undefined });
639 expect(onClose).toHaveBeenCalledOnce();
640 });
641
642 it("closes after returning a terminal event", async ({ expect }) => {
643 const onClose = vi.fn();
644 const terminalEvent = {
645 type: "workflow_completed",
646 instanceId: "completed-subscription",
647 eventId: 1,
648 timestamp: Date.now(),
649 output: "done",
650 } as const satisfies WorkflowSubscriptionEvent;
651 const subscription = new WorkflowSubscriptionTarget(
652 () =>
653 Promise.resolve({
654 done: false as const,
655 value: terminalEvent,
656 }),
657 onClose,
658 );
659
660 expect(await subscription.next()).toEqual({
661 done: false,
662 value: terminalEvent,
663 });
664 expect(onClose).toHaveBeenCalledOnce();
665 expect(await subscription.next()).toEqual({
666 done: true,
667 value: undefined,
668 });
669 });
670
671 it("streams historical events in log order through a disposable RPC target", async ({
672 expect,
673 }) => {
674 const id = uniqueId();
675 const binding = createBinding();
676 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
677
678 setTestWorkflowCallback(async (_event, step) => {
679 await step.do(
680 "subscription step",
681 {
682 retries: {
683 limit: 3,
684 delay: () => "2 seconds" as const,
685 backoff: "linear",
686 },
687 timeout: "30 seconds",
688 },
689 async () => "step output",
690 );
691 return "workflow output";
692 });
693
694 await binding.create({ id, params: { input: true } });
695 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
696
697 const instance = (await binding.get(id)) as WorkflowHandle;
698 using subscription = await instance.subscribe();
699 const events: Array<{
700 eventId: number;
701 type: string;
702 [key: string]: unknown;
703 }> = [];
704 while (true) {
705 const result = await subscription.next();
706 if (result.done) {
707 break;
708 }
709 events.push(result.value);
710 }
711
712 expect(events.map(({ type }) => type)).toEqual([
713 "workflow_queued",
714 "workflow_started",
715 "workflow_running",
716 "step_started",
717 "attempt_started",
718 "attempt_completed",
719 "step_completed",
720 "workflow_completed",
721 ]);
722 expect(events[2]).toMatchObject({
723 type: "workflow_running",
724 });
725 const firstEvent = events[0];
726 if (firstEvent === undefined) {
727 throw new Error("Expected subscription events");
728 }
729 expect(events.map(({ eventId }) => eventId)).toEqual(
730 Array.from(
731 { length: events.length },
732 (_, index) => firstEvent.eventId + index,
733 ),
734 );
735 expect(events).toContainEqual(
736 expect.objectContaining({
737 type: "workflow_started",
738 params: { input: true },
739 }),
740 );
741 expect(events).toContainEqual(
742 expect.objectContaining({
743 type: "step_started",
744 stepName: "subscription step-1",
745 config: {
746 retries: {
747 limit: 3,
748 delay: "[dynamic]",
749 backoff: "linear",
750 },
751 timeout: "30 seconds",
752 },
753 }),
754 );
755 expect(events).toContainEqual(
756 expect.objectContaining({
757 type: "step_completed",
758 stepName: "subscription step-1",
759 output: "step output",
760 }),
761 );
762 expect(events.at(-1)).toMatchObject({
763 type: "workflow_completed",
764 output: "workflow output",
765 });
766 expect(await subscription.next()).toEqual({
767 done: true,
768 value: undefined,
769 });
770 });
771
772 it("uses errored event names for Workflow and step failures", async ({
773 expect,
774 }) => {
775 const id = uniqueId();
776 const binding = createBinding();
777 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
778
779 setTestWorkflowCallback(async (_event, step) => {
780 await step.do(
781 "failing subscription step",
782 { retries: { limit: 0, delay: 0 } },
783 async () => {
784 throw new Error("subscription failure");
785 },
786 );
787 });
788
789 await binding.create({ id });
790 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_FAILURE);
791
792 const instance = (await binding.get(id)) as WorkflowHandle;
793 using subscription = await instance.subscribe();
794 const eventTypes: Array<string> = [];
795 while (true) {
796 const result = await subscription.next();
797 if (result.done) {
798 break;
799 }
800 eventTypes.push(result.value.type);
801 }
802
803 expect(eventTypes).toContain("attempt_errored");
804 expect(eventTypes).toContain("step_errored");
805 expect(eventTypes.at(-1)).toBe("workflow_errored");
806 });
807
808 it("includes the resolved rollback config in rollback step events", async ({
809 expect,
810 }) => {
811 const id = uniqueId();
812 const binding = createBinding();
813 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
814
815 setTestWorkflowCallback(async (_event, step) => {
816 await step.do("rollback subscription step", async () => "step output", {
817 rollback: async () => {},
818 rollbackConfig: {
819 retries: {
820 limit: 0,
821 delay: "1 second",
822 backoff: "constant",
823 },
824 timeout: "30 seconds",
825 },
826 });
827 throw new Error("trigger rollback");
828 });
829
830 await binding.create({ id });
831 await waitUntilLogEvent(engineStub, InstanceEvent.ROLLBACK_COMPLETE);
832
833 const instance = (await binding.get(id)) as WorkflowHandle;
834 using subscription = await instance.subscribe({
835 filter: ["rollback_step_started", "workflow_errored"],
836 });
837 expect(await subscription.next()).toMatchObject({
838 done: false,
839 value: {
840 type: "rollback_step_started",
841 stepName: "rollback subscription step-1",
842 config: {
843 retries: {
844 limit: 0,
845 delay: "1 second",
846 backoff: "constant",
847 },
848 timeout: "30 seconds",
849 },
850 },
851 });
852 });
853
854 it("returns stored structured and streamed step outputs", async ({
855 expect,
856 }) => {
857 const id = uniqueId();
858 const binding = createBinding();
859 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
860
861 setTestWorkflowCallback(async (_event, step) => {
862 await step.do("structured output", async () => ({ total: 1n }));
863 const stream = await step.do("stream output", async () => {
864 return new ReadableStream<Uint8Array>({
865 start(controller) {
866 controller.enqueue(
867 new TextEncoder().encode("streamed step output"),
868 );
869 controller.close();
870 },
871 });
872 });
873 await new Response(stream as ReadableStream<Uint8Array>).arrayBuffer();
874 await step.do(
875 "sensitive output",
876 { sensitive: "output" },
877 async () => "secret",
878 );
879 await step.do("undefined output", async () => undefined);
880 await step.do("null output", async () => null);
881 return "done";
882 });
883
884 await binding.create({ id });
885 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
886
887 const instance = (await binding.get(id)) as WorkflowHandle;
888 using subscription = await instance.subscribe({
889 filter: ["step_completed", "workflow_completed"],
890 });
891
892 const structuredResult = await subscription.next();
893 if (
894 structuredResult.done ||
895 structuredResult.value.type !== "step_completed"
896 ) {
897 throw new Error("Expected a structured step output event");
898 }
899 const structuredEvent = structuredResult.value;
900 expect(structuredEvent).toMatchObject({
901 type: "step_completed",
902 stepName: "structured output-1",
903 output: { total: 1n },
904 });
905
906 const streamResult = await subscription.next();
907 if (streamResult.done || streamResult.value.type !== "step_completed") {
908 throw new Error("Expected a streamed step output event");
909 }
910 const streamEvent = streamResult.value;
911 expect(streamEvent).toMatchObject({
912 type: "step_completed",
913 stepName: "stream output-1",
914 });
915 expect(streamEvent.output).toBeInstanceOf(ReadableStream);
916 expect(
917 await new Response(
918 streamEvent.output as ReadableStream<Uint8Array>,
919 ).text(),
920 ).toBe("streamed step output");
921
922 expect(await subscription.next()).toMatchObject({
923 done: false,
924 value: {
925 type: "step_completed",
926 stepName: "sensitive output-1",
927 output: "[REDACTED]",
928 },
929 });
930 const undefinedResult = await subscription.next();
931 expect(undefinedResult).toMatchObject({
932 done: false,
933 value: {
934 type: "step_completed",
935 stepName: "undefined output-1",
936 },
937 });
938 if (!undefinedResult.done) {
939 expect("output" in undefinedResult.value).toBe(false);
940 }
941 expect(await subscription.next()).toMatchObject({
942 done: false,
943 value: {
944 type: "step_completed",
945 stepName: "null output-1",
946 output: null,
947 },
948 });
949 expect(await subscription.next()).toMatchObject({
950 done: false,
951 value: { type: "workflow_completed", output: "done" },
952 });
953 });
954
955 it("returns full structured workflow outputs while keeping logs JSON-safe", async ({
956 expect,
957 }) => {
958 const id = uniqueId();
959 const binding = createBinding();
960 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
961 const createWorkflowOutput = () => ({
962 count: 42n,
963 typedArray: new Uint16Array([1, 2]),
964 arrayBuffer: new Uint8Array([3, 4, 5]).buffer,
965 completedAt: new Date("2026-09-09T12:00:00.000Z"),
966 values: new Map([["nested", new Set([6n, 7n])]]),
967 });
968
969 setTestWorkflowCallback(async () => createWorkflowOutput());
970 await binding.create({ id });
971 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
972
973 const instance = (await binding.get(id)) as WorkflowHandle;
974 using subscription = await instance.subscribe({
975 filter: ["workflow_completed"],
976 });
977 const result = await subscription.next();
978 if (result.done || result.value.type !== "workflow_completed") {
979 throw new Error("Expected a completed Workflow event");
980 }
981 expect(result.value.output).toEqual(createWorkflowOutput());
982
983 const logs = (await engineStub.readLogs()) as EngineLogs;
984 const completedLog = logs.logs.find(
985 (log) => log.event === InstanceEvent.WORKFLOW_SUCCESS,
986 );
987 expect(completedLog?.metadata.result).toEqual({
988 count: "[BigInt(42)]",
989 typedArray: "[Uint16Array(4 bytes)]",
990 arrayBuffer: "[ArrayBuffer(3 bytes)]",
991 completedAt: "2026-09-09T12:00:00.000Z",
992 values: {},
993 });
994 });
995
996 it("rejects when a stored streamed step output is corrupt", async ({
997 expect,
998 }) => {
999 const id = uniqueId();
1000 const binding = createBinding();
1001 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1002
1003 setTestWorkflowCallback(async (_event, step) => {
1004 const stream = await step.do("stream output", async () => {
1005 return new ReadableStream<Uint8Array>({
1006 start(controller) {
1007 controller.enqueue(new TextEncoder().encode("streamed output"));
1008 controller.close();
1009 },
1010 });
1011 });
1012 await new Response(stream as ReadableStream<Uint8Array>).arrayBuffer();
1013 });
1014
1015 await binding.create({ id });
1016 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
1017 await runInDurableObject(engineStub, (_engine, state) => {
1018 const step = state.storage.sql
1019 .exec<{ groupKey: string }>(
1020 "SELECT groupKey FROM states WHERE event = ? LIMIT 1",
1021 InstanceEvent.STEP_SUCCESS,
1022 )
1023 .one();
1024 if (step === null) {
1025 throw new Error("Expected a completed step");
1026 }
1027 state.storage.sql.exec(
1028 "DELETE FROM streaming_step_chunks WHERE cache_key = ?",
1029 step.groupKey,
1030 );
1031 });
1032
1033 await runInDurableObject(engineStub, async (engine) => {
1034 const subscription = await engine.subscribe({
1035 filter: ["step_completed"],
1036 });
1037 await expect(subscription.next()).rejects.toThrow(
1038 "Step has completed but its stored stream output is corrupt or incomplete",
1039 );
1040 expect(await subscription.next()).toEqual({
1041 done: true,
1042 value: undefined,
1043 });
1044 });
1045 });
1046
1047 it("returns done when the cursor is at or after a terminal event", async ({
1048 expect,
1049 }) => {
1050 const id = uniqueId();
1051 const binding = createBinding();
1052 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1053
1054 setTestWorkflowCallback(async () => "done");
1055 await binding.create({ id });
1056 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
1057
1058 const getEventId = async (event: InstanceEvent) => {
1059 const logs = (await engineStub.readDetailedLogs()) as Array<{
1060 id: number;
1061 event: InstanceEvent;
1062 }>;
1063 const eventId = logs.find((log) => log.event === event)?.id;
1064 if (eventId === undefined) {
1065 throw new Error(`Expected Workflow event ${event}`);
1066 }
1067 return eventId;
1068 };
1069 const expectDoneAtCursor = async (cursor: number) => {
1070 const instance = (await binding.get(id)) as WorkflowHandle;
1071 using subscription = await instance.subscribe({ cursor });
1072 const result = Promise.resolve(subscription.next());
1073 await expect(
1074 Promise.race([
1075 result,
1076 scheduler.wait(1000).then(() => "timed out" as const),
1077 ]),
1078 ).resolves.toEqual({ done: true, value: undefined });
1079 };
1080
1081 const terminalEventId = await getEventId(InstanceEvent.WORKFLOW_SUCCESS);
1082 await expectDoneAtCursor(terminalEventId);
1083
1084 await runInDurableObject(engineStub, (engine) => {
1085 engine.writeLog(InstanceEvent.__INTERNAL_PROD, null, null, {});
1086 });
1087 const laterInternalEventId = await getEventId(
1088 InstanceEvent.__INTERNAL_PROD,
1089 );
1090 await expectDoneAtCursor(laterInternalEventId);
1091 });
1092
1093 it("waits for live events and applies cursor and filter options", async ({
1094 expect,
1095 }) => {
1096 const id = uniqueId();
1097 const binding = createBinding();
1098 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1099
1100 setTestWorkflowCallback(async (_event, step) => {
1101 await step.waitForEvent("subscription wait", {
1102 type: "continue",
1103 timeout: "10 seconds",
1104 });
1105 return "done";
1106 });
1107
1108 await binding.create({ id });
1109 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1110
1111 const logs = (await engineStub.readDetailedLogs()) as Array<{
1112 id: number;
1113 event: InstanceEvent;
1114 }>;
1115 const startEventId = logs.find(
1116 ({ event }) => event === InstanceEvent.WORKFLOW_START,
1117 )?.id;
1118 if (startEventId === undefined) {
1119 throw new Error("Expected a workflow start event");
1120 }
1121
1122 const instance = (await binding.get(id)) as WorkflowHandle;
1123 using subscription = await instance.subscribe({
1124 cursor: startEventId,
1125 filter: ["workflow_queued", "wait_started", "workflow_completed"],
1126 });
1127 expect(await subscription.next()).toMatchObject({
1128 done: false,
1129 value: {
1130 type: "wait_started",
1131 stepName: "subscription wait-1",
1132 eventType: "continue",
1133 },
1134 });
1135
1136 const terminalEvent = subscription.next();
1137 await instance.sendEvent({ type: "continue", payload: null });
1138 expect(await terminalEvent).toMatchObject({
1139 done: false,
1140 value: { type: "workflow_completed", output: "done" },
1141 });
1142 expect(await subscription.next()).toEqual({
1143 done: true,
1144 value: undefined,
1145 });
1146 });
1147
1148 it("streams explicit Workflow status events", async ({ expect }) => {
1149 const id = uniqueId();
1150 const binding = createBinding();
1151 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1152
1153 setTestWorkflowCallback(async (_event, step) => {
1154 await step.waitForEvent("status subscription wait", {
1155 type: "continue",
1156 timeout: "10 seconds",
1157 });
1158 return "done";
1159 });
1160
1161 await binding.create({ id });
1162 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1163
1164 const instance = (await binding.get(id)) as WorkflowHandle;
1165 using subscription = await instance.subscribe({
1166 filter: [
1167 "workflow_queued",
1168 "workflow_running",
1169 "workflow_waiting",
1170 "workflow_waiting_for_pause",
1171 "workflow_paused",
1172 "workflow_completed",
1173 ],
1174 });
1175
1176 expect(await subscription.next()).toMatchObject({
1177 done: false,
1178 value: { type: "workflow_queued" },
1179 });
1180 expect(await subscription.next()).toMatchObject({
1181 done: false,
1182 value: { type: "workflow_running" },
1183 });
1184
1185 await engineStub.setStatus(0, id, InstanceStatus.Waiting);
1186 expect(await subscription.next()).toMatchObject({
1187 done: false,
1188 value: { type: "workflow_waiting" },
1189 });
1190
1191 await engineStub.setStatus(0, id, InstanceStatus.Queued);
1192 expect(await subscription.next()).toMatchObject({
1193 done: false,
1194 value: { type: "workflow_queued" },
1195 });
1196 await engineStub.setStatus(0, id, InstanceStatus.Running);
1197 expect(await subscription.next()).toMatchObject({
1198 done: false,
1199 value: { type: "workflow_running" },
1200 });
1201
1202 await engineStub.setStatus(0, id, InstanceStatus.WaitingForPause);
1203 expect(await subscription.next()).toMatchObject({
1204 done: false,
1205 value: { type: "workflow_waiting_for_pause" },
1206 });
1207 await engineStub.setStatus(0, id, InstanceStatus.Paused);
1208 expect(await subscription.next()).toMatchObject({
1209 done: false,
1210 value: { type: "workflow_paused" },
1211 });
1212
1213 await engineStub.setStatus(0, id, InstanceStatus.Running);
1214 expect(await subscription.next()).toMatchObject({
1215 done: false,
1216 value: { type: "workflow_running" },
1217 });
1218 const terminalEvent = subscription.next();
1219 await instance.sendEvent({ type: "continue", payload: null });
1220 expect(await terminalEvent).toMatchObject({
1221 done: false,
1222 value: { type: "workflow_completed", output: "done" },
1223 });
1224 expect(await subscription.next()).toEqual({
1225 done: true,
1226 value: undefined,
1227 });
1228
1229 const explicitStatusEventTypes = new Set<InstanceEvent>([
1230 InstanceEvent.WORKFLOW_QUEUED,
1231 InstanceEvent.WORKFLOW_RUNNING,
1232 InstanceEvent.WORKFLOW_PAUSED,
1233 InstanceEvent.WORKFLOW_WAITING_FOR_PAUSE,
1234 InstanceEvent.WORKFLOW_WAITING,
1235 ]);
1236 const statusEvents = (
1237 (await engineStub.readDetailedLogs()) as Array<{
1238 event: InstanceEvent;
1239 }>
1240 )
1241 .filter(({ event }) => explicitStatusEventTypes.has(event))
1242 .map(({ event }) => event);
1243 expect(statusEvents).toEqual([
1244 InstanceEvent.WORKFLOW_QUEUED,
1245 InstanceEvent.WORKFLOW_RUNNING,
1246 InstanceEvent.WORKFLOW_WAITING,
1247 InstanceEvent.WORKFLOW_QUEUED,
1248 InstanceEvent.WORKFLOW_RUNNING,
1249 InstanceEvent.WORKFLOW_WAITING_FOR_PAUSE,
1250 InstanceEvent.WORKFLOW_PAUSED,
1251 InstanceEvent.WORKFLOW_RUNNING,
1252 ]);
1253 });
1254
1255 it("rejects invalid subscription options", async ({ expect }) => {
1256 const id = uniqueId();
1257 const binding = createBinding();
1258 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1259
1260 setTestWorkflowCallback(async () => undefined);
1261 await binding.create({ id });
1262 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
1263
1264 const instance = await binding.get(id);
1265 const unsafeInstance = instance as unknown as {
1266 subscribe(options: unknown): Promise<WorkflowSubscription>;
1267 };
1268 const invalidOptions: Array<[description: string, options: unknown]> = [
1269 ["null", null],
1270 ["an array", []],
1271 ["a negative cursor", { cursor: -1 }],
1272 ["a fractional cursor", { cursor: 1.5 }],
1273 ["an unsafe cursor", { cursor: Number.MAX_SAFE_INTEGER + 1 }],
1274 ["a non-array filter", { filter: "workflow_started" }],
1275 ["a non-string event filter", { filter: [42] }],
1276 ["an unknown event filter", { filter: ["does_not_exist"] }],
1277 ["the internal event filter", { filter: ["internal"] }],
1278 ["an unknown option", { unexpected: true }],
1279 ];
1280
1281 for (const [description, options] of invalidOptions) {
1282 await expect(
1283 unsafeInstance.subscribe(options),
1284 description,
1285 ).rejects.toThrow("Invalid Workflow subscription options");
1286 }
1287 });
1288
1289 it("streams a persisted termination event after the Engine aborts", async ({
1290 expect,
1291 }) => {
1292 const id = uniqueId();
1293 const binding = createBinding();
1294 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1295
1296 setTestWorkflowCallback(async (_event, step) => {
1297 await step.waitForEvent("termination wait", {
1298 type: "never",
1299 timeout: "10 seconds",
1300 });
1301 });
1302
1303 await binding.create({ id });
1304 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1305 const instance = await binding.get(id);
1306 await instance.terminate();
1307
1308 const terminatedInstance = (await binding.get(id)) as WorkflowHandle;
1309 using subscription = await terminatedInstance.subscribe({
1310 filter: ["workflow_terminated"],
1311 });
1312
1313 expect(await subscription.next()).toMatchObject({
1314 done: false,
1315 value: { type: "workflow_terminated" },
1316 });
1317 });
1318 });
1319
1320 describe("sendEvent()", () => {
1321 it("should deliver event payload to a waiting workflow", async ({
1322 expect,
1323 }) => {
1324 const id = uniqueId();
1325 const binding = createBinding();
1326 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1327
1328 setTestWorkflowCallback(async (_event, step) => {
1329 const receivedEvent = await step.waitForEvent("wait-for-event", {
1330 type: "my-event-type",
1331 timeout: "2 seconds",
1332 });
1333 return receivedEvent;
1334 });
1335
1336 await binding.create({ id });
1337 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1338
1339 const instance = await binding.get(id);
1340 const eventPayload = { message: "hello", count: 42 };
1341
1342 await instance.sendEvent({
1343 type: "my-event-type",
1344 payload: eventPayload,
1345 });
1346
1347 await vi.waitUntil(
1348 async () => {
1349 const status = await instance.status();
1350 return status.status === "complete";
1351 },
1352 { timeout: 5000 },
1353 );
1354
1355 const status = await instance.status();
1356 expect(status.output).toMatchObject({
1357 payload: eventPayload,
1358 type: "my-event-type",
1359 });
1360 });
1361
1362 it("should handle multiple sequential events", async ({ expect }) => {
1363 const id = uniqueId();
1364 const binding = createBinding();
1365 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1366
1367 setTestWorkflowCallback(async (_event, step) => {
1368 const event1 = await step.waitForEvent("wait-1", {
1369 type: "event-type-1",
1370 timeout: "10 seconds",
1371 });
1372 const event2 = await step.waitForEvent("wait-2", {
1373 type: "event-type-2",
1374 timeout: "10 seconds",
1375 });
1376 return { first: event1.payload, second: event2.payload };
1377 });
1378
1379 await binding.create({ id });
1380 const instance = await binding.get(id);
1381
1382 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1383 await instance.sendEvent({
1384 type: "event-type-1",
1385 payload: { value: "first" },
1386 });
1387
1388 // Wait for the second waitForEvent
1389 await vi.waitUntil(
1390 async () => {
1391 const logs = (await engineStub.readLogs()) as EngineLogs;
1392 const waitStarts = logs.logs.filter(
1393 (log) => log.event === InstanceEvent.WAIT_START,
1394 );
1395 return waitStarts.length === 2;
1396 },
1397 { timeout: 5000 },
1398 );
1399
1400 await instance.sendEvent({
1401 type: "event-type-2",
1402 payload: { value: "second" },
1403 });
1404
1405 await vi.waitUntil(
1406 async () => {
1407 const status = await instance.status();
1408 return status.status === "complete";
1409 },
1410 { timeout: 5000 },
1411 );
1412
1413 const status = await instance.status();
1414 expect(status.output).toEqual({
1415 first: { value: "first" },
1416 second: { value: "second" },
1417 });
1418 });
1419 });
1420
1421 describe("terminate()", () => {
1422 it("should terminate a running workflow instance", async ({ expect }) => {
1423 const id = uniqueId();
1424 const binding = createBinding();
1425 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1426
1427 setTestWorkflowCallback(async (_event, step) => {
1428 await step.waitForEvent("wait-for-event", {
1429 type: "some-event",
1430 timeout: "1 second",
1431 });
1432 await step.do("should not be called", async () => {
1433 return "should not be called";
1434 });
1435 return "should never complete";
1436 });
1437
1438 await binding.create({ id });
1439 await waitUntilLogEvent(engineStub, InstanceEvent.WAIT_START);
1440
1441 const instance = await binding.get(id);
1442 await instance.terminate();
1443
1444 // Get a new stub since the engine was aborted
1445 const newEngineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1446
1447 const status = await runInDurableObject(newEngineStub, (engine) => {
1448 return engine.getStatus();
1449 });
1450 expect(status).toBe(InstanceStatus.Terminated);
1451
1452 const logs = (await newEngineStub.readLogs()) as EngineLogs;
1453 const hasTerminatedEvent = logs.logs.some(
1454 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
1455 );
1456 expect(hasTerminatedEvent).toBe(true);
1457
1458 // assert that step.do never started
1459 const hasStepStart = logs.logs.some(
1460 (log) => log.event === InstanceEvent.STEP_START,
1461 );
1462 expect(hasStepStart).toBe(false);
1463 });
1464 });
1465
1466 describe("restart()", () => {
1467 it("should restart a workflow instance", async ({ expect }) => {
1468 const id = uniqueId();
1469 const binding = createBinding();
1470 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1471
1472 setTestWorkflowCallback(async (_event, step) => {
1473 await step.sleep("sleep", 250);
1474 return "complete";
1475 });
1476
1477 await binding.create({ id });
1478 await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS);
1479
1480 let instance = await binding.get(id);
1481 let status = await instance.status();
1482 expect(status.status).toBe("complete");
1483
1484 // restart() aborts the old DO, gets a fresh stub, and calls attemptRestart()
1485 // The service binding (USER_WORKFLOW) survives the abort, so no re-setup needed
1486 await instance.restart();
1487
1488 const statusAfterRestart = await instance.status();
1489 expect(statusAfterRestart.status).toBe("running");
1490
1491 // Wait for the restarted workflow to complete via status polling
1492 await vi.waitUntil(
1493 async () => {
1494 const s = await instance.status();
1495 return s.status === "complete";
1496 },
1497 { timeout: 5000 },
1498 );
1499
1500 // Verify second run completed
1501 instance = await binding.get(id);
1502 status = await instance.status();
1503 expect(status.status).toBe("complete");
1504 });
1505 });
1506
1507 describe("pause()", () => {
1508 it("should pause a running workflow", async ({ expect }) => {
1509 const id = uniqueId();
1510 const binding = createBinding();
1511 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1512
1513 setTestWorkflowCallback(async (_event, step) => {
1514 await step.do("long-step", async () => {
1515 await scheduler.wait(500);
1516 return "result-1";
1517 });
1518 // step-2 should never run because pause takes effect after long-step
1519 await step.do("step-2", async () => "result-2");
1520 return "done";
1521 });
1522
1523 await binding.create({ id });
1524 await waitUntilLogEvent(engineStub, InstanceEvent.STEP_START);
1525
1526 const instance = await binding.get(id);
1527
1528 // Pause while long-step is in flight
1529 await instance.pause();
1530
1531 await vi.waitUntil(
1532 async () => {
1533 const s = await instance.status();
1534 return s.status === "paused";
1535 },
1536 { timeout: 5000 },
1537 );
1538
1539 const finalStatus = await instance.status();
1540 expect(finalStatus.status).toBe("paused");
1541 });
1542 });
1543
1544 describe("resume()", () => {
1545 it("should resume a paused workflow and complete it", async ({
1546 expect,
1547 }) => {
1548 const id = uniqueId();
1549 const binding = createBinding();
1550 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1551
1552 setTestWorkflowCallback(async (_event, step) => {
1553 await step.do("long-step", async () => {
1554 await scheduler.wait(500);
1555 return "result-1";
1556 });
1557 await step.do("step-2", async () => "result-2");
1558 return "all-done";
1559 });
1560
1561 await binding.create({ id });
1562 await waitUntilLogEvent(engineStub, InstanceEvent.STEP_START);
1563
1564 const instance = await binding.get(id);
1565
1566 // Pause while long-step is in flight
1567 await instance.pause();
1568
1569 await vi.waitUntil(
1570 async () => {
1571 const s = await instance.status();
1572 return s.status === "paused";
1573 },
1574 { timeout: 5000 },
1575 );
1576
1577 await instance.resume();
1578
1579 await vi.waitUntil(
1580 async () => {
1581 const s = await instance.status();
1582 return s.status === "complete";
1583 },
1584 { timeout: 5000 },
1585 );
1586
1587 const finalStatus = await instance.status();
1588 expect(finalStatus.status).toBe("complete");
1589 expect(finalStatus.output).toBe("all-done");
1590 });
1591
1592 it("should cancel a pending pause when resume is called before step finishes", async ({
1593 expect,
1594 }) => {
1595 const id = uniqueId();
1596 const binding = createBinding();
1597 const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id));
1598
1599 setTestWorkflowCallback(async (_event, step) => {
1600 await step.do("long-step", async () => {
1601 await scheduler.wait(1000);
1602 return "long-result";
1603 });
1604 await step.do("step-after", async () => "final-result");
1605 return "completed";
1606 });
1607
1608 await binding.create({ id });
1609 await waitUntilLogEvent(engineStub, InstanceEvent.STEP_START);
1610
1611 const instance = await binding.get(id);
1612
1613 // Pause while long-step is in flight — sets WaitingForPause
1614 await instance.pause();
1615
1616 const statusAfterPause = await instance.status();
1617 expect(statusAfterPause.status).toBe("waitingForPause");
1618
1619 // resume before the step finishes — this should cancel the pending pause
1620 await instance.resume();
1621
1622 // status should go back to Running
1623 const statusAfterResume = await instance.status();
1624 expect(statusAfterResume.status).toBe("running");
1625
1626 await vi.waitUntil(
1627 async () => {
1628 const s = await instance.status();
1629 return s.status === "complete";
1630 },
1631 { timeout: 5000 },
1632 );
1633
1634 const finalStatus = await instance.status();
1635 expect(finalStatus.status).toBe("complete");
1636 expect(finalStatus.output).toBe("completed");
1637 });
1638 });
1639});