Dispatch fan_out nodes per element on all six runtimes

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 `` `<id>[${_index}]` ``, not `"<id>"`. 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

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.
 

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"
   }

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<Env, Params> {
     async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
         const pipelineInput = event.payload;
@@ -223,14 +242,21 @@ export class BdrCampaignWorkflow extends WorkflowEntrypoint<Env, Params> {
         );
 
         // ─── Wave 9 ───
-        const personalize_email_result = await step.do(
-            "personalize_email",
-            { timeout: "10 minutes" },
-            async () => personalizeEmail({
-                contact: (exclusion_check_sequence_result as Record<string, unknown>)["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<string, unknown>)["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 ───

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",

examples/bdr-outreach/expected/runtimes/dbos-ts/src/main.ts

@@ -225,6 +225,25 @@ function unwrap<T>(settled: PromiseSettledResult<T>): 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<string, unknown>)["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<string, unknown>)["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(

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",

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]

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",