diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 348a3f5eb7..5f1c5ab994 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -1004,7 +1004,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. @@ -1796,7 +1804,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. @@ -2682,7 +2698,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. @@ -4570,6 +4594,10 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a \`\`datetime\`\` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -4615,6 +4643,10 @@ def workflow(func) # On replay the cached value is returned without re-executing \`\`fn\`\`. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# \`\`fn\`\`'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a \`\`datetime\`\` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. @@ -6508,8 +6540,12 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * * Inside a \`workflow()\`, calling a task dispatches it as a step. * Outside a workflow, the function body executes directly. + * + * A task runs as its own job, so its result is always encoded as JSON and + * decoded back before the caller sees it: a \`Date\` comes back as a string, a + * \`Map\` as \`{}\`. {@link JsonifiedFn} is that shape. */ -export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): T +export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): JsonifiedFn /** * Create a task that dispatches to a separate Windmill script. @@ -6540,7 +6576,15 @@ export function taskFlow(path: string, options?: TaskOptions): (...args: any[]) */ export function workflow(fn: (...args: any[]) => Promise) -export async function step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +export async function step(name: string, fn: () => T | Promise,): Promise>> export async function sleep(seconds: number): Promise @@ -6630,6 +6674,10 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a \`\`datetime\`\` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -6675,6 +6723,10 @@ def workflow(func) # On replay the cached value is returned without re-executing \`\`fn\`\`. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# \`\`fn\`\`'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a \`\`datetime\`\` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/python-client/wmill/tests/test_workflow.py b/python-client/wmill/tests/test_workflow.py index 54abff21ca..1cca524237 100644 --- a/python-client/wmill/tests/test_workflow.py +++ b/python-client/wmill/tests/test_workflow.py @@ -1,11 +1,47 @@ """Tests for the Workflow-as-Code SDK.""" import asyncio +import json import pytest +from datetime import datetime, timezone + from wmill.client import WorkflowCtx, _StepSuspend, TaskError, workflow, task, step, sleep, parallel, wait_for_approval, _run_workflow, _run_workflow_async +class _StubInlineClient: + """Stands in for the httpx client the inline fast path POSTs with. + + Decodes each request body, so a test sees exactly the JSON that reaches + ``/jobs/wac/inline_checkpoint`` — and therefore what a replay reads back. + """ + + def __init__(self): + self.posted = [] + + async def post(self, url, content=None): + self.posted.append(json.loads(content)) + + class _Response: + def raise_for_status(self): + pass + + return _Response() + + async def aclose(self): + pass + + +def _set_inline_fast_path_env(monkeypatch): + for var, val in ( + ("WM_JOB_ID", "job-1"), + ("WM_WORKSPACE", "admins"), + ("BASE_INTERNAL_URL", "http://localhost:8000"), + ("WM_TOKEN", "tok"), + ): + monkeypatch.setenv(var, val) + + @task async def extract_data(url: str): pass # body unused in workflow context @@ -974,31 +1010,14 @@ class TestRaisingInlineStepIsCheckpointed: the same ``TaskError`` a replay rebuilds from the marker — raising the original ``ValueError`` here would make ``except ValueError:`` catch on this run and miss on the next one.""" - for var, val in ( - ("WM_JOB_ID", "job-1"), - ("WM_WORKSPACE", "admins"), - ("BASE_INTERNAL_URL", "http://localhost:8000"), - ("WM_TOKEN", "tok"), - ): - monkeypatch.setenv(var, val) + _set_inline_fast_path_env(monkeypatch) - posted = [] - - class _StubResponse: - def raise_for_status(self): - pass - - class _StubClient: - async def post(self, url, json=None): - posted.append(json) - return _StubResponse() - - async def aclose(self): - pass + stub = _StubInlineClient() + posted = stub.posted async def run(): ctx = WorkflowCtx({}) - ctx._inline_http_client = _StubClient() + ctx._inline_http_client = stub with pytest.raises(TaskError, match="boom") as live: await ctx._run_inline_step("risky", self._boom) # ...and the replay of that very checkpoint raises the same thing. @@ -1031,6 +1050,52 @@ class TestRaisingInlineStepIsCheckpointed: assert r["result"] == 10 +class TestInlineStepRoundParity: + """The round that runs a ``step()`` body must see what a replay sees. + + The fast path returns the value it checkpointed, not the in-memory one: + a workflow branching on a datetime attribute or a tuple would otherwise + take one path on the round that ran the body and another on every replay, + which can change which tasks get dispatched, not just crash later. + """ + + CASES = [ + ("dt", lambda: datetime(2026, 1, 1, tzinfo=timezone.utc), "2026-01-01 00:00:00+00:00"), + ("pair", lambda: (1, 2), [1, 2]), + ("intkeys", lambda: {1: "a"}, {"1": "a"}), + ] + + def test_outside_a_workflow_returns_the_same_shape(self): + """No checkpoint, no replay — but a local run must not hand back a shape + a deployed one never produces, or testing a workflow locally proves + nothing. The async task path is the sharp edge: the wrapper is sync, so + the value has to be round-tripped after the await, not before.""" + + @task + async def make_pair(): + return (1, datetime(2026, 1, 1, tzinfo=timezone.utc)) + + assert asyncio.run(step("pair", lambda: (1, 2))) == [1, 2] + assert asyncio.run(make_pair()) == [1, "2026-01-01 00:00:00+00:00"] + + def test_live_round_matches_checkpoint_and_replay(self, monkeypatch): + _set_inline_fast_path_env(monkeypatch) + + async def run(): + for key, fn, expected in self.CASES: + stub = _StubInlineClient() + ctx = WorkflowCtx({}) + ctx._inline_http_client = stub + live = await ctx._run_inline_step(key, fn) + checkpointed = stub.posted[0]["result"] + assert checkpointed == expected + assert live == expected and type(live) is type(expected) + replayed = WorkflowCtx({"completed_steps": {key: checkpointed}}) + assert await replayed._run_inline_step(key, fn) == live + + asyncio.run(run()) + + # ===================================================================== # TASK OPTIONS TESTS # ===================================================================== diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index c68431ae80..682a2c4c9c 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -2702,6 +2702,13 @@ class TaskError(Exception): self.result = result +def _json_round_trip(value): + """Put a value through the checkpoint's encoding without checkpointing it, so + the paths that never persist anything still hand back the shape the ones that + do would. ``default=str`` matches the worker wrapper's encoder.""" + return json.loads(json.dumps(value, default=str)) + + def _step_error_marker(key: str, exc: BaseException) -> dict: """Serialize a failed ``step()`` body into the ``__wmill_error`` marker that task failures also use, so it can be stored in ``completed_steps``.""" @@ -2936,7 +2943,22 @@ class WorkflowCtx: _token = os.environ.get("WM_TOKEN") if _fast_path_enabled and _job_id and _workspace and _base and _token: _fast_path_ok = False + _replay_result = None try: + # ``default=str`` is the encoder the worker wrapper uses on the + # suspend path, so both arms checkpoint the same value — and a + # datetime or set takes the fast path instead of silently + # degrading to a suspend round. + _payload = _json_mod.dumps( + { + "key": key, + "result": result, + "started_at": started_at, + "duration_ms": duration_ms, + }, + default=str, + ) + _replay_result = _json_mod.loads(_payload)["result"] if self._inline_lock is None: self._inline_lock = _asyncio.Lock() # Lock wraps only the POST, not fn() above — concurrent @@ -2953,12 +2975,7 @@ class WorkflowCtx: ) _resp = await self._inline_http_client.post( f"{_base}/api/w/{_workspace}/jobs/wac/inline_checkpoint/{_job_id}", - json={ - "key": key, - "result": result, - "started_at": started_at, - "duration_ms": duration_ms, - }, + content=_payload, ) _resp.raise_for_status() _fast_path_ok = True @@ -2976,7 +2993,11 @@ class WorkflowCtx: # run and miss on the next. ``__cause__`` is for tracebacks only. if step_error is not None: raise _step_error_from_marker(result, name) from step_error - return result + # Return the round trip of what was checkpointed, never the + # in-memory value: handing back the live object would let the + # round that ran the body branch on a type — tuple, datetime — + # that no replay of it ever sees. + return _replay_result raise _StepSuspend({ "mode": "inline_checkpoint", @@ -3009,6 +3030,10 @@ def task( - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. - **Standalone**: executes the function body directly. + A task runs as its own job, so its result is always encoded as JSON and + decoded back before the caller sees it: a ``datetime`` comes back as a + string, a tuple as a list. + Usage:: @task @@ -3081,8 +3106,19 @@ def task( print(f"Task {func.__name__} ({child_job_id}) completed") return job_result - # Standalone — execute directly - return func(*args, **kwargs) + # Standalone — execute directly, but round-trip the result: a task's + # value crosses JSON in every other path, so a local run must agree. + # This wrapper is sync, so an ``async def`` task hands back a + # coroutine here — round-tripping that would serialize the coroutine + # object itself. + result = func(*args, **kwargs) + if _asyncio.iscoroutine(result): + + async def _round_trip_awaited(): + return _json_round_trip(await result) + + return _round_trip_awaited() + return _json_round_trip(result) wrapper._is_task = True wrapper._task_path = task_path @@ -3186,6 +3222,10 @@ async def step(name: str, fn): On replay the cached value is returned without re-executing ``fn``. Use for lightweight deterministic operations (timestamps, random IDs, config reads) that should not incur the overhead of a child job. + + ``fn``'s result is encoded as JSON and decoded back before it is returned, + so the round that runs the body sees the same types every replay sees: + a ``datetime`` comes back as a string, a tuple as a list. """ ctx: WorkflowCtx | None = _workflow_ctx.get(None) if ctx is not None: @@ -3193,7 +3233,9 @@ async def step(name: str, fn): result = fn() if _asyncio.iscoroutine(result): result = await result - return result + # Outside a workflow nothing is checkpointed, but round-trip anyway: running + # the script locally must not hand back a shape a deployed run never sees. + return _json_round_trip(result) async def sleep(seconds: int): diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index fb525e7b89..3d0124b2a8 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -1576,7 +1576,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. @@ -2370,6 +2378,10 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a \`\`datetime\`\` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -2415,6 +2427,10 @@ def workflow(func) # On replay the cached value is returned without re-executing \`\`fn\`\`. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# \`\`fn\`\`'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a \`\`datetime\`\` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. @@ -2502,8 +2518,12 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * * Inside a \`workflow()\`, calling a task dispatches it as a step. * Outside a workflow, the function body executes directly. + * + * A task runs as its own job, so its result is always encoded as JSON and + * decoded back before the caller sees it: a \`Date\` comes back as a string, a + * \`Map\` as \`{}\`. {@link JsonifiedFn} is that shape. */ -export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): T +export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): JsonifiedFn /** * Create a task that dispatches to a separate Windmill script. @@ -2534,7 +2554,15 @@ export function taskFlow(path: string, options?: TaskOptions): (...args: any[]) */ export function workflow(fn: (...args: any[]) => Promise) -export async function step(name: string, fn: () => T | Promise): Promise +/** + * Execute \`fn\` inline and checkpoint the result. On replay the cached value is + * returned without re-executing \`fn\`. + * + * \`fn\`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a \`Date\` + * comes back as a string, a \`Map\` as \`{}\`. {@link Jsonified} is that shape. + */ +export async function step(name: string, fn: () => T | Promise,): Promise>> export async function sleep(seconds: number): Promise @@ -2624,6 +2652,10 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a \`\`datetime\`\` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -2669,6 +2701,10 @@ def workflow(func) # On replay the cached value is returned without re-executing \`\`fn\`\`. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# \`\`fn\`\`'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a \`\`datetime\`\` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index 751b10e3a8..a70b5d08b4 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -1894,7 +1894,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. @@ -2688,6 +2696,10 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a ``datetime`` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -2733,6 +2745,10 @@ def workflow(func) # On replay the cached value is returned without re-executing ``fn``. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# ``fn``'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a ``datetime`` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index bef29c4403..ae13ae91a9 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -631,6 +631,10 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a ``datetime`` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -676,6 +680,10 @@ def workflow(func) # On replay the cached value is returned without re-executing ``fn``. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# ``fn``'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a ``datetime`` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/system_prompts/auto-generated/sdks/typescript.md b/system_prompts/auto-generated/sdks/typescript.md index dd98da4ea9..729e2ce108 100644 --- a/system_prompts/auto-generated/sdks/typescript.md +++ b/system_prompts/auto-generated/sdks/typescript.md @@ -459,7 +459,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index 3ef7d752e5..6d0c34deb4 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -33,6 +33,10 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a ``datetime`` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -78,6 +82,10 @@ def workflow(func) # On replay the cached value is returned without re-executing ``fn``. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# ``fn``'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a ``datetime`` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/system_prompts/auto-generated/sdks/wac-typescript.md b/system_prompts/auto-generated/sdks/wac-typescript.md index 3c3405611f..f8a2452942 100644 --- a/system_prompts/auto-generated/sdks/wac-typescript.md +++ b/system_prompts/auto-generated/sdks/wac-typescript.md @@ -31,8 +31,12 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * * Inside a `workflow()`, calling a task dispatches it as a step. * Outside a workflow, the function body executes directly. + * + * A task runs as its own job, so its result is always encoded as JSON and + * decoded back before the caller sees it: a `Date` comes back as a string, a + * `Map` as `{}`. {@link JsonifiedFn} is that shape. */ -export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): T +export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): JsonifiedFn /** * Create a task that dispatches to a separate Windmill script. @@ -63,7 +67,15 @@ export function taskFlow(path: string, options?: TaskOptions): (...args: any[]) */ export function workflow(fn: (...args: any[]) => Promise) -export async function step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +export async function step(name: string, fn: () => T | Promise,): Promise>> export async function sleep(seconds: number): Promise diff --git a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md index 015bd52be4..d7a757085a 100644 --- a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md @@ -630,7 +630,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. diff --git a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md index 2e6f37bd62..a5a7c02dab 100644 --- a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md @@ -630,7 +630,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. diff --git a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md index 2cfdf4e980..ea6ebe00aa 100644 --- a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md @@ -632,7 +632,15 @@ setWorkflowCtx(ctx: WorkflowCtx | null): void async sleep(seconds: number): Promise -async step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +async step(name: string, fn: () => T | Promise,): Promise>> /** * Create a task that dispatches to a separate Windmill script. diff --git a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md index 38cf7d33e6..b2b03b4f98 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -816,6 +816,10 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a ``datetime`` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -861,6 +865,10 @@ def workflow(func) # On replay the cached value is returned without re-executing ``fn``. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# ``fn``'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a ``datetime`` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md index 4f2aa46290..159b3636e3 100644 --- a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md +++ b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md @@ -272,8 +272,12 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * * Inside a `workflow()`, calling a task dispatches it as a step. * Outside a workflow, the function body executes directly. + * + * A task runs as its own job, so its result is always encoded as JSON and + * decoded back before the caller sees it: a `Date` comes back as a string, a + * `Map` as `{}`. {@link JsonifiedFn} is that shape. */ -export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): T +export function task Promise>(fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions,): JsonifiedFn /** * Create a task that dispatches to a separate Windmill script. @@ -304,7 +308,15 @@ export function taskFlow(path: string, options?: TaskOptions): (...args: any[]) */ export function workflow(fn: (...args: any[]) => Promise) -export async function step(name: string, fn: () => T | Promise): Promise +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +export async function step(name: string, fn: () => T | Promise,): Promise>> export async function sleep(seconds: number): Promise @@ -394,6 +406,10 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. # - **Standalone**: executes the function body directly. # +# A task runs as its own job, so its result is always encoded as JSON and +# decoded back before the caller sees it: a ``datetime`` comes back as a +# string, a tuple as a list. +# # Usage:: # # @task @@ -439,6 +455,10 @@ def workflow(func) # On replay the cached value is returned without re-executing ``fn``. # Use for lightweight deterministic operations (timestamps, random IDs, # config reads) that should not incur the overhead of a child job. +# +# ``fn``'s result is encoded as JSON and decoded back before it is returned, +# so the round that runs the body sees the same types every replay sees: +# a ``datetime`` comes back as a string, a tuple as a list. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. diff --git a/typescript-client/build.sh b/typescript-client/build.sh index 926ab962ae..7e1c52335c 100755 --- a/typescript-client/build.sh +++ b/typescript-client/build.sh @@ -40,7 +40,7 @@ cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" echo 'export type { DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts" echo "" >> "${script_dirpath}/src/index.ts" -echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, getApprovalUrls, type TaskOptions, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI, commitKafkaOffsets } from "./client";' >> "${script_dirpath}/src/index.ts" +echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, getApprovalUrls, type TaskOptions, type Jsonified, type JsonifiedFn, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI, commitKafkaOffsets } from "./client";' >> "${script_dirpath}/src/index.ts" # Build default export by combining client utilities + services # This preserves backward compatibility for `import wmill from "windmill-client"` diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 228fbdff62..4a31d2dfed 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1546,6 +1546,123 @@ function stepErrorFromMarker(marker: any, name: string): Error { return err; } +/** Values `JSON.stringify` cannot represent: it omits the property holding one. + * The type-level half of `checkpointableResult`, which nulls them at the top + * level, where there is no key to omit. */ +type JsonDropped = + | ((...args: any[]) => any) + // A class is a function too, but its type has only a construct signature. + | (abstract new (...args: any[]) => any) + | symbol; + +/** + * What a value looks like after the JSON round trip a checkpoint performs: + * `Date` → string, `Map`/`Set` → `{}`, `undefined` → null, methods gone. + * `NaN` and the infinities decode as null too, deliberately still typed + * `number`: `number | null` everywhere costs more than that case is worth. + * + * Mirrors `encodeCheckpointPayload` below — keep the two in step, or `step()` + * starts describing a value it does not return. + */ +export type Jsonified = + // `any` in, `any` out: distributing over it yields a useless union. + 0 extends 1 & T + ? any + : // `unknown` is the idiomatic annotation for a JSON blob, and it matches + // no branch below — without this it would reach `never`, which is + // assignable to everything and so hides real mismatches. + unknown extends T + ? unknown + : T extends string | number | boolean | null + ? T + : T extends undefined | void | symbol | ((...args: any[]) => any) + ? null + : T extends bigint + ? string + : T extends { toJSON(): infer R } + ? Jsonified + : // Neither has own enumerable entries, so both serialize to `{}`. + T extends ReadonlyMap | ReadonlySet + ? Record + : // Arrays before objects, and without an `as` clause: key + // remapping would drop the array/tuple shape. + T extends readonly any[] + ? { [K in keyof T]: Jsonified } + : T extends object + ? JsonifiedObject + : never; + +/** + * `JSON.stringify` keeps own enumerable string keys whose value it can + * represent. A key that can only hold an unrepresentable value is gone; one + * that merely might becomes optional, because it can come back missing — + * `| undefined` alone would still demand the key be there. + */ +type JsonifiedObject = Flatten< + { + [K in keyof T as K extends symbol + ? never + : [Exclude] extends [never] + ? never + : [T[K]] extends [Exclude] + ? K + : never]: Jsonified; + } & { + [K in keyof T as K extends symbol + ? never + : [Exclude] extends [never] + ? never + : [T[K]] extends [Exclude] + ? never + : K]?: Jsonified>; + } +>; + +/** Collapse the two halves above into one object, so hovering `Jsonified` shows + * a shape rather than an intersection. */ +type Flatten = { [K in keyof T]: T[K] }; + +/** Encode a checkpoint payload the way the worker wrapper does on the suspend + * path, so both arms record the same value for the same step. `undefined` maps + * to null; a bigint to its digits, which the wrapper gets instead from the + * `BigInt.prototype.toJSON` it installs and the SDK cannot count on. */ +function encodeCheckpointPayload(payload: Record): string { + return JSON.stringify(payload, (_key, value) => + typeof value === "undefined" + ? null + : typeof value === "bigint" + ? value.toString() + : value, + ); +} + +/** A whole result that `JSON.stringify` drops the key for leaves a checkpoint + * with no `result`, which neither the endpoint nor `WacOutput` accepts. Only + * the top level: nested, a dropped key is what every other bun path does. */ +function checkpointableResult(value: any): any { + return typeof value === "function" || typeof value === "symbol" ? null : value; +} + +/** Put a value through the checkpoint's encoding without checkpointing it, so + * the paths that never persist anything still hand back the shape the ones + * that do would. */ +function jsonRoundTrip(value: T): Jsonified { + return JSON.parse(encodeCheckpointPayload({ value: checkpointableResult(value) })).value; +} + +/** + * A task function as its callers see it. A task's result always crosses a JSON + * boundary — the child job's result is read back from the checkpoint, and the + * v1 path reads it back from the API — so only {@link Jsonified} of it survives. + * + * Rebuilding the signature costs some precision TypeScript cannot preserve: a + * generic task's type parameters instantiate at their constraints, and an + * overloaded one keeps only its last signature. Neither survives JSON anyway. + */ +export type JsonifiedFn Promise> = ( + ...args: Parameters +) => Promise>>>; + export interface TaskOptions { timeout?: number; tag?: string; @@ -1830,7 +1947,28 @@ export class WorkflowCtx { fastPathFlagRaw !== "no"; const jobId = getEnv("WM_JOB_ID"); const workspace = getEnv("WM_WORKSPACE"); + let payload: string | undefined; + let checkpointed: any; if (fastPathEnabled && jobId && workspace && OpenAPI.BASE && OpenAPI.TOKEN) { + try { + payload = encodeCheckpointPayload({ + key, + result: checkpointableResult(result), + started_at: startedAt, + duration_ms: durationMs, + }); + checkpointed = JSON.parse(payload).result; + } catch (e) { + // Circular reference or a throwing toJSON. The wrapper on the suspend + // path uses the same replacer and fails the same way, so falling + // through keeps the failure where it was before the fast path existed. + console.log( + `WAC v2 inline fast path could not serialize key ${key}, falling back to suspend: ${e}`, + ); + } + } + if (payload !== undefined) { + const body = payload; const chainTail = this._inlineChain.then(async () => { const ctrl = new AbortController(); const t = setTimeout(() => ctrl.abort(), 10_000); @@ -1843,12 +1981,7 @@ export class WorkflowCtx { "Content-Type": "application/json", Authorization: `Bearer ${OpenAPI.TOKEN}`, }, - body: JSON.stringify({ - key, - result, - started_at: startedAt, - duration_ms: durationMs, - }), + body, signal: ctrl.signal, }, ); @@ -1879,11 +2012,14 @@ export class WorkflowCtx { if (errored) { throw Object.assign(stepErrorFromMarker(result, name), { cause: stepError }); } - return result as T; + // Return the round trip of what was checkpointed, never the in-memory + // value: handing back the live object would let the round that ran the + // body branch on a type — Date, Map — that no replay of it ever sees. + return checkpointed as T; } } - this._raiseSuspend({ mode: "inline_checkpoint", steps: [], key, result, started_at: startedAt, duration_ms: durationMs }); + this._raiseSuspend({ mode: "inline_checkpoint", steps: [], key, result: checkpointableResult(result), started_at: startedAt, duration_ms: durationMs }); } /** Raise a suspend, parking it so a body that catches it cannot make it @@ -1936,12 +2072,25 @@ export async function sleep(seconds: number): Promise { await new Promise((r) => setTimeout(r, seconds * 1000)); } -export async function step(name: string, fn: () => T | Promise): Promise { +/** + * Execute `fn` inline and checkpoint the result. On replay the cached value is + * returned without re-executing `fn`. + * + * `fn`'s result is encoded as JSON and decoded back before it is returned, so + * the round that runs the body sees the same types every replay sees: a `Date` + * comes back as a string, a `Map` as `{}`. {@link Jsonified} is that shape. + */ +export async function step( + name: string, + fn: () => T | Promise, +): Promise>> { const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); if (ctx) { - return ctx._runInlineStep(name, fn); + return ctx._runInlineStep(name, fn) as Promise>>; } - return fn(); + // Outside a workflow nothing is checkpointed, but round-trip anyway: running + // the script locally must not hand back a shape a deployed run never sees. + return jsonRoundTrip(await fn()); } /** @@ -1953,12 +2102,16 @@ export async function step(name: string, fn: () => T | Promise): Promise Promise>( fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, maybeOptions?: TaskOptions, -): T { +): JsonifiedFn { let fn: T; let taskPath: string | undefined; let taskOptions: TaskOptions | undefined; @@ -2005,7 +2158,7 @@ export function task Promise>( if ((e as any)?.name === "StepSuspend" || e instanceof StepSuspend) throw e; ctx._raiseStepFailure(e); } - ctx._raiseSuspend({ mode: "step_complete", steps: [], result }); + ctx._raiseSuspend({ mode: "step_complete", steps: [], result: checkpointableResult(result) }); })(); } return stepResult; @@ -2036,10 +2189,11 @@ export function task Promise>( return r; })(); } else { - // Standalone — execute directly - return fn(...args); + // Standalone — execute directly, but round-trip the result: a task's + // value crosses JSON in every other path, so a local run must agree. + return Promise.resolve(fn(...args)).then(jsonRoundTrip); } - } as unknown as T; + } as unknown as JsonifiedFn; Object.defineProperty(wrapper, "name", { value: taskName }); (wrapper as any)._is_task = true; diff --git a/typescript-client/tests/workflow_inline_fast_path.test.ts b/typescript-client/tests/workflow_inline_fast_path.test.ts new file mode 100644 index 0000000000..54c548925d --- /dev/null +++ b/typescript-client/tests/workflow_inline_fast_path.test.ts @@ -0,0 +1,177 @@ +/** + * Round-parity tests for the WAC v2 inline fast path, against the real client. + * + * Run with: bun test typescript-client/tests/workflow_inline_fast_path.test.ts + * + * Unlike workflow.test.ts (which mirrors the SDK inline), these import + * client.ts itself — the fast path is what diverges between the round that + * runs a step body and its replays, so a mirror would not pin it. The two + * generated modules are stubbed so the import works without ./build.sh. + */ +import { expect, test, describe, mock, beforeEach } from "bun:test"; + +mock.module("../services.gen", () => ({ + ResourceService: {}, + VariableService: {}, + JobService: {}, + HelpersService: {}, + AppService: {}, + MetricsService: {}, + OidcService: {}, + UserService: {}, + KafkaTriggerService: {}, +})); +mock.module("../core/OpenAPI", () => ({ + OpenAPI: { BASE: "http://localhost:8000/api", TOKEN: "tok" }, +})); + +const { WorkflowCtx, step, task, setWorkflowCtx } = await import("../client.ts"); +import type { Jsonified } from "../client.ts"; + +process.env.WM_JOB_ID = "job-1"; +process.env.WM_WORKSPACE = "admins"; + +/** Last checkpoint POSTed by the fast path. */ +let posted: Record; + +beforeEach(() => { + posted = {}; + // @ts-ignore — stand in for the inline_checkpoint endpoint. + globalThis.fetch = async (_url: any, init: any) => { + const payload = JSON.parse(init.body); + posted[payload.key] = payload.result; + return new Response("{}", { status: 200 }); + }; +}); + +describe("inline step round parity", () => { + // Values JSON does not round-trip: the fast path used to hand the body the + // live object, so `result instanceof Date` was true on the round that ran + // the body and false on every replay of it. + const cases: Array<[string, () => any, any]> = [ + ["date", () => new Date("2026-01-01T00:00:00Z"), "2026-01-01T00:00:00.000Z"], + ["map", () => new Map([["a", 1]]), {}], + ["set", () => new Set([1, 2]), {}], + ["undefined-prop", () => ({ a: 1, b: undefined }), { a: 1, b: null }], + // A body that returns nothing — `step("notify", () => { sendEmail() })`. + ["nothing", () => undefined, null], + // `JSON.stringify` drops the key for these, and a checkpoint with no + // `result` is one neither the endpoint nor the worker can parse. + ["function", () => () => 1, null], + // Nested, though, a dropped key is what every other bun path produces — + // and what `Jsonified` describes. + ["nested-function", () => ({ a: 1, cb: () => 1 }), { a: 1 }], + // JSON has no non-finite numbers; `Jsonified` keeps calling these `number`. + ["non-finite", () => ({ nan: NaN, inf: Infinity }), { nan: null, inf: null }], + // A property holding an unrepresentable value comes back missing, not null + // — which is why `Jsonified` makes such a key optional. + ["union-symbol", () => ({ tag: Symbol("x"), keep: 1 }), { keep: 1 }], + ["class-value", () => ({ Klass: class W {}, keep: 1 }), { keep: 1 }], + ]; + + for (const [name, fn, expected] of cases) { + test(`${name}: the running round sees what the replay sees`, async () => { + const live = await new WorkflowCtx({} as any)._runInlineStep(name, fn); + expect(live).toEqual(expected); + // ...and it is exactly what got checkpointed, so the replay agrees. + expect(posted[name]).toEqual(expected); + const replayed = await new WorkflowCtx({ + completed_steps: { [name]: posted[name] }, + } as any)._runInlineStep(name, fn); + expect(replayed).toEqual(live); + }); + } + + test("outside a workflow, step() and task() return the same shape", async () => { + // No checkpoint, no replay — but a local run must not hand back a shape a + // deployed one never produces, or testing a workflow locally proves nothing. + const jobId = process.env.WM_JOB_ID; + delete process.env.WM_JOB_ID; // otherwise task() takes the v1 dispatch path + try { + expect(await step("pair", () => [1, new Date("2026-01-01T00:00:00Z")])).toEqual([ + 1, + "2026-01-01T00:00:00.000Z", + ]); + const makePair = task(async function makePair() { + return { at: new Date("2026-01-01T00:00:00Z") }; + }); + expect(await makePair()).toEqual({ at: "2026-01-01T00:00:00.000Z" }); + // No `BigInt.prototype.toJSON` outside the worker wrapper, so the + // encoder has to handle bigint itself or this throws. + expect(await step("big", () => ({ n: 2n ** 70n }))).toEqual({ + n: "1180591620717411303424", + }); + } finally { + process.env.WM_JOB_ID = jobId; + } + }); + + test("a child job's task result is normalized before it is reported", async () => { + // The child runs the task body and reports it as `step_complete`, which the + // worker parses into `WacOutput::Complete { result: Value }` — no serde + // default, so a result whose key JSON.stringify drops fails the job. + const ctx = new WorkflowCtx({ _executing_key: "returnsFn" } as any); + setWorkflowCtx(ctx); + try { + const returnsFn = task(async function returnsFn() { + return () => 1; + }); + const suspend: any = await returnsFn().then( + () => null, + (e: any) => e, + ); + expect(suspend?.dispatchInfo).toMatchObject({ mode: "step_complete", result: null }); + } finally { + setWorkflowCtx(null); + } + }); +}); + +// `Jsonified` must describe the values asserted above, or `step()` advertises a +// type it never returns. `bun test` strips types, and tsconfig only covers +// `src/`, so these are checked with `npx tsc --ignoreConfig --noEmit --strict +// --allowImportingTsExtensions ` (its bun:test / process errors are noise). +type Exact = (() => G extends A ? 1 : 2) extends () => G extends B ? 1 : 2 + ? true + : false; +const _assertType = (_: E) => {}; + +_assertType, string>>(true); +_assertType>, Record>>(true); +_assertType>, Record>>(true); +_assertType, { a: number; b: null }>>(true); +_assertType, null>>(true); +// Shapes JSON does preserve must survive untouched, methods aside. +_assertType, [number, string]>>(true); +_assertType< + Exact, { id: number; at: string }> +>(true); + +// `unknown` is the idiomatic JSON-blob annotation and must survive as itself: +// collapsing to `never` would make any downstream assignment typecheck. +_assertType, unknown>>(true); +_assertType>, Record>>(true); +_assertType, unknown[]>>(true); +_assertType, string>>(true); +// Symbol-valued properties are dropped by the encoder, like methods — but a +// property that only *might* hold one keeps its key, optional. +_assertType, { a: number }>>(true); +_assertType, { tag?: string }>>(true); +// The key can be missing at runtime, so the type has to accept it missing. +const _omittable: Jsonified<{ tag: string | symbol; keep: number }> = { keep: 1 }; +void _omittable; +_assertType, (string | null)[]>>(true); +// A class is a function at runtime, but its type has only a construct signature. +class _Widget { + x = 1; +} +_assertType, { keep: number }>>(true); + +// A task's result always crosses JSON too — the checkpoint on the workflow +// path, the API on the v1 path — while its arguments stay checked. +const _typedTask = task(async function _typedTask(id: number) { + return { id, at: new Date() }; +}); +_assertType< + Exact>, { id: number; at: string }> +>(true);