diff --git a/workbench/python/README.md b/workbench/python/README.md index d7ba8975ad..ca05f85c78 100644 --- a/workbench/python/README.md +++ b/workbench/python/README.md @@ -269,30 +269,85 @@ files in one process while the queue delivery takes an HTTP round trip, so ## Conformance baseline What the suite runs here is declared in `e2e-conformance.json`: the ported -fixtures, and — when there is one — an `unsupported` map naming individual tests -whose failure is a runtime gap rather than a missing fixture. There is none right -now. Both axes are ratchets: a claim that stops being true fails the run instead -of quietly skipping, so growing the file is the only way to move. -`ConformanceConfig` in `packages/core/e2e/utils.ts` spells out each direction. - -Current baseline: **9 passing, 128 skipped, of 137** on `world-local`, and -**8 of 156** on Vercel. It is one baseline, not two — the extra 19 collected on -Vercel are `e2e-agent.test.ts`, which that lane also picks up and skips whole, -and the ninth pass is `deploymentId: 'latest' is a no-op in non-Vercel worlds`, -which is local by definition. +fixtures, and an `unsupported` map naming individual tests whose failure is a +runtime gap rather than a missing fixture. Both axes are ratchets: a claim that +stops being true fails the run instead of quietly skipping, so growing the file is +the only way to move. `ConformanceConfig` in `packages/core/e2e/utils.ts` spells +out each direction. + +Current baseline: **32 passing, 105 skipped, of 137** on `world-local` — 27 +fixtures and 3 exempted tests. The Vercel lane collects 19 more tests +(`e2e-agent.test.ts`, which it also picks up and skips whole) and passes one +fewer, because `deploymentId: 'latest' is a no-op in non-Vercel worlds` is local +by definition. It is one baseline, not two. + +The three exemptions are two upstream causes, and both are ones this lane found +rather than predicted: + +- **`hookTokenReuseLoopWorkflow`** — `hook_created` and `hook_disposed` are + flushed with no ordering between them, so a run conflicts against its own + disposed hook. The event log says it outright: `hook_created` (round 0), + `hook_received` (round 0), `hook_conflict` (round 1), `hook_disposed` + (round 0). That is the shape upstream fixed as #2777 on the TypeScript side. +- **The two `FatalError` tests** — the step *lifecycle* is right, one attempt and + the run fails on it, but a thrown error does not keep its identity across the + event log. `step_failed.error` is written as text and comes back as + `RuntimeError`, so the workflow's `except` sees no `FatalError` and the failed + run's `errorCode` is `RuntimeError` where the driver wants `USER_ERROR`. + +Two things that used to be here are worth noting as gone, because both were +costing more than their own tests. A hook payload followed by a step used to +never create the step, which took out two fixtures and — since those were the +only fixtures producing an `encp` payload — also stopped this lane from checking +`encp` at all; both are fixed upstream and in the baseline now. And `encp` +itself, which is not the niche format its name suggests: on Vercel *every* hook +payload arrives sealed to the run's public key, because the driver resumes from +outside the run with no symmetric key, so a runtime that cannot read it fails +every hook fixture outright. ## What is missing This app is honest about being early. In rough order of how much it costs: -- **Most fixtures are simply not ported yet** — 66 tests across 52 fixtures. - They are not blocked on one thing anymore: the largest blocks are hooks (19 - tests, where vercel-py's `BaseHook.wait()` has a different shape than the - async-iterable hook the fixtures use), streams (11, where vercel-py now has - `read_stream` / `get_writable` and nothing here uses them yet), - `setAttributes` (9, no Python equivalent), and `FatalError` / - `RetryableError` (7 — `FatalError` is exported now, `RetryableError` has no - Python counterpart at all). +- **Most fixtures are still not ported** — 40 tests across 28 fixtures. What + changed is not the size but the shape: every one now names a missing API, where + the list used to include "ordinary porting". Two entries that used to be here + are gone for opposite reasons — streams, because `get_writable()` landed and + five fixtures use it now, including two that forward a writable into a child + run; and hooks-as-a-design-problem, because that premise was wrong (`HookEvent` + is both awaitable and async-iterable, so `for await (const p of hook)` ports + unchanged, and five hook fixtures are in). + + What is left, largest first: + + | tests | blocked on | + | --- | --- | + | 18 | hooks — `metadata` on `BaseHook.wait()` (6 alone), `getConflict()` (6 alone), both (4), other (2) | + | 9 | `setAttributes` — no Python equivalent at all, and spec version 4, which Python does not claim | + | 3 | distributed abort | + | 3 | `start()` or `fetch` from a workflow body — the sandbox denies both by design, so these are not gaps | + | 2 | error identity across the event log — the same cause as two of the three exemptions | + | 2 | invoking a step id that was never registered — see below, it is a gap of its own | + | 3 | one name each: `RetryableError` + `StepInfo.step_started_at`, `getWorkflowMetadata`, a step returning a `ReadableStream` | + | 1 | the webhook route | + + **`metadata` is the best-value item on that table by a distance**: 6 tests for + one keyword argument. `HookCreatedEventData.metadata` already exists on the + wire, the `Hook` entity carries it, and `HookCreatedEvent.payloads()` already + puts it in the key-resolution path — the only thing missing is somewhere to + *pass* it, since `BaseHook.wait()` takes `token` and nothing else and the flush + hard-codes `HookCreatedEventData(token=s.token)`. +- **An unregistered step id is a `KeyError`, not a failed step.** Found while + checking whether `stepNotRegistered{Catchable,Uncaught}` were portable: they are + not, and the reason is a gap rather than a missing API. + `Workflows._get_step` is `return self._steps[step_name]`, called *before* the + `try` that writes `step_failed`, so an unknown name raises out of the queue + handler and the delivery 500s. TypeScript raises a `StepNotRegisteredError` that + fails the step with "is not registered", which is what lets the workflow catch + it and the run complete — which is exactly what those two tests assert. Worth + more than its two tests: in a real app this is what a bundling mistake looks + like, and the difference between the two behaviours is a clear failed run versus + an endless redelivery loop. - **The `.well-known/workflow/v1` surface lives in `app.py`, not the SDK**, and reaching it needs three `vercel.workflow._internal` imports (`workflow_entrypoint`, `FLOW_ROUTE`, and the `HTTPRequest` base), none of diff --git a/workbench/python/e2e-conformance.json b/workbench/python/e2e-conformance.json index 55f8e276dc..20a3ab8fa8 100644 --- a/workbench/python/e2e-conformance.json +++ b/workbench/python/e2e-conformance.json @@ -10,10 +10,36 @@ "language": "python", "fixtures": [ "addTenWorkflow", + "errorFatalCatchable", + "errorRetryDisabled", + "errorRetryFatal", + "errorRetrySuccess", + "hookTokenReuseLoopWorkflow", + "hookWithSleepFinalStepWorkflow", + "hookWithSleepWorkflow", + "metadataFromHelperWorkflow", "nullByteWorkflow", + "outputStreamInsideStepWorkflow", + "outputStreamWorkflow", "parallelSleepWorkflow", "promiseAllWorkflow", + "promiseAnyWorkflow", + "promiseRaceStressTestWorkflow", + "promiseRaceWorkflow", + "retainedInterleavingWorkflow", "sleepInLoopWorkflow", - "sleepingWorkflow" - ] + "sleepWinsRaceWorkflow", + "sleepWithSequentialStepsWorkflow", + "sleepingWorkflow", + "spawnWorkflowFromStepWorkflow", + "stepWinsRaceWorkflow", + "utf8StreamWorkflow", + "writableForwardedFromStepWorkflow", + "writableForwardedFromWorkflowWorkflow" + ], + "unsupported": { + "hookTokenReuseLoopWorkflow - same run recreates a hook with the same token after dispose()": "vercel-py flushes `hook_created` and `hook_disposed` from one task group with no ordering between them, so a run conflicts against its own disposed hook \u2014 the shape upstream fixed as vercel/workflow#2777. Confirmed by the event log rather than read off the code: `hook_created` (round 0), `hook_received` (round 0), `hook_conflict` (round 1), `hook_disposed` (round 0) \u2014 round 1's registration was validated a beat before round 0's disposal landed. The failure the driver reports is downstream of that: the fixture returns early on the conflict, so the hook it was waiting for is gone and `waitForHook` 404s with `HookNotFoundError`.", + "FatalError fails immediately without retries": "The step lifecycle is right \u2014 `FatalError` burns exactly one attempt \u2014 but the run's `errorCode` is not. vercel-py writes `run_failed.code` as `type(e).__name__`, and by then the step's `FatalError` has been rehydrated into a `RuntimeError`, so the run carries `RuntimeError` where the test expects `USER_ERROR`.", + "FatalError can be caught and detected with FatalError.is()": "A thrown error does not keep its identity across the event log: vercel-py writes `step_failed.error` as text and the workflow side raises `RuntimeError()`, so the `FatalError` the step raised is no longer one by the time the body catches it. Upstream needs the serialized-error pipeline (`SerializedData` in place of the pre-#1851 `StructuredError` shape) before this and the two `RoundTrip` fixtures can pass." + } } diff --git a/workbench/python/pyproject.toml b/workbench/python/pyproject.toml index 681bf5e382..375ac10a32 100644 --- a/workbench/python/pyproject.toml +++ b/workbench/python/pyproject.toml @@ -24,6 +24,15 @@ dependencies = [ # `@workflow/world-vercel` encrypts it whenever it can resolve a per-run key, # which on a deployment it always can, with no opt-out. "vercel", + # And `vercel-workflow` as well, which the note above says to add "if they + # differ" — they do, as of this rev. A `[tool.uv.sources]` entry only applies + # to a direct requirement, so without this line the git pin below reaches + # `vercel` and every sibling still comes from PyPI. That was harmless while + # `4814d61f` and the 0.9.0 release were byte-identical; the rev pinned now is + # one commit past it and that commit is *entirely* inside + # `src/vercel-workflow`, so pinning `vercel` alone would have installed the + # release and silently tested the code without the fix. + "vercel-workflow", # vercel-py carries uvicorn as a dev-only dependency, so declare it here. "uvicorn>=0.30", ] @@ -65,7 +74,8 @@ dependencies = [ # builder then reject under `--locked`. `uv run` re-locks too, which is why # the `dev` script passes it as well. [tool.uv.sources] -vercel = { git = "https://github.com/vercel/vercel-py", rev = "4814d61fa2ac1074ea5d67aede8f159ba1daae9b", subdirectory = "src/vercel" } +vercel = { git = "https://github.com/vercel/vercel-py", rev = "6a10dd4c03457efce22c28986bafecbb925624a7", subdirectory = "src/vercel" } +vercel-workflow = { git = "https://github.com/vercel/vercel-py", rev = "6a10dd4c03457efce22c28986bafecbb925624a7", subdirectory = "src/vercel-workflow" } # How `@vercel/python` builds this app. Reached only because `vercel.json` # declares `pyproject.toml` as the build src, which puts the builder in diff --git a/workbench/python/uv.lock b/workbench/python/uv.lock index 8e7c91090b..e3a7be3d3f 100644 --- a/workbench/python/uv.lock +++ b/workbench/python/uv.lock @@ -568,7 +568,7 @@ wheels = [ [[package]] name = "vercel" version = "0.10.0" -source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=4814d61fa2ac1074ea5d67aede8f159ba1daae9b#4814d61fa2ac1074ea5d67aede8f159ba1daae9b" } +source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=6a10dd4c03457efce22c28986bafecbb925624a7#6a10dd4c03457efce22c28986bafecbb925624a7" } dependencies = [ { name = "anyio" }, { name = "httpx" }, @@ -687,7 +687,7 @@ wheels = [ [[package]] name = "vercel-workflow" version = "0.9.0" -source = { registry = "https://pypi.org/simple" } +source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel-workflow&rev=6a10dd4c03457efce22c28986bafecbb925624a7#6a10dd4c03457efce22c28986bafecbb925624a7" } dependencies = [ { name = "anyio" }, { name = "cbor2" }, @@ -699,10 +699,6 @@ dependencies = [ { name = "vercel-oidc" }, { name = "vercel-queue" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/df/91/e852e0803d45c06c0f493d1b96fe8b1e6a7246f971889d47182e7722b547/vercel_workflow-0.9.0.tar.gz", hash = "sha256:88f3483e2ea3595e02db48741af6de8e545dd574146453295d66a5a6a06d44da", size = 103391, upload-time = "2026-08-14T17:03:16.59Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/a8/9f/84892f5e2bc4d4d44a2823d1fd5dc1f72fe91e9921c419631d4fd56df71c/vercel_workflow-0.9.0-py3-none-any.whl", hash = "sha256:9da5ee97f0460e87ec185b107166c66e57d93e11d82e6953a22ed994888e93af", size = 112718, upload-time = "2026-08-14T17:03:15.258Z" }, -] [[package]] name = "websockets" @@ -827,10 +823,12 @@ source = { virtual = "." } dependencies = [ { name = "uvicorn" }, { name = "vercel" }, + { name = "vercel-workflow" }, ] [package.metadata] requires-dist = [ { name = "uvicorn", specifier = ">=0.30" }, - { name = "vercel", git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=4814d61fa2ac1074ea5d67aede8f159ba1daae9b" }, + { name = "vercel", git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=6a10dd4c03457efce22c28986bafecbb925624a7" }, + { name = "vercel-workflow", git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel-workflow&rev=6a10dd4c03457efce22c28986bafecbb925624a7" }, ] diff --git a/workbench/python/workflows/99_e2e.py b/workbench/python/workflows/99_e2e.py index 3cd1e0d51c..13129ff42e 100644 --- a/workbench/python/workflows/99_e2e.py +++ b/workbench/python/workflows/99_e2e.py @@ -23,10 +23,25 @@ """ import asyncio +import dataclasses +import json import random +import re import time - -from vercel.workflow import Workflows, sleep, time_ns +from typing import Any, Awaitable, TypeVar + +from vercel.workflow import ( + BaseHook, + FatalError, + Run, + WorkflowWritable, + Workflows, + get_step_metadata, + get_writable, + sleep, + start, + time_ns, +) # `as_vercel_job=False` because `app.py` wires the queue entrypoints itself: the # default constructor creates them and discards the HTTP handlers, and calling @@ -151,3 +166,739 @@ async def sleepInLoopWorkflow() -> dict: await sleep(sleepMs) return {"timestamps": timestamps, "totalElapsed": timestamps[-1] - timestamps[0]} + + +########################################################## +# Racing suspensions +# +# `Promise.race` and `Promise.any` have no asyncio spelling that takes bare +# awaitables, so the five race fixtures below share these two helpers. Both +# resolve ties by the order the awaitables were passed rather than by set +# iteration order — `asyncio.wait` returns a `set`, and a workflow body has to +# be deterministic across replays, so picking `next(iter(done))` would be a +# latent replay divergence the moment two steps land in the same turn. +# +# Nothing cancels the losers, matching JS: a race that resolves leaves the +# other steps running, the body returns, and `_run_in_loop` cancels the +# orphaned tasks on its way out. Their step invocations are already in flight +# and complete against a run that has finished, exactly as they do on the +# TypeScript side. + +_T = TypeVar("_T") + + +async def _race(*awaitables: Awaitable[_T]) -> _T: + """`Promise.race`: settle with the first to settle, error included.""" + tasks = [asyncio.ensure_future(a) for a in awaitables] + done, _pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for task in tasks: + if task in done: + return task.result() + raise AssertionError("asyncio.wait returned no completed task") + + +async def _any(*awaitables: Awaitable[_T]) -> _T: + """`Promise.any`: the first to *succeed*; failures are skipped.""" + tasks = [asyncio.ensure_future(a) for a in awaitables] + pending = set(tasks) + while pending: + done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED) + for task in tasks: + if task in done and task.exception() is None: + return task.result() + raise RuntimeError("all awaitables rejected") + + +########################################################## +# promiseRaceWorkflow — 99_e2e.ts:62 +# promiseAnyWorkflow — 99_e2e.ts:79 +# +# The counterpart to `promiseAllWorkflow`: three steps suspend in one turn, but +# the body resumes on the first one instead of all three. That is the case +# where replay and completion overlap — the run finishes while two step +# invocations are still outstanding. +# +# `promiseAnyWorkflow` additionally needs a step failure to be *skippable* +# rather than fatal. Python surfaces a failed step to the body as a +# `RuntimeError` carrying the step's error text (see the note on +# `errorRetryDisabled` below), which is all `_any` needs — it only asks whether +# the task raised. + + +@app.step +async def specificDelay(delay: int, v: str) -> str: + await asyncio.sleep(delay / 1000) + return v.upper() + + +@app.workflow +async def promiseRaceWorkflow() -> str: + return await _race( + specificDelay(10_000, "a"), + specificDelay(100, "b"), # "b" should always win + specificDelay(20_000, "c"), + ) + + +@app.step +async def stepThatFails() -> str: + raise FatalError("step failed") + + +@app.workflow +async def promiseAnyWorkflow() -> str: + return await _any( + stepThatFails(), + specificDelay(100, "b"), # "b" should always win + specificDelay(6_000, "c"), + ) + + +########################################################## +# sleepWinsRaceWorkflow — 99_e2e.ts:225 +# stepWinsRaceWorkflow — 99_e2e.ts:236 +# +# A race between the two suspension kinds. `sleep` resumes from a timer the +# world owns and a step resumes from an event the step handler writes, so these +# assert that the orchestrator resolves whichever lands first without waiting +# for the other — the driver bounds `durationMs` at 5s against a 10s loser. +# +# Both are listed under `unsupported` in `../e2e-conformance.json` even though +# they pick the right winner every time, and the 5s bound is why: these are the +# only two tests in the suite that bound a run's *elapsed* time from above, so +# they are the only two that notice the ~5s `world-local` waits before +# redelivering a first delivery the app 500'd on a run row it could not read +# yet. That is the same upstream gap as the `resilient start` entry, and all +# three exemptions come out together when it is fixed. + + +@app.step +async def delayMsStep(ms: int, label: str) -> str: + await asyncio.sleep(ms / 1000) + return label + + +async def _sleepThen(duration: str, label: str) -> str: + """`sleep(...).then(() => label)` — a plain coroutine, not a step.""" + await sleep(duration) + return label + + +@app.workflow +async def sleepWinsRaceWorkflow() -> dict: + startTime = time_ns() // 1_000_000 + winner = await _race(delayMsStep(10_000, "step"), _sleepThen("1s", "sleep")) + endTime = time_ns() // 1_000_000 + return {"winner": winner, "durationMs": endTime - startTime} + + +@app.workflow +async def stepWinsRaceWorkflow() -> dict: + startTime = time_ns() // 1_000_000 + winner = await _race(delayMsStep(1_000, "step"), _sleepThen("10s", "sleep")) + endTime = time_ns() // 1_000_000 + return {"winner": winner, "durationMs": endTime - startTime} + + +########################################################## +# promiseRaceStressTestWorkflow — 99_e2e.ts:478 +# +# Five steps 5s apart, raced and retired one at a time, so the body re-enters +# `_race` over a shrinking set across five separate replays. Each replay has to +# rebuild the same suspensions in the same order and match them to the events +# already in the log; a single misaligned slot shows up as a missing or +# duplicated entry in the returned list. + + +@app.step +async def promiseRaceStressTestDelayStep(dur: int, resp: int) -> int: + await asyncio.sleep(dur / 1000) + return resp + + +@app.workflow +async def promiseRaceStressTestWorkflow() -> list: + # Tasks rather than coroutines: unlike a JS promise, a coroutine can only + # be awaited once, and every loop iteration races the survivors again. + promises = { + i: asyncio.ensure_future(promiseRaceStressTestDelayStep(1000 * 5 * i, i)) + for i in range(5) + } + done = [] + + while promises: + res = await _race(*promises.values()) + done.append(res) + del promises[res] + + return done + + +########################################################## +# errorRetrySuccess — 99_e2e.ts:1279 +# errorRetryDisabled — 99_e2e.ts:1329 +# +# The two halves of the step retry policy: the default (`DEFAULT_MAX_RETRIES`, +# 3 in both SDKs) and `max_retries=0`. `get_step_metadata().attempt` is what +# makes them observable from inside the step, and it is the same 1-based +# counter the driver reads back off the step entity through the CLI. +# +# These two work because neither inspects the exception. `errorRetryDisabled` +# reads its attempt number back out of the message *text*, which is the one part +# of a thrown error that survives the event log — see the note on +# `errorFatalCatchable` below for what does not. +# +# `errorRetryCustomDelay`, the third fixture in the same TypeScript block, is +# not here at all: it needs `RetryableError(retryAfter=…)`, which vercel-py does +# not export, and `StepInfo` carries no `step_started_at` for the test's +# duration assertion either. + + +@app.step +async def retryUntilAttempt3() -> int: + attempt = get_step_metadata().attempt + if attempt < 3: + raise RuntimeError(f"Failed on attempt {attempt}") + return attempt + + +@app.workflow +async def errorRetrySuccess() -> dict: + return {"finalAttempt": await retryUntilAttempt3()} + + +@app.step(max_retries=0) +async def throwWithNoRetries() -> None: + raise RuntimeError(f"Failed on attempt {get_step_metadata().attempt}") + + +@app.workflow +async def errorRetryDisabled() -> dict: + try: + await throwWithNoRetries() + return {"failed": False, "attempt": None} + except Exception as e: + match = re.search(r"attempt (\d+)", str(e)) + return {"failed": True, "attempt": int(match.group(1)) if match else None} + + +########################################################## +# outputStreamWorkflow — 99_e2e.ts:355 +# outputStreamInsideStepWorkflow — 99_e2e.ts:403 +# utf8StreamWorkflow — 99_e2e.ts:441 +# +# Run-scoped streams, which are the one part of the protocol that does not +# travel through the event log: chunks are appended to a separate per-run log +# that a reader tails live. The driver reads them with `run.getReadable()`, so +# these fixtures assert that Python's framing and payload encoding are the ones +# `@workflow/core`'s `getDeserializeStream` expects. +# +# Both spellings of the same API are covered, because they take different paths +# through the SDK. `get_writable()` in the *workflow body* returns a +# `WorkflowStreamHandle` — the body replays and has no network, so it cannot +# write — and passing that handle into a step's arguments is what turns it into +# a writer, via the serialization layer. `get_writable()` in a *step* returns +# the writer directly. The two have to name the same stream for the workflow to +# be able to hand one to a step at all. +# +# A `bytes` chunk arrives on the TypeScript side as a `Uint8Array` and anything +# else as its devalue value, which is why the driver reads chunk 0 as binary +# and chunk 1 as an object. +# +# Nothing closes a stream implicitly in either SDK, and the driver asserts the +# reader sees `done` — hence the explicit `stepCloseOutputStream` at the end of +# each fixture. + + +@app.step +async def stepWithOutputStreamBinary(writable: WorkflowWritable, text: str) -> None: + await writable.write(text.encode()) + + +@app.step +async def stepWithOutputStreamObject(writable: WorkflowWritable, obj: Any) -> None: + await writable.write(obj) + + +@app.step +async def stepCloseOutputStream(writable: WorkflowWritable) -> None: + await writable.close() + + +@app.workflow +async def outputStreamWorkflow() -> str: + writable = get_writable() + namedWritable = get_writable(namespace="test") + await sleep("1s") + await stepWithOutputStreamBinary(writable, "Hello, world!") + await sleep("1s") + await stepWithOutputStreamBinary(namedWritable, "Hello, named stream!") + await sleep("1s") + await stepWithOutputStreamObject(writable, {"foo": "test"}) + await sleep("1s") + await stepWithOutputStreamObject(namedWritable, {"foo": "bar"}) + await sleep("1s") + await stepCloseOutputStream(writable) + await stepCloseOutputStream(namedWritable) + return "done" + + +@app.step +async def stepWithOutputStreamInsideStep(text: str) -> None: + await get_writable().write(text.encode()) + + +@app.step +async def stepWithNamedOutputStreamInsideStep(namespace: str, obj: Any) -> None: + await get_writable(namespace=namespace).write(obj) + + +@app.step +async def stepCloseOutputStreamInsideStep(namespace: str | None = None) -> None: + await get_writable(namespace=namespace).close() + + +@app.workflow +async def outputStreamInsideStepWorkflow() -> str: + await sleep("1s") + await stepWithOutputStreamInsideStep("Hello from step!") + await sleep("1s") + await stepWithNamedOutputStreamInsideStep( + "step-ns", {"message": "Hello from named stream in step!"} + ) + await sleep("1s") + await stepWithOutputStreamInsideStep("Second message") + await sleep("1s") + await stepWithNamedOutputStreamInsideStep("step-ns", {"counter": 42}) + await sleep("1s") + await stepCloseOutputStreamInsideStep() + await stepCloseOutputStreamInsideStep("step-ns") + return "done" + + +@app.step +async def stepWriteUtf8Text(writable: WorkflowWritable, text: str) -> None: + await writable.write(text.encode()) + + +@app.step +async def stepWriteUtf8Json(writable: WorkflowWritable, value: Any) -> None: + await writable.write(json.dumps(value, ensure_ascii=False).encode()) + + +@app.workflow +async def utf8StreamWorkflow() -> str: + writable = get_writable() + await sleep("1s") + await stepWriteUtf8Text(writable, "Hello, world!") + await stepWriteUtf8Text(writable, "Café — naïve résumé") + await stepWriteUtf8Text(writable, "你好，世界！🌍✨") + await stepWriteUtf8Text(writable, "مرحبا بالعالم") + await stepWriteUtf8Json(writable, {"greeting": "안녕하세요", "emoji": "🎉"}) + await stepCloseOutputStream(writable) + return "done" + + +########################################################## +# errorRetryFatal — 99_e2e.ts:1293 +# errorFatalCatchable — 99_e2e.ts:1347 +# +# `FatalError` is the one retry-control error vercel-py exports, and the step +# handler honours it: `fatal or attempt >= max_retries + 1` is what decides +# whether to write `step_failed` instead of `step_retrying`, so a step that +# raises it burns exactly one attempt. +# +# Both tests are under `unsupported`, and for the half of each that Python +# cannot do rather than the half it can. What works is the lifecycle above — the +# one attempt, and the run failing on it. What does not is the error's +# *identity*: `step_failed.error` is written as text and comes back to the +# workflow body as `RuntimeError()`, so the `except` below sees no +# `FatalError`, and the resulting `run_failed.code` is `type(e).__name__` on +# that `RuntimeError` where the driver expects `USER_ERROR`. Upstream needs the +# serialized-error pipeline (`SerializedData` in place of the pre-#1851 +# `StructuredError` shape); the two `RoundTrip` fixtures in the same TypeScript +# block wait on the same change and are not ported at all, since they assert on +# the cause chain and there would be nothing to assert. + + +@app.step +async def throwFatalError() -> None: + raise FatalError("Fatal step error") + + +@app.workflow +async def errorRetryFatal() -> str: + await throwFatalError() + return "never reached" + + +@app.workflow +async def errorFatalCatchable() -> dict: + try: + await throwFatalError() + return {"caught": False, "isFatal": False} + except Exception as e: + return {"caught": True, "isFatal": isinstance(e, FatalError)} + + +########################################################## +# metadataFromHelperWorkflow — 99_e2e.ts:3148 +# +# Upstream's #1577 regression test: the metadata accessors have to work from a +# helper defined at module level rather than inline in the step body. The +# mechanism differs — `AsyncLocalStorage` there, a `contextvars.ContextVar` +# here — but the failure mode it guards against is the same one, a context +# that only propagates as far as the decorated function. +# +# Only the step half is checked. `getStepMetadata()`'s counterpart +# `getWorkflowMetadata()` has no Python equivalent, so `workflowRunId` comes +# off `StepInfo.run_id`, which is the same run id the TypeScript fixture reads +# out of the workflow metadata. That also makes `workflowAndStepMetadataWorkflow` +# — which asserts the two metadata objects against each other — unportable for +# now, so it is not in this file. + + +async def _withStrictMetadataCheck(fn): + stepMetadata = get_step_metadata() + return await fn(), stepMetadata + + +@app.step +async def metadataHelperStep(label: str) -> dict: + async def _produce() -> str: + return label + + _result, stepMetadata = await _withStrictMetadataCheck(_produce) + + return { + "label": label, + "workflowRunId": stepMetadata.run_id, + "stepId": stepMetadata.step_id, + "attempt": stepMetadata.attempt, + } + + +@app.workflow +async def metadataFromHelperWorkflow(label: str) -> dict: + return await metadataHelperStep(label) + + +########################################################## +# spawnWorkflowFromStepWorkflow — 99_e2e.ts:1112 +# +# A run that starts another run. `start()` is a world write, so it can only +# happen in a step — the workflow body replays and its sandbox has no network, +# which is the same restriction the TypeScript fixture states in a comment. +# Waiting for the child is a step for the same reason. +# +# `Run(run_id).return_value()` polls the child's status; the TypeScript +# `getRun(runId).returnValue` is the same shape. Both hold the parent's step +# open for as long as the child takes, which is the caveat the TS fixture's +# `fibonacciWorkflow` neighbour documents at length — worth remembering before +# porting that one, since its recursion needs the worker pool to be deep enough +# for every waiting parent. + + +@app.step +async def doubleValue(value: int) -> int: + return value * 2 + + +@app.workflow +async def childWorkflow(value: int) -> dict: + return {"childResult": await doubleValue(value), "originalValue": value} + + +@app.step +async def spawnChildWorkflow(value: int) -> str: + childRun = await start(childWorkflow, value) + return childRun.run_id + + +@app.step +async def awaitWorkflowResult(runId: str) -> Any: + return await Run(runId).return_value() + + +@app.workflow +async def spawnWorkflowFromStepWorkflow(inputValue: int) -> dict: + childRunId = await spawnChildWorkflow(inputValue) + childResult = await awaitWorkflowResult(childRunId) + return { + "parentInput": inputValue, + "childRunId": childRunId, + "childResult": childResult, + } + + +########################################################## +# hookWithSleepWorkflow — 99_e2e.ts:2960 +# hookWithSleepFinalStepWorkflow — 99_e2e.ts:2995 +# hookTokenReuseLoopWorkflow — 99_e2e.ts:990 +# sleepWithSequentialStepsWorkflow — 99_e2e.ts:3071 +# +# The four fixtures in the TypeScript file's hook/sleep-interaction cluster, +# and the first hooks on this side. The bucket they used to sit in was labelled +# "vercel-py's `BaseHook.wait()` has a different shape than the async-iterable +# the fixtures use", which turns out to be wrong: `HookEvent` implements both +# `__await__` (one payload) and `__aiter__` / `__anext__` (a stream of them), so +# `for await (const p of hook)` ports to `async for payload in hook` directly. +# +# What is really different is the payload type. `Hook.set_result` requires the +# class handed to `wait()` to be a dataclass or a pydantic model, and calls +# `hook_cls(**raw)` on the plain JSON the resumer sent — so the port is a +# dataclass with a default per optional field, and the fixture's structural +# type becomes a declared one. The declaration has to stay loose in the same +# places the TypeScript type is optional: the driver resumes with `{type, id}` +# on one payload and `{type, done}` on another, and a required field would +# raise on whichever call omitted it. +# +# Three translation traps, each of which cost a debugging round here: +# +# - **`using hook` is not `try/finally`.** A Python workflow body unwinds +# through a `_SuspendException` on *every* suspension, so a `finally` around +# an `await` runs once per turn rather than once at scope exit. Disposing a +# hook there deletes the suspension before the orchestrator can flush its +# `hook_created`, and the run stalls with no hook for the driver to resume. +# Dispose on the normal path only. +# - **`void sleep('1d')`** is `asyncio.ensure_future(sleep("1d"))`. The wait is +# created and never completes; the body returns first and the orphaned task +# is cancelled with the loop. +# - **A step takes the payload as a dict**, not as the dataclass: keeping the +# step signature `dict` avoids registering a serializer for a type that only +# exists to satisfy `set_result`. +# +# `sleepWithSequentialStepsWorkflow` is the cluster's control and has no hook in +# it at all — a fire-and-forget sleep plus three sequential steps. It is ported +# here rather than with the other sleep fixtures because its whole purpose is to +# be read next to the two above: it passes, which is what makes their failures +# specific to hooks rather than to a pending wait. +# +# Two of the four are under `unsupported`, and both look like real orchestrator +# defects rather than missing API — see `../e2e-conformance.json` for what was +# observed. + + +@dataclasses.dataclass +class SleepHookPayload(BaseHook): + type: str + id: int | None = None + done: bool | None = None + + +@app.step +async def processPayload(payload: dict) -> dict: + return {"processed": True, "type": payload["type"], "id": payload.get("id")} + + +@app.workflow +async def hookWithSleepWorkflow(token: str) -> list: + hook = SleepHookPayload.wait(token=token) + + # Concurrent sleep that won't complete during the test + asyncio.ensure_future(sleep("1d")) + + results = [] + async for payload in hook: + results.append(await processPayload(dataclasses.asdict(payload))) + if payload.done: + break + + hook.dispose() + return results + + +@app.workflow +async def hookWithSleepFinalStepWorkflow(token: str) -> dict: + hook = SleepHookPayload.wait(token=token) + asyncio.ensure_future(sleep("1d")) + + seen = [] + finalResult = None + async for payload in hook: + if payload.id is not None: + seen.append(payload.id) + if payload.done: + finalResult = await processPayload(dataclasses.asdict(payload)) + break + + hook.dispose() + return {"seen": seen, "finalResult": finalResult} + + +@dataclasses.dataclass +class ReuseHookPayload(BaseHook): + message: str + + +@app.workflow +async def hookTokenReuseLoopWorkflow(token: str, rounds: int) -> dict: + received = [] + for round in range(rounds): + hook = ReuseHookPayload.wait(token=token) + + # `hook.getConflict()` has no Python equivalent. A conflict is still + # observable, just later and as an exception: `HookConflictEvent` + # resolves the hook's future with a `RuntimeError`, so the same two + # outcomes come out of one `await` instead of two. + try: + payload = await hook + except RuntimeError as e: + if "already in use" not in str(e): + raise + return {"received": received, "conflictRound": round} + + received.append(payload.message) + hook.dispose() + + return {"received": received, "conflictRound": None} + + +@app.step +async def addNumbers(a: int, b: int) -> int: + return a + b + + +@app.workflow +async def sleepWithSequentialStepsWorkflow() -> dict: + shouldCancel = False + + async def _cancelAfterSleep() -> None: + nonlocal shouldCancel + await sleep("1d") + shouldCancel = True + + asyncio.ensure_future(_cancelAfterSleep()) + + a = await addNumbers(1, 2) + b = await addNumbers(a, 3) + c = await addNumbers(b, 4) + return {"a": a, "b": b, "c": c, "shouldCancel": shouldCancel} + + +########################################################## +# writableForwardedFromWorkflowWorkflow — 99_e2e.ts:3515 +# writableForwardedFromStepWorkflow — 99_e2e.ts:3540 +# +# A stream reference crossing a *run* boundary: the parent hands its writable to +# a child run as part of that run's input, and the driver then reads the bytes +# off the **parent's** stream. So the handle has to survive `start()`'s input +# serialization, arrive in another run's body, and still name the stream it came +# from rather than the child's own — which is why `WorkflowStreamHandle` carries +# a run id instead of deriving one from the ambient run. +# +# The two variants differ in where the parent's `get_writable()` is called, and +# they are not the same path through the SDK. Variant 1 calls it in the workflow +# body, so a *handle* is serialized into a step's arguments, revived there as a +# writer, and serialized again into the child's input. Variant 2 calls it inside +# the step that also calls `start()`, so what gets forwarded is the step-context +# writer directly. Both have to land on the same stream. +# +# `start()` lives in a step for the usual reason: it is a world write, and the +# workflow body replays with no network. The TypeScript fixture says the same +# thing in a comment, which is a good sign the restriction is protocol-shaped +# rather than Python-shaped. + + +@app.step +async def writeBytesToWritable(writable: WorkflowWritable, payload: str) -> None: + await writable.write(payload.encode()) + + +@app.workflow +async def writableForwardedChildWorkflow( + parentWritable: WorkflowWritable, payload: str +) -> str: + await writeBytesToWritable(parentWritable, payload) + return "child-done" + + +@app.step +async def startChildWithWorkflowWritable( + parentWritable: WorkflowWritable, payload: str +) -> str: + childRun = await start(writableForwardedChildWorkflow, parentWritable, payload) + # Let the child finish writing before the parent is allowed to close. + await childRun.return_value() + return childRun.run_id + + +@app.workflow +async def writableForwardedFromWorkflowWorkflow(payload: str) -> dict: + writable = get_writable() + childRunId = await startChildWithWorkflowWritable(writable, payload) + await stepCloseOutputStream(writable) + return {"childRunId": childRunId} + + +@app.step +async def startChildWithStepWritable(payload: str) -> str: + writable = get_writable() + childRun = await start(writableForwardedChildWorkflow, writable, payload) + await childRun.return_value() + await writable.close() + return childRun.run_id + + +@app.workflow +async def writableForwardedFromStepWorkflow(payload: str) -> dict: + return {"childRunId": await startChildWithStepWritable(payload)} + + +########################################################## +# retainedInterleavingWorkflow — 99_e2e.ts:264 +# +# Every suspension kind this app can produce, in one body, with the exact +# composite result asserted — so a dropped, duplicated or misordered boundary +# fails loudly rather than shifting a number nobody checks. +# +# Upstream wrote it for VM retention (`WORKFLOW_RETAINED_VM`): primitive step +# arguments keep the retained VM, a non-primitive argument demotes the boundary +# to a cold replay. Python has no such VM, so `unwrapValue`'s object argument is +# just an object argument here. That does not make the fixture pointless on this +# side — what the test actually asserts is that nine values come back right +# across a step / gather / race / sleep / hook interleaving, and that is the +# same claim in any language. +# +# The hook is created with a token and no metadata, which is the one hook shape +# vercel-py can express today, and it is awaited *concurrently with a step* — +# `asyncio.gather(hook, add(...))`, since `HookEvent` is awaitable. Dispose on +# the normal path only; see the hook cluster above for why `finally` would break +# it. + + +@app.step +async def unwrapValue(box: dict) -> int: + return box["value"] + + +@dataclasses.dataclass +class DeltaPayload(BaseHook): + delta: int + + +@app.workflow +async def retainedInterleavingWorkflow(token: str) -> dict: + a = await add(1, 2) + b = await unwrapValue({"value": a}) + c, d = await asyncio.gather(add(b, 10), add(b, 20)) + e, f = await asyncio.gather(unwrapValue({"value": c}), add(d, 1)) + winner = await _race(delayMsStep(100, "step"), _sleepThen("30s", "sleep")) + await sleep("1s") + + hook = DeltaPayload.wait(token=token) + payload, g = await asyncio.gather(hook, add(e + f, 100)) + h = await add(g, payload.delta) + hook.dispose() + + return { + "a": a, + "b": b, + "c": c, + "d": d, + "e": e, + "f": f, + "winner": winner, + "g": g, + "h": h, + }