diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 4e8cc464d5..fff7df1153 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -1106,6 +1106,9 @@ async sleep(seconds: number): Promise * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> @@ -1949,6 +1952,9 @@ async sleep(seconds: number): Promise * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> @@ -2898,6 +2904,9 @@ async sleep(seconds: number): Promise * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> @@ -5073,7 +5082,9 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -5159,12 +5170,15 @@ def workflow(func) # \`\`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. +# +# Called from inside a task body, \`\`fn\`\` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after \`\`seconds\`\`. -# Outside a workflow, falls back to \`\`asyncio.sleep\`\`. +# Outside a workflow, or inside a task body, falls back to \`\`asyncio.sleep\`\`. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. @@ -7436,6 +7450,8 @@ async def enrich(customer_id: str): In TypeScript, prefer assigning each task to a named top-level const. In Python, prefer top-level async functions decorated with \`@task()\` or \`@task\`. +Only the workflow body dispatches steps. A task runs as a single job, so a task or \`step()\` called from inside a task body runs in place in that job, as a plain function call: no child job, no checkpoint, and its task options (retry, timeout, tag) do not apply. \`sleep()\` there waits in the job, and \`taskScript()\` / \`taskFlow()\` / \`waitForApproval()\` (\`task_script()\` / \`task_flow()\` / \`wait_for_approval()\`) throw. To get separate jobs, call the tasks one after another from the workflow body and pass results between them. + For existing scripts: \`\`\`typescript @@ -7601,8 +7617,9 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * const call_api = task(fetchOrders, { retry: { attempts: 3, delay: 30, multiplier: 2 } }); * * Inside a \`workflow()\`, calling a task dispatches it as a step. - * Outside a workflow, the function body executes directly and - * {@link TaskOptions} — retry included — does not apply. + * Outside a workflow, or called from inside another task, the function body + * executes directly and {@link TaskOptions} — retry included — does not apply: + * a task runs as a single job, so the tasks it calls run in that same job. * * 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 @@ -7646,6 +7663,9 @@ export function workflow(fn: (...args: any[]) => Promise) * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ export async function step(name: string, fn: () => T | Promise,): Promise>> @@ -7744,7 +7764,9 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -7830,12 +7852,15 @@ def workflow(func) # \`\`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. +# +# Called from inside a task body, \`\`fn\`\` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after \`\`seconds\`\`. -# Outside a workflow, falls back to \`\`asyncio.sleep\`\`. +# Outside a workflow, or inside a task body, falls back to \`\`asyncio.sleep\`\`. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/python-client/wmill/tests/test_workflow.py b/python-client/wmill/tests/test_workflow.py index ceed39bfeb..d32ae1458b 100644 --- a/python-client/wmill/tests/test_workflow.py +++ b/python-client/wmill/tests/test_workflow.py @@ -7,7 +7,7 @@ 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 +from wmill.client import WorkflowCtx, _StepSuspend, TaskError, workflow, task, task_script, step, sleep, parallel, wait_for_approval, _run_workflow, _run_workflow_async class _FakeItems(dict): @@ -752,6 +752,84 @@ class TestChildModeParallel: assert result["result"] == 8 +class TestSdkCallsInsideTaskBody: + """A task body runs as one child job, so the SDK calls it makes cannot be + dispatched as steps of their own.""" + + def _run_child(self, wf, key): + async def bounded(): + return await asyncio.wait_for( + _run_workflow_async(wf, {"completed_steps": {}, "_executing_key": key}, {}), + timeout=2, + ) + + return asyncio.run(bounded()) + + def test_nested_task_and_step_run_in_the_tasks_own_job(self): + @task + async def inner(x: str): + return f"inner:{x}" + + @task + def sync_leaf(x: str): + return f"sync:{x}" + + # A sync body calls its tasks without awaiting, whatever called it. + @task + def sync_inner(x: str): + return sync_leaf(x).upper().lower() + + @task + async def outer(x: str): + via_task = await inner(x) + via_sync = await sync_inner(x) + via_step = await step("s", lambda: "stepped") + return f"{via_task}+{via_sync}+{via_step}" + + @workflow + async def wf(): + return await outer("a") + + result = self._run_child(wf, "outer") + assert result == {"type": "complete", "result": "inner:a+sync:a+stepped"} + + def test_sibling_task_of_the_workflow_body_is_not_run(self): + # The workflow body keeps replaying beside the task body, and its own + # calls must stay parked: running them would execute a sibling task in + # this job. + ran = [] + + @task + async def inner(x: str): + ran.append(x) + return x + + @task + async def outer(x: str): + await asyncio.sleep(0) + return await inner("nested") + + @workflow + async def wf(): + return await asyncio.gather(outer("a"), inner("sibling")) + + result = self._run_child(wf, "outer") + assert result == {"type": "complete", "result": "nested"} + assert ran == ["nested"] + + def test_step_that_would_need_a_dispatch_fails_instead_of_hanging(self): + @task + async def dispatching(): + return await task_script("f/other/script")() + + @workflow + async def wf(): + return await dispatching() + + with pytest.raises(RuntimeError, match="cannot be called from inside a task"): + self._run_child(wf, "dispatching") + + class TestKeyDeterminism: def test_keys_consistent_across_replays(self): r1 = _run_workflow(det_wf, {}, {"n": 3}) diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 97da60aadc..824db9741f 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -2766,6 +2766,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]: import asyncio as _asyncio import contextvars as _contextvars +import inspect as _inspect import sys as _sys import traceback as _traceback @@ -2993,6 +2994,64 @@ _workflow_ctx: _contextvars.ContextVar["WorkflowCtx"] = _contextvars.ContextVar( "_workflow_ctx" ) +# Set while a task body runs in its child job: ``"async"`` or ``"sync"`` after +# the kind of function the body is. A flag on the ctx cannot stand in: under +# ``asyncio.gather`` the workflow body keeps replaying beside the task body, and +# its calls must still park rather than run a sibling task's body in this job. +_in_task_body: _contextvars.ContextVar[Optional[str]] = _contextvars.ContextVar( + "_in_task_body", default=None +) + + +def _dispatch_ctx() -> "WorkflowCtx | None": + """The ctx SDK calls dispatch through, or None where they run in place: + outside a workflow, and inside a task body, which is one job and has no + rounds of its own to dispatch or checkpoint a step in.""" + if _in_task_body.get(): + return None + return _workflow_ctx.get(None) + + +async def _resolved(value): + return value + + +def _call_in_task_body(func, caller: str, args, kwargs): + """Run a task called from inside a task body in place, round-tripping its + result as a dispatched one would be. + + The workflow body awaits every task, sync ones included, so an async caller + gets an awaitable back either way; a sync caller gets the value. The scope + is re-marked for the nested body, which may be of the other kind.""" + + async def run_async(coro): + token = _in_task_body.set("async") + try: + return _json_round_trip(await coro) + finally: + _in_task_body.reset(token) + + if _inspect.iscoroutinefunction(func): + return run_async(func(*args, **kwargs)) + token = _in_task_body.set("sync") + try: + result = func(*args, **kwargs) + finally: + _in_task_body.reset(token) + if _asyncio.iscoroutine(result): + return run_async(result) + result = _json_round_trip(result) + return _resolved(result) if caller == "async" else result + + +def _assert_not_inside_task_body(what: str) -> None: + if _in_task_body.get(): + raise RuntimeError( + f"{what} cannot be called from inside a task: a task runs as a single " + "job and cannot suspend or dispatch steps. Call it from the @workflow " + "body instead." + ) + class WorkflowCtx: """Internal context for workflow replay/suspension. @@ -3117,15 +3176,20 @@ class WorkflowCtx: raise _StepSuspend({"mode": "sleep", "key": key, "seconds": seconds, "steps": []}) async def _resolved(self, value): - return value + return await _resolved(value) async def _execute_directly(self, func, **kwargs): + token = _in_task_body.set( + "async" if _inspect.iscoroutinefunction(func) else "sync" + ) try: result = func(**kwargs) if _asyncio.iscoroutine(result): result = await result except Exception as exc: raise _StepFailure(exc) from exc + finally: + _in_task_body.reset(token) raise _StepSuspend({"mode": "step_complete", "steps": [], "result": result}) async def _never_resolve(self): @@ -3367,7 +3431,9 @@ def task( - **v2 (inside @workflow)**: dispatches as a checkpoint step. - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. - - **Standalone**: executes the function body directly. + - **Standalone, or called from inside another v2 task**: executes the + function body directly. A task runs as a single job, so the tasks it + calls run in that same job, without their own options. 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 @@ -3447,12 +3513,16 @@ def task( @functools.wraps(func) def wrapper(*args, **kwargs): # WAC v2: inside a @workflow context - ctx = _workflow_ctx.get(None) + ctx = _dispatch_ctx() if ctx is not None: script = task_path if task_path else task_name merged = _merge_args(args, kwargs) return ctx._next_step(task_name, script, func, _task_options=_task_opts, **merged) + caller = _in_task_body.get() + if caller is not None: + return _call_in_task_body(func, caller, args, kwargs) + # WAC v1: running inside a Windmill job but not in a @workflow if ( os.environ.get("WM_JOB_ID") is not None @@ -3531,6 +3601,7 @@ def task_script( _opts = {k: v for k, v in {"timeout": timeout, "tag": tag, "cache_ttl": cache_ttl, "priority": priority, "concurrent_limit": concurrency_limit, "concurrency_key": concurrency_key, "concurrency_time_window_s": concurrency_time_window_s, "retry": _checked_retry(retry)}.items() if v is not None} or None def wrapper(**kwargs): + _assert_not_inside_task_body(f'task_script("{path}")') ctx = _workflow_ctx.get(None) if ctx is not None: return ctx._next_step(name, path, dispatch_type="script", _task_options=_opts, **kwargs) @@ -3570,6 +3641,7 @@ def task_flow( _opts = {k: v for k, v in {"timeout": timeout, "tag": tag, "cache_ttl": cache_ttl, "priority": priority, "concurrent_limit": concurrency_limit, "concurrency_key": concurrency_key, "concurrency_time_window_s": concurrency_time_window_s, "retry": _checked_retry(retry)}.items() if v is not None} or None def wrapper(**kwargs): + _assert_not_inside_task_body(f'task_flow("{path}")') ctx = _workflow_ctx.get(None) if ctx is not None: return ctx._next_step(name, path, dispatch_type="flow", _task_options=_opts, **kwargs) @@ -3604,8 +3676,11 @@ async def step(name: str, 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 ``datetime`` comes back as a string, a tuple as a list. + + Called from inside a task body, ``fn`` runs in place and nothing is + checkpointed: a task is a single job and is never replayed. """ - ctx: WorkflowCtx | None = _workflow_ctx.get(None) + ctx = _dispatch_ctx() if ctx is not None: return await ctx._run_inline_step(name, fn) result = fn() @@ -3620,9 +3695,9 @@ async def sleep(seconds: int): """Server-side sleep — suspend the workflow for the given duration without holding a worker. Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. - Outside a workflow, falls back to ``asyncio.sleep``. + Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. """ - ctx: WorkflowCtx | None = _workflow_ctx.get(None) + ctx = _dispatch_ctx() if ctx is not None: return await ctx._sleep(seconds) await _asyncio.sleep(seconds) @@ -3660,6 +3735,7 @@ async def wait_for_approval( await step("notify", lambda: send_email(urls["resume"], urls["cancel"])) result = await wait_for_approval(key="manager", timeout=3600) """ + _assert_not_inside_task_body("wait_for_approval()") ctx: WorkflowCtx | None = _workflow_ctx.get(None) if ctx is not None: return await ctx._wait_for_approval( diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index cfa3415e35..393b724f6b 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -1427,6 +1427,8 @@ async def enrich(customer_id: str): In TypeScript, prefer assigning each task to a named top-level const. In Python, prefer top-level async functions decorated with \`@task()\` or \`@task\`. +Only the workflow body dispatches steps. A task runs as a single job, so a task or \`step()\` called from inside a task body runs in place in that job, as a plain function call: no child job, no checkpoint, and its task options (retry, timeout, tag) do not apply. \`sleep()\` there waits in the job, and \`taskScript()\` / \`taskFlow()\` / \`waitForApproval()\` (\`task_script()\` / \`task_flow()\` / \`wait_for_approval()\`) throw. To get separate jobs, call the tasks one after another from the workflow body and pass results between them. + For existing scripts: \`\`\`typescript @@ -2029,6 +2031,9 @@ async sleep(seconds: number): Promise * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> @@ -2871,7 +2876,9 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -2957,12 +2964,15 @@ def workflow(func) # \`\`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. +# +# Called from inside a task body, \`\`fn\`\` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after \`\`seconds\`\`. -# Outside a workflow, falls back to \`\`asyncio.sleep\`\`. +# Outside a workflow, or inside a task body, falls back to \`\`asyncio.sleep\`\`. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. @@ -3084,8 +3094,9 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * const call_api = task(fetchOrders, { retry: { attempts: 3, delay: 30, multiplier: 2 } }); * * Inside a \`workflow()\`, calling a task dispatches it as a step. - * Outside a workflow, the function body executes directly and - * {@link TaskOptions} — retry included — does not apply. + * Outside a workflow, or called from inside another task, the function body + * executes directly and {@link TaskOptions} — retry included — does not apply: + * a task runs as a single job, so the tasks it calls run in that same job. * * 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 @@ -3129,6 +3140,9 @@ export function workflow(fn: (...args: any[]) => Promise) * \`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. + * + * Called from inside a task body, \`fn\` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ export async function step(name: string, fn: () => T | Promise,): Promise>> @@ -3227,7 +3241,9 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -3313,12 +3329,15 @@ def workflow(func) # \`\`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. +# +# Called from inside a task body, \`\`fn\`\` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after \`\`seconds\`\`. -# Outside a workflow, falls back to \`\`asyncio.sleep\`\`. +# Outside a workflow, or inside a task body, falls back to \`\`asyncio.sleep\`\`. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index 14ca869180..a4064c3d9e 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -1906,6 +1906,9 @@ async sleep(seconds: number): Promise * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> @@ -2748,7 +2751,9 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -2834,12 +2839,15 @@ def workflow(func) # ``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. +# +# Called from inside a task body, ``fn`` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. -# Outside a workflow, falls back to ``asyncio.sleep``. +# Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index 13a93ed063..1964c88685 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -669,7 +669,9 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -755,12 +757,15 @@ def workflow(func) # ``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. +# +# Called from inside a task body, ``fn`` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. -# Outside a workflow, falls back to ``asyncio.sleep``. +# Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/system_prompts/auto-generated/sdks/typescript.md b/system_prompts/auto-generated/sdks/typescript.md index 55df35ccf7..da9e79ec94 100644 --- a/system_prompts/auto-generated/sdks/typescript.md +++ b/system_prompts/auto-generated/sdks/typescript.md @@ -463,6 +463,9 @@ async sleep(seconds: number): Promise * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index a8e98a9d4f..6dd307dd24 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -36,7 +36,9 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -122,12 +124,15 @@ def workflow(func) # ``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. +# +# Called from inside a task body, ``fn`` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. -# Outside a workflow, falls back to ``asyncio.sleep``. +# Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/system_prompts/auto-generated/sdks/wac-typescript.md b/system_prompts/auto-generated/sdks/wac-typescript.md index 608d75d18c..dd04a64de0 100644 --- a/system_prompts/auto-generated/sdks/wac-typescript.md +++ b/system_prompts/auto-generated/sdks/wac-typescript.md @@ -67,8 +67,9 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * const call_api = task(fetchOrders, { retry: { attempts: 3, delay: 30, multiplier: 2 } }); * * Inside a `workflow()`, calling a task dispatches it as a step. - * Outside a workflow, the function body executes directly and - * {@link TaskOptions} — retry included — does not apply. + * Outside a workflow, or called from inside another task, the function body + * executes directly and {@link TaskOptions} — retry included — does not apply: + * a task runs as a single job, so the tasks it calls run in that same job. * * 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 @@ -112,6 +113,9 @@ export function workflow(fn: (...args: any[]) => Promise) * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ export async function step(name: string, fn: () => T | Promise,): 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 6ad44db068..055b1613b2 100644 --- a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md @@ -659,6 +659,9 @@ async sleep(seconds: number): Promise * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> 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 a5f599a946..beaf7817bb 100644 --- a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md @@ -671,6 +671,9 @@ async sleep(seconds: number): Promise * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> 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 ce0c610da7..afefb7e987 100644 --- a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md @@ -661,6 +661,9 @@ async sleep(seconds: number): Promise * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ async step(name: string, fn: () => T | Promise,): Promise>> 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 4946ed0f6e..ee50b6d62d 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -877,7 +877,9 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -963,12 +965,15 @@ def workflow(func) # ``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. +# +# Called from inside a task body, ``fn`` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. -# Outside a workflow, falls back to ``asyncio.sleep``. +# Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. 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 13d305c74e..09789b388a 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 @@ -143,6 +143,8 @@ async def enrich(customer_id: str): In TypeScript, prefer assigning each task to a named top-level const. In Python, prefer top-level async functions decorated with `@task()` or `@task`. +Only the workflow body dispatches steps. A task runs as a single job, so a task or `step()` called from inside a task body runs in place in that job, as a plain function call: no child job, no checkpoint, and its task options (retry, timeout, tag) do not apply. `sleep()` there waits in the job, and `taskScript()` / `taskFlow()` / `waitForApproval()` (`task_script()` / `task_flow()` / `wait_for_approval()`) throw. To get separate jobs, call the tasks one after another from the workflow body and pass results between them. + For existing scripts: ```typescript @@ -308,8 +310,9 @@ export async function getResumeUrls(approver?: string, flowLevel?: boolean): Pro * const call_api = task(fetchOrders, { retry: { attempts: 3, delay: 30, multiplier: 2 } }); * * Inside a `workflow()`, calling a task dispatches it as a step. - * Outside a workflow, the function body executes directly and - * {@link TaskOptions} — retry included — does not apply. + * Outside a workflow, or called from inside another task, the function body + * executes directly and {@link TaskOptions} — retry included — does not apply: + * a task runs as a single job, so the tasks it calls run in that same job. * * 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 @@ -353,6 +356,9 @@ export function workflow(fn: (...args: any[]) => Promise) * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ export async function step(name: string, fn: () => T | Promise,): Promise>> @@ -451,7 +457,9 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # # - **v2 (inside @workflow)**: dispatches as a checkpoint step. # - **v1 (WM_JOB_ID set, no @workflow)**: dispatches via HTTP API. -# - **Standalone**: executes the function body directly. +# - **Standalone, or called from inside another v2 task**: executes the +# function body directly. A task runs as a single job, so the tasks it +# calls run in that same job, without their own options. # # 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 @@ -537,12 +545,15 @@ def workflow(func) # ``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. +# +# Called from inside a task body, ``fn`` runs in place and nothing is +# checkpointed: a task is a single job and is never replayed. async def step(name: str, fn) # Server-side sleep — suspend the workflow for the given duration without holding a worker. # # Inside a @workflow, the parent job suspends and auto-resumes after ``seconds``. -# Outside a workflow, falls back to ``asyncio.sleep``. +# Outside a workflow, or inside a task body, falls back to ``asyncio.sleep``. async def sleep(seconds: int) # Suspend the workflow and wait for an external approval. diff --git a/system_prompts/base/workflow-as-code.md b/system_prompts/base/workflow-as-code.md index 0e529f6803..efc29df360 100644 --- a/system_prompts/base/workflow-as-code.md +++ b/system_prompts/base/workflow-as-code.md @@ -95,6 +95,8 @@ async def enrich(customer_id: str): In TypeScript, prefer assigning each task to a named top-level const. In Python, prefer top-level async functions decorated with `@task()` or `@task`. +Only the workflow body dispatches steps. A task runs as a single job, so a task or `step()` called from inside a task body runs in place in that job, as a plain function call: no child job, no checkpoint, and its task options (retry, timeout, tag) do not apply. `sleep()` there waits in the job, and `taskScript()` / `taskFlow()` / `waitForApproval()` (`task_script()` / `task_flow()` / `wait_for_approval()`) throw. To get separate jobs, call the tasks one after another from the workflow body and pass results between them. + For existing scripts: ```typescript diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 3f7b03117a..da957239cf 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1832,6 +1832,52 @@ export function setWorkflowCtx(ctx: WorkflowCtx | null) { Reflect.set(globalThis, "__wmill_wf_ctx", ctx); } +type TaskBodyScope = { run(store: true, fn: () => R): R; getStore(): true | undefined }; + +/** Marks the async context of a task body running in its child job. A flag on + * the ctx cannot stand in: the workflow body keeps replaying beside the task + * body (`Promise.all([a(), b()])`), and its calls must still park rather than + * run a sibling task's body in this job. Kept on `globalThis`, like the ctx, + * so a bundle holding two copies of this module shares one scope. Resolved + * through `process.getBuiltinModule` because a `node:` import would break + * browser bundles of this package. */ +function taskBodyScope(): TaskBodyScope | null { + const existing = Reflect.get(globalThis, "__wmill_wf_task_body"); + if (existing !== undefined) return existing; + let scope: TaskBodyScope | null = null; + try { + const AsyncLocalStorage = (globalThis as any).process?.getBuiltinModule?.("node:async_hooks")?.AsyncLocalStorage; + if (AsyncLocalStorage) scope = new AsyncLocalStorage(); + } catch {} + Reflect.set(globalThis, "__wmill_wf_task_body", scope); + return scope; +} + +function runTaskBody(fn: () => R): R { + const scope = taskBodyScope(); + return scope ? scope.run(true, fn) : fn(); +} + +function insideTaskBody(): boolean { + return taskBodyScope()?.getStore() === true; +} + +/** The ctx SDK calls dispatch through, or null where they run in place: outside + * a workflow, and inside a task body, which is one job and has no rounds of its + * own to dispatch or checkpoint a step in. */ +function dispatchCtx(): WorkflowCtx | null { + if (insideTaskBody()) return null; + return _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); +} + +function assertNotInsideTaskBody(what: string): void { + if (insideTaskBody()) { + throw new Error( + `${what} cannot be called from inside a task: a task runs as a single job and cannot suspend or dispatch steps. Call it from the workflow() body instead.`, + ); + } +} + export class WorkflowCtx { private completed: Record; @@ -2304,11 +2350,11 @@ export class WorkflowCtx { } export async function sleep(seconds: number): Promise { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + const ctx = dispatchCtx(); if (ctx) { return ctx._sleep(seconds) as Promise; } - // Outside workflow context, just wait locally + // Outside a workflow, or inside a task body, just wait locally await new Promise((r) => setTimeout(r, seconds * 1000)); } @@ -2319,12 +2365,15 @@ export async function sleep(seconds: number): Promise { * `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. + * + * Called from inside a task body, `fn` runs in place and nothing is + * checkpointed: a task is a single job and is never replayed. */ export async function step( name: string, fn: () => T | Promise, ): Promise>> { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + const ctx = dispatchCtx(); if (ctx) { return ctx._runInlineStep(name, fn) as Promise>>; } @@ -2342,8 +2391,9 @@ export async function step( * const call_api = task(fetchOrders, { retry: { attempts: 3, delay: 30, multiplier: 2 } }); * * Inside a `workflow()`, calling a task dispatches it as a step. - * Outside a workflow, the function body executes directly and - * {@link TaskOptions} — retry included — does not apply. + * Outside a workflow, or called from inside another task, the function body + * executes directly and {@link TaskOptions} — retry included — does not apply: + * a task runs as a single job, so the tasks it calls run in that same job. * * 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 @@ -2377,7 +2427,7 @@ export function task Promise>( // .then() which throws StepSuspend and empties pending before the caller // can flush. const wrapper = function (...args: any[]) { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + const ctx = dispatchCtx(); if (ctx) { // Inside a workflow with checkpoint/replay context — dispatch as step const script = taskPath ?? taskName; @@ -2397,7 +2447,7 @@ export function task Promise>( return (async () => { let result: any; try { - result = await fn(...args); + result = await runTaskBody(() => fn(...args)); } catch (e) { if (isSuspendSignal(e, StepSuspend)) throw e; ctx._raiseStepFailure(e); @@ -2406,7 +2456,7 @@ export function task Promise>( })(); } return stepResult; - } else if (getEnv("WM_JOB_ID") && !getEnv("WM_FLOW_JOB_ID")) { + } else if (!insideTaskBody() && getEnv("WM_JOB_ID") && !getEnv("WM_FLOW_JOB_ID")) { // Inside a Windmill root job without checkpoint context — v1 HTTP dispatch // WM_FLOW_JOB_ID is set on child jobs, so we skip dispatch for those return (async () => { @@ -2456,7 +2506,8 @@ export function taskScript(path: string, options?: TaskOptions): (...args: any[] assertUsableRetry(options?.retry); const name = path.split("/").pop() || path; const wrapper = function (...args: any[]) { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + assertNotInsideTaskBody(`taskScript("${path}")`); + const ctx = dispatchCtx(); if (ctx) { const kwargs = args.length === 1 && typeof args[0] === "object" && args[0] !== null ? args[0] @@ -2482,7 +2533,8 @@ export function taskFlow(path: string, options?: TaskOptions): (...args: any[]) assertUsableRetry(options?.retry); const name = path.split("/").pop() || path; const wrapper = function (...args: any[]) { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + assertNotInsideTaskBody(`taskFlow("${path}")`); + const ctx = dispatchCtx(); if (ctx) { const kwargs = args.length === 1 && typeof args[0] === "object" && args[0] !== null ? args[0] @@ -2535,7 +2587,8 @@ export function waitForApproval(options?: { skin?: "detailed" | "minimal"; description?: string | object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> { - const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); + assertNotInsideTaskBody("waitForApproval()"); + const ctx = dispatchCtx(); if (!ctx) { throw new Error("waitForApproval can only be called inside a workflow()"); } diff --git a/typescript-client/tests/workflow_nested_task.test.ts b/typescript-client/tests/workflow_nested_task.test.ts new file mode 100644 index 0000000000..39ac2185d8 --- /dev/null +++ b/typescript-client/tests/workflow_nested_task.test.ts @@ -0,0 +1,91 @@ +/** + * A task body runs as one child job, so the SDK calls it makes cannot be + * dispatched as steps of their own. Against the real client. + * + * Run with: bun test typescript-client/tests/workflow_nested_task.test.ts + */ +import { expect, test, describe, mock, afterEach } 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, StepSuspend, task, taskScript, step, setWorkflowCtx } = await import( + "../client.ts" +); + +let ran: string[] = []; + +const inner = task(async function inner(x: string) { + ran.push("inner"); + return `inner:${x}`; +}); + +const outer = task(async function outer(x: string) { + ran.push("outer"); + await Promise.resolve(); + const viaTask = await inner(x); + const viaStep = await step("s", () => "stepped"); + return `${viaTask}+${viaStep}`; +}); + +const dispatching = task(async function dispatching() { + return await taskScript("f/other/script")({}); +}); + +/** Run `body` as the child round executing `key`, and hand back what it raised. */ +async function childRound(key: string, body: () => Promise): Promise { + setWorkflowCtx(new WorkflowCtx({ completed_steps: {}, _executing_key: key })); + const hung = new Promise((r) => setTimeout(() => r("hung"), 500)); + return await Promise.race([body().then(() => "returned", (e) => e), hung]); +} + +afterEach(() => { + ran = []; + setWorkflowCtx(null); +}); + +describe("SDK calls inside a task body", () => { + test("a nested task and step run in the task's own job", async () => { + const raised = await childRound("outer", async () => { + await outer("a"); + }); + + expect(raised).toBeInstanceOf(StepSuspend); + expect(raised.dispatchInfo).toMatchObject({ + mode: "step_complete", + result: "inner:a+stepped", + }); + }); + + // The workflow body keeps replaying beside the task body, and its own calls + // must stay parked: running them would execute a sibling task in this job. + test("a sibling task of the workflow body is not run", async () => { + const raised = await childRound("outer", async () => { + await Promise.all([outer("a"), inner("sibling")]); + }); + + expect(raised.dispatchInfo).toMatchObject({ mode: "step_complete" }); + expect(ran).toEqual(["outer", "inner"]); + }); + + test("a step that would need a dispatch fails instead of hanging", async () => { + const raised = await childRound("dispatching", async () => { + await dispatching(); + }); + + expect(raised).toBeInstanceOf(Error); + expect(raised.message).toContain("cannot be called from inside a task"); + }); +});