From 66451339f05e95d73b31960168c72f2a777f3248 Mon Sep 17 00:00:00 2001 From: Trevor Hudson Date: Thu, 30 Jul 2026 14:32:39 -0700 Subject: [PATCH 1/2] Dispatch fan_out nodes per element on all six runtimes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `fan_out` node means "invoke once per element of an upstream list". Only DBOS did that; python, temporal, cloudflare, dbos-ts and inngest passed the whole list in a single invocation. That was not a missing feature so much as a runtime leaking into node semantics. The same pipeline.yaml produced two incompatible contracts for the code a user fills in: BDR's `personalize_email` judge declares `contact` as one contact and its Zod/Pydantic schema says so, but five of six runtimes handed it the entire array. Fill in the stub against DBOS, re-target Temporal, and it breaks — which invariant #1 (the IR is runtime-agnostic) exists to prevent. Which input is the fanned list now resolves in the language-neutral `_common.fan_out_element_param`, not in a Python-only helper. Two adapters disagreeing about that would make one pipeline mean two different things, so it cannot live in a per-language module. Precedence is unchanged (fan_out edge marker > incoming-edge source > only node-bound param) and ambiguity is still an emit-time error, never a guess. Per runtime: DBOS enqueues a durable step per element (unchanged); python uses pool.map; temporal gathers one activity execution each; cloudflare and inngest Promise.all over per-element steps; dbos-ts allSettled + unwrap. All preserve input order. Two traps this surfaced, both documented in CLAUDE.md: - Cloudflare and Inngest key a durable step by NAME, so element steps are index-suffixed. A constant name makes Cloudflare serve element 0's cached result for all N — the run succeeds, the output has the right length, and every entry is identical. DBOS is exempt: it identifies a step by execution order. - A parkable (MCP-bound) fan_out node on Cloudflare runs its elements sequentially, same reason a parkable step leaves its parallel wave — waitForEvent inside a promise combinator is undocumented and its timeout throws, which would reject every sibling. Also collapses temporal's three dispatch shapes onto one `_execute_activity_expr`. The wave branch had previously been a separate copy that silently omitted retry_policy; one renderer means a field cannot be added to one shape and forgotten in the others. Emitted code routes the fanned list through a guard (`fanOutList` / `_fan_out_list`) naming the node and the IR reference. Found by running it: the first live fan_out run died with "Cannot read properties of undefined (reading 'map')" — no node, no reference, in generated code the user never wrote. Python's "'NoneType' object is not iterable" is no better, so both languages get it; diagnostics are part of the emitted contract and a guard on one runtime only is its own parity gap. Testing. tests/test_fan_out_parity.py asserts the contract across all six, and is built to fail against batch dispatch rather than merely pass against the new output: each check requires the element param to bind the loop variable AND the whole-list expression to be absent. Verified empirically — against origin/main the five broken runtimes fail every behavioral assertion while every dbos parametrization passes. The live e2e mocks were themselves masking this. Three TS suites had `exclusion_check_sequence` returning `passed: []`, so BDR's fan was a correct no-op that no assertion caught. Mock values now derive from the IR via `_helpers.fan_out_source_keys`, and the suites assert the fan ran once per element with each invocation receiving exactly one. Inngest's memoization check is now fan-aware (== FAN_OUT_ELEMENTS for fan_out nodes, == 1 otherwise), which is what catches a step-name collapse. 1159 fast + 27 slow pass; ruff, strict mypy, sanity-check clean. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01SBuSDL7Svm7iDzgmhTpDSW --- CLAUDE.md | 71 ++++- README.md | 9 +- .../runtimes/cloudflare/.rote-manifest.json | 2 +- .../runtimes/cloudflare/src/workflow.ts | 42 ++- .../runtimes/dbos-ts/.rote-manifest.json | 2 +- .../expected/runtimes/dbos-ts/src/main.ts | 34 ++- .../runtimes/dbos/.rote-manifest.json | 2 +- .../expected/runtimes/dbos/main.py | 25 +- .../runtimes/inngest/.rote-manifest.json | 2 +- .../runtimes/inngest/src/inngest/pipeline.ts | 38 ++- .../runtimes/temporal/.rote-manifest.json | 2 +- .../expected/runtimes/temporal/workflow.py | 43 ++- src/rote/adapters/_common.py | 58 ++++ src/rote/adapters/_py_common.py | 98 +++--- src/rote/adapters/_ts_common.py | 62 ++++ src/rote/adapters/cloudflare.py | 105 ++++++- src/rote/adapters/dbos.py | 6 +- src/rote/adapters/dbos_ts.py | 46 +++ src/rote/adapters/inngest.py | 79 ++++- src/rote/adapters/python.py | 54 +++- src/rote/adapters/temporal.py | 134 ++++++--- tests/_helpers.py | 44 +++ tests/test_cloudflare_adapter.py | 8 + tests/test_cloudflare_e2e.py | 58 +++- tests/test_dbos_adapter.py | 18 +- tests/test_dbos_ts_e2e.py | 39 ++- tests/test_fan_out_parity.py | 278 ++++++++++++++++++ tests/test_inngest_adapter.py | 8 + tests/test_inngest_e2e.py | 51 +++- 29 files changed, 1239 insertions(+), 179 deletions(-) create mode 100644 tests/test_fan_out_parity.py diff --git a/CLAUDE.md b/CLAUDE.md index cc8e308..a60a7bb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -596,6 +596,52 @@ empirically against `inngest` 4.11.0 + `inngest-cli` 1.34.0: register it — the e2e sends an explicit `PUT` to the app's serve handler to force registration. See `tests/test_inngest_e2e.py`. +### A fan_out element's step name must carry its index + +Cloudflare and Inngest key a durable step by its **name**, so a fan_out +node emits `` `[${_index}]` ``, not `""`. Reusing one name for +every element is the nastiest failure shape available here: Cloudflare +returns element 0's *cached* result for all N elements, so the run +succeeds, the output has the right length, and every entry is +identical — no error, anywhere. Inngest's memoization collapses the fan +to a single execution the same way (its e2e asserts +`count == FAN_OUT_ELEMENTS` for fan_out nodes and `== 1` for the rest, +which is what catches it). DBOS is the exception on purpose: both its +Python and TS SDKs identify a step by execution order, so repeated +calls to one registered step are already distinct — no suffix needed, +and adding one would be meaningless rather than harmful. + +Second, on Cloudflare only: a **parkable** fan_out node (MCP-bound) +runs its elements *sequentially*, not in `Promise.all`, for the same +reason a parkable step leaves its parallel wave — `waitForEvent` inside +a promise combinator is undocumented and its timeout *throws*, which +would reject every other element. + +### An undefined fanned list must name the node, on every runtime + +Emitted code routes the fanned list through a guard — `fanOutList` in +TS, `_fan_out_list` in Python — that throws naming the node id and the +IR reference. Without it the failure is `Cannot read properties of +undefined (reading 'map')` (TS) or `'NoneType' object is not iterable` +(Python): no node, no reference, inside generated code the user never +wrote. Found by running it — the first live fan_out run on Cloudflare +died exactly that way, and the cause is almost always an upstream node +that didn't return the key the IR says it does. + +Both guards are emitted only when the pipeline has a fan_out node, and +they are *behavioral* emitted code like `_serialize`: if they drift, the +same broken pipeline reports differently per target. Diagnostics are +part of the emitted contract — a guard on one runtime and not another is +its own parity gap. + +Related trap in the **tests**: a live e2e's mocks must return the fanned +key as a non-empty list. Three TS e2e suites had `exclusion_check_sequence` +returning `passed: []`, which made BDR's fan a correct no-op — the node +never ran and every assertion still passed. `tests/_helpers.py`'s +`fan_out_source_keys` now derives those mock values from the IR using the +adapters' own resolution, so a mock cannot disagree with the emitted code +about which input is the list. + ### Eval pricing: no hardcoded prices, and two traps found empirically `rote.eval.pricing` fetches current models + official prices at eval @@ -810,17 +856,22 @@ Don't waste time debugging stubs. These are intentional. - The BDR example's `extracted/*.py` modules raise `NotImplementedError` — users fill them in with real API client code; the compiler produces scaffolding, not production code -- `fan_out` nodes receive the whole upstream list in one invocation on - every adapter EXCEPT DBOS, which dispatches one enqueued durable step - per element (element param resolved by - `rote.adapters._py_common.fan_out_binding` — fan_out edge marker > - incoming-edge source > only node-bound param; ambiguity is an - emit-time error). Per-element dispatch on the other adapters is the - remaining enhancement — until then their judge signatures must accept - the batch - **Working end-to-end:** +- `fan_out` on ALL SIX runtimes: a node marked `fan_out: true` is + invoked once per element of its bound list, and the result binds as a + list in input order. Which input is the list is resolved by + `rote.adapters._common.fan_out_element_param` (fan_out edge marker > + incoming-edge source > only node-bound param; ambiguity is an + emit-time error, never a guess) — it lives in the *language-neutral* + common module on purpose: two adapters disagreeing about which input + is the list would make one pipeline mean two different things. Per + runtime: DBOS enqueues one durable step per element; python uses + `pool.map`; temporal `asyncio.gather` over one activity execution + each; Cloudflare and Inngest `Promise.all` over per-element steps; + DBOS-TS `Promise.allSettled` + `unwrap`. See the fan_out gotchas + below for the two traps. + - `agent_loop` on ALL runtimes. Python got the real loop first; the three TypeScript runtimes now emit one too — MCP tools bound through whichever MCP helper that runtime already emits, `loop_body` sub-nodes @@ -1041,7 +1092,7 @@ Don't waste time debugging stubs. These are intentional. pipeline with cross-process gate signaling via `DBOSClient`; real judge usage captured through the emitted `$ROTE_USAGE_LOG` hook; measurements appended to `~/.local/share/rote/eval-corpus.jsonl`) -- 1146 tests (1119 fast + 27 slow). Run with `pytest tests/` (fast +- 1187 tests (1160 fast + 27 slow). Run with `pytest tests/` (fast only — what runs by default). Slow tests cover the runtime e2e suites (Temporal, Cloudflare, DBOS, DBOS-TS, Inngest, MCP-over-stdio); the TS ones require a Node toolchain, DBOS-TS diff --git a/README.md b/README.md index 1dda9a3..04eba77 100644 --- a/README.md +++ b/README.md @@ -378,9 +378,8 @@ plain-Python subprocess); those need a Node toolchain / Docker, so they run locally with `pytest tests/ -m slow`, not in CI. Known gaps: the extracted modules are `NotImplementedError` stubs -you fill in with real API-client code, a Restate adapter is planned, and -`fan_out` nodes currently receive the whole upstream list in one -invocation (per-element dispatch is a planned enhancement). Published on +you fill in with real API-client code, and a Restate adapter is planned. +Published on PyPI as [`rote-cli`](https://pypi.org/project/rote-cli/) via tag-driven Trusted Publishing ([docs/releasing.md](docs/releasing.md)). @@ -461,9 +460,7 @@ In rough priority order: Cloudflare; a separate node makes the short-circuit uniform. 3. **More example skills**: BDR is one shape; research-heavy, retrieval-heavy, and code-review skills stress the IR differently. -4. **`fan_out` per-element dispatch**: currently the whole upstream list - arrives in one invocation. -5. **The compiler compiling itself**: `rote-compile` is a SKILL.md; +4. **The compiler compiling itself**: `rote-compile` is a SKILL.md; pointing `rote compile` at it should crystallize its rubric-grade pieces and leave only the genuinely fuzzy judgments in the loop. diff --git a/examples/bdr-outreach/expected/runtimes/cloudflare/.rote-manifest.json b/examples/bdr-outreach/expected/runtimes/cloudflare/.rote-manifest.json index dfd9687..cf87206 100644 --- a/examples/bdr-outreach/expected/runtimes/cloudflare/.rote-manifest.json +++ b/examples/bdr-outreach/expected/runtimes/cloudflare/.rote-manifest.json @@ -21,7 +21,7 @@ "src/signatures/_roteInference.ts": "525c82572076a4a5be6a3611cf54fa8a3de17b22b23606ef9288eac694eebd74", "src/signatures/personalize_email.ts": "3f2c36d103afcba9ef1320fe212cc08802fa3b5c623b1252ce830b12f5f4445e", "src/signatures/vet_contact.ts": "89046fa5fb1c6646188c195507dd89927f9a4b009c600b0614db86735bcb3b99", - "src/workflow.ts": "61bd7c4f10725c883d1ff2c155ec254c1f6ed3405ec68995b182fca2f62f3888", + "src/workflow.ts": "9f6d4139495d9e19423f8eb03e7cbf98f4578115ad1a04baf9de359e716191c7", "tsconfig.json": "11ebc79fa7946d07d6cc0b555485988ec0de25e1183a7c654660f3bdfd4163f3", "wrangler.jsonc": "fa4f9733ea58ffeaa4e9ab94812ccd7b89f361ca4b7e68929fe8c1cd8b8f2ca2" } diff --git a/examples/bdr-outreach/expected/runtimes/cloudflare/src/workflow.ts b/examples/bdr-outreach/expected/runtimes/cloudflare/src/workflow.ts index f9c940a..e2e53a6 100644 --- a/examples/bdr-outreach/expected/runtimes/cloudflare/src/workflow.ts +++ b/examples/bdr-outreach/expected/runtimes/cloudflare/src/workflow.ts @@ -67,6 +67,25 @@ function authEventType(err: unknown, fallback: string): string { return match ? `rote_auth_${match[1]}` : fallback; } +/** The list a fan_out node dispatches over. + * + * A bare `.map()` on a missing key fails with "Cannot read properties + * of undefined (reading 'map')", which names neither the node nor the + * reference — in generated code that is close to undebuggable, and the + * cause is almost always an upstream node that didn't return the key + * the IR says it does. One call turns that into an actionable message. + * Found by running it: the first live fan_out run hit exactly this. */ +function fanOutList(value: unknown, nodeId: string, ref: string): unknown[] { + if (!Array.isArray(value)) { + const got = value === null ? "null" : typeof value; + throw new Error( + `fan_out node '${nodeId}' expected an array from '${ref}', got ${got}. ` + + `The upstream node must return that key as a list.`, + ); + } + return value; +} + export class BdrCampaignWorkflow extends WorkflowEntrypoint { async run(event: WorkflowEvent, step: WorkflowStep) { const pipelineInput = event.payload; @@ -223,14 +242,21 @@ export class BdrCampaignWorkflow extends WorkflowEntrypoint { ); // ─── Wave 9 ─── - const personalize_email_result = await step.do( - "personalize_email", - { timeout: "10 minutes" }, - async () => personalizeEmail({ - contact: (exclusion_check_sequence_result as Record)["passed"], - intel: target_research_result, - campaign_type: pipelineInput["campaign_type"], - }, this.env), + // fan_out: personalize_email runs once per element of the bound list, + // each as its own durable step (names are index-suffixed — + // Cloudflare caches step results by name). + const personalize_email_result = await Promise.all( + fanOutList( + (exclusion_check_sequence_result as Record)["passed"], + "personalize_email", + "exclusion_check_sequence.output.passed", + ).map((_item, _index) => + step.do( + `personalize_email[${_index}]`, + { timeout: "10 minutes" }, + async () => personalizeEmail({ contact: _item, intel: target_research_result, campaign_type: pipelineInput["campaign_type"] }, this.env), + ), + ), ); // ─── Wave 10 ─── diff --git a/examples/bdr-outreach/expected/runtimes/dbos-ts/.rote-manifest.json b/examples/bdr-outreach/expected/runtimes/dbos-ts/.rote-manifest.json index ce2c2ca..69580f3 100644 --- a/examples/bdr-outreach/expected/runtimes/dbos-ts/.rote-manifest.json +++ b/examples/bdr-outreach/expected/runtimes/dbos-ts/.rote-manifest.json @@ -16,7 +16,7 @@ "src/extracted/pre_enrollment_report.ts": "900e205e7ab818c5059db4fa5e9b669ce355f59ca4be364a1cedb90d5ed27dae", "src/extracted/target_research.ts": "007cbf0cee3c1fc5f30554c107120c21dbb235ae654474a1b008d8b6e74e8599", "src/extracted/taxonomy_lookup.ts": "2957ddacc25475f96a63e3e6fecc779f81a8a8b987093464a4edd0e01e0ec6a5", - "src/main.ts": "4e33346d0bd0bc664308e44e32b07a79c68f0735aa92977febf1cb9992c11239", + "src/main.ts": "062450041e627884edcb769bf3536bc2a970afbceb7530f882ada90c911a5d21", "src/signatures/_roteInference.ts": "525c82572076a4a5be6a3611cf54fa8a3de17b22b23606ef9288eac694eebd74", "src/signatures/personalize_email.ts": "cff8a6f39d10141d1318ec0ddf169f52cc2b8a85b140e579064278b414b2d9d1", "src/signatures/vet_contact.ts": "a91199a2ed230c33027b6c212a748ac7b8327ffd4ec68508aa4873e1ef56b28a", diff --git a/examples/bdr-outreach/expected/runtimes/dbos-ts/src/main.ts b/examples/bdr-outreach/expected/runtimes/dbos-ts/src/main.ts index 894b9f9..bfacbf1 100644 --- a/examples/bdr-outreach/expected/runtimes/dbos-ts/src/main.ts +++ b/examples/bdr-outreach/expected/runtimes/dbos-ts/src/main.ts @@ -225,6 +225,25 @@ function unwrap(settled: PromiseSettledResult): T { return settled.value; } +/** The list a fan_out node dispatches over. + * + * A bare `.map()` on a missing key fails with "Cannot read properties + * of undefined (reading 'map')", which names neither the node nor the + * reference — in generated code that is close to undebuggable, and the + * cause is almost always an upstream node that didn't return the key + * the IR says it does. One call turns that into an actionable message. + * Found by running it: the first live fan_out run hit exactly this. */ +function fanOutList(value: unknown, nodeId: string, ref: string): unknown[] { + if (!Array.isArray(value)) { + const got = value === null ? "null" : typeof value; + throw new Error( + `fan_out node '${nodeId}' expected an array from '${ref}', got ${got}. ` + + `The upstream node must return that key as a list.`, + ); + } + return value; +} + /** * End-to-end BDR outreach campaign workflow for pharma/biotech research */ @@ -310,13 +329,16 @@ export const runPipeline = DBOS.registerWorkflow( ); // ─── Wave 9 ─── - const personalize_email_result = await personalizeEmailStep( - { - contact: (exclusion_check_sequence_result as Record)["passed"], - intel: target_research_result, - campaign_type: pipelineInput["campaign_type"], - }, + // fan_out: personalize_email runs once per element of the bound list, + // each as its own durable step. + const personalize_email_settled = await Promise.allSettled( + fanOutList( + (exclusion_check_sequence_result as Record)["passed"], + "personalize_email", + "exclusion_check_sequence.output.passed", + ).map((_item) => personalizeEmailStep({ contact: _item, intel: target_research_result, campaign_type: pipelineInput["campaign_type"] })), ); + const personalize_email_result = personalize_email_settled.map(unwrap); // ─── Wave 10 ─── const create_sales_template_result = await createSalesTemplateStep( diff --git a/examples/bdr-outreach/expected/runtimes/dbos/.rote-manifest.json b/examples/bdr-outreach/expected/runtimes/dbos/.rote-manifest.json index 679ca7f..942d83d 100644 --- a/examples/bdr-outreach/expected/runtimes/dbos/.rote-manifest.json +++ b/examples/bdr-outreach/expected/runtimes/dbos/.rote-manifest.json @@ -10,7 +10,7 @@ "extracted/report.py": "e5d374b40fac4a6be9cd7a1c519848bfda18a094187ebb2297edd247df113707", "extracted/taxonomy.py": "44b3bab5a9dde17f170b436157cc349e70a2388a065a4b413a1fe9bf453d437a", "extracted/zoominfo.py": "9509863b14f2626fcd8154752c09e0bd7a5a4c6d5dc16071e33aac72eeb0b194", - "main.py": "f6314ca406b2524c7abd79d98cf72a33e09c17b0e185a3489c27449e581b3d43", + "main.py": "5427a43bf0622484be3f1d954a082c0a33b4063697f83b113f59105f68b4c4d3", "signatures/__init__.py": "fd4f3602f2a868d6b07b48232059ce49416f94e6a4d23aa42993c2b015a6600d", "signatures/_rote_inference.py": "52d800912b61bbead97a7a5cd609e032a3ddee5ba8c47613406ddd5957ecb97f", "signatures/personalize_email.py": "e2d906e425fa1c230a49c42e9ffa6043737f5eefc82cbe506fd3ef2c58b30a83", diff --git a/examples/bdr-outreach/expected/runtimes/dbos/main.py b/examples/bdr-outreach/expected/runtimes/dbos/main.py index 3d20533..ee98cbc 100644 --- a/examples/bdr-outreach/expected/runtimes/dbos/main.py +++ b/examples/bdr-outreach/expected/runtimes/dbos/main.py @@ -69,6 +69,25 @@ def _serialize(obj: Any) -> Any: return obj +def _fan_out_list(value: object, node_id: str, ref: str) -> list: + """The list a fan_out node dispatches over. + + Iterating a missing key raises "'NoneType' object is not iterable", + which names neither the node nor the reference — in generated code + that is close to undebuggable, and the cause is almost always an + upstream node that didn't return the key the IR says it does. The + TypeScript runtimes emit the same guard (`fanOutList`), so a missing + key reports identically on every target. + """ + if not isinstance(value, list): + raise TypeError( + f"fan_out node {node_id!r} expected a list from {ref!r}, got " + f"{type(value).__name__}. The upstream node must return that " + f"key as a list." + ) + return value + + @DBOS.step(name="target_research", retries_allowed=True, max_attempts=3, backoff_rate=2.0) def target_research(payload: dict) -> dict: """Run external research (Bright Data web search, ClinicalTrials.gov) @@ -373,7 +392,11 @@ def run_pipeline(pipeline_input: dict) -> dict: # list, each as its own enqueued durable step. personalize_email_payloads = [ {"contact": _item, "campaign_type": pipeline_input["campaign_type"], "intel": target_research_result} - for _item in exclusion_check_sequence_result["passed"] + for _item in _fan_out_list( + exclusion_check_sequence_result["passed"], + "personalize_email", + "exclusion_check_sequence.output.passed", + ) ] personalize_email_handles = [queue.enqueue(personalize_email, _p) for _p in personalize_email_payloads] personalize_email_result = [_h.get_result() for _h in personalize_email_handles] diff --git a/examples/bdr-outreach/expected/runtimes/inngest/.rote-manifest.json b/examples/bdr-outreach/expected/runtimes/inngest/.rote-manifest.json index 2dd480c..5c4505f 100644 --- a/examples/bdr-outreach/expected/runtimes/inngest/.rote-manifest.json +++ b/examples/bdr-outreach/expected/runtimes/inngest/.rote-manifest.json @@ -17,7 +17,7 @@ "src/extracted/taxonomy_lookup.ts": "2957ddacc25475f96a63e3e6fecc779f81a8a8b987093464a4edd0e01e0ec6a5", "src/index.ts": "e362bc56c1ea4d3d48f0efd9befb08b713d969cce7061b43b224f9f91d035dc1", "src/inngest/client.ts": "2312f84ecf800d6a37be7f978492199b7b8ef52f2223353a73cc5a63c3917643", - "src/inngest/pipeline.ts": "b1f1afe4a8a13d626546c727cac1995fd4e0063fd2486881fc07663e7b64d927", + "src/inngest/pipeline.ts": "8cacdf7ce70a9374a56eed8024c4c3944d2d869c3b732e7c4864e948e43d3ef4", "src/signatures/_roteInference.ts": "525c82572076a4a5be6a3611cf54fa8a3de17b22b23606ef9288eac694eebd74", "src/signatures/personalize_email.ts": "259ed4f1c775666e7668fe7600edead3f346f0348f1e54eb5a85e0aaea5c8a57", "src/signatures/vet_contact.ts": "23bbeba6416117887c2afbc8e70c584f57150dbad8c612b8f6c35a9fa32429b2", diff --git a/examples/bdr-outreach/expected/runtimes/inngest/src/inngest/pipeline.ts b/examples/bdr-outreach/expected/runtimes/inngest/src/inngest/pipeline.ts index 8141848..f5c8752 100644 --- a/examples/bdr-outreach/expected/runtimes/inngest/src/inngest/pipeline.ts +++ b/examples/bdr-outreach/expected/runtimes/inngest/src/inngest/pipeline.ts @@ -42,6 +42,25 @@ function requireEnv(name: string): string { return value; } +/** The list a fan_out node dispatches over. + * + * A bare `.map()` on a missing key fails with "Cannot read properties + * of undefined (reading 'map')", which names neither the node nor the + * reference — in generated code that is close to undebuggable, and the + * cause is almost always an upstream node that didn't return the key + * the IR says it does. One call turns that into an actionable message. + * Found by running it: the first live fan_out run hit exactly this. */ +function fanOutList(value: unknown, nodeId: string, ref: string): unknown[] { + if (!Array.isArray(value)) { + const got = value === null ? "null" : typeof value; + throw new Error( + `fan_out node '${nodeId}' expected an array from '${ref}', got ${got}. ` + + `The upstream node must return that key as a list.`, + ); + } + return value; +} + /** * End-to-end BDR outreach campaign workflow for pharma/biotech research */ @@ -137,15 +156,22 @@ export const runPipeline = inngest.createFunction( })); // ─── Wave 9 ─── - const personalize_email_result = await step.run("personalize_email", async () => personalizeEmail({ - contact: (exclusion_check_sequence_result as Record)["passed"], - intel: target_research_result, - campaign_type: pipelineInput["campaign_type"], - }, { + // fan_out: personalize_email runs once per element of the bound list, + // each as its own durable step (ids are index-suffixed to stay + // unique within the run). + const personalize_email_result = await Promise.all( + fanOutList( + (exclusion_check_sequence_result as Record)["passed"], + "personalize_email", + "exclusion_check_sequence.output.passed", + ).map((_item, _index) => + step.run(`personalize_email[${_index}]`, async () => personalizeEmail({ contact: _item, intel: target_research_result, campaign_type: pipelineInput["campaign_type"] }, { ANTHROPIC_API_KEY: requireEnv("ANTHROPIC_API_KEY"), ROTE_MODEL_PERSONALIZE_EMAIL: process.env.ROTE_MODEL_PERSONALIZE_EMAIL, ROTE_BASE_URL_PERSONALIZE_EMAIL: process.env.ROTE_BASE_URL_PERSONALIZE_EMAIL, - })); + })), + ), + ); // ─── Wave 10 ─── // IR retry policy: max 3 (exponential). Inngest v4 diff --git a/examples/bdr-outreach/expected/runtimes/temporal/.rote-manifest.json b/examples/bdr-outreach/expected/runtimes/temporal/.rote-manifest.json index 702b8dc..33c9b6b 100644 --- a/examples/bdr-outreach/expected/runtimes/temporal/.rote-manifest.json +++ b/examples/bdr-outreach/expected/runtimes/temporal/.rote-manifest.json @@ -7,6 +7,6 @@ "signatures/_rote_inference.py": "52d800912b61bbead97a7a5cd609e032a3ddee5ba8c47613406ddd5957ecb97f", "signatures/personalize_email.py": "904149801bb1f17eea6ae1db51885d790ae93bc5656fd2b51744e5db26b7df93", "signatures/vet_contact.py": "58ed0d55e2e9cd47848e0f483075cf996bc267bd68986f9d7d0b2db6b2a6667c", - "workflow.py": "ebdc5361368968b498e43f1fafd3b06c2dcade0a4803fee13ffabf5e5eb02ab3" + "workflow.py": "56ca1bfcf70e13f54a432175c160a9a14c884f76f2b9ea78f1ee3a8345c1c457" } } diff --git a/examples/bdr-outreach/expected/runtimes/temporal/workflow.py b/examples/bdr-outreach/expected/runtimes/temporal/workflow.py index 2f59a9f..46d62ab 100644 --- a/examples/bdr-outreach/expected/runtimes/temporal/workflow.py +++ b/examples/bdr-outreach/expected/runtimes/temporal/workflow.py @@ -39,6 +39,24 @@ def _parse_minutes(s: str) -> float: return float(s) +def _fan_out_list(value: object, node_id: str, ref: str) -> list: + """The list a fan_out node dispatches over. + + Iterating a missing key raises "'NoneType' object is not iterable", + which names neither the node nor the reference — in generated code + that is close to undebuggable, and the cause is almost always an + upstream node that didn't return the key the IR says it does. The + TypeScript runtimes emit the same guard (`fanOutList`), so a missing + key reports identically on every target. + """ + if not isinstance(value, list): + raise TypeError( + f"fan_out node {node_id!r} expected a list from {ref!r}, got " + f"{type(value).__name__}. The upstream node must return that " + f"key as a list." + ) + return value + @workflow.defn(name="BdrCampaign_ada0f771") class BdrCampaignWorkflow: """Compiled workflow for bdr-campaign.""" @@ -154,14 +172,23 @@ async def run(self, pipeline_input: dict) -> dict: ) # ─── Wave 9 ─── - personalize_email_result = await workflow.execute_activity( - "personalize_email", - { - "contact": exclusion_check_sequence_result["passed"], - "intel": target_research_result, - "campaign_type": pipeline_input["campaign_type"], - }, - start_to_close_timeout=timedelta(minutes=_parse_minutes("5m")), + # fan_out: personalize_email runs once per element of the bound + # list, each as its own activity execution. + personalize_email_result = list( + await asyncio.gather( + *( + workflow.execute_activity( + "personalize_email", + {"contact": _item, "campaign_type": pipeline_input["campaign_type"], "intel": target_research_result}, + start_to_close_timeout=timedelta(minutes=_parse_minutes("5m")), + ) + for _item in _fan_out_list( + exclusion_check_sequence_result["passed"], + "personalize_email", + "exclusion_check_sequence.output.passed", + ) + ) + ) ) # ─── Wave 10 ─── diff --git a/src/rote/adapters/_common.py b/src/rote/adapters/_common.py index 33147db..47cae77 100644 --- a/src/rote/adapters/_common.py +++ b/src/rote/adapters/_common.py @@ -14,6 +14,7 @@ import hashlib import json import re +from collections.abc import Sequence from pathlib import Path from rote.ir import Node, NodeKind, Pipeline, parse_input_ref @@ -422,6 +423,63 @@ def check_input_refs_available(node: Node, available: set[str]) -> None: ) +def fan_out_element_param(node: Node, pipeline: Pipeline) -> str: + """Which ``inputs`` param carries the list a ``fan_out`` node fans over. + + Every other input is shared verbatim by all invocations. Which param + is the list, in precedence order: + + 1. the param bound to the source of an incoming ``fan_out: true`` + edge (the IR's explicit marker); + 2. else the param bound to a node with any incoming edge — inputs + may also reference nodes *without* an edge (shared context, e.g. + BDR's ``intel``), which is what makes "the only node-bound + param" too naive; + 3. else the only node-bound param. + + Ambiguity after all three is an emit-time error, never a guess: + dispatching over the wrong list would silently judge the wrong + things. + + This lives in the language-neutral common module because the answer + is a property of the IR, not of the target language — every adapter + must fan over the *same* input or the same pipeline would mean + different things on different runtimes. + """ + if not node.inputs: + raise ValueError(f"fan_out node {node.id!r} has no inputs: nothing to fan over") + node_bound = { + param: parsed.node_id + for param, ref in node.inputs.items() + if (parsed := parse_input_ref(ref)).node_id is not None + } + edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id} + fan_edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id and e.fan_out} + + for sources in (fan_edge_sources, edge_sources): + matching = sorted(p for p, src in node_bound.items() if src in sources) + if len(matching) == 1: + return matching[0] + if len(node_bound) != 1: + raise ValueError( + f"fan_out node {node.id!r}: cannot identify the element param — " + f"node-bound inputs {sorted(node_bound)} and incoming edges " + f"{sorted(edge_sources)} don't single one out; mark the list edge " + f"with `fan_out: true`" + ) + return next(iter(node_bound)) + + +def fan_out_nodes(wave: Sequence[Node]) -> tuple[list[Node], list[Node]]: + """Split a wave into ``(fan_out_nodes, plain_nodes)``, order preserved. + + A ``fan_out`` node dispatches once per element of its bound list, so + it never shares the single/parallel payload shapes an adapter emits + for plain nodes. + """ + return [n for n in wave if n.fan_out], [n for n in wave if not n.fan_out] + + # ───────── MCP-capability refusal ───────── diff --git a/src/rote/adapters/_py_common.py b/src/rote/adapters/_py_common.py index 7d13166..7a60d38 100644 --- a/src/rote/adapters/_py_common.py +++ b/src/rote/adapters/_py_common.py @@ -30,6 +30,7 @@ DEFAULT_AGENT_MAX_ITERATIONS, EmitWriter, _to_pascal_case, + fan_out_element_param, safe_docstring_line, ) from rote.ir import LLMSignature, Node, NodeKind, Pipeline, parse_input_ref @@ -93,52 +94,65 @@ def _payload_literal(node: Node, indent: str) -> str: return "\n".join(lines) -def fan_out_binding(node: Node, pipeline: Pipeline) -> tuple[str, str, dict[str, str]]: - """``(element_param, list_expr, scalar_exprs)`` for a ``fan_out`` node. +_FAN_OUT_LIST_HELPER_PY = '''\ +def _fan_out_list(value: object, node_id: str, ref: str) -> list: + """The list a fan_out node dispatches over. - The element parameter is the one bound to the upstream *list* the - node fans over (ir-schema.md); every other input is shared verbatim - by all invocations. Which param that is, in precedence order: - - 1. the param bound to the source of an incoming ``fan_out: true`` - edge (the IR's explicit marker); - 2. else the param bound to a node with any incoming edge — inputs - may also reference nodes *without* an edge (shared context, e.g. - BDR's ``intel``), which is what makes "the only node-bound - param" too naive; - 3. else the only node-bound param. - - Ambiguity after all three is an emit-time error, never a guess: - dispatching over the wrong list would silently judge the wrong - things. + Iterating a missing key raises "'NoneType' object is not iterable", + which names neither the node nor the reference — in generated code + that is close to undebuggable, and the cause is almost always an + upstream node that didn't return the key the IR says it does. The + TypeScript runtimes emit the same guard (`fanOutList`), so a missing + key reports identically on every target. """ - if not node.inputs: - raise ValueError(f"fan_out node {node.id!r} has no inputs: nothing to fan over") - node_bound = { - param: parsed.node_id - for param, ref in node.inputs.items() - if (parsed := parse_input_ref(ref)).node_id is not None - } - edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id} - fan_edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id and e.fan_out} - - for sources in (fan_edge_sources, edge_sources): - matching = sorted(p for p, src in node_bound.items() if src in sources) - if len(matching) == 1: - element_param = matching[0] - break - else: - if len(node_bound) != 1: - raise ValueError( - f"fan_out node {node.id!r}: cannot identify the element param — " - f"node-bound inputs {sorted(node_bound)} and incoming edges " - f"{sorted(edge_sources)} don't single one out; mark the list edge " - f"with `fan_out: true`" - ) - element_param = next(iter(node_bound)) + if not isinstance(value, list): + raise TypeError( + f"fan_out node {node_id!r} expected a list from {ref!r}, got " + f"{type(value).__name__}. The upstream node must return that " + f"key as a list." + ) + return value +''' + + +def fan_out_list_helper() -> str: + """The ``_fan_out_list`` guard emitted into Python runtimes that fan out. + Behavioral emitted code, like ``_serialize``: if the runtimes' guards + drift, the same broken pipeline reports differently per target. + """ + return _FAN_OUT_LIST_HELPER_PY + + +def fan_out_binding( + node: Node, pipeline: Pipeline, indent: str = "" +) -> tuple[str, str, dict[str, str]]: + """``(element_param, list_expr, scalar_exprs)`` for a ``fan_out`` node. + + The Python rendering of :func:`fan_out_element_param` — which input + is the list is a language-neutral property of the IR, so it is + resolved there and only the expressions are built here. + + ``indent`` is the column the ``_fan_out_list(`` token sits at; the + guard call wraps against it rather than running long on one line. + """ + element_param = fan_out_element_param(node, pipeline) + # Narrowing only: fan_out_element_param raises on a node with no + # inputs, so reaching here means there is something to fan over. + assert node.inputs is not None scalars = {p: _ref_to_python_expr(r) for p, r in node.inputs.items() if p != element_param} - return element_param, _ref_to_python_expr(node.inputs[element_param]), scalars + ref = node.inputs[element_param] + inner = indent + " " + # json.dumps, not !r: emitted Python uses double quotes throughout + # (the repo's ruff-format style), same as _py_literal. + list_expr = ( + "_fan_out_list(\n" + f"{inner}{_ref_to_python_expr(ref)},\n" + f"{inner}{json.dumps(node.id)},\n" + f"{inner}{json.dumps(ref)},\n" + f"{indent})" + ) + return element_param, list_expr, scalars # ───────── JSON Schema → Pydantic source ───────── diff --git a/src/rote/adapters/_ts_common.py b/src/rote/adapters/_ts_common.py index e3abcac..98889ba 100644 --- a/src/rote/adapters/_ts_common.py +++ b/src/rote/adapters/_ts_common.py @@ -40,6 +40,7 @@ DEFAULT_AGENT_MAX_ITERATIONS, _to_camel_case, _to_pascal_case, + fan_out_element_param, safe_block_comment_line, ) from rote.ir import LLMSignature, Node, NodeKind, Pipeline, parse_input_ref @@ -593,6 +594,67 @@ def payload_ts_literal(node: Node, indent: str) -> str: return "\n".join(lines) +FAN_OUT_LIST_HELPER_TS = """\ +/** The list a fan_out node dispatches over. + * + * A bare `.map()` on a missing key fails with "Cannot read properties + * of undefined (reading 'map')", which names neither the node nor the + * reference — in generated code that is close to undebuggable, and the + * cause is almost always an upstream node that didn't return the key + * the IR says it does. One call turns that into an actionable message. + * Found by running it: the first live fan_out run hit exactly this. */ +function fanOutList(value: unknown, nodeId: string, ref: string): unknown[] { + if (!Array.isArray(value)) { + const got = value === null ? "null" : typeof value; + throw new Error( + `fan_out node '${nodeId}' expected an array from '${ref}', got ${got}. ` + + `The upstream node must return that key as a list.`, + ); + } + return value; +} +""" + + +def fan_out_ts_binding(node: Node, pipeline: Pipeline, indent: str = "") -> tuple[str, str]: + """``(payload_literal, list_expr)`` for a ``fan_out`` node in TypeScript. + + The payload is a one-line object binding the element param to the + loop variable ``_item`` and every other input to its shared + expression — the TS rendering of + :func:`rote.adapters._common.fan_out_element_param`, which decides + *which* input is the list. + + The list expression goes through ``fanOutList`` + (:data:`FAN_OUT_LIST_HELPER_TS`), which both narrows the type — node + results are ``Record`` or ``never``, so neither + ``.map`` nor ``for…of`` compiles against them — and reports a + missing upstream key by node id instead of as a bare TypeError. + + ``indent`` is the column the ``fanOutList(`` token sits at. The call + wraps against it instead of running to ~180 characters on one line — + emitted code is meant to be read and reviewed. + """ + element_param = fan_out_element_param(node, pipeline) + entries = [] + for param, input_ref in node.inputs.items() if node.inputs else []: + key = param if _TS_IDENT_RE.fullmatch(param) else json.dumps(param) + value = "_item" if param == element_param else ref_to_ts_expr(input_ref) + entries.append(f"{key}: {value}") + payload = "{ " + ", ".join(entries) + " }" + assert node.inputs is not None + ref = node.inputs[element_param] + inner = indent + " " + list_expr = ( + "fanOutList(\n" + f"{inner}{ref_to_ts_expr(ref)},\n" + f"{inner}{json.dumps(node.id)},\n" + f"{inner}{json.dumps(ref)},\n" + f"{indent})" + ) + return payload, list_expr + + # ───────── Pipeline queries ───────── diff --git a/src/rote/adapters/cloudflare.py b/src/rote/adapters/cloudflare.py index 24b74c4..bbdae1e 100644 --- a/src/rote/adapters/cloudflare.py +++ b/src/rote/adapters/cloudflare.py @@ -42,6 +42,7 @@ _pipeline_hash, _to_camel_case, check_input_refs_available, + fan_out_nodes, pipeline_identity, safe_block_comment_line, workflow_class_name, @@ -52,6 +53,7 @@ from rote.adapters._ts_common import ( AI_UTILS_NPM_VERSION, ANTHROPIC_SDK_NPM_VERSION, + FAN_OUT_LIST_HELPER_TS, MCP_SDK_NPM_VERSION, ROTE_INFERENCE_HELPER_TS, ROTE_MCP_WORKERS_HELPER_TS, @@ -60,6 +62,7 @@ emit_agent_loop_module, emit_ts_signature_module, emit_workers_mcp_call_module, + fan_out_ts_binding, llm_clients, mcp_backed_nodes, module_imports, @@ -181,7 +184,12 @@ def auth_event_type(server: str) -> str: def _emit_step_call_mcp_parkable( - node: Node, cfg: CloudflareAdapterConfig, *, pipeline: Pipeline + node: Node, + cfg: CloudflareAdapterConfig, + *, + pipeline: Pipeline, + payload: str | None = None, + name_suffix: str = "", ) -> str: """MCP-backed dispatch: auth failures park the instance durably. @@ -194,10 +202,15 @@ def _emit_step_call_mcp_parkable( behavior; and events sent before the wait starts are buffered per-instance, so `rote mcp release` can blast every non-terminal instance without a race. + + ``payload`` and ``name_suffix`` are supplied by the fan_out path: + one element per call, and a step-name fragment (``[${_index}]``) + that keeps each element's step name unique within the instance. """ fn_name = _to_camel_case(node.id) config = _step_config_literal(node, cfg) - payload = payload_ts_literal(node, indent=" " * 24) + if payload is None: + payload = payload_ts_literal(node, indent=" " * 24) nid = node.id if node.mcp is not None: # One binding, one server: the release event is known at emit time. @@ -220,6 +233,10 @@ def _emit_step_call_mcp_parkable( " // so `rote mcp release ` releases the right instances.\n" ) label = "agent loop" if node.mcp is None else "MCP-backed" + # A fan_out element's step names carry its index; every step name + # must be unique within an instance or the elements collapse onto + # one cached step. + first_name = f"`{nid}{name_suffix}`" if name_suffix else json.dumps(nid) return ( f" // ─── {nid} ({label}): park on dead credentials ───\n" f"{release_hint}" @@ -228,8 +245,8 @@ def _emit_step_call_mcp_parkable( f" try {{\n" f" {nid}_result = (await step.do(\n" f" {nid}_attempt === 0\n" - f" ? {json.dumps(nid)}\n" - f" : `{nid} (auth retry ${{{nid}_attempt}})`,\n" + f" ? {first_name}\n" + f" : `{nid}{name_suffix} (auth retry ${{{nid}_attempt}})`,\n" f" {config},\n" f" async () => {{\n" f" try {{\n" @@ -252,7 +269,7 @@ def _emit_step_call_mcp_parkable( f" // consecutive failure. Fresh step names per attempt.\n" f" if ({nid}_attempt % 2 === 1) {{\n" f" await step.waitForEvent(\n" - f" `{nid} auth wait ${{{nid}_attempt}}`,\n" + f" `{nid}{name_suffix} auth wait ${{{nid}_attempt}}`,\n" f' {{ type: {event_expr}, timeout: "30 days" }},\n' f" );\n" f" }}\n" @@ -341,6 +358,67 @@ def _emit_parallel_step_calls(nodes: list[Node], cfg: CloudflareAdapterConfig) - return "\n".join(lines) + "\n" +def _emit_fan_out_parallel(node: Node, cfg: CloudflareAdapterConfig, *, pipeline: Pipeline) -> str: + """Emit a plain ``fan_out`` node as one ``step.do`` per element. + + Same ``Promise.all`` reasoning as a parallel wave: it is + Cloudflare's documented concurrency primitive, each element persists + under its own step name, and a rejection means that element already + exhausted its own retries. + + Step names are index-suffixed. Cloudflare caches a step's result by + name, so reusing one name for every element would return the first + element's result for all of them — the failure would look like a + correct run with suspiciously uniform output. + """ + payload, list_expr = fan_out_ts_binding(node, pipeline, indent=" " * 12) + pass_env = node.kind in (NodeKind.LLM_JUDGE, NodeKind.AGENT_LOOP) + args = f"{payload}, this.env" if pass_env else payload + return "\n".join( + [ + f" // fan_out: {node.id} runs once per element of the bound list,", + " // each as its own durable step (names are index-suffixed —", + " // Cloudflare caches step results by name).", + f" const {node.id}_result = await Promise.all(", + f" {list_expr}.map((_item, _index) =>", + " step.do(", + f" `{node.id}[${{_index}}]`,", + f" {_step_config_literal(node, cfg)},", + f" async () => {_to_camel_case(node.id)}({args}),", + " ),", + " ),", + " );", + ] + ) + + +def _emit_fan_out_parkable(node: Node, cfg: CloudflareAdapterConfig, *, pipeline: Pipeline) -> str: + """Emit a parkable ``fan_out`` node: one element per iteration. + + Sequential, for the same reason a parkable step leaves its parallel + wave — it can suspend on ``waitForEvent``, whose behavior inside a + promise combinator is undocumented and whose timeout *throws*, which + would reject every other element. + """ + payload, list_expr = fan_out_ts_binding(node, pipeline, indent=" " * 8) + inner = _emit_step_call_mcp_parkable( + node, cfg, pipeline=pipeline, payload=payload, name_suffix="[${_index}]" + ) + return "\n".join( + [ + f" // fan_out: {node.id} runs once per element of the bound list.", + " // Parkable steps stay sequential — a waitForEvent inside a", + " // promise combinator would reject every sibling on timeout.", + f" const {node.id}_results: Record[] = [];", + f" for (const [_index, _item] of {list_expr}.entries()) {{", + textwrap.indent(inner.rstrip("\n"), " "), + f" {node.id}_results.push({node.id}_result);", + " }", + f" const {node.id}_result = {node.id}_results;", + ] + ) + + def _emit_hitl_gate(node: Node, cfg: CloudflareAdapterConfig) -> str: assert node.signal is not None _validate_signal_name(node.signal, node.id) @@ -508,7 +586,14 @@ def _is_parkable(node: Node) -> bool: parkable = [n for n in wave if _is_parkable(n)] plain = [n for n in wave if n.kind is not NodeKind.HITL_GATE and not _is_parkable(n)] - for node in plain + parkable: + # fan_out nodes dispatch once per element of their bound list — + # they never share the single/parallel payload shapes below, but + # they keep the parkable/plain split (a parkable fan_out runs its + # elements sequentially, for the same waitForEvent reason). + fanned_parkable, parkable = fan_out_nodes(parkable) + fanned_plain, plain = fan_out_nodes(plain) + + for node in plain + parkable + fanned_plain + fanned_parkable: check_input_refs_available(node, available) if len(plain) > 1: @@ -527,6 +612,12 @@ def _is_parkable(node: Node) -> bool: _emit_step_call_mcp_parkable(node, cfg, pipeline=pipeline).rstrip("\n") ) + for node in fanned_plain: + body_lines.append(_emit_fan_out_parallel(node, cfg, pipeline=pipeline)) + + for node in fanned_parkable: + body_lines.append(_emit_fan_out_parkable(node, cfg, pipeline=pipeline)) + for node in gates: body_lines.append(_emit_hitl_gate(node, cfg).rstrip("\n")) @@ -552,6 +643,8 @@ def _is_parkable(node: Node) -> bool: ) helper_block = f"{_STEP_NEEDS_AUTH_HELPER}\n" if parks_on_auth else "" + if any(n.fan_out for n in pipeline.nodes): + helper_block += f"{FAN_OUT_LIST_HELPER_TS}\n" return header + imports + "\n" + env_block + helper_block + class_block diff --git a/src/rote/adapters/dbos.py b/src/rote/adapters/dbos.py index b3ab71c..5d4e783 100644 --- a/src/rote/adapters/dbos.py +++ b/src/rote/adapters/dbos.py @@ -88,6 +88,7 @@ _signature_path_parts, agent_loop_call, fan_out_binding, + fan_out_list_helper, resolve_extracted_source, serialize_helper, write_signature_package, @@ -521,7 +522,7 @@ def _emit_workflow_body(pipeline: Pipeline, cfg: DbosAdapterConfig) -> str: lines.append(f" {node.id}_result = {node.id}_handle.get_result()") for node in fan_out_nodes: - element_param, list_expr, scalars = fan_out_binding(node, pipeline) + element_param, list_expr, scalars = fan_out_binding(node, pipeline, indent=" " * 8) payload_items = "".join( f', "{param}": {expr}' for param, expr in sorted(scalars.items()) ) @@ -660,6 +661,9 @@ def emit_main(pipeline: Pipeline, cfg: DbosAdapterConfig | None = None) -> str: " portable across the system database." ) + if any(n.fan_out for n in pipeline.nodes): + header += "\n\n" + fan_out_list_helper() + if mcp_backed: header += "\n\n" + textwrap.dedent( '''\ diff --git a/src/rote/adapters/dbos_ts.py b/src/rote/adapters/dbos_ts.py index 870c6bc..6425dde 100644 --- a/src/rote/adapters/dbos_ts.py +++ b/src/rote/adapters/dbos_ts.py @@ -96,10 +96,12 @@ _to_camel_case, _to_pascal_case, check_input_refs_available, + fan_out_nodes, safe_block_comment_line, ) from rote.adapters._ts_common import ( ANTHROPIC_SDK_NPM_VERSION, + FAN_OUT_LIST_HELPER_TS, MCP_SDK_NPM_VERSION, REQUIRE_ENV_HELPER, ROTE_INFERENCE_HELPER_TS, @@ -108,6 +110,7 @@ emit_node_agent_loop_module, emit_node_tsconfig, emit_ts_signature_module, + fan_out_ts_binding, judge_env_arg, llm_clients, mcp_backed_nodes, @@ -377,6 +380,38 @@ def _emit_hitl_wait(node: Node, pipeline: Pipeline) -> str: ) +def _emit_fan_out(node: Node, pipeline: Pipeline, cfg: DbosTsAdapterConfig) -> list[str]: + """Emit a ``fan_out`` node as one durable step per element. + + Unlike Cloudflare and Inngest, DBOS identifies a step by its + execution order within the workflow rather than by a caller-supplied + name, so calling the same registered step function once per element + needs no index-suffixed id — the elements simply become consecutive + durable steps. + + ``Promise.allSettled`` for the same reason the parallel-wave branch + uses it: a bare ``Promise.all`` can crash the Node process on + unhandled rejections. ``allSettled`` and ``map`` both preserve + order, so ``_result[i]`` is element ``i``. + """ + fn_name = _to_camel_case(node.id) + payload, list_expr = fan_out_ts_binding(node, pipeline, indent=" " * 12) + call = f"{fn_name}Step({payload})" + if _is_mcp_backed(node, cfg): + assert node.mcp is not None + # Each element parks independently; one release signal wakes the + # workflow and runWithAuthPark retries the element that parked. + call = f"runWithAuthPark(() => {call}, {json.dumps(node.mcp.server)})" + return [ + f" // fan_out: {node.id} runs once per element of the bound list,", + " // each as its own durable step.", + f" const {node.id}_settled = await Promise.allSettled(", + f" {list_expr}.map((_item) => {call}),", + " );", + f" const {node.id}_result = {node.id}_settled.map(unwrap);", + ] + + def _emit_workflow_body(pipeline: Pipeline, cfg: DbosTsAdapterConfig) -> tuple[str, bool]: """Render the workflow function body; returns (body, uses_unwrap).""" waves = _execution_waves(pipeline) @@ -399,6 +434,12 @@ def _emit_workflow_body(pipeline: Pipeline, cfg: DbosTsAdapterConfig) -> tuple[s lines.append("") lines.append(f" // ─── Wave {wave_idx} ───") + # fan_out nodes dispatch once per element of their bound list — + # they never share the single/parallel payload shapes below. + fanned, non_hitl = fan_out_nodes(non_hitl) + if fanned: + uses_unwrap = True + if len(non_hitl) == 1: node = non_hitl[0] fn_name = _to_camel_case(node.id) @@ -468,6 +509,9 @@ def _emit_workflow_body(pipeline: Pipeline, cfg: DbosTsAdapterConfig) -> tuple[s else: lines.append(f" const {node.id}_result = unwrap({node.id}_settled);") + for node in fanned: + lines.extend(_emit_fan_out(node, pipeline, cfg)) + for gate in hitl: lines.append(_emit_hitl_wait(gate, pipeline).rstrip("\n")) @@ -646,6 +690,8 @@ def emit_main(pipeline: Pipeline, cfg: DbosTsAdapterConfig | None = None) -> str body, uses_unwrap = _emit_workflow_body(pipeline, cfg) unwrap_block = f"\n{_UNWRAP_HELPER.rstrip(chr(10))}\n" if uses_unwrap else "" + if any(n.fan_out for n in pipeline.nodes): + unwrap_block += f"\n{FAN_OUT_LIST_HELPER_TS.rstrip(chr(10))}\n" workflow_block = ( "// ───────── Workflow ─────────\n" diff --git a/src/rote/adapters/inngest.py b/src/rote/adapters/inngest.py index f8b76b8..a7f4790 100644 --- a/src/rote/adapters/inngest.py +++ b/src/rote/adapters/inngest.py @@ -106,10 +106,12 @@ _pipeline_hash, _to_camel_case, check_input_refs_available, + fan_out_nodes, safe_block_comment_line, ) from rote.adapters._ts_common import ( ANTHROPIC_SDK_NPM_VERSION, + FAN_OUT_LIST_HELPER_TS, MCP_SDK_NPM_VERSION, REQUIRE_ENV_HELPER, ROTE_INFERENCE_HELPER_TS, @@ -118,6 +120,7 @@ emit_node_agent_loop_module, emit_node_tsconfig, emit_ts_signature_module, + fan_out_ts_binding, judge_env_arg, llm_clients, mcp_backed_nodes, @@ -419,18 +422,36 @@ def _is_mcp_backed(node: Node, cfg: InngestAdapterConfig) -> bool: ) -def _step_call_expr(node: Node, payload_indent: str) -> str: - """The ``step.run("", async () => fn(payload))`` expression, unterminated.""" +def _step_call_expr( + node: Node, + payload_indent: str, + *, + payload: str | None = None, + step_id_expr: str | None = None, +) -> str: + """The ``step.run("", async () => fn(payload))`` expression, unterminated. + + ``payload`` and ``step_id_expr`` are overridden by the fan_out path, + which binds one element per call and needs a per-element step id. + """ fn_name = _to_camel_case(node.id) - payload = payload_ts_literal(node, indent=payload_indent) + if payload is None: + payload = payload_ts_literal(node, indent=payload_indent) if node.kind is NodeKind.LLM_JUDGE: call = f"{fn_name}({payload}, {judge_env_arg(node)})" else: call = f"{fn_name}({payload})" - return f"step.run({json.dumps(node.id)}, async () => {call})" + return f"step.run({step_id_expr or json.dumps(node.id)}, async () => {call})" -def _parkable_call_expr(node: Node, pipeline: Pipeline, payload_indent: str) -> str: +def _parkable_call_expr( + node: Node, + pipeline: Pipeline, + payload_indent: str, + *, + payload: str | None = None, + step_id_expr: str | None = None, +) -> str: """The ``runParkable(step, ...)`` expression for an MCP-backed node. Auth failures suspend the run on the pipeline's ``rote.auth.`` @@ -438,14 +459,46 @@ def _parkable_call_expr(node: Node, pipeline: Pipeline, payload_indent: str) -> """ assert node.mcp is not None fn_name = _to_camel_case(node.id) - payload = payload_ts_literal(node, indent=payload_indent) + if payload is None: + payload = payload_ts_literal(node, indent=payload_indent) release_event = auth_event_name(pipeline, node.mcp.server) return ( - f"runParkable(step, {json.dumps(node.id)}, {json.dumps(release_event)}, " + f"runParkable(step, {step_id_expr or json.dumps(node.id)}, " + f"{json.dumps(release_event)}, " f"{json.dumps(node.mcp.server)}, async () => {fn_name}({payload}))" ) +def _fan_out_lines(node: Node, pipeline: Pipeline, cfg: InngestAdapterConfig) -> list[str]: + """Emit a ``fan_out`` node as one durable step per element. + + Step ids are index-suffixed because Inngest requires them unique + within a run — reusing one id for every element would collapse the + whole fan into a single memoized step. + + ``.map`` preserves order, so ``_result[i]`` corresponds to + element ``i`` of the bound list. + """ + payload, list_expr = fan_out_ts_binding(node, pipeline, indent=" " * 12) + step_id_expr = f"`{node.id}[${{_index}}]`" + if _is_mcp_backed(node, cfg): + expr = _parkable_call_expr( + node, pipeline, " " * 16, payload=payload, step_id_expr=step_id_expr + ) + else: + expr = _step_call_expr(node, " " * 16, payload=payload, step_id_expr=step_id_expr) + return [ + f" // fan_out: {node.id} runs once per element of the bound list,", + " // each as its own durable step (ids are index-suffixed to stay", + " // unique within the run).", + f" const {node.id}_result = await Promise.all(", + f" {list_expr}.map((_item, _index) =>", + f" {expr},", + " ),", + " );", + ] + + _RUN_PARKABLE_HELPER = """\ // ───────── MCP auth parking ───────── // @@ -590,6 +643,10 @@ def _emit_workflow_body(pipeline: Pipeline, fn_retries: int, cfg: InngestAdapter lines.append("") lines.append(f" // ─── Wave {wave_idx} ───") + # fan_out nodes dispatch once per element of their bound list — + # they never share the single/parallel payload shapes below. + fanned, non_hitl = fan_out_nodes(non_hitl) + if len(non_hitl) == 1: node = non_hitl[0] comment = _node_policy_comment(node, fn_retries, indent=" " * 8) @@ -622,6 +679,12 @@ def _emit_workflow_body(pipeline: Pipeline, fn_retries: int, cfg: InngestAdapter lines.append(f" {expr},") lines.append(" ]);") + for node in fanned: + comment = _node_policy_comment(node, fn_retries, indent=" " * 8) + if comment: + lines.append(comment.rstrip("\n")) + lines.extend(_fan_out_lines(node, pipeline, cfg)) + for gate in hitl: lines.append(_emit_hitl_wait(gate, pipeline).rstrip("\n")) @@ -734,6 +797,8 @@ def emit_pipeline_ts(pipeline: Pipeline, cfg: InngestAdapterConfig | None = None sections.append(helper_block.strip("\n")) if mcp_backed: sections.append(_RUN_PARKABLE_HELPER.rstrip("\n")) + if any(n.fan_out for n in pipeline.nodes): + sections.append(FAN_OUT_LIST_HELPER_TS.rstrip("\n")) sections.append(function_block) return "\n\n".join(sections) diff --git a/src/rote/adapters/python.py b/src/rote/adapters/python.py index 2b03eba..522dc58 100644 --- a/src/rote/adapters/python.py +++ b/src/rote/adapters/python.py @@ -65,6 +65,7 @@ _pipeline_hash, _to_pascal_case, check_input_refs_available, + fan_out_nodes, refuse_mcp_only_nodes, safe_docstring_line, ) @@ -74,6 +75,8 @@ _payload_literal, _signature_path_parts, agent_loop_call, + fan_out_binding, + fan_out_list_helper, resolve_extracted_source, serialize_helper, write_signature_package, @@ -288,13 +291,43 @@ def _emit_node_agent_loop(node: Node, cfg: PythonAdapterConfig) -> str: def _has_parallel_wave(pipeline: Pipeline) -> bool: - return any(len(wave) > 1 for wave in _execution_waves(pipeline)) + """Whether main.py needs ``ThreadPoolExecutor``. + + Two independent reasons: a wave with more than one node, or any + ``fan_out`` node (whose per-element calls run on a pool even when it + is alone in its wave). + """ + waves = _execution_waves(pipeline) + return any(len(wave) > 1 for wave in waves) or any(n.fan_out for w in waves for n in w) def _has_retry(pipeline: Pipeline) -> bool: return any(n.retry and n.retry.max > 0 for n in pipeline.nodes) +def _emit_fan_out_call(node: Node, pipeline: Pipeline) -> str: + """Emit a ``fan_out`` node as one call per element of its bound list. + + ``pool.map`` rather than ``submit``/``result``: it preserves input + order, so ``_result[i]`` corresponds to element ``i`` — a + guarantee downstream nodes rely on when they zip a fanned result + back against the list it came from. + """ + element_param, list_expr, scalars = fan_out_binding(node, pipeline, indent=" " * 8) + shared = "".join(f', "{param}": {expr}' for param, expr in sorted(scalars.items())) + return "\n".join( + [ + f" # fan_out: {node.id} runs once per element of the bound list.", + f" {node.id}_payloads = [", + f' {{"{element_param}": _item{shared}}}', + f" for _item in {list_expr}", + " ]", + " with ThreadPoolExecutor() as pool:", + f" {node.id}_result = list(pool.map({node.id}, {node.id}_payloads))", + ] + ) + + def _emit_pipeline_body(pipeline: Pipeline) -> str: waves = _execution_waves(pipeline) lines: list[str] = [] @@ -310,8 +343,12 @@ def _emit_pipeline_body(pipeline: Pipeline) -> str: lines.append("") lines.append(f" # ─── Wave {wave_idx} ───") - if len(wave) == 1: - node = wave[0] + # fan_out nodes dispatch once per element of their bound list — + # they never share the single/parallel payload shapes below. + fanned, plain = fan_out_nodes(wave) + + if len(plain) == 1: + node = plain[0] payload = _payload_literal(node, indent=" " * 8) if payload == "{}": lines.append(f" {node.id}_result = {node.id}({{}})") @@ -319,11 +356,11 @@ def _emit_pipeline_body(pipeline: Pipeline) -> str: lines.append(f" {node.id}_result = {node.id}(") lines.append(f" {payload}") lines.append(" )") - else: + elif len(plain) > 1: lines.append(" # Independent nodes: run concurrently on a thread pool and") lines.append(" # join before the next wave starts.") lines.append(" with ThreadPoolExecutor() as pool:") - for node in wave: + for node in plain: payload = _payload_literal(node, indent=" " * 12) if payload == "{}": lines.append(f" {node.id}_future = pool.submit({node.id}, {{}})") @@ -332,9 +369,12 @@ def _emit_pipeline_body(pipeline: Pipeline) -> str: lines.append(f" {node.id},") lines.append(f" {payload},") lines.append(" )") - for node in wave: + for node in plain: lines.append(f" {node.id}_result = {node.id}_future.result()") + for node in fanned: + lines.append(_emit_fan_out_call(node, pipeline)) + available.update(n.id for n in wave) lines.append("") @@ -421,6 +461,8 @@ def emit_main(pipeline: Pipeline, cfg: PythonAdapterConfig | None = None) -> str "Node functions return JSON-serializable payloads so the final\n" " result prints cleanly and pastes into other systems." ) + if any(n.fan_out for n in pipeline.nodes): + serialize_block += "\n\n" + fan_out_list_helper() node_parts: list[str] = [] for node in pipeline.nodes: diff --git a/src/rote/adapters/temporal.py b/src/rote/adapters/temporal.py index 8187a19..5aabbd7 100644 --- a/src/rote/adapters/temporal.py +++ b/src/rote/adapters/temporal.py @@ -35,6 +35,7 @@ _pipeline_hash, _to_pascal_case, check_input_refs_available, + fan_out_nodes, refuse_mcp_only_nodes, safe_docstring_line, ) @@ -43,6 +44,8 @@ _payload_literal, _signature_path_parts, agent_loop_call, + fan_out_binding, + fan_out_list_helper, write_signature_package, ) from rote.ir import Node, NodeKind, Pipeline @@ -358,39 +361,73 @@ def {gate.signal}(self, payload: dict) -> None: return init_block + "\n".join(handler_blocks) -def _emit_wave_call(node: Node, cfg: TemporalAdapterConfig) -> str: - """Emit a multi-line call to ``workflow.execute_activity`` for one node. - - Produces something like:: - - foo_result = await workflow.execute_activity( - "foo", - { - "brief": pipeline_input, - "intel": target_research_result, - }, - start_to_close_timeout=timedelta(minutes=...), - retry_policy=RetryPolicy(...), - ) +def _execute_activity_expr( + node: Node, cfg: TemporalAdapterConfig, payload: str, indent: str +) -> str: + """The ``workflow.execute_activity(...)`` expression for one node. + + The first line carries no indentation (the caller prefixes it with + an assignment or a comma-separated position); continuation lines and + the closing paren are indented against ``indent``. + + This is the single renderer behind all three dispatch shapes — + lone node, parallel wave, and fan_out. It exists because the wave + and single-node branches were once separate copies, and the wave + copy quietly omitted ``retry_policy``: a node lost its declared + retry budget merely by gaining a sibling, in exactly the parallel + fetch waves where flaky network calls live. One renderer means a + new field cannot be added to one shape and forgotten in the others. """ + inner = indent + " " + timeout = _activity_timeout(node, cfg.default_activity_timeout) + lines = [ + "workflow.execute_activity(", + f'{inner}"{node.id}",', + f"{inner}{payload},", + f'{inner}start_to_close_timeout=timedelta(minutes=_parse_minutes("{timeout}")),', + ] + retry = _retry_policy_args(node) + if retry: + lines.append(f"{inner}retry_policy={retry},") + lines.append(f"{indent})") + return "\n".join(lines) + + +def _emit_wave_call(node: Node, cfg: TemporalAdapterConfig) -> str: + """Emit an awaited activity execution for a node alone in its wave.""" if node.kind is NodeKind.HITL_GATE: # HITL gates are not activities — they're handled separately. return "" - - timeout = _activity_timeout(node, cfg.default_activity_timeout) - retry = _retry_policy_args(node) payload = _payload_literal(node, indent=" " * 12) + expr = _execute_activity_expr(node, cfg, payload, indent=" " * 8) + return f" {node.id}_result = await {expr}\n" - lines = [ - f" {node.id}_result = await workflow.execute_activity(", - f' "{node.id}",', - f" {payload},", - f' start_to_close_timeout=timedelta(minutes=_parse_minutes("{timeout}")),', - ] - if retry: - lines.append(f" retry_policy={retry},") - lines.append(" )") - return "\n".join(lines) + "\n" + +def _emit_fan_out_call(node: Node, pipeline: Pipeline, cfg: TemporalAdapterConfig) -> str: + """Emit a ``fan_out`` node as one activity execution per element. + + ``asyncio.gather`` preserves argument order, so ``_result[i]`` + corresponds to element ``i`` of the bound list — the same ordering + guarantee the other adapters give. + """ + element_param, list_expr, scalars = fan_out_binding(node, pipeline, indent=" " * 20) + shared = "".join(f', "{param}": {expr}' for param, expr in sorted(scalars.items())) + payload = f'{{"{element_param}": _item{shared}}}' + expr = _execute_activity_expr(node, cfg, payload, indent=" " * 20) + return "\n".join( + [ + f" # fan_out: {node.id} runs once per element of the bound", + " # list, each as its own activity execution.", + f" {node.id}_result = list(", + " await asyncio.gather(", + " *(", + f" {expr}", + f" for _item in {list_expr}", + " )", + " )", + " )", + ] + ) def _emit_hitl_block(node: Node) -> str: @@ -432,33 +469,27 @@ def _emit_workflow_run(pipeline: Pipeline, cfg: TemporalAdapterConfig) -> str: body_lines.append("") body_lines.append(f" # ─── Wave {wave_idx} ───") - if len(non_hitl) == 1: - body_lines.append(_emit_wave_call(non_hitl[0], cfg).rstrip("\n")) - elif len(non_hitl) > 1: + # fan_out nodes dispatch once per element of their bound list — + # they never share the single/parallel payload shapes below. + fanned, plain = fan_out_nodes(non_hitl) + + if len(plain) == 1: + body_lines.append(_emit_wave_call(plain[0], cfg).rstrip("\n")) + elif len(plain) > 1: # Parallel via asyncio.gather body_lines.append(" (") - for n in non_hitl: + for n in plain: body_lines.append(f" {n.id}_result,") body_lines.append(" ) = await asyncio.gather(") - for n in non_hitl: - timeout = _activity_timeout(n, cfg.default_activity_timeout) + for n in plain: payload = _payload_literal(n, indent=" " * 16) - timeout_line = ( - " start_to_close_timeout=" - f'timedelta(minutes=_parse_minutes("{timeout}")),\n' - ) - # A node must not lose its declared retry budget just for - # having a sibling in its wave — parallel fetch waves are - # exactly where the flaky network calls live. - retry = _retry_policy_args(n) - retry_line = f" retry_policy={retry},\n" if retry else "" - body_lines.append( - f" workflow.execute_activity(\n" - f' "{n.id}",\n' - f" {payload},\n" + timeout_line + retry_line + " )," - ) + expr = _execute_activity_expr(n, cfg, payload, indent=" " * 12) + body_lines.append(f" {expr},") body_lines.append(" )") + for n in fanned: + body_lines.append(_emit_fan_out_call(n, pipeline, cfg)) + for h in hitl: body_lines.append(_emit_hitl_block(h).rstrip("\n")) @@ -527,7 +558,16 @@ def _parse_minutes(s: str) -> float: return float(s[:-1]) * 60 * 24 return float(s) + ''' + ).lstrip("\n") + ) + + if any(n.fan_out for n in pipeline.nodes): + parts.append("\n" + fan_out_list_helper() + "\n") + parts.append( + textwrap.dedent( + f''' @workflow.defn(name="{versioned_workflow_name}") class {class_name}: """Compiled workflow for {pipeline.name}.""" diff --git a/tests/_helpers.py b/tests/_helpers.py index 393042b..99ff299 100644 --- a/tests/_helpers.py +++ b/tests/_helpers.py @@ -74,3 +74,47 @@ def mini_pipeline(node: Node) -> Pipeline: entry_nodes=[node.id], exit_nodes=[node.id], ) + + +# ───────── fan_out mock support (live e2e) ───────── +# +# Every live e2e replaces node implementations with recording mocks. A +# fan_out node dispatches once per element of an upstream list, so the +# mock standing in for that upstream node has to return a real list — +# otherwise the fan either dispatches zero times (empty list, node +# silently never runs) or trips the fanOutList guard (missing key). +# +# Derived from the IR rather than hardcoded per suite: the three TS e2e +# files had already encoded `exclusion_check_sequence` returning +# `passed: []`, which made the BDR fan a no-op that no assertion caught. + +#: Elements per fanned list in mocked runs. Must exceed 1, or +#: per-element dispatch is indistinguishable from batch dispatch. +FAN_OUT_ELEMENTS = 3 + + +def fan_out_element(index: int) -> dict[str, int]: + """One canned element of a mocked fanned list.""" + return {"fanElement": index} + + +def fan_out_source_keys(pipeline: Pipeline) -> dict[str, dict[str, list[dict[str, int]]]]: + """``{upstream_node: {field: [element, ...]}}`` for every fan_out node. + + Uses the adapters' own element-param resolution, so a mock can never + disagree with the emitted code about which input is the fanned list. + """ + from rote.adapters._common import fan_out_element_param + from rote.ir import parse_input_ref + + extra: dict[str, dict[str, list[dict[str, int]]]] = {} + for node in pipeline.nodes: + if not node.fan_out or not node.inputs: + continue + parsed = parse_input_ref(node.inputs[fan_out_element_param(node, pipeline)]) + if parsed.node_id is None or parsed.field is None: + continue + extra.setdefault(parsed.node_id, {})[parsed.field] = [ + fan_out_element(i) for i in range(FAN_OUT_ELEMENTS) + ] + return extra diff --git a/tests/test_cloudflare_adapter.py b/tests/test_cloudflare_adapter.py index 1f7e749..3417d99 100644 --- a/tests/test_cloudflare_adapter.py +++ b/tests/test_cloudflare_adapter.py @@ -249,6 +249,14 @@ def test_workflow_has_step_do_for_every_non_hitl_node( continue if node.id in nested_ids: continue + if node.fan_out: + # One step per element, so the name is an indexed template + # literal rather than a constant — asserting the constant + # form here would silently accept a batch dispatch. + assert f"`{node.id}[${{_index}}]`" in src, ( + f"Missing per-element step.do call for fan_out node {node.id!r}" + ) + continue # Indentation differs by dispatch form: sequential (12), inside a # Promise.all wave (16), and inside a park-on-auth retry loop # (20, where the step name is a ternary on the attempt counter). diff --git a/tests/test_cloudflare_e2e.py b/tests/test_cloudflare_e2e.py index 5ae8865..53be690 100644 --- a/tests/test_cloudflare_e2e.py +++ b/tests/test_cloudflare_e2e.py @@ -36,8 +36,9 @@ import pytest +from rote.adapters._common import fan_out_element_param from rote.adapters.cloudflare import CloudflareAdapter -from rote.ir import Pipeline +from rote.ir import Pipeline, parse_input_ref REPO_ROOT = Path(__file__).resolve().parent.parent BDR_PIPELINE_YAML = REPO_ROOT / "examples" / "bdr-outreach" / "expected" / "pipeline.yaml" @@ -118,7 +119,35 @@ def _to_camel_case(s: str) -> str: return parts[0] + "".join(p.capitalize() for p in parts[1:]) -def _write_test_overlay(out_dir: Path) -> None: +#: How many elements each fan_out node's upstream list carries in the +#: mocked run. More than one, or per-element dispatch is +#: indistinguishable from batch dispatch. +FAN_OUT_ELEMENTS = 3 + + +def _fan_out_source_keys(pipeline: Pipeline) -> dict[str, dict[str, list[dict]]]: + """``{upstream_node: {field: [element, ...]}}`` for every fan_out node. + + A generic echo mock returns only ``{mocked, node, received}``, so the + key a fan_out node fans over comes back ``undefined`` and the run + dies on it. Derive the (node, field) pairs from the IR — the same + resolution the adapters use — and give each one a real list. + """ + extra: dict[str, dict[str, list[dict]]] = {} + for node in pipeline.nodes: + if not node.fan_out or not node.inputs: + continue + param = fan_out_element_param(node, pipeline) + parsed = parse_input_ref(node.inputs[param]) + if parsed.node_id is None or parsed.field is None: + continue + extra.setdefault(parsed.node_id, {})[parsed.field] = [ + {"fanElement": i} for i in range(FAN_OUT_ELEMENTS) + ] + return extra + + +def _write_test_overlay(out_dir: Path, pipeline: Pipeline) -> None: """Replace stubs with echo mocks. The emitted ``extracted/*.ts`` and ``signatures/*.ts`` modules all @@ -128,7 +157,12 @@ def _write_test_overlay(out_dir: Path) -> None: progress through its waves. The mock echoes the input it received (``received``) so the test can assert that data-flow threading delivered real payloads between steps. + + Nodes feeding a fan_out node additionally return that node's list + key, so the fan dispatches for real instead of tripping the + ``fanOutList`` guard. """ + fan_keys = _fan_out_source_keys(pipeline) src = out_dir / "src" for sub in ("extracted", "signatures"): d = src / sub @@ -144,12 +178,16 @@ def _write_test_overlay(out_dir: Path) -> None: continue node_id = f.stem fn = _to_camel_case(node_id) + extra = "".join( + f", {json.dumps(field)}: {json.dumps(value)}" + for field, value in sorted(fan_keys.get(node_id, {}).items()) + ) f.write_text( f"export async function {fn}(" f"input?: unknown, _env?: unknown" f"): Promise> {{\n" f" return {{ mocked: true, node: {json.dumps(node_id)}, " - f"received: input ?? null }};\n" + f"received: input ?? null{extra} }};\n" f"}}\n", encoding="utf-8", ) @@ -241,7 +279,7 @@ def wrangler_dev_session(bdr_pipeline: Pipeline, tmp_path_factory: pytest.TempPa out = tmp_path_factory.mktemp("cf-workflow-e2e") CloudflareAdapter().emit(bdr_pipeline, out) - _write_test_overlay(out) + _write_test_overlay(out, bdr_pipeline) npm_proc = subprocess.run( ["npm", "install", "--no-audit", "--no-fund"], @@ -447,7 +485,17 @@ def _step_output(state: dict, node: str) -> dict: # A whole-output binding: create_sales_template received # personalize_email's full step result. sales_template_received = _step_output(final, "create_sales_template")["received"] - assert sales_template_received["personalizations"]["node"] == "personalize_email" + personalizations = sales_template_received["personalizations"] + assert isinstance(personalizations, list), ( + f"personalize_email is a fan_out node, so downstream must receive one " + f"result per element; got {type(personalizations).__name__}" + ) + assert len(personalizations) == FAN_OUT_ELEMENTS + assert all(p["node"] == "personalize_email" for p in personalizations) + # Each invocation got ONE element, not the whole list. + assert [p["received"]["contact"] for p in personalizations] == [ + {"fanElement": i} for i in range(FAN_OUT_ELEMENTS) + ] assert sales_template_received["campaign_name"] == brief["drug_brand"] # Pipeline input field selection at the end of the chain. diff --git a/tests/test_dbos_adapter.py b/tests/test_dbos_adapter.py index 0b7d2c1..be8040f 100644 --- a/tests/test_dbos_adapter.py +++ b/tests/test_dbos_adapter.py @@ -765,7 +765,15 @@ def test_fan_out_node_dispatches_per_element() -> None: assert '{"post": _item, "brief": pipeline_input["brief"]}' in src, ( "element param bound per item, scalars shared" ) - assert 'for _item in filter_posts_result["published_posts"]' in src + # The bound list goes through the _fan_out_list guard, which names + # the node and the IR reference when an upstream key is missing. + assert ( + "for _item in _fan_out_list(\n" + ' filter_posts_result["published_posts"],\n' + ' "judge_content",\n' + ' "filter_posts.output.published_posts",\n' + " )" + ) in src assert "judge_content_handles = [queue.enqueue(judge_content, _p)" in src assert "judge_content_result = [_h.get_result() for _h in judge_content_handles]" in src # The whole-list single-call shape must be gone. @@ -805,7 +813,13 @@ def _impl_node(node_id: str) -> Node: ) element_param, list_expr, scalars = fan_out_binding(judge, pipeline) assert element_param == "contact" - assert list_expr == 'passed_contacts_result["passed"]' + assert list_expr == ( + "_fan_out_list(\n" + ' passed_contacts_result["passed"],\n' + ' "personalize",\n' + ' "passed_contacts.output.passed",\n' + ")" + ) assert scalars == { "intel": "research_result", "campaign": 'pipeline_input["campaign"]', diff --git a/tests/test_dbos_ts_e2e.py b/tests/test_dbos_ts_e2e.py index a401938..2d6cbed 100644 --- a/tests/test_dbos_ts_e2e.py +++ b/tests/test_dbos_ts_e2e.py @@ -48,6 +48,7 @@ from rote.adapters._common import _execution_waves, _to_camel_case from rote.adapters.dbos_ts import DbosTsAdapter from rote.ir import NodeKind, Pipeline +from tests._helpers import FAN_OUT_ELEMENTS, fan_out_element, fan_out_source_keys REPO_ROOT = Path(__file__).resolve().parent.parent BDR_PIPELINE_YAML = REPO_ROOT / "examples" / "bdr-outreach" / "expected" / "pipeline.yaml" @@ -147,8 +148,17 @@ def test_node_modules_contains_expected_packages(emitted_dir: Path) -> None: } -def _mock_output(node_id: str) -> dict: - return _MOCK_OUTPUTS.get(node_id, {"mocked": True, "node": node_id}) +def _mock_output(node_id: str, pipeline: Pipeline) -> dict: + """The canned output for a node, with any fanned list filled in. + + A node feeding a fan_out node MUST return that list non-empty: + the hardcoded `passed: []` below made BDR's fan dispatch zero + times, so personalize_email silently never ran and the live + tests still passed. + """ + out = dict(_MOCK_OUTPUTS.get(node_id, {"mocked": True, "node": node_id})) + out.update(fan_out_source_keys(pipeline).get(node_id, {})) + return out _MOCK_MODULE = """\ @@ -241,7 +251,7 @@ def _write_test_overlay(out_dir: Path, pipeline: Pipeline) -> None: fn=_to_camel_case(node.id), extra=extra, node_id=json.dumps(node.id), - output=json.dumps(_mock_output(node.id)), + output=json.dumps(_mock_output(node.id, pipeline)), ) (out_dir / "src" / sub / f"{node.id}.ts").write_text(src, encoding="utf-8") (out_dir / "src" / "e2e.ts").write_text(_DRIVER_TS, encoding="utf-8") @@ -518,7 +528,7 @@ def test_workflow_executes_through_hitl_gates( # The fan-in loop consumed both wave-1 results plus an input field. loop_payload = payloads["lead_generation_loop"] assert loop_payload["brief"] == brief - assert loop_payload["taxonomy"] == _mock_output("taxonomy_lookup") + assert loop_payload["taxonomy"] == _mock_output("taxonomy_lookup", bdr_pipeline) assert loop_payload["target_quota"] == brief["target_quota"] # The first HITL gate's resume payload flowed into hubspot_upsert via @@ -529,11 +539,30 @@ def test_workflow_executes_through_hitl_gates( # `contacts: hubspot_upsert.output.upserted`. assert payloads["exclusion_check_dnc"] == {"contacts": [{"vid": "hs-1"}]} + # ── fan_out: personalize_email ran once per surviving contact ── + # `payloads` is keyed by node so it keeps only the last record; count + # the raw records instead, and check each invocation got exactly ONE + # element rather than the whole list. + fan_records = [r for r in _recorded(record_path) if r["node"] == "personalize_email"] + assert len(fan_records) == FAN_OUT_ELEMENTS, ( + f"fan_out node personalize_email must run once per element of " + f"exclusion_check_sequence.passed; ran {len(fan_records)} time(s)" + ) + assert sorted( + (r["payload"]["contact"] for r in fan_records), key=lambda c: c["fanElement"] + ) == [fan_out_element(i) for i in range(FAN_OUT_ELEMENTS)], ( + "each fan_out invocation must receive one element, not the batch" + ) + # The report node received a fan-in of upstream results: # pipeline input field + two different upstream nodes. report_payload = payloads["pre_enrollment_report"] assert report_payload["campaign_name"] == brief["drug_brand"] - assert report_payload["passed_contacts"] == [] + # exclusion_check_sequence.passed is the list personalize_email + # fans over, so the mock returns it non-empty (see _mock_output). + assert report_payload["passed_contacts"] == [ + fan_out_element(i) for i in range(FAN_OUT_ELEMENTS) + ] assert report_payload["template_ids"] == ["t1", "t2"] # ── Durability: the gates were checkpointed recv steps in the system diff --git a/tests/test_fan_out_parity.py b/tests/test_fan_out_parity.py new file mode 100644 index 0000000..d6a56b0 --- /dev/null +++ b/tests/test_fan_out_parity.py @@ -0,0 +1,278 @@ +"""Cross-runtime contract: a ``fan_out`` node dispatches once per element. + +Every adapter must fan over the *same* input and hand each invocation a +single element. Until 0.12.x only DBOS did; the other five passed the +whole upstream list in one call, which meant the same ``pipeline.yaml`` +produced two incompatible contracts for a user-filled stub — a per-post +judge on DBOS and a whole-post-list judge everywhere else. That is a +runtime leaking into node semantics, which invariant #1 forbids. + +The assertions here are written to fail against batch dispatch, not just +to pass against the current output: each one checks that the element +param is bound to the loop variable AND that the whole-list expression +is *absent* from the payload. A test that only asserted "the list +expression appears somewhere" would pass either way, since per-element +dispatch iterates that same expression. +""" + +from __future__ import annotations + +import ast +import re + +import pytest + +from rote.adapters.cloudflare import emit_workflow as emit_cloudflare +from rote.adapters.dbos import emit_main as emit_dbos +from rote.adapters.dbos_ts import emit_main as emit_dbos_ts +from rote.adapters.inngest import emit_pipeline_ts as emit_inngest +from rote.adapters.python import emit_main as emit_python +from rote.adapters.temporal import emit_workflow as emit_temporal +from rote.ir import LLMSignature, Node, NodeKind, Pipeline + +# ───────── Fixture: list-producer → fan_out judge ───────── +# +# Uses signature_spec (not the legacy Python-path form) so the identical +# pipeline is emittable by all six adapters — the whole point being a +# comparison across runtimes. + +_SPEC = LLMSignature( + input_schema={ + "type": "object", + "properties": {"post": {"type": "string"}, "brief": {"type": "string"}}, + "required": ["post"], + }, + output_schema={ + "type": "object", + "properties": {"score": {"type": "number"}}, + "required": ["score"], + }, + prompt="Score this post: {{ post }}", +) + + +def _fan_out_pipeline() -> Pipeline: + return Pipeline( + name="fan", + input={"type": "In", "required": [], "optional": []}, + nodes=[ + Node( + id="filter_posts", + kind=NodeKind.PURE_FUNCTION, + description="produce the list", + impl="filters.py:filter_posts", + inputs={"raw": "pipeline.input.raw"}, + ), + Node( + id="judge_content", + kind=NodeKind.LLM_JUDGE, + description="score ONE post", + signature_spec=_SPEC, + fan_out=True, + inputs={ + # the list to fan over + "post": "filter_posts.output.published_posts", + # shared by every invocation + "brief": "pipeline.input.brief", + }, + ), + ], + edges=[{"from": "filter_posts", "to": "judge_content", "fan_out": True}], + entry_nodes=["filter_posts"], + exit_nodes=["judge_content"], + ) + + +# The expression each language uses for the whole upstream list. If this +# appears as the *value bound to the element param*, the adapter is +# batching. +_PY_LIST_EXPR = 'filter_posts_result["published_posts"]' +_TS_LIST_EXPR = '(filter_posts_result as Record)["published_posts"]' + +_EMITTERS = { + "python": (emit_python, "py"), + "temporal": (emit_temporal, "py"), + "dbos": (emit_dbos, "py"), + "cloudflare": (emit_cloudflare, "ts"), + "dbos-ts": (emit_dbos_ts, "ts"), + "inngest": (emit_inngest, "ts"), +} + + +@pytest.fixture(scope="module") +def emitted() -> dict[str, tuple[str, str]]: + """``{runtime: (source, language)}`` for one shared fan_out pipeline.""" + pipeline = _fan_out_pipeline() + return {name: (fn(pipeline), lang) for name, (fn, lang) in _EMITTERS.items()} + + +@pytest.mark.parametrize("runtime", sorted(_EMITTERS)) +def test_element_param_binds_one_element_not_the_whole_list( + runtime: str, emitted: dict[str, tuple[str, str]] +) -> None: + """The judge receives ONE post, on every runtime. + + The negative half is what makes this test worth having: batch + dispatch binds `post` straight to the list expression, and that is + exactly the string asserted absent. + """ + src, lang = emitted[runtime] + if lang == "py": + bound_to_element = '"post": _item' in src + bound_to_list = f'"post": {_PY_LIST_EXPR}' in src + else: + bound_to_element = "post: _item" in src + bound_to_list = f"post: {_TS_LIST_EXPR}" in src + + assert bound_to_element, ( + f"{runtime}: fan_out node's element param 'post' is not bound to the " + f"per-element loop variable — it must receive one element, not the batch" + ) + assert not bound_to_list, ( + f"{runtime}: fan_out node's element param 'post' is bound to the whole " + f"upstream list; this is the batch dispatch fan_out is supposed to replace" + ) + + +@pytest.mark.parametrize("runtime", sorted(_EMITTERS)) +def test_non_element_inputs_are_shared_by_every_invocation( + runtime: str, emitted: dict[str, tuple[str, str]] +) -> None: + """`brief` is not the fanned list, so every element gets it verbatim.""" + src, lang = emitted[runtime] + expected = ( + '"brief": pipeline_input["brief"]' if lang == "py" else 'brief: pipelineInput["brief"]' + ) + assert expected in src, f"{runtime}: shared input 'brief' missing from the fan_out payload" + + +@pytest.mark.parametrize("runtime", sorted(_EMITTERS)) +def test_the_bound_list_is_iterated(runtime: str, emitted: dict[str, tuple[str, str]]) -> None: + """Per-element dispatch means iterating the list, not passing it.""" + src, lang = emitted[runtime] + # The guard call wraps against each adapter's own nesting depth, so + # match the shape rather than a fixed indent. + if lang == "py": + pattern = r"for _item in _fan_out_list\(\s*" + re.escape(_PY_LIST_EXPR) + "," + iterated = re.search(pattern, src) is not None + else: + pattern = r"fanOutList\(\s*" + re.escape(_TS_LIST_EXPR) + "," + iterated = re.search(pattern, src) is not None and (").map(" in src or ").entries()" in src) + assert iterated, f"{runtime}: the fanned list is never iterated per element" + + +@pytest.mark.parametrize("runtime", sorted(_EMITTERS)) +def test_a_missing_upstream_key_names_the_node( + runtime: str, emitted: dict[str, tuple[str, str]] +) -> None: + """Every runtime reports a missing fanned key the same way. + + Regression for a live failure: the first real fan_out run on + Cloudflare died with "Cannot read properties of undefined (reading + 'map')" — no node id, no reference, inside generated code the user + never wrote. Python's bare "'NoneType' object is not iterable" is + no better. Diagnostics are part of the emitted contract, so a guard + on one runtime and not another is its own parity gap. + """ + src, lang = emitted[runtime] + helper, message = ( + ("def _fan_out_list(", "expected a list from") + if lang == "py" + else ("function fanOutList(", "expected an array from") + ) + assert helper in src, f"{runtime}: fan_out guard helper not emitted" + assert message in src, f"{runtime}: guard does not explain what it expected" + # The guard must name the node and the IR reference it came from. + assert "judge_content" in src and "filter_posts.output.published_posts" in src + + +@pytest.mark.parametrize("runtime", ["cloudflare", "inngest"]) +def test_per_element_step_names_are_unique( + runtime: str, emitted: dict[str, tuple[str, str]] +) -> None: + """Cloudflare and Inngest key a durable step by its name. + + Reusing one name for every element is the nastiest possible failure + here: Cloudflare returns the first element's cached result for all + of them, so the run *succeeds* with silently uniform output. DBOS is + exempt — it identifies a step by execution order, not by name. + """ + src, _ = emitted[runtime] + assert "judge_content[${_index}]" in src, ( + f"{runtime}: per-element steps must carry the element index in their " + f"name; a constant name collapses the fan onto one cached step" + ) + + +@pytest.mark.parametrize("runtime", sorted(_EMITTERS)) +def test_fanned_result_is_a_list(runtime: str, emitted: dict[str, tuple[str, str]]) -> None: + """Downstream nodes see one result per element, in input order.""" + src, lang = emitted[runtime] + if lang == "py": + # list(pool.map(...)) / list(await asyncio.gather(...)) / + # [_h.get_result() for _h in ...] + assert re.search(r"judge_content_result = (list\(|\[)", src), ( + f"{runtime}: fanned result must bind as a list of per-element results" + ) + else: + array_forms = ( + r"await Promise\.all\(", # cloudflare (plain), inngest + r"judge_content_settled\.map\(", # dbos-ts (allSettled + unwrap) + r"judge_content_results", # cloudflare (parkable, sequential) + ) + pattern = r"judge_content_result = (" + "|".join(array_forms) + ")" + assert re.search(pattern, src), ( + f"{runtime}: fanned result must bind as an array of per-element results" + ) + + +@pytest.mark.parametrize("runtime", ["python", "temporal", "dbos"]) +def test_emitted_python_parses(runtime: str, emitted: dict[str, tuple[str, str]]) -> None: + """The fan_out emission is syntactically valid Python.""" + src, _ = emitted[runtime] + ast.parse(src) + + +def test_all_six_runtimes_agree_on_which_input_is_the_list() -> None: + """The fanned input is a property of the IR, not of the target. + + Resolution lives in the language-neutral common module precisely so + two adapters cannot disagree about which input is the list — that + would make one pipeline mean two different things. + """ + from rote.adapters._common import fan_out_element_param + + pipeline = _fan_out_pipeline() + judge = pipeline.nodes[1] + assert fan_out_element_param(judge, pipeline) == "post" + + +def test_ambiguous_fan_out_is_an_emit_time_error() -> None: + """Two edge-fed node-bound inputs and no marker → refuse, never guess. + + Picking the wrong list silently judges the wrong things, which is + far worse than failing to emit. + """ + from rote.adapters._common import fan_out_element_param + + pipeline = Pipeline( + name="ambiguous", + input={"type": "In", "required": [], "optional": []}, + nodes=[ + Node(id="a", kind=NodeKind.PURE_FUNCTION, description="d", impl="m.py:a"), + Node(id="b", kind=NodeKind.PURE_FUNCTION, description="d", impl="m.py:b"), + Node( + id="j", + kind=NodeKind.LLM_JUDGE, + description="d", + signature_spec=_SPEC, + fan_out=True, + inputs={"post": "a.output.xs", "brief": "b.output.ys"}, + ), + ], + edges=[{"from": "a", "to": "j"}, {"from": "b", "to": "j"}], + entry_nodes=["a", "b"], + exit_nodes=["j"], + ) + with pytest.raises(ValueError, match="cannot identify the element param"): + fan_out_element_param(pipeline.nodes[2], pipeline) diff --git a/tests/test_inngest_adapter.py b/tests/test_inngest_adapter.py index dd0af3c..54887ac 100644 --- a/tests/test_inngest_adapter.py +++ b/tests/test_inngest_adapter.py @@ -237,6 +237,14 @@ def test_step_run_for_every_non_hitl_top_level_node( for node in bdr_pipeline.nodes: if node.kind is NodeKind.HITL_GATE or node.id in nested_ids: continue + if node.fan_out: + # One step per element, so the id is an indexed template + # literal rather than a constant — asserting the constant + # form here would silently accept a batch dispatch. + assert f"step.run(`{node.id}[${{_index}}]`" in pipeline_src, ( + f"Missing per-element step.run call for fan_out node {node.id!r}" + ) + continue assert f'step.run("{node.id}"' in pipeline_src, ( f"Missing step.run call for node {node.id!r} in pipeline.ts" ) diff --git a/tests/test_inngest_e2e.py b/tests/test_inngest_e2e.py index fd42488..4e2b94f 100644 --- a/tests/test_inngest_e2e.py +++ b/tests/test_inngest_e2e.py @@ -59,6 +59,7 @@ from rote.adapters._common import _execution_waves, _to_camel_case from rote.adapters.inngest import InngestAdapter, gate_event_name, trigger_event_name from rote.ir import NodeKind, Pipeline +from tests._helpers import FAN_OUT_ELEMENTS, fan_out_element, fan_out_source_keys REPO_ROOT = Path(__file__).resolve().parent.parent BDR_PIPELINE_YAML = REPO_ROOT / "examples" / "bdr-outreach" / "expected" / "pipeline.yaml" @@ -153,8 +154,17 @@ def test_node_modules_contains_expected_packages(emitted_dir: Path) -> None: } -def _mock_output(node_id: str) -> dict: - return _MOCK_OUTPUTS.get(node_id, {"mocked": True, "node": node_id}) +def _mock_output(node_id: str, pipeline: Pipeline) -> dict: + """The canned output for a node, with any fanned list filled in. + + A node feeding a fan_out node MUST return that list non-empty: + the hardcoded `passed: []` below made BDR's fan dispatch zero + times, so personalize_email silently never ran and the live + tests still passed. + """ + out = dict(_MOCK_OUTPUTS.get(node_id, {"mocked": True, "node": node_id})) + out.update(fan_out_source_keys(pipeline).get(node_id, {})) + return out _MOCK_MODULE = """\ @@ -180,7 +190,7 @@ def _write_test_overlay(out_dir: Path, pipeline: Pipeline) -> None: fn=_to_camel_case(node.id), extra=extra, node_id=json.dumps(node.id), - output=json.dumps(_mock_output(node.id)), + output=json.dumps(_mock_output(node.id, pipeline)), ) (out_dir / "src" / sub / f"{node.id}.ts").write_text(src, encoding="utf-8") @@ -516,7 +526,7 @@ def test_workflow_executes_through_hitl_gates( # The fan-in loop consumed both wave-1 results plus an input field. loop_payload = payloads["lead_generation_loop"] assert loop_payload["brief"] == brief - assert loop_payload["taxonomy"] == _mock_output("taxonomy_lookup") + assert loop_payload["taxonomy"] == _mock_output("taxonomy_lookup", bdr_pipeline) assert loop_payload["target_quota"] == brief["target_quota"] # The first HITL gate's resume payload flowed into hubspot_upsert via @@ -527,19 +537,44 @@ def test_workflow_executes_through_hitl_gates( # `contacts: hubspot_upsert.output.upserted`. assert payloads["exclusion_check_dnc"] == {"contacts": [{"vid": "hs-1"}]} + # ── fan_out: personalize_email ran once per surviving contact ── + # `payloads` is keyed by node so it keeps only the last record; count + # the raw records instead, and check each invocation got exactly ONE + # element rather than the whole list. + fan_records = [r for r in _recorded(record_path) if r["node"] == "personalize_email"] + assert len(fan_records) == FAN_OUT_ELEMENTS, ( + f"fan_out node personalize_email must run once per element of " + f"exclusion_check_sequence.passed; ran {len(fan_records)} time(s)" + ) + assert sorted( + (r["payload"]["contact"] for r in fan_records), key=lambda c: c["fanElement"] + ) == [fan_out_element(i) for i in range(FAN_OUT_ELEMENTS)], ( + "each fan_out invocation must receive one element, not the batch" + ) + # The report node received a fan-in of upstream results: # pipeline input field + two different upstream nodes. report_payload = payloads["pre_enrollment_report"] assert report_payload["campaign_name"] == brief["drug_brand"] - assert report_payload["passed_contacts"] == [] + # exclusion_check_sequence.passed is the list personalize_email + # fans over, so the mock returns it non-empty (see _mock_output). + assert report_payload["passed_contacts"] == [ + fan_out_element(i) for i in range(FAN_OUT_ELEMENTS) + ] assert report_payload["template_ids"] == ["t1", "t2"] # Every non-HITL top-level node ran exactly once — memoized steps # are not re-executed across the executor's re-invocations of the - # handler (the durable-execution contract). + # handler (the durable-execution contract). A fan_out node is the + # one exception: it legitimately runs once per element, and its + # per-element step ids are what keep those from memoizing onto each + # other (a constant id would collapse them to a single execution). + fan_out_ids = {n.id for n in bdr_pipeline.nodes if n.fan_out} node_counts: dict[str, int] = {} for r in _recorded(record_path): node_counts[r["node"]] = node_counts.get(r["node"], 0) + 1 - assert all(c == 1 for c in node_counts.values()), ( - f"steps must execute exactly once (memoization); counts: {node_counts}" + expected_counts = {nid: (FAN_OUT_ELEMENTS if nid in fan_out_ids else 1) for nid in node_counts} + assert node_counts == expected_counts, ( + f"steps must execute exactly once (memoization), or once per element " + f"for fan_out nodes; counts: {node_counts}" ) From 4fa180ba97579fdad8fcfaeb88171d424a31670b Mon Sep 17 00:00:00 2001 From: Trevor Hudson Date: Thu, 30 Jul 2026 15:49:08 -0700 Subject: [PATCH 2/2] Close three test gaps found by mutation-testing the adapters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up audit to the fan_out work. The recurring failure mode in this repo is a test that observes something *adjacent* to the behavior, so this measures the suite instead of trusting it: apply a semantic regression to an adapter, run the suite, and anything that still passes is an untested behavior. 18 mutations (retry policy, timeouts, step naming, payload threading, MCP allowlist narrowing, the ANTHROPIC_API_KEY scrub, invariants #4 and #7); 15 caught, 3 survived. **temporal: per-node `timeout:` was ignored and nothing noticed.** Making `_activity_timeout` return the default unconditionally left the whole suite green. A step declaring a 30s budget silently got 5m — five times too long on a wedged call, with nothing in the emitted code hinting why. **cloudflare: same gap on HITL gates**, and worse there, because `waitForEvent` THROWS on timeout: a gate inheriting the 7d default instead of its declared 1h turns a fast-fail approval window into a week-long hang, and the reverse fails a legitimately long wait. Both new tests carry a negative half — asserting only that the declared value appears would still pass if the adapter emitted it for every node, which is exactly how the original gap hid. Writing the cloudflare one immediately surfaced a real emitted-output bug: `1h` rendered as `"1 hours"`. Cloudflare's WorkflowDuration accepts either spelling so nothing broke, but emitted code is a reviewed artifact and "1 hours" reads as a defect. Now singularized. **The third survivor is an equivalent mutant, not a gap.** Swapping the tier order in `fan_out_element_param` cannot change the answer: fan_edge_sources ⊆ edge_sources, and a tier is accepted only when it singles out exactly one param, so if the broader tier singles one out it is necessarily the same one. Commented in place so the next sweep doesn't chase it. The precedence test added alongside is still real — verified by a mutation that IS observable (always take the alphabetically-first param), which it catches. **Also: the temporal e2e's fan was a no-op.** Its `exclusion_check_sequence` mock returned `passed: []`, so personalize_email never ran — the same vacuous-fixture bug just fixed in the three TS suites, missed because that suite passed and I only chased failures. Its `CAPTURED_PAYLOADS` dict is keyed by node and so keeps only the last call, structurally unable to observe a fan at all; an ordered CAPTURE_LOG now sits alongside it. Verified by mutating the snapshot the e2e actually imports back to batch dispatch: it fails with "expected 3 activity executions, got 1". That last point is worth knowing: the temporal e2e imports the committed `examples/.../runtimes/temporal/workflow.py` snapshot, not fresh emission, so mutating the adapter does not reach it. The snapshot is regenerated by the adapter tests' fixtures (which emit into the repo on purpose, so `git diff` shows emission changes at review). Audited two more patterns and found no open gaps: value-taking flags asserted only by presence (`--output-last-message`, `--append-system-prompt`) turn out to have their values checked elsewhere, and the remaining empty-collection fixtures are inert because nothing iterates them. CLAUDE.md gains a "Testing discipline" section with the five defects of this shape, the three habits that catch them, and the instruction to verify a new test by breaking the code rather than by inspection. 1162 fast + 27 slow pass; ruff, strict mypy, sanity-check clean. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01SBuSDL7Svm7iDzgmhTpDSW --- CLAUDE.md | 49 +++++++++++++++++++++++++++- src/rote/adapters/_common.py | 19 ++++++++++- tests/test_cloudflare_adapter.py | 49 ++++++++++++++++++++++++++++ tests/test_fan_out_parity.py | 55 +++++++++++++++++++++++++++++++- tests/test_temporal_adapter.py | 49 ++++++++++++++++++++++++++++ tests/test_temporal_e2e.py | 34 ++++++++++++++++++-- 6 files changed, 250 insertions(+), 5 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a60a7bb..37cd812 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -798,6 +798,53 @@ If a change of yours would violate any of them, stop and reconsider. --- +## Testing discipline: make assertions capable of failing + +Five real defects in this repo shipped past a green suite because a +test observed something *adjacent* to the behavior. This is the single +most common failure mode here, so it gets its own section. + +| Defect | What the test checked | What it should have checked | +| --- | --- | --- | +| bare `--tools` killed the whole subscription lane | `"--tools" in command` | the flag's **value** | +| DBOS e2e mocked agent loops via an unreachable branch | that the overlay file was written | that the mock was actually *called* | +| BDR's fan was a no-op in 4 live e2e suites | that the workflow completed | that the fanned node ran **N** times | +| per-node `timeout:` ignored (temporal, cloudflare gates) | that *a* timeout was emitted | that the **declared** one was, and the default was not | +| `1h` emitted as `"1 hours"` | nothing — no per-node timeout test existed | — | + +Three habits that catch these: + +- **Assert the value, not the presence.** A flag, key, or node id + appearing in the output says nothing about what it was set to. +- **Give a negative half.** Pair "the declared value is present" with + "the default is absent" — otherwise emitting both still passes. +- **Make fixtures non-degenerate.** An empty list makes a fan, a loop, + or a retry a *correct* no-op that no assertion can catch. Fixture + collections that drive iteration count carry >1 element + (`tests/_helpers.py::FAN_OUT_ELEMENTS`), and their values are derived + from the IR (`fan_out_source_keys`) so a mock cannot disagree with the + emitted code about which input is which. + +**Verify a new test by breaking the code.** Not by inspection — apply +the regression the test is meant to catch and confirm it fails. Any +test written for a bug fix should be run against the *unfixed* code +once. When the fix spans runtimes, check the split: the fan_out parity +suite fails on the five broken runtimes and passes on `dbos`, which +already worked, proving it recognizes correct behavior rather than just +matching new strings. + +A periodic mutation sweep is the systematic version, and it works well +here because emission is pure template substitution — mutate an adapter, +run `pytest tests/`, and any mutation that survives is an untested +behavior. The sweep that produced the table above ran 18 mutations +(retry policy, timeouts, step naming, payload threading, MCP allowlist +narrowing, the `ANTHROPIC_API_KEY` scrub, invariant #4 and #7) and +caught 15. Watch for **equivalent mutants** — the fan_out tier ordering +in `fan_out_element_param` survives because it is provably unobservable, +not because it is untested; that one is commented in place. + +--- + ## Workflow expectations ### Running tests @@ -1092,7 +1139,7 @@ Don't waste time debugging stubs. These are intentional. pipeline with cross-process gate signaling via `DBOSClient`; real judge usage captured through the emitted `$ROTE_USAGE_LOG` hook; measurements appended to `~/.local/share/rote/eval-corpus.jsonl`) -- 1187 tests (1160 fast + 27 slow). Run with `pytest tests/` (fast +- 1189 tests (1162 fast + 27 slow). Run with `pytest tests/` (fast only — what runs by default). Slow tests cover the runtime e2e suites (Temporal, Cloudflare, DBOS, DBOS-TS, Inngest, MCP-over-stdio); the TS ones require a Node toolchain, DBOS-TS diff --git a/src/rote/adapters/_common.py b/src/rote/adapters/_common.py index 47cae77..7908ea7 100644 --- a/src/rote/adapters/_common.py +++ b/src/rote/adapters/_common.py @@ -306,12 +306,20 @@ def ir_duration_to_human(s: str) -> str: Cloudflare's step config accepts directly. Strings that don't match the IR shorthand pattern are passed through unchanged (they're assumed to already be in an acceptable form). + + A value of exactly 1 is singularized ('1 hour', not '1 hours'). + Cloudflare's ``WorkflowDuration`` accepts either spelling, so this + is legibility rather than correctness — but emitted code is read by + humans, and '1 hours' in a reviewed artifact looks like a bug. """ s = s.strip() m = _IR_DURATION_RE.fullmatch(s) if not m: return s - return f"{m.group(1)} {_UNIT_TO_HUMAN[m.group(2)]}" + value, unit = m.group(1), _UNIT_TO_HUMAN[m.group(2)] + if float(value) == 1: + unit = unit.removesuffix("s") + return f"{value} {unit}" _UNIT_TO_SECONDS = { @@ -456,6 +464,15 @@ def fan_out_element_param(node: Node, pipeline: Pipeline) -> str: edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id} fan_edge_sources = {e.from_ for e in pipeline.edges if e.to == node.id and e.fan_out} + # The tiers are tried narrowest-first, but the ORDER is provably + # unobservable: fan_edge_sources ⊆ edge_sources by construction, so + # matching(fan) ⊆ matching(edge), and a tier is accepted only when it + # singles out exactly one param. If the broader tier singles one out, + # it is necessarily the same one. (Mutation testing flags swapping + # these as a surviving mutant — it is equivalent, not a test gap.) + # The tiers still earn their place: they express which signal the + # author gave, and tier 2 is what handles a marker pointing at a node + # no input references. for sources in (fan_edge_sources, edge_sources): matching = sorted(p for p, src in node_bound.items() if src in sources) if len(matching) == 1: diff --git a/tests/test_cloudflare_adapter.py b/tests/test_cloudflare_adapter.py index 3417d99..1c61c4f 100644 --- a/tests/test_cloudflare_adapter.py +++ b/tests/test_cloudflare_adapter.py @@ -34,6 +34,7 @@ _pipeline_hash, _to_camel_case, _validate_signal_name, + emit_workflow, ) from rote.ir import LLMSignature, Node, NodeKind, Pipeline, RetryPolicy from tests._helpers import assert_no_mcp_in_ts @@ -84,6 +85,11 @@ def test_ir_duration_to_cf_conversion() -> None: assert _ir_duration_to_cf("7d") == "7 days" assert _ir_duration_to_cf("2h") == "2 hours" assert _ir_duration_to_cf("250ms") == "250 milliseconds" + # Exactly 1 singularizes — emitted code is read by humans, and + # "1 hours" in a reviewed artifact reads as a bug. + assert _ir_duration_to_cf("1h") == "1 hour" + assert _ir_duration_to_cf("1d") == "1 day" + assert _ir_duration_to_cf("1s") == "1 second" # Already human-readable: pass-through. assert _ir_duration_to_cf("10 minutes") == "10 minutes" @@ -975,3 +981,46 @@ def test_parallel_wave_keeps_per_node_step_config() -> None: assert 'timeout: "9 minutes"' in src assert "retries: { limit: 4," in src + + +def test_hitl_gate_uses_its_declared_timeout_not_the_default() -> None: + """A gate's `timeout:` must reach its `waitForEvent` config. + + Found by mutation testing: hardcoding the adapter default left the + suite green. This one is worse than it looks on Cloudflare, because + `waitForEvent` THROWS on timeout rather than returning — a gate that + silently inherits a 7-day default instead of its declared 1-hour + budget turns a fast-fail approval window into a week-long hang, and + the reverse turns a legitimate week-long wait into a failed run. + + Asserting only that the pinned value appears would still pass if the + adapter emitted it for every gate, so both gates are checked. + """ + pinned = Node( + id="quick_gate", + kind=NodeKind.HITL_GATE, + description="d", + signal="quick_approved", + timeout="1h", + ) + defaulted = Node( + id="slow_gate", + kind=NodeKind.HITL_GATE, + description="d", + signal="slow_approved", + ) + pipeline = Pipeline( + name="gates", + input={"type": "In", "required": [], "optional": []}, + nodes=[pinned, defaulted], + edges=[{"from": "quick_gate", "to": "slow_gate"}], + entry_nodes=["quick_gate"], + exit_nodes=["slow_gate"], + ) + src = emit_workflow(pipeline, CloudflareAdapterConfig(default_hitl_timeout="7d")) + + quick = src[src.index('"quick_gate"') : src.index('"slow_gate"')] + slow = src[src.index('"slow_gate"') :] + assert '"1 hour"' in quick, "a gate's declared timeout must reach waitForEvent" + assert '"7 days"' in slow, "a gate without a timeout falls back to the default" + assert '"7 days"' not in quick diff --git a/tests/test_fan_out_parity.py b/tests/test_fan_out_parity.py index d6a56b0..564fc3a 100644 --- a/tests/test_fan_out_parity.py +++ b/tests/test_fan_out_parity.py @@ -28,7 +28,7 @@ from rote.adapters.inngest import emit_pipeline_ts as emit_inngest from rote.adapters.python import emit_main as emit_python from rote.adapters.temporal import emit_workflow as emit_temporal -from rote.ir import LLMSignature, Node, NodeKind, Pipeline +from rote.ir import Edge, LLMSignature, Node, NodeKind, Pipeline # ───────── Fixture: list-producer → fan_out judge ───────── # @@ -276,3 +276,56 @@ def test_ambiguous_fan_out_is_an_emit_time_error() -> None: ) with pytest.raises(ValueError, match="cannot identify the element param"): fan_out_element_param(pipeline.nodes[2], pipeline) + + +def test_the_fan_out_edge_marker_beats_a_plain_edge() -> None: + """Two edge-fed inputs, one edge marked `fan_out: true` — the marker wins. + + The precedence tiers only differ when more than one node-bound input + is edge-fed, so a single-edge fixture exercises tier 1 and tier 2 + identically and cannot tell them apart. Found by mutation testing: + swapping the tier order left the whole suite green, because nothing + covered the case the ordering exists for. + + Getting this wrong fans over the wrong upstream list, which judges + the wrong things silently rather than failing. + """ + from rote.adapters._common import fan_out_element_param + + judge = Node( + id="j", + kind=NodeKind.LLM_JUDGE, + description="d", + signature_spec=_SPEC, + fan_out=True, + inputs={"post": "posts.output.items", "brief": "briefs.output.items"}, + ) + pipeline = Pipeline( + name="two-edges", + input={"type": "In", "required": [], "optional": []}, + nodes=[ + Node(id="posts", kind=NodeKind.PURE_FUNCTION, description="d", impl="m.py:posts"), + Node(id="briefs", kind=NodeKind.PURE_FUNCTION, description="d", impl="m.py:briefs"), + judge, + ], + # BOTH feed the judge; only `posts` is marked as the fanned list. + edges=[ + {"from": "posts", "to": "j", "fan_out": True}, + {"from": "briefs", "to": "j"}, + ], + entry_nodes=["posts", "briefs"], + exit_nodes=["j"], + ) + assert fan_out_element_param(judge, pipeline) == "post" + + # And symmetrically, so the test cannot pass by dict/sort order: + # move the marker to the other edge and the answer must follow it. + flipped = pipeline.model_copy( + update={ + "edges": [ + Edge.model_validate({"from": "posts", "to": "j"}), + Edge.model_validate({"from": "briefs", "to": "j", "fan_out": True}), + ] + } + ) + assert fan_out_element_param(judge, flipped) == "brief" diff --git a/tests/test_temporal_adapter.py b/tests/test_temporal_adapter.py index 82f0477..bf1ab79 100644 --- a/tests/test_temporal_adapter.py +++ b/tests/test_temporal_adapter.py @@ -24,6 +24,7 @@ _execution_waves, _pipeline_hash, _to_pascal_case, + emit_workflow, ) from rote.ir import Node, NodeKind, Pipeline from tests._helpers import mini_pipeline @@ -548,3 +549,51 @@ def test_refuses_mcp_only_node_with_actionable_error(tmp_path: Path) -> None: def test_emit_activities_also_refuses_mcp_only_node() -> None: with pytest.raises(ValueError, match="no MCP backend"): TemporalAdapter().emit_activities(_mcp_only_pipeline()) + + +def test_per_node_timeout_reaches_the_activity_call() -> None: + """A node's declared `timeout:` must beat the adapter default. + + Found by mutation testing: making `_activity_timeout` return the + default unconditionally left the entire suite green. A pipeline that + declares a 30-second budget for a fast step and gets the 5-minute + default instead hangs five times too long on a wedged call, with + nothing in the emitted code hinting why. + + Both halves matter — asserting only that "30s" appears would still + pass if the adapter emitted it for every node. + """ + pinned = Node( + id="fast_step", + kind=NodeKind.PURE_FUNCTION, + description="d", + impl="m.py:fast", + timeout="30s", + ) + defaulted = Node( + id="slow_step", + kind=NodeKind.PURE_FUNCTION, + description="d", + impl="m.py:slow", + ) + pipeline = Pipeline( + name="timeouts", + input={"type": "In", "required": [], "optional": []}, + nodes=[pinned, defaulted], + edges=[{"from": "fast_step", "to": "slow_step"}], + entry_nodes=["fast_step"], + exit_nodes=["slow_step"], + ) + cfg = TemporalAdapterConfig(default_activity_timeout="5m") + src = emit_workflow(pipeline, cfg) + + fast_call = src[src.index('"fast_step"') : src.index('"slow_step"')] + slow_call = src[src.index('"slow_step"') :] + assert '_parse_minutes("30s")' in fast_call, ( + "a node's declared timeout must reach its execute_activity call" + ) + assert '_parse_minutes("5m")' in slow_call, ( + "a node without a timeout must fall back to the adapter default" + ) + # The pinned node must NOT also carry the default. + assert '_parse_minutes("5m")' not in fast_call diff --git a/tests/test_temporal_e2e.py b/tests/test_temporal_e2e.py index 719e05a..b04ae04 100644 --- a/tests/test_temporal_e2e.py +++ b/tests/test_temporal_e2e.py @@ -29,6 +29,8 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import UnsandboxedWorkflowRunner, Worker +from tests._helpers import FAN_OUT_ELEMENTS, fan_out_element + REPO_ROOT = Path(__file__).resolve().parent.parent BDR_EXAMPLE_PKG_ROOT = REPO_ROOT / "examples" / "bdr-outreach" @@ -46,10 +48,15 @@ # not just that the orchestration completes. CAPTURED_PAYLOADS: dict[str, dict] = {} +#: Every (node, payload) in call order. CAPTURED_PAYLOADS is keyed by +#: node so it keeps only the last call — structurally unable to see a +#: fan_out node's N invocations. +CAPTURE_LOG: list[tuple[str, dict]] = [] def _capture(name: str, payload: dict) -> None: CAPTURED_PAYLOADS[name] = payload + CAPTURE_LOG.append((name, payload)) @activity.defn(name="target_research") @@ -132,11 +139,17 @@ async def mock_exclusion_check_recent(payload: dict) -> dict: @activity.defn(name="exclusion_check_sequence") async def mock_exclusion_check_sequence(payload: dict) -> dict: - return {"passed": [], "excluded": []} + # Non-empty: personalize_email fans over this list, and an empty + # one makes the fan a correct no-op that no assertion can catch. + return { + "passed": [fan_out_element(i) for i in range(FAN_OUT_ELEMENTS)], + "excluded": [], + } @activity.defn(name="personalize_email") async def mock_personalize_email(payload: dict) -> dict: + _capture("personalize_email", payload) return {"opening_line": "Hi there,", "ta_callout": "rare disease"} @@ -219,6 +232,7 @@ async def test_bdr_workflow_runs_to_completion(bdr_workflow_class) -> None: # n right payloads to the mocked activities along the way. """ CAPTURED_PAYLOADS.clear() + CAPTURE_LOG.clear() # A complete brief matching the pipeline's input contract. Values # reuse the fictionalized examples already present in the IR's @@ -301,5 +315,21 @@ async def test_bdr_workflow_runs_to_completion(bdr_workflow_class) -> None: # n # pipeline input field + two different upstream nodes. report_payload = CAPTURED_PAYLOADS["pre_enrollment_report"] assert report_payload["campaign_name"] == brief["drug_brand"] - assert report_payload["passed_contacts"] == [] + assert report_payload["passed_contacts"] == [ + fan_out_element(i) for i in range(FAN_OUT_ELEMENTS) + ] assert report_payload["template_ids"] == ["t1", "t2"] + + # ─── fan_out: personalize_email ran once per surviving contact ─── + # CAPTURED_PAYLOADS is keyed by node, so it can only ever show one + # invocation; the ordered log is what makes a fan observable. + fan_calls = [p for (n, p) in CAPTURE_LOG if n == "personalize_email"] + assert len(fan_calls) == FAN_OUT_ELEMENTS, ( + f"personalize_email is a fan_out node over " + f"exclusion_check_sequence.passed; expected {FAN_OUT_ELEMENTS} " + f"activity executions, got {len(fan_calls)}" + ) + # Each invocation got ONE contact, not the whole list, and the + # non-fanned inputs are shared verbatim across all of them. + assert sorted(c["contact"]["fanElement"] for c in fan_calls) == list(range(FAN_OUT_ELEMENTS)) + assert {c["campaign_type"] for c in fan_calls} == {brief["campaign_type"]}