From 8d9d81f15a8e3925f0c137f257e0f24ded91c8df Mon Sep 17 00:00:00 2001 From: Fantix King Date: Fri, 14 Aug 2026 17:19:05 -0400 Subject: [PATCH 1/3] [e2e] Port 18 more fixtures, and pin the hook-resume fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `9 passed | 128 skipped` → `29 passed | 108 skipped` of 137. 6 fixtures → 24, and 3 exempted tests. Rebased onto main rather than replayed: the base PR merged with a newer pin and a different `[tool.uv.sources]` shape than the one this branch walked through, so the intermediate pin bumps described a journey that no longer exists. Their findings live in the README and in the upstream gap notes; only the endpoint is a commit now. **Fixtures.** Races (5) — `Promise.race` / `Promise.any` have no asyncio spelling that takes bare awaitables, so two helpers supply them, resolving ties by argument order rather than by `asyncio.wait`'s set iteration order, which would be a latent replay divergence the first time two steps landed in the same turn. Streams (3), both spellings: a handle the workflow body passes into a step's arguments, and `get_writable()` called inside the step. Retries (4), child runs, metadata-from-helper, and the hook/sleep cluster (4). **Hooks were never the shape problem.** The bucket was parked on "`BaseHook.wait()` has a different shape than the async-iterable hook the fixtures use". `HookEvent` implements `__await__` *and* `__aiter__` / `__anext__`, so `for await (const p of hook)` ports to `async for payload in hook` unchanged. The real constraint is narrower: `Hook.set_result` calls `hook_cls(**raw)`, so a fixture's structural TypeScript type becomes a declared dataclass — loose in exactly the fields the fixture leaves optional, since the driver resumes with `{type, id}` on one payload and `{type, done}` on the next. One translation trap earned its comment in the fixture file: `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 instead of 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. **Pin.** `6a10dd4`, one commit above main for the hook-resume delivery fix. Both `vercel` *and* `vercel-workflow` are git-pinned, which the existing note in `pyproject.toml` says to check for and which matters here for the first time: that commit is entirely inside `src/vercel-workflow`, and a source entry only applies to a direct requirement, so pinning `vercel` alone would have installed PyPI's 0.9.0 and silently tested the code without the fix. Verified with the diff recipe that note documents — the installed tree matches the rev byte for byte. **Three exemptions left, two causes, both found by running this.** `hookTokenReuseLoopWorkflow` is now confirmed from the event log rather than inferred from the source: `hook_created` (round 0), `hook_received` (round 0), `hook_conflict` (round 1), `hook_disposed` (round 0) — the disposal lands after the next registration is validated, so the run conflicts against its own disposed hook, which is vercel/workflow#2777's shape. The two `FatalError` tests need a thrown error to survive the event log with its identity intact. Stability: 4 consecutive full-suite runs plus 6 targeted runs of the hook pair that used to flake, all green. Co-Authored-By: Claude Opus 5 (1M context) --- workbench/python/README.md | 79 +++- workbench/python/e2e-conformance.json | 27 +- workbench/python/pyproject.toml | 12 +- workbench/python/uv.lock | 12 +- workbench/python/workflows/99_e2e.py | 627 +++++++++++++++++++++++++- 5 files changed, 726 insertions(+), 31 deletions(-) diff --git a/workbench/python/README.md b/workbench/python/README.md index d7ba8975ad..cd4f649c2f 100644 --- a/workbench/python/README.md +++ b/workbench/python/README.md @@ -269,30 +269,71 @@ 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: **29 passing, 108 skipped, of 137** on `world-local` — 24 +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** — roughly 39 tests across 31 fixtures. + What changed is not the size but the shape: every one of them now names a + missing API, where the list used to include "ordinary porting". Streams are + gone from it entirely (`get_writable` landed and three fixtures use it), and so + is the largest old entry — hooks were said to need a design decision because + `BaseHook.wait()` "has a different shape than the async-iterable hook the + fixtures use", which is simply not so: `HookEvent` implements `__await__` *and* + `__aiter__` / `__anext__`, so `for await (const p of hook)` ports to + `async for payload in hook` unchanged, and four hook fixtures are in on that + basis. + + What is left, by size: hooks again (11 fixtures / 15 tests) but for two named + things — `metadata` on `BaseHook.wait()`, where the wire model + `HookCreatedEventData` already carries the field and only the authoring API + cannot fill it, and a `getConflict()` equivalent, which also needs + `hook_conflict`'s `conflictingRunId` to stop being parsed away. Then + `setAttributes` (9 tests, no Python equivalent at all, and spec version 4, + which Python does not claim), distributed abort (3), and single names: + `RetryableError`, `getWorkflowMetadata`, a `ReadableStream` returned from a + step, and invoking a step id that was never registered. + + Three fixtures are unportable by design rather than by gap: `fetchWorkflow` and + the two that call `start()` from a workflow body do what the Python sandbox + denies on purpose. `spawnWorkflowFromStepWorkflow` is the supported shape and is + ported. - **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..a6a0342b21 100644 --- a/workbench/python/e2e-conformance.json +++ b/workbench/python/e2e-conformance.json @@ -10,10 +10,33 @@ "language": "python", "fixtures": [ "addTenWorkflow", + "errorFatalCatchable", + "errorRetryDisabled", + "errorRetryFatal", + "errorRetrySuccess", + "hookTokenReuseLoopWorkflow", + "hookWithSleepFinalStepWorkflow", + "hookWithSleepWorkflow", + "metadataFromHelperWorkflow", "nullByteWorkflow", + "outputStreamInsideStepWorkflow", + "outputStreamWorkflow", "parallelSleepWorkflow", "promiseAllWorkflow", + "promiseAnyWorkflow", + "promiseRaceStressTestWorkflow", + "promiseRaceWorkflow", "sleepInLoopWorkflow", - "sleepingWorkflow" - ] + "sleepWinsRaceWorkflow", + "sleepWithSequentialStepsWorkflow", + "sleepingWorkflow", + "spawnWorkflowFromStepWorkflow", + "stepWinsRaceWorkflow", + "utf8StreamWorkflow" + ], + "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..8abc91a6c8 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,611 @@ 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} From be423be5ade6a7bb775f0e3d2e212338d712663c Mon Sep 17 00:00:00 2001 From: Fantix King Date: Fri, 14 Aug 2026 22:11:14 -0400 Subject: [PATCH 2/3] [e2e] Port the three fixtures that needed nothing from upstream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `29 passed | 108 skipped` → `32 passed | 105 skipped`. 24 fixtures → 27, no pin change, no SDK change. Answering "what is portable *today*" rather than "what is blocked": - **`writableForwardedFromWorkflowWorkflow` / `writableForwardedFromStepWorkflow`** — new on main, and the most interesting of the three: 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 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. It does — `WorkflowStreamHandle` carries a run id rather than deriving one from the ambient run. Both variants pass, and they are genuinely different paths: one serializes a workflow-context handle into a step's arguments and forwards it from there, the other forwards the step-context writer directly. - **`retainedInterleavingWorkflow`** — every suspension kind this app can produce in one body, with the exact composite result asserted. Upstream wrote it for VM retention, which Python does not have; what the test actually asserts is that nine values survive a step / gather / race / sleep / hook interleaving, and that claim is language-independent. It is also the first fixture here to await a hook *concurrently with a step* (`asyncio.gather(hook, add(...))`, since `HookEvent` is awaitable). The fourth candidate turned out to be a gap, not a port, and is worth more than the two tests it costs. `stepNotRegistered{Catchable,Uncaught}` invoke a step id that was never registered; `Workflows._get_step` is `return self._steps[step_name]`, called *before* the `try` that writes `step_failed`, so an unknown name raises `KeyError` out of the queue handler and the delivery 500s and redelivers. TypeScript raises `StepNotRegisteredError`, fails the step with "is not registered", and lets the workflow catch it — which is what both tests assert. In a real app this is what a bundling mistake looks like, and the difference is a clear failed run versus an endless redelivery loop. Recorded in the README. Remaining backlog is 40 tests across 28 fixtures, and the README now carries it as a table ordered by what blocks it. The top row is hooks (18), and within it `metadata` on `BaseHook.wait()` is 6 tests for one keyword argument — the field already exists on the wire model, on the `Hook` entity, and in `payloads()`; there is just nowhere to pass it. Three full-suite runs green. Co-Authored-By: Claude Opus 5 (1M context) --- workbench/python/README.md | 66 +++++++------ workbench/python/e2e-conformance.json | 5 +- workbench/python/workflows/99_e2e.py | 128 ++++++++++++++++++++++++++ 3 files changed, 172 insertions(+), 27 deletions(-) diff --git a/workbench/python/README.md b/workbench/python/README.md index cd4f649c2f..ca05f85c78 100644 --- a/workbench/python/README.md +++ b/workbench/python/README.md @@ -275,7 +275,7 @@ stops being true fails the run instead of quietly skipping, so growing the file the only way to move. `ConformanceConfig` in `packages/core/e2e/utils.ts` spells out each direction. -Current baseline: **29 passing, 108 skipped, of 137** on `world-local` — 24 +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 @@ -309,31 +309,45 @@ every hook fixture outright. This app is honest about being early. In rough order of how much it costs: -- **Most fixtures are still not ported** — roughly 39 tests across 31 fixtures. - What changed is not the size but the shape: every one of them now names a - missing API, where the list used to include "ordinary porting". Streams are - gone from it entirely (`get_writable` landed and three fixtures use it), and so - is the largest old entry — hooks were said to need a design decision because - `BaseHook.wait()` "has a different shape than the async-iterable hook the - fixtures use", which is simply not so: `HookEvent` implements `__await__` *and* - `__aiter__` / `__anext__`, so `for await (const p of hook)` ports to - `async for payload in hook` unchanged, and four hook fixtures are in on that - basis. - - What is left, by size: hooks again (11 fixtures / 15 tests) but for two named - things — `metadata` on `BaseHook.wait()`, where the wire model - `HookCreatedEventData` already carries the field and only the authoring API - cannot fill it, and a `getConflict()` equivalent, which also needs - `hook_conflict`'s `conflictingRunId` to stop being parsed away. Then - `setAttributes` (9 tests, no Python equivalent at all, and spec version 4, - which Python does not claim), distributed abort (3), and single names: - `RetryableError`, `getWorkflowMetadata`, a `ReadableStream` returned from a - step, and invoking a step id that was never registered. - - Three fixtures are unportable by design rather than by gap: `fetchWorkflow` and - the two that call `start()` from a workflow body do what the Python sandbox - denies on purpose. `spawnWorkflowFromStepWorkflow` is the supported shape and is - ported. +- **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 a6a0342b21..20a3ab8fa8 100644 --- a/workbench/python/e2e-conformance.json +++ b/workbench/python/e2e-conformance.json @@ -26,13 +26,16 @@ "promiseAnyWorkflow", "promiseRaceStressTestWorkflow", "promiseRaceWorkflow", + "retainedInterleavingWorkflow", "sleepInLoopWorkflow", "sleepWinsRaceWorkflow", "sleepWithSequentialStepsWorkflow", "sleepingWorkflow", "spawnWorkflowFromStepWorkflow", "stepWinsRaceWorkflow", - "utf8StreamWorkflow" + "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`.", diff --git a/workbench/python/workflows/99_e2e.py b/workbench/python/workflows/99_e2e.py index 8abc91a6c8..13129ff42e 100644 --- a/workbench/python/workflows/99_e2e.py +++ b/workbench/python/workflows/99_e2e.py @@ -774,3 +774,131 @@ async def _cancelAfterSleep() -> None: 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, + } From e80d802c6ce87157e7ad8e647c81c91668dc0ae7 Mon Sep 17 00:00:00 2001 From: Fantix King Date: Sat, 15 Aug 2026 11:43:52 -0400 Subject: [PATCH 3/3] [e2e] Pin hook metadata and port the three fixtures it unblocked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `32 passed | 105 skipped` → `36 passed | 101 skipped`. 27 fixtures → 30. Pin moves to `c64712f`, the metadata commit stacked on lazy hook resume. `metadata` was the best-value item on the backlog table — 6 tests for one keyword argument — and it delivered 5 of them. `hookWorkflow` (2), `hookDisposeTestWorkflow` (1) and one of `hookCleanupTestWorkflow`'s three (1), plus one more elsewhere. Metadata is how a run tells its resumer what it is waiting for, and the suite leans on it hard: `hookWorkflow`'s driver resumes with `customData: hook.metadata?.customData` and then asserts the workflow saw that exact value, so a missing field does not weaken the test, it fails it. One translation detail is load-bearing and has a comment on it. `hookWorkflow`'s payload is a **pydantic model** where the other two are dataclasses, because the driver sends `done` only on the last payload and the test asserts the first two come back with `done` *absent* — `undefined`, not `false` and not `null`. A dataclass materializes every optional field, so `asdict` would report `done: None` and the assertion would fail on the difference between "not sent" and "sent as null". `model_dump(exclude_unset=True)` reproduces what the resumer actually sent, which is the property the test is about. The sixth test is exempted, and diagnosing it turned up a defect worth more than the test. `concurrent hook token conflict` expects the second run to *fail*; it hangs instead. The world answers the loser's `hook_created` with a `hook_conflict` event rather than an `EntityConflictError`, and the flush's `create_hook` closure ignores the response — so the hook is still in `context.suspensions`, the "no suspensions" re-invoke does not fire, and with no pending `Wait` the handler returns `None`. Nothing re-delivers. The log ends at `run_created`, `run_started`, `hook_conflict` and the run sits in `running` forever. vercel-py *has* a `HookConflictEvent` consumer that raises exactly the message the test wants; it never gets a delivery to run in. Two smaller things sit behind that: the error's identity has to survive the log, and `HookConflictEventData` keeps only `token` where TypeScript also carries `conflictingRunId`. Backlog is 35 tests across 25 fixtures. The hooks row is down from 18 to 12 and is now all `getConflict()`, which is not metadata's size: it needs a suspension kind Python does not have — commit the registration and resume without waiting for a payload — and the adopt/supersede fixtures need `conflictingRunId` kept first. Two full-suite runs green. Co-Authored-By: Claude Opus 5 (1M context) --- workbench/python/README.md | 36 +++++++---- workbench/python/e2e-conformance.json | 6 +- workbench/python/pyproject.toml | 4 +- workbench/python/uv.lock | 8 +-- workbench/python/workflows/99_e2e.py | 89 +++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 20 deletions(-) diff --git a/workbench/python/README.md b/workbench/python/README.md index ca05f85c78..aea800530a 100644 --- a/workbench/python/README.md +++ b/workbench/python/README.md @@ -275,20 +275,30 @@ stops being true fails the run instead of quietly skipping, so growing the file 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 +Current baseline: **36 passing, 101 skipped, of 137** on `world-local` — 30 +fixtures and 4 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: +The four exemptions are three upstream causes, and all three 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. +- **`concurrent hook token conflict`** — a hook token conflict does not fail the + loser, it *hangs* it. The world answers the loser's `hook_created` with a + `hook_conflict` event rather than an error, the flush ignores the response, and + with the hook still suspended and no `Wait` pending the handler simply returns. + Nothing re-delivers, so the log ends at `run_created`, `run_started`, + `hook_conflict` and the run sits in `running` — vercel-py *has* a + `HookConflictEvent` consumer that raises the right message, it just never gets + a delivery to run in. Two smaller things sit behind that one: the error's + identity has to survive the log, and `HookConflictEventData` keeps only `token` + where TypeScript also carries `conflictingRunId`. - **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 @@ -309,7 +319,7 @@ every hook fixture outright. This app is honest about being early. In rough order of how much it costs: -- **Most fixtures are still not ported** — 40 tests across 28 fixtures. What +- **Most fixtures are still not ported** — 35 tests across 25 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 @@ -322,21 +332,21 @@ This app is honest about being early. In rough order of how much it costs: | tests | blocked on | | --- | --- | - | 18 | hooks — `metadata` on `BaseHook.wait()` (6 alone), `getConflict()` (6 alone), both (4), other (2) | + | 12 | hooks — `getConflict()` (6 alone), `getConflict()` + a conflicting run reaching a terminal state (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 | error identity across the event log — the same cause as two of the four 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)`. + `metadata` used to head that table at 6 tests for one keyword argument; it + landed, and it bought 5 (the sixth needs the conflict handling above). What is + left of the hooks row is `getConflict()`, which is not the same size: it needs a + suspension kind Python does not have — "commit the hook registration and resume + without waiting for a payload" — and, for the four fixtures that adopt or + supersede an owner, it needs `conflictingRunId` to stop being parsed away first. - **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. diff --git a/workbench/python/e2e-conformance.json b/workbench/python/e2e-conformance.json index 20a3ab8fa8..8cf7673831 100644 --- a/workbench/python/e2e-conformance.json +++ b/workbench/python/e2e-conformance.json @@ -14,9 +14,12 @@ "errorRetryDisabled", "errorRetryFatal", "errorRetrySuccess", + "hookCleanupTestWorkflow", + "hookDisposeTestWorkflow", "hookTokenReuseLoopWorkflow", "hookWithSleepFinalStepWorkflow", "hookWithSleepWorkflow", + "hookWorkflow", "metadataFromHelperWorkflow", "nullByteWorkflow", "outputStreamInsideStepWorkflow", @@ -40,6 +43,7 @@ "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." + "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.", + "concurrent hook token conflict - two workflows cannot use the same hook token simultaneously": "Three things, and the first is the one that matters: a hook token conflict does not fail the run, it *hangs* it. The world answers the loser's `hook_created` with a `hook_conflict` event rather than an `EntityConflictError`, and the flush's `create_hook` closure ignores the response; the hook is still in `context.suspensions` so the 'no suspensions' re-invoke does not fire, and with no pending `Wait` the handler returns `None`. Nothing re-delivers, so the log ends at `run_created`, `run_started`, `hook_conflict` and the run sits in `running` forever \u2014 vercel-py's `HookConflictEvent` consumer, which raises the right message, never gets a delivery to run in. Then, once the run does fail: the test also needs `HookConflictError.is(cause)` to survive the event log (the `FatalError` exemptions' cause) and `cause.conflictingRunId`, which `HookConflictEventData` parses away \u2014 it keeps only `token` where TypeScript carries both." } } diff --git a/workbench/python/pyproject.toml b/workbench/python/pyproject.toml index 375ac10a32..dddf38f2d2 100644 --- a/workbench/python/pyproject.toml +++ b/workbench/python/pyproject.toml @@ -74,8 +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 = "6a10dd4c03457efce22c28986bafecbb925624a7", subdirectory = "src/vercel" } -vercel-workflow = { git = "https://github.com/vercel/vercel-py", rev = "6a10dd4c03457efce22c28986bafecbb925624a7", subdirectory = "src/vercel-workflow" } +vercel = { git = "https://github.com/vercel/vercel-py", rev = "c64712f942e6979578257429bca9bf33809819d1", subdirectory = "src/vercel" } +vercel-workflow = { git = "https://github.com/vercel/vercel-py", rev = "c64712f942e6979578257429bca9bf33809819d1", 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 e3a7be3d3f..a74334b455 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=6a10dd4c03457efce22c28986bafecbb925624a7#6a10dd4c03457efce22c28986bafecbb925624a7" } +source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=c64712f942e6979578257429bca9bf33809819d1#c64712f942e6979578257429bca9bf33809819d1" } dependencies = [ { name = "anyio" }, { name = "httpx" }, @@ -687,7 +687,7 @@ wheels = [ [[package]] name = "vercel-workflow" version = "0.9.0" -source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel-workflow&rev=6a10dd4c03457efce22c28986bafecbb925624a7#6a10dd4c03457efce22c28986bafecbb925624a7" } +source = { git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel-workflow&rev=c64712f942e6979578257429bca9bf33809819d1#c64712f942e6979578257429bca9bf33809819d1" } dependencies = [ { name = "anyio" }, { name = "cbor2" }, @@ -829,6 +829,6 @@ dependencies = [ [package.metadata] requires-dist = [ { name = "uvicorn", specifier = ">=0.30" }, - { 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" }, + { name = "vercel", git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel&rev=c64712f942e6979578257429bca9bf33809819d1" }, + { name = "vercel-workflow", git = "https://github.com/vercel/vercel-py?subdirectory=src%2Fvercel-workflow&rev=c64712f942e6979578257429bca9bf33809819d1" }, ] diff --git a/workbench/python/workflows/99_e2e.py b/workbench/python/workflows/99_e2e.py index 13129ff42e..f9c09e4a06 100644 --- a/workbench/python/workflows/99_e2e.py +++ b/workbench/python/workflows/99_e2e.py @@ -30,6 +30,8 @@ import time from typing import Any, Awaitable, TypeVar +import pydantic + from vercel.workflow import ( BaseHook, FatalError, @@ -902,3 +904,90 @@ async def retainedInterleavingWorkflow(token: str) -> dict: "g": g, "h": h, } + + +########################################################## +# hookWorkflow — 99_e2e.ts:126 +# hookCleanupTestWorkflow — 99_e2e.ts:628 +# hookDisposeTestWorkflow — 99_e2e.ts:946 +# +# The three fixtures that needed hook *metadata* and nothing else. Metadata is +# how a run tells its resumer what it is waiting for: attached once when the hook +# is registered, read back off the hook entity rather than out of a payload. The +# suite leans on it hard — `hookWorkflow`'s driver resumes with +# `customData: hook.metadata?.customData` and then asserts the workflow saw that +# exact value, so a missing metadata field does not weaken the test, it fails it. +# +# `hookWorkflow`'s payload is a **pydantic model** rather than a dataclass, and +# that is load-bearing rather than a style choice. The driver sends `done` only +# on the last payload, and the test asserts the first two come back with `done` +# *absent* — `undefined`, not `false` and not `null`. A dataclass materializes +# every optional field, so `dataclasses.asdict` would report `done: None` and the +# assertion would fail on the difference between "not sent" and "sent as null". +# `model_dump(exclude_unset=True)` reproduces what the resumer actually sent, +# which is the property the test is really about. The other two fixtures have no +# optional fields and stay dataclasses. +# +# `using hook` becomes an explicit `dispose()` on the normal path — never a +# `finally`; see the hook cluster above for why. Where that dispose lands matters +# only in `hookDisposeTestWorkflow`, and there it is the whole point: it releases +# the token *before* the 5s sleep, so another run can claim it while this one is +# still going. In the other two the run completes right after, which frees the +# token anyway, so the call is a formality kept for symmetry with the fixture. + + +class HookPayload(BaseHook, pydantic.BaseModel): + message: str + customData: str + done: bool | None = None + + +@app.workflow +async def hookWorkflow(token: str, customData: str) -> list: + hook = HookPayload.wait(token=token, metadata={"customData": customData}) + + payloads = [] + async for payload in hook: + payloads.append(payload.model_dump(exclude_unset=True)) + if payload.done: + break + + hook.dispose() + return payloads + + +@dataclasses.dataclass +class MessagePayload(BaseHook): + message: str + customData: str + + +@app.workflow +async def hookCleanupTestWorkflow(token: str, customData: str) -> dict: + hook = MessagePayload.wait(token=token, metadata={"customData": customData}) + payload = await hook + hook.dispose() + return { + "message": payload.message, + "customData": payload.customData, + "hookCleanupTestData": "workflow_completed", + } + + +@app.workflow +async def hookDisposeTestWorkflow(token: str, customData: str) -> dict: + hook = MessagePayload.wait(token=token, metadata={"customData": customData}) + payload = await hook + message, customDataResult = payload.message, payload.customData + + # Releases the token here rather than at run completion, which is what lets + # the test's second run claim it while this one is still sleeping. + hook.dispose() + await sleep("5s") + + return { + "message": message, + "customData": customDataResult, + "disposed": True, + "hookDisposeTestData": "workflow_completed", + }