test(testgen): async channel pipeline (scheduler differential)

Generate a self-contained async program: a producer sends k values, a consumer
receives exactly k (matched counts + buffered channel => no deadlock), counts
them, and returns the count, which main prints. Exercises the cooperative
scheduler, `task_group` / `async let`, `Channel`/`Sender`/`Receiver`, `await`,
`take`, and `?` end-to-end across every backend — including the compiled-Rust
lowering. Completion and the printed count are deterministic, so it is safe for
the value differential (this is the determinism guard the plan called for).

Smoke (>=95% accept, vm==jit) and full N-way incl. compiled (50 cases) green.

Note: while building this, the framework surfaced a minor lowering bug — a source
file named after a Rust keyword (e.g. `async.rss`) lowers to `async::main()`,
invalid Rust. Not triggered by the generator (it uses non-keyword names); filed
for a separate fix (sanitize keyword module names).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Haofei

crates/rss-testgen/src/generator.rs

@@ -98,6 +98,11 @@ impl<'a> Generator<'a> {
     // -- top level -------------------------------------------------------
 
     fn program(&mut self) -> GeneratedProgram {
+        // Occasionally emit a self-contained async channel pipeline instead.
+        if self.seed.choice(5) == 0 {
+            return self.gen_async_program();
+        }
+
         // Always declare `features: local` so generated `local` bindings are
         // allowed; the feature is harmless when unused.
         let mut source = String::from("features: local\n\n");
@@ -139,6 +144,54 @@ impl<'a> Generator<'a> {
         }
     }
 
+    /// A self-contained async channel pipeline: a producer sends `k` values, a
+    /// consumer receives exactly `k` (so there is no deadlock), counts them, and
+    /// returns the count, which `main` prints. Exercises the cooperative
+    /// scheduler, `task_group` / `async let`, channels, `await`, and `?` across
+    /// every backend. Send/receive counts are matched and the channel is buffered
+    /// (capacity >= k), so completion and the printed count are deterministic.
+    fn gen_async_program(&mut self) -> GeneratedProgram {
+        let k = 1 + self.seed.choice(4); // 1..=4 items
+
+        let mut produce = String::from(
+            "async fn produce(sender: read Sender<Int>) -> Result<Unit, ChannelError> {\n",
+        );
+        for i in 0..k {
+            let value = self.seed.range_i64(0, 1000);
+            produce.push_str(&format!("    local s{i} = {value}\n"));
+            produce.push_str(&format!(
+                "    await Sender.send(sender: read sender, value: take s{i})?\n"
+            ));
+        }
+        produce.push_str("    return Ok(Unit)\n}\n\n");
+
+        let mut consume = String::from(
+            "async fn consume(receiver: read Receiver<Int>) -> Result<Int, ChannelError> {\n    let mut n = 0\n",
+        );
+        for _ in 0..k {
+            consume.push_str("    let _ = await Receiver.recv(receiver: read receiver)?\n");
+            consume.push_str("    n = n + 1\n");
+        }
+        consume.push_str("    return Ok(n)\n}\n\n");
+
+        let main = "fn main() -> Result<Unit, ChannelError> {\n    \
+             let mut channel = Channel.bounded<Int>(capacity: 4)?\n    \
+             let sender = Channel.sender(channel: read channel)\n    \
+             let receiver = Channel.receiver(channel: mut channel)?\n    \
+             task_group {\n        \
+             async let producer = produce(sender: read sender)\n        \
+             async let consumer = consume(receiver: read receiver)\n        \
+             await producer?\n        \
+             let count = await consumer?\n        \
+             Log.write(message: read String.from_int(value: count))\n    }\n    \
+             return Ok(Unit)\n}\n";
+
+        GeneratedProgram {
+            source: format!("features: async, local\n\n{produce}{consume}{main}"),
+            is_async: true,
+        }
+    }
+
     // -- type declarations -----------------------------------------------
 
     fn gen_struct_decl(&mut self, index: usize, out: &mut String) -> StructDef {