diff --git a/CLAUDE.md b/CLAUDE.md index cc8e308..37cd812 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 @@ -752,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 @@ -810,17 +903,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 +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`) -- 1146 tests (1119 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/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..7908ea7 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 @@ -305,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 = { @@ -422,6 +431,72 @@ 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} + + # 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: + 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..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" @@ -249,6 +255,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). @@ -967,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_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..564fc3a --- /dev/null +++ b/tests/test_fan_out_parity.py @@ -0,0 +1,331 @@ +"""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 Edge, 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) + + +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_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}" ) 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"]}