xco

test_event.c (96706B)

---

 /*
  * test_event.c — exercises xco_latch_t, xco_select_event_t, xco_runtime_t.
  *
  * Drives hand-coded xco_step state machines through the event substrate
  * — no coroutines required. A "waiter" is the canonical pattern: try
  * the event, park if not ready, re-try after wake.
  */
 
 #include "xco.h"
 
 #include <assert.h>
 #include <stdio.h>
 
 /* ---- Waiter: blocks on one event, captures its value -------------- */
 
 typedef struct {
     xco_mach_t      base;
     xco_event_t     *e;
     xco_runtime_t   *rt;
     xco_waker_t sw;
     int          phase;
     uintptr_t    got;
 } waiter_t;
 
 static xco_step_result_t waiter_step(xco_mach_t *s, uintptr_t v) {
     waiter_t *w = (waiter_t *)s;
     switch (w->phase) {
     case 0: {
         uintptr_t out;
         if (xco_event_poll(w->e, &out, NULL)) {
             w->got   = out;
             w->phase = 2;
             return (xco_step_result_t){out, XCO_STEP_DEAD};
         }
         xco_waker_init(&w->sw, w->rt, &w->base);
         xco_event_poll(w->e, NULL, &w->sw.base);
         w->phase = 1;
         return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
     }
     case 1:
         /* v is the value handed in by the fire site (latch payload, or
          * for select, the winning index). No re-try needed. */
         w->got   = v;
         w->phase = 2;
         return (xco_step_result_t){v, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void waiter_init(waiter_t *w, xco_runtime_t *rt, xco_event_t *e) {
     w->base  = (xco_mach_t){.step = waiter_step, .status = XCO_STEP_INIT};
     w->e     = e;
     w->rt    = rt;
     w->phase = 0;
     w->got   = 0;
 }
 
 /* ---- Latch tests --------------------------------------------------- */
 
 static void test_latch_wake(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
 
     waiter_t w; waiter_init(&w, &rt, &l.base);
 
     xco_step_result_t r = xco_step(&w.base, 0);
     assert(r.status == XCO_STEP_SUSPENDED);
     assert(rt.head == NULL);          /* parked, not queued */
     assert(l.waiters == &w.sw.base);  /* armed on the latch */
 
     xco_latch_set(&l, 42);
     assert(l.waiters == NULL);        /* drained on fire */
     assert(rt.head != NULL);          /* fire enqueued the step */
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 42);
     assert(rt.head == NULL);
 }
 
 static void test_latch_already_set(void) {
     /* Already-set latch: try succeeds inline, no park, no queueing. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
     xco_latch_set(&l, 7);
 
     waiter_t w; waiter_init(&w, &rt, &l.base);
 
     xco_step_result_t r = xco_step(&w.base, 0);
     assert(r.status == XCO_STEP_DEAD);
     assert(r.value  == 7);
     assert(w.got    == 7);
     assert(rt.head  == NULL);
 }
 
 static void test_latch_multi_waiter(void) {
     /* One xco_latch_set wakes every waiter. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
 
     waiter_t a, b, c;
     waiter_init(&a, &rt, &l.base);
     waiter_init(&b, &rt, &l.base);
     waiter_init(&c, &rt, &l.base);
 
     xco_step(&a.base, 0);
     xco_step(&b.base, 0);
     xco_step(&c.base, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_SUSPENDED);
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
     assert(xco_mach_status(&c.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_set(&l, 99);
     xco_rt_run(&rt, 0);
 
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD && a.got == 99);
     assert(xco_mach_status(&b.base) == XCO_STEP_DEAD && b.got == 99);
     assert(xco_mach_status(&c.base) == XCO_STEP_DEAD && c.got == 99);
 }
 
 static void test_latch_set_idempotent(void) {
     xco_latch_t l; xco_latch_init(&l);
     xco_latch_set(&l, 1);
     xco_latch_set(&l, 2);  /* ignored */
 
     uintptr_t v = 0;
     assert(xco_event_poll(&l.base, &v, NULL));
     assert(v == 1);
 }
 
 static void test_latch_unpark(void) {
     /* Manually park then unpark — verify the waiter is removed cleanly. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
 
     xco_waker_t sw1, sw2, sw3;
     /* Step pointer is just a sentinel here; we never run them. */
     xco_waker_init(&sw1, &rt, (xco_mach_t *)0x1);
     xco_waker_init(&sw2, &rt, (xco_mach_t *)0x2);
     xco_waker_init(&sw3, &rt, (xco_mach_t *)0x3);
     xco_event_poll(&l.base, NULL, &sw1.base);
     xco_event_poll(&l.base, NULL, &sw2.base);
     xco_event_poll(&l.base, NULL, &sw3.base);
 
     /* Remove the middle one. */
     xco_event_unpark(&l.base, &sw2.base);
     assert(sw2.base.next == NULL);
 
     /* Unpark of an already-removed waiter is a no-op. */
     xco_event_unpark(&l.base, &sw2.base);
 
     /* sw1 and sw3 still on the list. */
     int seen1 = 0, seen3 = 0;
     for (xco_waiter_t *w = l.waiters; w; w = w->next) {
         if (w == &sw1.base) seen1 = 1;
         if (w == &sw3.base) seen3 = 1;
         assert(w != &sw2.base);
     }
     assert(seen1 && seen3);
 
     /* Drain so we don't leave dangling armed waiters. */
     xco_event_unpark(&l.base, &sw1.base);
     xco_event_unpark(&l.base, &sw3.base);
     assert(l.waiters == NULL);
 }
 
 /* ---- Select tests -------------------------------------------------- */
 
 static void test_select_winner(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t a, b, c;
     xco_latch_init(&a); xco_latch_init(&b); xco_latch_init(&c);
 
     xco_select_event_t s;
     xco_select_input_t inputs[3];
     xco_event_t *srcs[3] = {&a.base, &b.base, &c.base};
     xco_select_event_init(&s, inputs, 3, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &s.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(a.waiters && b.waiters && c.waiters);  /* all armed */
 
     xco_latch_set(&b, 0xBBB);
     xco_rt_run(&rt, 0);
 
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 1);                /* b's index */
 
     /* Losers were disarmed by xco_select_input_fire. */
     assert(a.waiters == NULL);
     assert(c.waiters == NULL);
 
     /* Re-trying the winning input yields its actual payload. */
     uintptr_t v;
     assert(xco_event_poll(&b.base, &v, NULL));
     assert(v == 0xBBB);
 
     xco_select_event_deinit(&s);
 }
 
 static void test_select_fast_path(void) {
     /* Input already set at init: fire immediately, never park anyone. */
     xco_latch_t a, b;
     xco_latch_init(&a); xco_latch_init(&b);
     xco_latch_set(&a, 0xAAA);
 
     xco_select_event_t s;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&a.base, &b.base};
     xco_select_event_init(&s, inputs, 2, srcs);
 
     uintptr_t v;
     assert(xco_event_poll(&s.done.base, &v, NULL));
     assert(v == 0);                /* a wins */
     assert(b.waiters == NULL);     /* nothing parked on the loser */
 
     xco_select_event_deinit(&s);
 }
 
 static void test_select_deinit_unparks(void) {
     /* Selecting then dropping (no input fires) must release input waiters. */
     xco_latch_t a, b;
     xco_latch_init(&a); xco_latch_init(&b);
 
     xco_select_event_t s;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&a.base, &b.base};
     xco_select_event_init(&s, inputs, 2, srcs);
     assert(a.waiters != NULL);
     assert(b.waiters != NULL);
 
     xco_select_event_deinit(&s);
     assert(a.waiters == NULL);
     assert(b.waiters == NULL);
 }
 
 static void test_select_compose(void) {
     /* select(select(A, B), C): firing A propagates through both layers. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t a, b, c;
     xco_latch_init(&a); xco_latch_init(&b); xco_latch_init(&c);
 
     xco_select_event_t inner;
     xco_select_input_t in_inputs[2];
     xco_event_t *in_srcs[2] = {&a.base, &b.base};
     xco_select_event_init(&inner, in_inputs, 2, in_srcs);
 
     xco_select_event_t outer;
     xco_select_input_t out_inputs[2];
     xco_event_t *out_srcs[2] = {&inner.done.base, &c.base};
     xco_select_event_init(&outer, out_inputs, 2, out_srcs);
 
     waiter_t w; waiter_init(&w, &rt, &outer.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_set(&a, 0);
     xco_rt_run(&rt, 0);
 
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);                /* outer winner: inner (index 0) */
 
     uintptr_t v;
     assert(xco_event_poll(&inner.done.base, &v, NULL));
     assert(v == 0);                    /* inner winner: a (index 0) */
 
     /* The unrelated source c was disarmed when outer fired. */
     assert(c.waiters == NULL);
     /* And b was disarmed when inner fired. */
     assert(b.waiters == NULL);
 
     xco_select_event_deinit(&outer);
     xco_select_event_deinit(&inner);
 }
 
 /* ---- All-of -------------------------------------------------------- */
 
 static void test_allof_basic(void) {
     /* Three latches: done fires only after all three set. Storage is the
      * same xco_select_event_t / xco_select_input_t — only the init call differs. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t a, b, c;
     xco_latch_init(&a); xco_latch_init(&b); xco_latch_init(&c);
 
     xco_select_event_t s;
     xco_select_input_t inputs[3];
     xco_event_t *srcs[3] = {&a.base, &b.base, &c.base};
     xco_allof_event_init(&s, inputs, 3, srcs);
     assert(!s.done.set);
     assert(a.waiters && b.waiters && c.waiters);
 
     waiter_t w; waiter_init(&w, &rt, &s.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_set(&a, 0xAAA);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);   /* still waiting */
     assert(!s.done.set);
 
     xco_latch_set(&b, 0xBBB);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(!s.done.set);
 
     xco_latch_set(&c, 0xCCC);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 2);                                 /* closing index = c */
 
     assert(inputs[0].value == 0xAAA);
     assert(inputs[1].value == 0xBBB);
     assert(inputs[2].value == 0xCCC);
 
     xco_select_event_deinit(&s);
 }
 
 static void test_allof_fast_path_partial(void) {
     /* Some ready at init: those are captured inline, no parking. */
     xco_latch_t a, b;
     xco_latch_init(&a); xco_latch_init(&b);
     xco_latch_set(&a, 0xAAA);
 
     xco_select_event_t s;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&a.base, &b.base};
     xco_allof_event_init(&s, inputs, 2, srcs);
 
     assert(!s.done.set);                   /* still need b */
     assert(a.waiters == NULL);              /* a was inline */
     assert(b.waiters == &inputs[1].w);      /* b is parked */
     assert(inputs[0].value == 0xAAA);
 
     xco_latch_set(&b, 0xBBB);
     assert(s.done.set);
     assert(inputs[1].value == 0xBBB);
     uintptr_t v;
     assert(xco_event_poll(&s.done.base, &v, NULL));
     assert(v == 1);                         /* b closed the wait */
 
     xco_select_event_deinit(&s);
 }
 
 static void test_allof_fast_path_all(void) {
     /* All ready at init: done fires immediately, no waiter is ever parked. */
     xco_latch_t a, b;
     xco_latch_init(&a); xco_latch_init(&b);
     xco_latch_set(&a, 0xAAA);
     xco_latch_set(&b, 0xBBB);
 
     xco_select_event_t s;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&a.base, &b.base};
     xco_allof_event_init(&s, inputs, 2, srcs);
 
     assert(s.done.set);
     uintptr_t v;
     assert(xco_event_poll(&s.done.base, &v, NULL));
     assert(v == 1);                         /* last input's index */
     assert(inputs[0].value == 0xAAA);
     assert(inputs[1].value == 0xBBB);
 
     xco_select_event_deinit(&s);
 }
 
 static void test_allof_empty(void) {
     /* n=0: fires immediately with payload 0. */
     xco_select_event_t s;
     xco_allof_event_init(&s, NULL, 0, NULL);
     assert(s.done.set);
     uintptr_t v;
     assert(xco_event_poll(&s.done.base, &v, NULL));
     assert(v == 0);
     xco_select_event_deinit(&s);
 }
 
 static void test_allof_deinit_partial(void) {
     /* One input fired, one still parked: deinit must release the parked
      * waiter without disturbing the fired one. */
     xco_latch_t a, b;
     xco_latch_init(&a); xco_latch_init(&b);
 
     xco_select_event_t s;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&a.base, &b.base};
     xco_allof_event_init(&s, inputs, 2, srcs);
 
     xco_latch_set(&a, 0xAAA);
     assert(!s.done.set);
     assert(a.waiters == NULL);              /* a's waiter detached on fire */
     assert(b.waiters == &inputs[1].w);      /* b still parked */
 
     xco_select_event_deinit(&s);
     assert(b.waiters == NULL);               /* released */
 }
 
 static void test_allof_compose_with_select(void) {
     /* An allof-mode event is still a xco_select_event_t — feed it into a
      * select-mode wait. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t a, b, c;
     xco_latch_init(&a); xco_latch_init(&b); xco_latch_init(&c);
 
     xco_select_event_t all;
     xco_select_input_t all_inputs[2];
     xco_event_t *all_srcs[2] = {&a.base, &b.base};
     xco_allof_event_init(&all, all_inputs, 2, all_srcs);
 
     xco_select_event_t sel;
     xco_select_input_t sel_inputs[2];
     xco_event_t *sel_srcs[2] = {&all.done.base, &c.base};
     xco_select_event_init(&sel, sel_inputs, 2, sel_srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     /* Fire both halves of the allof; the select then sees its first input ready. */
     xco_latch_set(&a, 1);
     xco_latch_set(&b, 2);
     xco_rt_run(&rt, 0);
 
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);                     /* allof side won the select */
     assert(all.done.set);
     assert(c.waiters == NULL);               /* select disarmed the loser */
 
     xco_select_event_deinit(&sel);
     xco_select_event_deinit(&all);
 }
 
 /* ---- Channel ------------------------------------------------------- */
 
 /* Sender state machine: poll; if blocked, the poll already parked us. */
 typedef struct {
     xco_mach_t           base;
     xco_chan_t           *c;
     xco_runtime_t        *rt;
     xco_chan_send_waiter_t csw;
     uintptr_t         value;
     int               phase;
     bool              done;
 } sender_t;
 
 static xco_step_result_t sender_step(xco_mach_t *s, uintptr_t v) {
     sender_t *snd = (sender_t *)s;
     (void)v;
     switch (snd->phase) {
     case 0:
         xco_chan_send_waiter_init(&snd->csw, snd->rt, &snd->base);
         switch (xco_chan_send_poll(snd->c, snd->value, &snd->csw)) {
         case XCO_SEND_DELIVERED:
         case XCO_SEND_CLOSED:
             snd->done  = true;
             snd->phase = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_SEND_BLOCKED:
             snd->phase = 1;
             return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
         }
         __builtin_unreachable();
     case 1:
         snd->done  = true;
         snd->phase = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void sender_init(sender_t *snd, xco_runtime_t *rt, xco_chan_t *c, uintptr_t value) {
     snd->base  = (xco_mach_t){.step = sender_step, .status = XCO_STEP_INIT};
     snd->c     = c;
     snd->rt    = rt;
     snd->value = value;
     snd->phase = 0;
     snd->done  = false;
 }
 
 static void test_chan_send_blocks_until_recv(void) {
     /* Sender arrives first, parks; receiver pulls and both finish. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     sender_t snd; sender_init(&snd, &rt, &c, 0xDEADBEEF);
     xco_step(&snd.base, 0);
     assert(xco_mach_status(&snd.base) == XCO_STEP_SUSPENDED);
     assert(c.send_head == &snd.csw.sw.base);
     assert(c.send_tail == &snd.csw.sw.base);
 
     waiter_t r; waiter_init(&r, &rt, &c.recv);
     xco_step_result_t rr = xco_step(&r.base, 0);
     assert(rr.status == XCO_STEP_DEAD);
     assert(rr.value  == 0xDEADBEEF);
     assert(r.got     == 0xDEADBEEF);
     assert(c.send_head == NULL);
 
     /* Sender's resumption is queued by the recv-side fire. */
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&snd.base) == XCO_STEP_DEAD);
     assert(snd.done);
 }
 
 static void test_chan_recv_blocks_until_send(void) {
     /* Receiver arrives first, parks; sender delivers inline. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     waiter_t r; waiter_init(&r, &rt, &c.recv);
     xco_step(&r.base, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_SUSPENDED);
     assert(c.recv_head == &r.sw.base);
 
     assert(xco_chan_send_poll(&c, 0xCAFE, NULL) == XCO_SEND_DELIVERED);
     assert(c.recv_head == NULL);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_DEAD);
     assert(r.got == 0xCAFE);
 }
 
 static void test_chan_send_try_no_recv(void) {
     /* No receiver parked: pure-try poll reports BLOCKED without parking. */
     xco_chan_t c; xco_chan_init(&c);
     assert(xco_chan_send_poll(&c, 1, NULL) == XCO_SEND_BLOCKED);
     assert(c.send_head == NULL && c.recv_head == NULL);
 }
 
 static void test_chan_fifo(void) {
     /* Three senders park; three receives pull values in arrival order. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     sender_t s[3];
     for (int i = 0; i < 3; i++) {
         sender_init(&s[i], &rt, &c, (uintptr_t)(100 + i));
         xco_step(&s[i].base, 0);
         assert(xco_mach_status(&s[i].base) == XCO_STEP_SUSPENDED);
     }
     assert(c.send_head == &s[0].csw.sw.base);
     assert(c.send_tail == &s[2].csw.sw.base);
 
     for (int i = 0; i < 3; i++) {
         uintptr_t v;
         assert(xco_event_poll(&c.recv, &v, NULL));
         assert(v == (uintptr_t)(100 + i));
     }
     assert(c.send_head == NULL);
 
     xco_rt_run(&rt, 0);
     for (int i = 0; i < 3; i++) {
         assert(xco_mach_status(&s[i].base) == XCO_STEP_DEAD);
         assert(s[i].done);
     }
 }
 
 static void test_chan_send_unpark(void) {
     /* Cancel a parked sender — list is repaired; other waiters intact. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     sender_t a, b, d;
     sender_init(&a, &rt, &c, 1);
     sender_init(&b, &rt, &c, 2);
     sender_init(&d, &rt, &c, 3);
     xco_step(&a.base, 0); xco_step(&b.base, 0); xco_step(&d.base, 0);
     assert(c.send_head == &a.csw.sw.base);
     assert(c.send_tail == &d.csw.sw.base);
 
     xco_chan_send_unpark(&c, &b.csw);                 /* middle */
     /* Idempotent: removing again is a no-op. */
     xco_chan_send_unpark(&c, &b.csw);
 
     /* Order preserved: a, then d. */
     uintptr_t v;
     assert(xco_event_poll(&c.recv, &v, NULL) && v == 1);
     assert(xco_event_poll(&c.recv, &v, NULL) && v == 3);
     assert(c.send_head == NULL);
 
     /* a and d resume; b stays SUSPENDED (its waiter was unparked
      * without firing). Drain so it doesn't dangle. */
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&d.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
 }
 
 static void test_chan_select_recv(void) {
     /* select(xco_chan_recv, latch). Sender delivers; select fires with
      * xco_chan_recv's index, and the value is captured in inputs[winner].value. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
     xco_latch_t   l;  xco_latch_init(&l);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&c.recv, &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(c.recv_head != NULL);   /* select's input waiter parked here */
 
     assert(xco_chan_send_poll(&c, 0xABCDEF, NULL) == XCO_SEND_DELIVERED);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);                          /* xco_chan_recv won (index 0) */
     assert(inputs[0].value == 0xABCDEF);         /* captured value */
 
     /* Loser was disarmed; deinit is safe (no-op since fired). */
     assert(l.waiters == NULL);
     xco_select_event_deinit(&sel);
 }
 
 static void test_chan_recv_fifo(void) {
     /* Three receivers park; three sends deliver in arrival order.
      * The two waitlists are mutually exclusive (try-then-park is
      * atomic in single-threaded use), so the channel state at any
      * moment has at most one side queued. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     waiter_t r[3];
     for (int i = 0; i < 3; i++) {
         waiter_init(&r[i], &rt, &c.recv);
         xco_step(&r[i].base, 0);
         assert(xco_mach_status(&r[i].base) == XCO_STEP_SUSPENDED);
     }
     assert(c.recv_head == &r[0].sw.base);
     assert(c.recv_tail == &r[2].sw.base);
     assert(c.send_head == NULL);                  /* never both sides */
 
     for (int i = 0; i < 3; i++) {
         assert(xco_chan_send_poll(&c, (uintptr_t)(200 + i), NULL) == XCO_SEND_DELIVERED);
     }
     assert(c.recv_head == NULL);
 
     xco_rt_run(&rt, 0);
     for (int i = 0; i < 3; i++) {
         assert(xco_mach_status(&r[i].base) == XCO_STEP_DEAD);
         assert(r[i].got == (uintptr_t)(200 + i));
     }
 }
 
 /* ---- Send op (selectable send) ------------------------------------- */
 
 static void test_chan_send_op_inline(void) {
     /* Receiver already parked: op_init delivers immediately, done set. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     waiter_t r; waiter_init(&r, &rt, &c.recv);
     xco_step(&r.base, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_SUSPENDED);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xFEED);
     assert(op.done.set);                          /* delivered inline */
     assert(c.send_head == NULL);                  /* nothing parked */
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_DEAD);
     assert(r.got == 0xFEED);
 
     xco_chan_send_op_deinit(&op);
 }
 
 static void test_chan_send_op_blocks(void) {
     /* No receiver: op parks. Receiver arrives later, op.done fires. */
     xco_chan_t c; xco_chan_init(&c);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xBEAD);
     assert(!op.done.set);
     assert(c.send_head == &op.qsw.sw.base);
 
     /* Recv pulls value and fires op's waiter → sets op.done. */
     uintptr_t v;
     assert(xco_event_poll(&c.recv, &v, NULL));
     assert(v == 0xBEAD);
     assert(op.done.set);
     assert(c.send_head == NULL);
 
     xco_chan_send_op_deinit(&op);
 }
 
 static void test_chan_select_send(void) {
     /* select(send_op, latch). Receiver pulls; send wins. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
     xco_latch_t   timeout; xco_latch_init(&timeout);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0x5EED);
     assert(!op.done.set);                         /* parked, no recv */
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&op.done.base, &timeout.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     /* A receiver arrives and pulls. op.done fires → select fires. */
     uintptr_t v;
     assert(xco_event_poll(&c.recv, &v, NULL));
     assert(v == 0x5EED);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);                           /* send op won */
     assert(timeout.waiters == NULL);              /* loser disarmed */
 
     xco_select_event_deinit(&sel);
     xco_chan_send_op_deinit(&op);
 }
 
 static void test_chan_select_send_loses(void) {
     /* select(send_op, latch). Latch fires first; send is canceled
      * cleanly via op_deinit (chan side disarmed, no dangling waiter). */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
     xco_latch_t   l;  xco_latch_init(&l);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xABBA);
     assert(c.send_head == &op.qsw.sw.base);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&op.done.base, &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
 
     xco_latch_set(&l, 0xDEAD);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 1);                           /* latch (index 1) won */
     assert(inputs[1].value == 0xDEAD);
 
     /* send didn't happen — op still parked on chan. deinit unparks. */
     assert(c.send_head == &op.qsw.sw.base);
     xco_select_event_deinit(&sel);
     xco_chan_send_op_deinit(&op);
     assert(c.send_head == NULL);
 }
 
 static void test_chan_select_recv_fast_path(void) {
     /* Sender already parked when select arms: fast path captures value. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     sender_t snd; sender_init(&snd, &rt, &c, 0x12345);
     xco_step(&snd.base, 0);
     assert(xco_mach_status(&snd.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_t l; xco_latch_init(&l);
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&c.recv, &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     /* Fast path fired immediately. */
     uintptr_t winner;
     assert(xco_event_poll(&sel.done.base, &winner, NULL));
     assert(winner == 0);
     assert(inputs[0].value == 0x12345);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&snd.base) == XCO_STEP_DEAD);
     xco_select_event_deinit(&sel);
 }
 
 /* ---- Cancellation -------------------------------------------------- */
 
 static void test_cancel_basic(void) {
     /* The alias is a thin rename over latch: not-set, set, idempotent set,
      * and xco_event_poll yields the latch's payload of 0. */
     xco_cancel_t c; xco_cancel_init(&c);
     assert(!xco_cancel_is_set(&c));
     xco_cancel_set(&c);
     assert(xco_cancel_is_set(&c));
     xco_cancel_set(&c);                              /* idempotent */
     assert(xco_cancel_is_set(&c));
 
     uintptr_t v = 0xBADD;
     assert(xco_event_poll(xco_cancel_event(&c), &v, NULL));
     assert(v == 0);
 }
 
 static void test_wait_or_cancel_ev_wins(void) {
     /* Event resolves first: waiter sees XCO_WAIT_OK and inputs[0].value
      * carries the event's payload. The cancel side is disarmed. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
     xco_cancel_t  c;  xco_cancel_init(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &l.base, &c);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(l.waiters && c.waiters);              /* both armed */
 
     xco_latch_set(&l, 0xF00D);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_OK);
     assert(inputs[XCO_WAIT_OK].value == 0xF00D);
     assert(c.waiters == NULL);                   /* cancel disarmed */
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_wait_or_cancel_cancel_wins(void) {
     /* Cancel fires while parked: waiter sees XCO_WAIT_CANCELLED, ev disarmed. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
     xco_cancel_t  c;  xco_cancel_init(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &l.base, &c);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_cancel_set(&c);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_CANCELLED);
     assert(l.waiters == NULL);                   /* ev disarmed */
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_wait_or_cancel_already_cancelled(void) {
     /* Pre-set cancel: select fast-path fires XCO_WAIT_CANCELLED at init time;
      * ev is never parked, so deinit has nothing to disarm. */
     xco_latch_t  l;  xco_latch_init(&l);
     xco_cancel_t c;  xco_cancel_init(&c);
     xco_cancel_set(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &l.base, &c);
 
     uintptr_t v;
     assert(xco_event_poll(&sel.done.base, &v, NULL));
     assert(v == XCO_WAIT_CANCELLED);
     assert(l.waiters == NULL);                   /* never parked */
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_wait_or_cancel_ev_precedes_cancel(void) {
     /* Both ready at init: ev wins (it's checked first in the fast path).
      * This is the right semantic — if the work has already resolved,
      * a concurrent cancel doesn't retroactively undo it. */
     xco_latch_t  l;  xco_latch_init(&l);
     xco_cancel_t c;  xco_cancel_init(&c);
     xco_latch_set(&l, 0x600D);
     xco_cancel_set(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &l.base, &c);
 
     uintptr_t v;
     assert(xco_event_poll(&sel.done.base, &v, NULL));
     assert(v == XCO_WAIT_OK);
     assert(inputs[XCO_WAIT_OK].value == 0x600D);
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_wait_or_cancel_chan_recv(void) {
     /* Waiter blocks on xco_chan_recv via xco_wait_or_cancel; cancel fires while
      * parked. After resume + deinit, the chan's recv list must be empty
      * so a future send doesn't deliver to a freed waiter. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    ch; xco_chan_init(&ch);
     xco_cancel_t  c;  xco_cancel_init(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &ch.recv, &c);
     assert(ch.recv_head == &inputs[0].w);        /* select's input parked */
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_cancel_set(&c);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_CANCELLED);
     assert(ch.recv_head == NULL);                /* xco_select_input_fire disarmed */
 
     xco_select_event_deinit(&sel);
     /* No stale waiter lingering: pure-try poll reports BLOCKED, fires nothing. */
     assert(xco_chan_send_poll(&ch, 1, NULL) == XCO_SEND_BLOCKED);
 }
 
 static void test_wait_or_cancel_send_op(void) {
     /* Selectable send under cancel: cancel wins, the send_op is still
      * parked on the chan — xco_chan_send_op_deinit must release it so the
      * sender's value is not silently dropped from the chan's list. */
     xco_chan_t   ch; xco_chan_init(&ch);
     xco_cancel_t c;  xco_cancel_init(&c);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &ch, 0xABBA);
     assert(!op.done.set);
     assert(ch.send_head == &op.qsw.sw.base);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &op.done.base, &c);
 
     xco_cancel_set(&c);                              /* fires sel synchronously */
     uintptr_t v;
     assert(xco_event_poll(&sel.done.base, &v, NULL));
     assert(v == XCO_WAIT_CANCELLED);
 
     /* Send didn't happen — op still parked on the chan. The select
      * winning doesn't drain the chan's send list; deinit does. */
     assert(ch.send_head == &op.qsw.sw.base);
     xco_select_event_deinit(&sel);
     xco_chan_send_op_deinit(&op);
     assert(ch.send_head == NULL);
 }
 
 /* ---- Timers -------------------------------------------------------- */
 
 static void test_timer_basic(void) {
     /* Insert one timer; advance past its deadline; fire payload is the
      * deadline value. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timer_t t;
     xco_timer_init(&t, &h.base, 100);
     assert(!xco_timer_fired(&t));
     assert(t.in_heap);
 
     /* Not yet expired: advance is a no-op. */
     xco_timers_advance(&h.base, 50);
     assert(!xco_timer_fired(&t));
 
     /* Expired: fires. */
     xco_timers_advance(&h.base, 100);
     assert(xco_timer_fired(&t));
     assert(!t.in_heap);
 
     uintptr_t v;
     assert(xco_event_poll(xco_timer_event(&t), &v, NULL));
     assert(v == 100);
 
     /* Idempotent deinit (already fired). */
     xco_timer_deinit(&t);
 }
 
 static void test_timer_peek(void) {
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     assert(xco_timers_peek(&h.base) == UINT64_MAX);
 
     xco_timer_t a, b, c;
     xco_timer_init(&a, &h.base, 300);
     xco_timer_init(&b, &h.base, 100);
     xco_timer_init(&c, &h.base, 200);
 
     assert(xco_timers_peek(&h.base) == 100);  /* earliest deadline */
 
     xco_timers_advance(&h.base, 250);
     /* b and c fired; a remains. */
     assert(xco_timer_fired(&b));
     assert(xco_timer_fired(&c));
     assert(!xco_timer_fired(&a));
     assert(xco_timers_peek(&h.base) == 300);
 
     xco_timer_deinit(&a);
     xco_timer_deinit(&b);
     xco_timer_deinit(&c);
 }
 
 static void test_timer_cancel(void) {
     /* Cancel before fire; later advance must not fire it. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timer_t a, b, c;
     xco_timer_init(&a, &h.base, 100);
     xco_timer_init(&b, &h.base, 200);
     xco_timer_init(&c, &h.base, 300);
 
     xco_timer_deinit(&b);                        /* cancel middle */
     assert(!b.in_heap);
     assert(!xco_timer_fired(&b));
 
     xco_timers_advance(&h.base, 1000);
     assert(xco_timer_fired(&a));
     assert(!xco_timer_fired(&b));                /* cancelled — never fired */
     assert(xco_timer_fired(&c));
 
     xco_timer_deinit(&a);
     xco_timer_deinit(&c);
 }
 
 static void test_timer_park_wake(void) {
     /* A waiter parked on a timer wakes when the timer fires. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timer_t t;
     xco_timer_init(&t, &h.base, 500);
 
     waiter_t w; waiter_init(&w, &rt, xco_timer_event(&t));
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(t.done.waiters == &w.sw.base);
 
     xco_timers_advance(&h.base, 500);
     /* fire enqueued the step. */
     assert(rt.head != NULL);
     xco_rt_run(&rt, 500);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 500);                    /* deadline as payload */
 
     xco_timer_deinit(&t);
 }
 
 static void test_timer_select(void) {
     /* Compose a timer into a select. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timer_t   t; xco_timer_init(&t, &h.base, 200);
     xco_latch_t   l; xco_latch_init(&l);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {xco_timer_event(&t), &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_timers_advance(&h.base, 200);
     xco_rt_run(&rt, 200);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);                      /* timer (input 0) won */
     assert(inputs[0].value == 200);          /* deadline captured */
     assert(l.waiters == NULL);               /* loser disarmed */
 
     xco_select_event_deinit(&sel);
     xco_timer_deinit(&t);
 }
 
 static void test_timer_pairing_heap_order(void) {
     /* Insert many timers with non-sorted deadlines; advance must pop
      * them in deadline order. Stress the heap structure. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     enum { N = 32 };
     xco_timer_t ts[N];
     /* A scrambled sequence — interleave halves so the insert order
      * exercises the heap's restructuring. */
     uint64_t deadlines[N] = {
         17,  3, 25,  9, 31,  1, 28, 12,
         20,  7, 22,  5, 30, 14, 27, 10,
         16,  2, 24,  8, 32, 11, 29, 13,
         19,  6, 21,  4, 26, 15, 23, 18,
     };
 
     for (int i = 0; i < N; i++) xco_timer_init(&ts[i], &h.base, deadlines[i]);
 
     /* Advance one tick at a time; verify exactly one timer fires per
      * matching deadline (since deadlines 1..32 are unique). */
     int fired = 0;
     for (uint64_t now = 1; now <= N; now++) {
         xco_timers_advance(&h.base, now);
         int fired_now = 0;
         for (int i = 0; i < N; i++) {
             if (xco_timer_fired(&ts[i]) && ts[i].deadline == now) fired_now++;
         }
         assert(fired_now == 1);
         fired++;
     }
     assert(fired == N);
     for (int i = 0; i < N; i++) assert(xco_timer_fired(&ts[i]));
 }
 
 static void test_timer_cancel_stress(void) {
     /* Insert N timers; cancel every other one; advance and verify only
      * the survivors fire, in order. Exercises non-root cancel paths. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     enum { N = 16 };
     xco_timer_t ts[N];
     /* Deadlines: 1..N. Inserted in order so the heap shape is a long
      * single-spine; cancels hit non-root nodes. */
     for (int i = 0; i < N; i++) xco_timer_init(&ts[i], &h.base, (uint64_t)(i + 1));
 
     /* Cancel even indices (deadlines 2,4,6,...). */
     for (int i = 0; i < N; i += 2) xco_timer_deinit(&ts[i]);
 
     xco_timers_advance(&h.base, N);
     for (int i = 0; i < N; i++) {
         if (i % 2 == 0) assert(!xco_timer_fired(&ts[i]));
         else            assert( xco_timer_fired(&ts[i]));
     }
 
     for (int i = 1; i < N; i += 2) xco_timer_deinit(&ts[i]);
 }
 
 static void test_timer_deinit_idempotent(void) {
     /* Deinit after fire is a no-op; deinit twice (without fire) is too. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timer_t a, b;
     xco_timer_init(&a, &h.base, 10);
     xco_timer_init(&b, &h.base, 20);
 
     xco_timers_advance(&h.base, 10);
     assert(xco_timer_fired(&a) && !xco_timer_fired(&b));
     xco_timer_deinit(&a);                        /* already fired */
     xco_timer_deinit(&a);                        /* twice — still no-op */
 
     xco_timer_deinit(&b);                        /* cancel before fire */
     xco_timer_deinit(&b);                        /* twice — must not corrupt */
     xco_timers_advance(&h.base, 1000);
     assert(!xco_timer_fired(&b));                /* never fired */
 }
 
 /* ---- Timeout ------------------------------------------------------- */
 
 static void test_timeout_fires(void) {
     /* Bare timeout: advance past deadline, cancel becomes set. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timeout_t to;
     xco_timeout_init(&to, &h.base, 100);
     assert(!xco_cancel_is_set(&to.cancel));
 
     xco_timers_advance(&h.base, 100);
     assert(xco_cancel_is_set(&to.cancel));        /* bridge fired */
     assert(xco_timer_fired(&to.timer));
 
     xco_timeout_deinit(&to);                      /* idempotent */
 }
 
 static void test_timeout_deinit_before_fire(void) {
     /* Deinit a timeout before its deadline: cancel must remain unset
      * and the timer must be removed from the source so a later advance
      * doesn't fire freed memory. */
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_timeout_t to;
     xco_timeout_init(&to, &h.base, 100);
     xco_timeout_deinit(&to);
     assert(!xco_cancel_is_set(&to.cancel));
 
     /* Advance past the original deadline: nothing left to fire. */
     xco_timers_advance(&h.base, 1000);
     assert(!xco_cancel_is_set(&to.cancel));
 }
 
 static void test_timeout_with_wait_or_cancel(void) {
     /* Canonical pattern: a waiter blocks on (ev, timeout.cancel). The
      * timeout fires first; waiter resumes with XCO_WAIT_CANCELLED. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_latch_t ev; xco_latch_init(&ev);
     xco_timeout_t to;
     xco_timeout_init(&to, &h.base, 50);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &ev.base, &to.cancel);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     /* xco_rt_run advances the timer source itself: now=50 fires the timer,
      * which fires the bridge, which sets cancel, which fires sel,
      * which enqueues the waiter — all in one xco_rt_run call. */
     xco_rt_run(&rt, 50);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_CANCELLED);
 
     xco_select_event_deinit(&sel);
     xco_timeout_deinit(&to);
 }
 
 static void test_timeout_ev_wins(void) {
     /* Event wins the race; timeout should be deinit'd cleanly without
      * having fired. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_latch_t ev; xco_latch_init(&ev);
     xco_timeout_t to;
     xco_timeout_init(&to, &h.base, 1000);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, &ev.base, &to.cancel);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_set(&ev, 0xF00D);
     xco_rt_run(&rt, 50);                          /* now < deadline; timer untouched */
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_OK);
     assert(inputs[XCO_WAIT_OK].value == 0xF00D);
     assert(!xco_cancel_is_set(&to.cancel));
 
     xco_select_event_deinit(&sel);
     xco_timeout_deinit(&to);
     /* Subsequent advance must not fire anything (timer was removed). */
     xco_rt_run(&rt, 10000);
     assert(!xco_cancel_is_set(&to.cancel));
 }
 
 /* ---- xco_rt_run + timer integration ------------------------------------ */
 
 static void test_rt_run_advances_timers(void) {
     /* A waiter parked on a timer; xco_rt_run with now=deadline advances the
      * source, fires the timer, drains the waiter — all in one call. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_timer_t t; xco_timer_init(&t, &h.base, 42);
     waiter_t w; waiter_init(&w, &rt, xco_timer_event(&t));
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(rt.head == NULL);                  /* parked, not queued */
 
     xco_rt_run(&rt, 42);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 42);
     assert(rt.head == NULL);
 
     xco_timer_deinit(&t);
 }
 
 static void test_rt_run_no_timers(void) {
     /* Without an attached timer source, xco_rt_run(rt, now) is the pure
      * drainer regardless of `now`. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
 
     waiter_t w; waiter_init(&w, &rt, &l.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_latch_set(&l, 7);
     xco_rt_run(&rt, 99999);                       /* now ignored */
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 7);
 }
 
 /* ---- Semaphore ----------------------------------------------------- */
 
 /* Acquirer state machine: one-shot acquire of 1 permit. Mirrors waiter_t
  * but specialized for the "permit handed at fire time" semantics — the
  * resume is itself the proof of acquisition. */
 typedef struct {
     xco_mach_t      base;
     xco_semaphore_t *sem;
     xco_runtime_t   *rt;
     xco_waker_t sw;
     int          phase;
     bool         got;
 } sem_acquirer_t;
 
 static xco_step_result_t sem_acquirer_step(xco_mach_t *s, uintptr_t v) {
     sem_acquirer_t *a = (sem_acquirer_t *)s;
     (void)v;
     switch (a->phase) {
     case 0:
         if (xco_event_poll(xco_semaphore_event(a->sem), NULL, NULL)) {
             a->got   = true;
             a->phase = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         }
         xco_waker_init(&a->sw, a->rt, &a->base);
         xco_event_poll(xco_semaphore_event(a->sem), NULL, &a->sw.base);
         a->phase = 1;
         return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
     case 1:
         a->got   = true;
         a->phase = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void sem_acquirer_init(sem_acquirer_t *a, xco_runtime_t *rt, xco_semaphore_t *s) {
     a->base  = (xco_mach_t){.step = sem_acquirer_step, .status = XCO_STEP_INIT};
     a->sem   = s;
     a->rt    = rt;
     a->phase = 0;
     a->got   = false;
 }
 
 static void test_semaphore_inline_acquire(void) {
     /* permits > 0: try succeeds and decrements. */
     xco_semaphore_t s; xco_semaphore_init(&s, 2);
     assert(xco_event_poll(xco_semaphore_event(&s), NULL, NULL));
     assert(s.permits == 1);
     assert(xco_event_poll(xco_semaphore_event(&s), NULL, NULL));
     assert(s.permits == 0);
     assert(!xco_event_poll(xco_semaphore_event(&s), NULL, NULL));
     assert(s.permits == 0);
 }
 
 static void test_semaphore_park_then_release(void) {
     /* Empty sem: acquirer parks; release wakes it with a permit. */
     xco_runtime_t   rt; xco_rt_init(&rt);
     xco_semaphore_t s;  xco_semaphore_init(&s, 0);
 
     sem_acquirer_t a; sem_acquirer_init(&a, &rt, &s);
     xco_step(&a.base, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_SUSPENDED);
     assert(s.head == &a.sw.base);
 
     xco_semaphore_release(&s, 1);
     /* Release prefers parked waiters: permit went straight to a, count stays 0. */
     assert(s.permits == 0);
     assert(s.head == NULL);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD);
     assert(a.got);
 }
 
 static void test_semaphore_fifo(void) {
     /* Three parked acquirers; release(2) wakes the first two in order;
      * the third stays parked. release(1) again wakes it. */
     xco_runtime_t   rt; xco_rt_init(&rt);
     xco_semaphore_t s;  xco_semaphore_init(&s, 0);
 
     sem_acquirer_t a[3];
     for (int i = 0; i < 3; i++) {
         sem_acquirer_init(&a[i], &rt, &s);
         xco_step(&a[i].base, 0);
     }
     assert(s.head == &a[0].sw.base);
     assert(s.tail == &a[2].sw.base);
 
     xco_semaphore_release(&s, 2);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a[0].base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&a[1].base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&a[2].base) == XCO_STEP_SUSPENDED);
     assert(s.head == &a[2].sw.base);
 
     xco_semaphore_release(&s, 1);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a[2].base) == XCO_STEP_DEAD);
     assert(s.head == NULL);
     assert(s.permits == 0);
 }
 
 static void test_semaphore_release_overflow_to_count(void) {
     /* release(n) where n > waiters: hand to all waiters first, then add
      * the leftover to permits. */
     xco_runtime_t   rt; xco_rt_init(&rt);
     xco_semaphore_t s;  xco_semaphore_init(&s, 0);
 
     sem_acquirer_t a; sem_acquirer_init(&a, &rt, &s);
     xco_step(&a.base, 0);
 
     xco_semaphore_release(&s, 5);
     /* a got 1; 4 leftover sat as permits. */
     assert(s.permits == 4);
     xco_rt_run(&rt, 0);
     assert(a.got);
     assert(s.permits == 4);
 }
 
 static void test_semaphore_select_acquire(void) {
     /* Compose a sem acquire with a cancel via xco_wait_or_cancel: cancel wins. */
     xco_runtime_t   rt; xco_rt_init(&rt);
     xco_semaphore_t s;  xco_semaphore_init(&s, 0);
     xco_cancel_t    c;  xco_cancel_init(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, xco_semaphore_event(&s), &c);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
     assert(s.head != NULL);                  /* sem-side waiter parked */
 
     xco_cancel_set(&c);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_CANCELLED);
     assert(s.head == NULL);                  /* xco_select_input_fire detached us */
 
     xco_select_event_deinit(&sel);
     /* No leak: a future release does not over-grant past the count. */
     xco_semaphore_release(&s, 1);
     assert(s.permits == 1);
 }
 
 static void test_semaphore_binary_mutex(void) {
     /* Binary semaphore as mutex: init permits=1; first take succeeds inline,
      * second parks, release on the way out wakes the waiter. */
     xco_runtime_t   rt; xco_rt_init(&rt);
     xco_semaphore_t mu; xco_semaphore_init(&mu, 1);
 
     /* Holder takes it inline. */
     assert(xco_event_poll(xco_semaphore_event(&mu), NULL, NULL));
     assert(mu.permits == 0);
 
     /* Contender parks. */
     sem_acquirer_t b; sem_acquirer_init(&b, &rt, &mu);
     xco_step(&b.base, 0);
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
 
     /* Holder releases on exit. */
     xco_semaphore_release(&mu, 1);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&b.base) == XCO_STEP_DEAD);
     assert(b.got);
 }
 
 /* ---- Queue --------------------------------------------------------- */
 
 /* Queue sender: same shape as the chan sender_t. */
 typedef struct {
     xco_mach_t            base;
     xco_queue_t           *q;
     xco_runtime_t         *rt;
     xco_queue_send_waiter_t qsw;
     uintptr_t          value;
     int                phase;
     bool               done;
 } queue_sender_t;
 
 static xco_step_result_t queue_sender_step(xco_mach_t *s, uintptr_t v) {
     queue_sender_t *snd = (queue_sender_t *)s;
     (void)v;
     switch (snd->phase) {
     case 0:
         xco_queue_send_waiter_init(&snd->qsw, snd->rt, &snd->base);
         switch (xco_queue_send_poll(snd->q, snd->value, &snd->qsw)) {
         case XCO_QSEND_ACCEPTED:
         case XCO_QSEND_CLOSED:
             snd->done  = true;
             snd->phase = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_QSEND_BLOCKED:
             snd->phase = 1;
             return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
         }
         __builtin_unreachable();
     case 1:
         snd->done  = true;
         snd->phase = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void queue_sender_init(queue_sender_t *s, xco_runtime_t *rt, xco_queue_t *q, uintptr_t value) {
     s->base  = (xco_mach_t){.step = queue_sender_step, .status = XCO_STEP_INIT};
     s->q     = q;
     s->rt    = rt;
     s->value = value;
     s->phase = 0;
     s->done  = false;
 }
 
 static void test_queue_block_buffered(void) {
     /* cap=3, BLOCK: three sends fill the buffer; three recvs drain in order. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[3];
     xco_queue_t   q; xco_queue_init(&q, buf, 3, XCO_QUEUE_BLOCK);
 
     assert(xco_queue_send_poll(&q, 10, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 20, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 30, NULL) == XCO_QSEND_ACCEPTED);
     assert(q.len == 3);
     /* Buffer full under BLOCK: pure-try poll reports BLOCKED. */
     assert(xco_queue_send_poll(&q, 40, NULL) == XCO_QSEND_BLOCKED);
 
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 10);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 20);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 30);
     assert(!xco_event_poll(xco_queue_recv_event(&q), &v, NULL));   /* empty */
     assert(q.len == 0);
     (void)rt;
 }
 
 static void test_queue_block_sender_parks(void) {
     /* Buffer fills, next sender parks; a recv unblocks it. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_BLOCK);
 
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 2, NULL) == XCO_QSEND_ACCEPTED);
 
     queue_sender_t s; queue_sender_init(&s, &rt, &q, 3);
     xco_step(&s.base, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_SUSPENDED);
     assert(q.send_head == &s.qsw.sw.base);
 
     /* Recv pops 1; 3 should slot into the buffer and the parked sender wakes. */
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 1);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_DEAD);
     assert(s.done);
     assert(q.send_head == NULL);
 
     /* Buffer should now hold 2, 3. */
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 2);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 3);
     assert(q.len == 0);
 }
 
 static void test_queue_drop_newest(void) {
     /* When full, send_try returns true but silently discards. */
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_DROP_NEWEST);
 
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 2, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 3, NULL) == XCO_QSEND_ACCEPTED);    /* full → drop 3 */
     assert(xco_queue_send_poll(&q, 4, NULL) == XCO_QSEND_ACCEPTED);    /* still full → drop 4 */
 
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 1);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 2);
     assert(!xco_event_poll(xco_queue_recv_event(&q), &v, NULL));
 }
 
 static void test_queue_drop_oldest(void) {
     /* When full, send_try returns true and evicts head, pushes new tail. */
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_DROP_OLDEST);
 
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 2, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 3, NULL) == XCO_QSEND_ACCEPTED);    /* full → evict 1, buffer = [2, 3] */
     assert(xco_queue_send_poll(&q, 4, NULL) == XCO_QSEND_ACCEPTED);    /* full → evict 2, buffer = [3, 4] */
 
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 3);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 4);
     assert(!xco_event_poll(xco_queue_recv_event(&q), &v, NULL));
 }
 
 static void test_queue_direct_handoff(void) {
     /* Receiver parked; send_try hands off directly, bypassing the buffer.
      * Works the same regardless of policy — exercise BLOCK here. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_BLOCK);
 
     waiter_t r; waiter_init(&r, &rt, xco_queue_recv_event(&q));
     xco_step(&r.base, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_SUSPENDED);
     assert(q.recv_head == &r.sw.base);
 
     assert(xco_queue_send_poll(&q, 0xCAFE, NULL) == XCO_QSEND_ACCEPTED);
     assert(q.len == 0);                /* didn't touch the buffer */
     assert(q.recv_head == NULL);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_DEAD);
     assert(r.got == 0xCAFE);
 }
 
 static void test_queue_recv_blocks_then_drains_buffered(void) {
     /* Receivers wait when empty. After buffer warms up, recvs pop FIFO
      * — no value-skipping when both sides race past one another. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_BLOCK);
 
     /* 2 receivers parked. */
     waiter_t r1, r2;
     waiter_init(&r1, &rt, xco_queue_recv_event(&q));
     waiter_init(&r2, &rt, xco_queue_recv_event(&q));
     xco_step(&r1.base, 0);
     xco_step(&r2.base, 0);
 
     /* Two sends → direct handoff to the two parked receivers, not buffered. */
     assert(xco_queue_send_poll(&q, 100, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 200, NULL) == XCO_QSEND_ACCEPTED);
     assert(q.len == 0);
 
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r1.base) == XCO_STEP_DEAD && r1.got == 100);
     assert(xco_mach_status(&r2.base) == XCO_STEP_DEAD && r2.got == 200);
 }
 
 static void test_queue_select_recv(void) {
     /* Queue recv composes into select. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     xco_latch_t   l; xco_latch_init(&l);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {xco_queue_recv_event(&q), &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     assert(xco_queue_send_poll(&q, 0xBEEF, NULL) == XCO_QSEND_ACCEPTED);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);
     assert(inputs[0].value == 0xBEEF);
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_queue_send_unpark(void) {
     /* Cancel a parked sender; remaining FIFO order intact. */
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
 
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);   /* fills buffer */
 
     queue_sender_t a, b, d;
     queue_sender_init(&a, &rt, &q, 2);
     queue_sender_init(&b, &rt, &q, 3);
     queue_sender_init(&d, &rt, &q, 4);
     xco_step(&a.base, 0);
     xco_step(&b.base, 0);
     xco_step(&d.base, 0);
 
     xco_queue_send_unpark(&q, &b.qsw);
     xco_queue_send_unpark(&q, &b.qsw);   /* idempotent */
 
     /* Drain: 1, then a's 2 (slots into buffer when 1 popped), then d's 4. */
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 1);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 2);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&d.base) == XCO_STEP_DEAD);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 4);
     /* b never sent its value, stayed SUSPENDED. */
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
 }
 
 /* ---- Broadcast ----------------------------------------------------- */
 
 /* Subscriber: parks on the broadcast, captures fire payload, re-parks
  * on resume. n_seen counts how many publishes it received. */
 typedef struct {
     xco_mach_t      base;
     xco_broadcast_t *b;
     xco_runtime_t   *rt;
     xco_waker_t sw;
     uintptr_t    last;
     int          n_seen;
     int          target;     /* unsubscribe after this many */
 } bcast_sub_t;
 
 static xco_step_result_t bcast_sub_step(xco_mach_t *s, uintptr_t v) {
     bcast_sub_t *b = (bcast_sub_t *)s;
     if (b->n_seen > 0) {
         /* Resume from a fire — capture and decide whether to re-park. */
         b->last = v;
     }
     if (b->n_seen >= b->target) {
         return (xco_step_result_t){b->last, XCO_STEP_DEAD};
     }
     /* Re-park: the runtime returned sw fully detached, so xco_event_poll
      * works without re-init. (Init was done once in bcast_sub_init.) */
     xco_event_poll(xco_broadcast_event(b->b), NULL, &b->sw.base);
     b->n_seen++;
     return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
 }
 
 static void bcast_sub_init(bcast_sub_t *s, xco_runtime_t *rt, xco_broadcast_t *b, int target) {
     s->base   = (xco_mach_t){.step = bcast_sub_step, .status = XCO_STEP_INIT};
     s->b      = b;
     s->rt     = rt;
     s->last   = 0;
     s->n_seen = 0;
     s->target = target;
     xco_waker_init(&s->sw, rt, &s->base);
 }
 
 static void test_broadcast_try_always_false(void) {
     /* No "ready now" state; subscribers always wait for the next publish. */
     xco_broadcast_t b; xco_broadcast_init(&b);
     assert(!xco_event_poll(xco_broadcast_event(&b), NULL, NULL));
     xco_broadcast_publish(&b, 7);
     /* Even after a publish, try is still false — value is read via accessor. */
     assert(!xco_event_poll(xco_broadcast_event(&b), NULL, NULL));
     assert(xco_broadcast_has_value(&b));
     assert(xco_broadcast_value(&b) == 7);
 }
 
 static void test_broadcast_publish_wakes_all(void) {
     /* Three subscribers parked; one publish wakes all with the value. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_broadcast_t b; xco_broadcast_init(&b);
 
     bcast_sub_t s[3];
     for (int i = 0; i < 3; i++) {
         bcast_sub_init(&s[i], &rt, &b, 1);  /* one publish then exit */
         xco_step(&s[i].base, 0);
         assert(xco_mach_status(&s[i].base) == XCO_STEP_SUSPENDED);
     }
 
     xco_broadcast_publish(&b, 0xBEEF);
     assert(b.waiters == NULL);   /* drained */
 
     xco_rt_run(&rt, 0);
     for (int i = 0; i < 3; i++) {
         assert(xco_mach_status(&s[i].base) == XCO_STEP_DEAD);
         assert(s[i].last == 0xBEEF);
     }
 }
 
 static void test_broadcast_rearm(void) {
     /* Subscriber parks for two publishes; receives both values in order. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_broadcast_t b; xco_broadcast_init(&b);
 
     bcast_sub_t s; bcast_sub_init(&s, &rt, &b, 2);
     xco_step(&s.base, 0);
 
     xco_broadcast_publish(&b, 11);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_SUSPENDED);   /* re-armed */
     assert(s.last == 11);
 
     xco_broadcast_publish(&b, 22);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_DEAD);
     assert(s.last == 22);
 }
 
 static void test_broadcast_missed_publish(void) {
     /* Subscriber not parked at publish time misses it; the next publish
      * wakes them with that value (no replay, no coalescing into one). */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_broadcast_t b; xco_broadcast_init(&b);
 
     /* Publish before anyone subscribes — value is in the slot but no one wakes. */
     xco_broadcast_publish(&b, 1);
     assert(xco_broadcast_value(&b) == 1);
 
     bcast_sub_t s; bcast_sub_init(&s, &rt, &b, 1);
     xco_step(&s.base, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_SUSPENDED);   /* didn't see the prior */
 
     xco_broadcast_publish(&b, 2);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_DEAD);
     assert(s.last == 2);
 }
 
 static void test_broadcast_unpark(void) {
     /* xco_event_unpark removes a subscriber cleanly. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_broadcast_t b; xco_broadcast_init(&b);
 
     xco_waker_t sw1, sw2, sw3;
     xco_waker_init(&sw1, &rt, (xco_mach_t *)0x1);
     xco_waker_init(&sw2, &rt, (xco_mach_t *)0x2);
     xco_waker_init(&sw3, &rt, (xco_mach_t *)0x3);
     xco_event_poll(xco_broadcast_event(&b), NULL, &sw1.base);
     xco_event_poll(xco_broadcast_event(&b), NULL, &sw2.base);
     xco_event_poll(xco_broadcast_event(&b), NULL, &sw3.base);
 
     xco_event_unpark(xco_broadcast_event(&b), &sw2.base);
     xco_event_unpark(xco_broadcast_event(&b), &sw2.base);     /* idempotent */
 
     int seen1 = 0, seen3 = 0;
     for (xco_waiter_t *w = b.waiters; w; w = w->next) {
         if (w == &sw1.base) seen1 = 1;
         if (w == &sw3.base) seen3 = 1;
         assert(w != &sw2.base);
     }
     assert(seen1 && seen3);
 
     xco_event_unpark(xco_broadcast_event(&b), &sw1.base);
     xco_event_unpark(xco_broadcast_event(&b), &sw3.base);
 }
 
 /* ---- Task (state-machine xco_step) ------------------------------------ */
 
 /* A countdown SM that completes after N steps, returning a final value.
  * On every step it checks task->cancel via xco_event_poll and bails early
  * (winding down to XCO_STEP_DEAD) if it's set. Demonstrates the cooperation
  * pattern: cancel is a signal, the SM owns the unwind. */
 typedef struct {
     xco_mach_t   base;
     xco_task_t   *task;
     int       remaining;
     uintptr_t final;
 } taskbody_t;
 
 static xco_step_result_t taskbody_step(xco_mach_t *s, uintptr_t v) {
     taskbody_t *cd = (taskbody_t *)s;
     (void)v;
     if (xco_task_is_cancelled(cd->task)) {
         xco_task_done(cd->task, 0);                /* cooperative unwind */
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     if (cd->remaining-- == 0) {
         xco_task_done(cd->task, cd->final);
         return (xco_step_result_t){cd->final, XCO_STEP_DEAD};
     }
     return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
 }
 
 static void taskbody_init(taskbody_t *cd, xco_task_t *t, int n, uintptr_t final) {
     cd->base      = (xco_mach_t){.step = taskbody_step, .status = XCO_STEP_INIT};
     cd->task      = t;
     cd->remaining = n;
     cd->final     = final;
 }
 
 static void test_task_state_machine_join(void) {
     /* Drive a countdown SM as a task; join via xco_task_done_event. */
     xco_runtime_t rt; xco_rt_init(&rt);
 
     taskbody_t cd;
     xco_task_t      t;
     taskbody_init(&cd, &t, 3, 0xAAAA);
     xco_task_init(&t, &cd.base);
 
     assert(!xco_task_finished(&t));
 
     /* Pump it manually. */
     while (xco_mach_status(&cd.base) != XCO_STEP_DEAD) {
         xco_step(&cd.base, 0);
     }
     assert(xco_task_finished(&t));
 
     /* A latched join: try the done event for the return value. */
     uintptr_t v;
     assert(xco_event_poll(xco_task_done_event(&t), &v, NULL));
     assert(v == 0xAAAA);
 }
 
 static void test_task_state_machine_cancel(void) {
     /* Cancel mid-flight; SM observes and unwinds. Joiner sees done with
      * the cooperative-unwind sentinel value (here, 0). */
     xco_runtime_t rt; xco_rt_init(&rt);
 
     taskbody_t cd;
     xco_task_t      t;
     taskbody_init(&cd, &t, 100, 0xAAAA);
     xco_task_init(&t, &cd.base);
 
     /* A second xco_step waiting for completion via xco_wait_or_cancel-style
      * shape — here, just a direct waiter on the done event. */
     waiter_t joiner; waiter_init(&joiner, &rt, xco_task_done_event(&t));
     xco_step(&joiner.base, 0);
     assert(xco_mach_status(&joiner.base) == XCO_STEP_SUSPENDED);
 
     /* Step the task a few times. */
     xco_step(&cd.base, 0);
     xco_step(&cd.base, 0);
     assert(!xco_task_finished(&t));
 
     xco_cancel_set(xco_task_cancel(&t));
     /* Next step observes cancel and dies. */
     xco_step(&cd.base, 0);
     assert(xco_mach_status(&cd.base) == XCO_STEP_DEAD);
     assert(xco_task_finished(&t));
 
     /* Joiner wakes with the done payload (0 from the cooperative unwind). */
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&joiner.base) == XCO_STEP_DEAD);
     assert(joiner.got == 0);
 }
 
 /* ---- Countdown ----------------------------------------------------- */
 
 static void test_countdown_basic(void) {
     xco_countdown_t cd; xco_countdown_init(&cd, 3);
     assert(!xco_countdown_fired(&cd));
     xco_countdown_done(&cd);
     xco_countdown_done(&cd);
     assert(!xco_countdown_fired(&cd));
     xco_countdown_done(&cd);
     assert(xco_countdown_fired(&cd));
 
     uintptr_t v = 0xBAD;
     assert(xco_event_poll(xco_countdown_event(&cd), &v, NULL));
     assert(v == 0);
 }
 
 static void test_countdown_zero_init(void) {
     /* n=0: latch fires immediately. */
     xco_countdown_t cd; xco_countdown_init(&cd, 0);
     assert(xco_countdown_fired(&cd));
 }
 
 static void test_countdown_add(void) {
     xco_countdown_t cd; xco_countdown_init(&cd, 1);
     xco_countdown_add(&cd, 2);
     xco_countdown_done(&cd);
     xco_countdown_done(&cd);
     assert(!xco_countdown_fired(&cd));
     xco_countdown_done(&cd);
     assert(xco_countdown_fired(&cd));
 }
 
 static void test_countdown_park_wake(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_countdown_t cd; xco_countdown_init(&cd, 2);
 
     waiter_t w; waiter_init(&w, &rt, xco_countdown_event(&cd));
     xco_step(&w.base, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_SUSPENDED);
 
     xco_countdown_done(&cd);
     assert(rt.head == NULL);
     xco_countdown_done(&cd);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);
 }
 
 /* ---- Notify -------------------------------------------------------- */
 
 static void test_notify_try_always_false(void) {
     xco_notify_t n; xco_notify_init(&n);
     assert(!xco_event_poll(xco_notify_event(&n), NULL, NULL));
     xco_notify_one(&n);                          /* empty: no-op */
     xco_notify_all(&n);                          /* empty: no-op */
     assert(!xco_event_poll(xco_notify_event(&n), NULL, NULL));
 }
 
 static void test_notify_one_fifo(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_notify_t  n;  xco_notify_init(&n);
 
     waiter_t a, b, c;
     waiter_init(&a, &rt, xco_notify_event(&n));
     waiter_init(&b, &rt, xco_notify_event(&n));
     waiter_init(&c, &rt, xco_notify_event(&n));
     xco_step(&a.base, 0); xco_step(&b.base, 0); xco_step(&c.base, 0);
 
     xco_notify_one(&n);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
 
     xco_notify_one(&n);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&b.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&c.base) == XCO_STEP_SUSPENDED);
 
     xco_notify_one(&n);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&c.base) == XCO_STEP_DEAD);
     assert(n.head == NULL);
 
     xco_notify_one(&n);                          /* empty: no-op */
     assert(rt.head == NULL);
 }
 
 static void test_notify_all(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_notify_t  n;  xco_notify_init(&n);
 
     waiter_t a, b, c;
     waiter_init(&a, &rt, xco_notify_event(&n));
     waiter_init(&b, &rt, xco_notify_event(&n));
     waiter_init(&c, &rt, xco_notify_event(&n));
     xco_step(&a.base, 0); xco_step(&b.base, 0); xco_step(&c.base, 0);
 
     xco_notify_all(&n);
     assert(n.head == NULL);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&a.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&b.base) == XCO_STEP_DEAD);
     assert(xco_mach_status(&c.base) == XCO_STEP_DEAD);
 }
 
 static void test_notify_select(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_notify_t  n;  xco_notify_init(&n);
     xco_latch_t   l;  xco_latch_init(&l);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {xco_notify_event(&n), &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
 
     xco_notify_one(&n);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == 0);
     assert(l.waiters == NULL);
 
     xco_select_event_deinit(&sel);
 }
 
 static void test_notify_unpark(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_notify_t  n;  xco_notify_init(&n);
 
     xco_waker_t s1, s2, s3;
     xco_waker_init(&s1, &rt, (xco_mach_t *)0x1);
     xco_waker_init(&s2, &rt, (xco_mach_t *)0x2);
     xco_waker_init(&s3, &rt, (xco_mach_t *)0x3);
     xco_event_poll(xco_notify_event(&n), NULL, &s1.base);
     xco_event_poll(xco_notify_event(&n), NULL, &s2.base);
     xco_event_poll(xco_notify_event(&n), NULL, &s3.base);
 
     xco_event_unpark(xco_notify_event(&n), &s2.base);
     xco_event_unpark(xco_notify_event(&n), &s2.base);  /* idempotent */
 
     /* FIFO head is s1; pop s1, then s3 remains. */
     xco_notify_one(&n);
     assert(n.head == &s3.base);
     xco_event_unpark(xco_notify_event(&n), &s3.base);
     assert(n.head == NULL);
 }
 
 /* ---- Mutex --------------------------------------------------------- */
 
 static void test_mutex_basic(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_mutex_t   mu; xco_mutex_init(&mu);
 
     assert(xco_event_poll(xco_mutex_event(&mu), NULL, NULL));
     assert(!xco_event_poll(xco_mutex_event(&mu), NULL, NULL));
 
     sem_acquirer_t b; sem_acquirer_init(&b, &rt, &mu);
     xco_step(&b.base, 0);
     assert(xco_mach_status(&b.base) == XCO_STEP_SUSPENDED);
 
     xco_mutex_release(&mu);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&b.base) == XCO_STEP_DEAD);
     assert(b.got);
 }
 
 /* ---- Queue send op ------------------------------------------------- */
 
 static void test_queue_send_op_inline(void) {
     /* BLOCK with room: init delivers immediately, payload 1. */
     uintptr_t buf[2];
     xco_queue_t   q; xco_queue_init(&q, buf, 2, XCO_QUEUE_BLOCK);
 
     xco_queue_send_op_t op;
     xco_queue_send_op_init(&op, &q, 0xAA);
     assert(op.done.set);
     uintptr_t v;
     assert(xco_event_poll(&op.done.base, &v, NULL));
     assert(v == 1);
     assert(q.len == 1);
 
     xco_queue_send_op_deinit(&op);
 }
 
 static void test_queue_send_op_blocks(void) {
     /* BLOCK full: parks. Recv frees a slot, sender drains, op fires. */
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
 
     xco_queue_send_op_t op;
     xco_queue_send_op_init(&op, &q, 2);
     assert(!op.done.set);
     assert(q.send_head == &op.qsw.sw.base);
 
     uintptr_t v;
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 1);
     assert(op.done.set);
     uintptr_t pv;
     assert(xco_event_poll(&op.done.base, &pv, NULL));
     assert(pv == 1);
     assert(xco_event_poll(xco_queue_recv_event(&q), &v, NULL) && v == 2);
 
     xco_queue_send_op_deinit(&op);
 }
 
 static void test_queue_send_op_drop_inline(void) {
     /* DROP_NEWEST: op resolves inline regardless of fullness. */
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_DROP_NEWEST);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
 
     xco_queue_send_op_t op;
     xco_queue_send_op_init(&op, &q, 2);            /* dropped */
     assert(op.done.set);
     uintptr_t v;
     assert(xco_event_poll(&op.done.base, &v, NULL));
     assert(v == 1);
 
     xco_queue_send_op_deinit(&op);
 }
 
 static void test_queue_send_op_select_loses(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     xco_latch_t   l; xco_latch_init(&l);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
 
     xco_queue_send_op_t op;
     xco_queue_send_op_init(&op, &q, 2);
     assert(q.send_head == &op.qsw.sw.base);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_event_t *srcs[2] = {&op.done.base, &l.base};
     xco_select_event_init(&sel, inputs, 2, srcs);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
 
     xco_latch_set(&l, 0xCAFE);
     xco_rt_run(&rt, 0);
     assert(w.got == 1);
     assert(q.send_head == &op.qsw.sw.base);
 
     xco_select_event_deinit(&sel);
     xco_queue_send_op_deinit(&op);
     assert(q.send_head == NULL);
 }
 
 /* ---- Chan close ---------------------------------------------------- */
 
 typedef struct {
     xco_mach_t       base;
     xco_chan_t       *c;
     xco_runtime_t    *rt;
     xco_waker_t  sw;
     int           phase;
     xco_recv_status_t status;
     uintptr_t     value;
 } chan_rx_t;
 
 static xco_step_result_t chan_rx_step(xco_mach_t *s, uintptr_t v) {
     chan_rx_t *r = (chan_rx_t *)s;
     (void)v;
     switch (r->phase) {
     case 0: {
         xco_recv_status_t st = xco_chan_recv(r->c, &r->value);
         if (st != XCO_RECV_EMPTY) {
             r->status = st;
             r->phase  = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         }
         xco_waker_init(&r->sw, r->rt, &r->base);
         xco_event_poll(&r->c->recv, NULL, &r->sw.base);
         r->phase = 1;
         return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
     }
     case 1:
         r->status = xco_chan_recv(r->c, &r->value);
         r->phase  = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void chan_rx_init(chan_rx_t *r, xco_runtime_t *rt, xco_chan_t *c) {
     r->base   = (xco_mach_t){.step = chan_rx_step, .status = XCO_STEP_INIT};
     r->c      = c;
     r->rt     = rt;
     r->phase  = 0;
     r->status = XCO_RECV_EMPTY;
     r->value  = 0;
 }
 
 static void test_chan_close_empty(void) {
     xco_chan_t c; xco_chan_init(&c);
     assert(!xco_chan_is_closed(&c));
     xco_chan_close(&c);
     assert(xco_chan_is_closed(&c));
     xco_chan_close(&c);                              /* idempotent */
 
     uintptr_t v;
     assert(xco_chan_recv(&c, &v) == XCO_RECV_CLOSED);
 }
 
 static void test_chan_close_wakes_receiver(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     chan_rx_t r; chan_rx_init(&r, &rt, &c);
     xco_step(&r.base, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_SUSPENDED);
     assert(c.recv_head == &r.sw.base);
 
     xco_chan_close(&c);
     assert(c.recv_head == NULL);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_DEAD);
     assert(r.status == XCO_RECV_CLOSED);
 }
 
 typedef struct {
     xco_mach_t           base;
     xco_chan_t           *c;
     xco_runtime_t        *rt;
     xco_chan_send_waiter_t csw;
     uintptr_t         value;
     int               phase;
     bool              done;
     bool              delivered;
 } chan_tx_t;
 
 static xco_step_result_t chan_tx_step(xco_mach_t *s, uintptr_t v) {
     chan_tx_t *snd = (chan_tx_t *)s;
     (void)v;
     switch (snd->phase) {
     case 0:
         xco_chan_send_waiter_init(&snd->csw, snd->rt, &snd->base);
         switch (xco_chan_send_poll(snd->c, snd->value, &snd->csw)) {
         case XCO_SEND_DELIVERED:
             snd->done      = true;
             snd->delivered = true;
             snd->phase     = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_SEND_CLOSED:
             snd->done      = true;
             snd->delivered = false;
             snd->phase     = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_SEND_BLOCKED:
             snd->phase = 1;
             return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
         }
         __builtin_unreachable();
     case 1:
         snd->done      = true;
         snd->delivered = snd->csw.delivered;
         snd->phase     = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void chan_tx_init(chan_tx_t *snd, xco_runtime_t *rt, xco_chan_t *c, uintptr_t value) {
     snd->base      = (xco_mach_t){.step = chan_tx_step, .status = XCO_STEP_INIT};
     snd->c         = c;
     snd->rt        = rt;
     snd->value     = value;
     snd->phase     = 0;
     snd->done      = false;
     snd->delivered = false;
 }
 
 static void test_chan_close_drains_senders(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_chan_t    c;  xco_chan_init(&c);
 
     chan_tx_t s[3];
     for (int i = 0; i < 3; i++) {
         chan_tx_init(&s[i], &rt, &c, (uintptr_t)(100 + i));
         xco_step(&s[i].base, 0);
     }
     assert(c.send_head == &s[0].csw.sw.base);
 
     xco_chan_close(&c);
     assert(c.send_head == NULL);
     xco_rt_run(&rt, 0);
     for (int i = 0; i < 3; i++) {
         assert(xco_mach_status(&s[i].base) == XCO_STEP_DEAD);
         assert(s[i].done);
         assert(!s[i].delivered);
     }
 }
 
 static void test_chan_send_try_after_close(void) {
     xco_chan_t c; xco_chan_init(&c);
     xco_chan_close(&c);
     assert(xco_chan_send_poll(&c, 1, NULL) == XCO_SEND_CLOSED);
 }
 
 static void test_chan_send_op_close_inline(void) {
     xco_chan_t c; xco_chan_init(&c);
     xco_chan_close(&c);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xABBA);
     assert(op.done.set);
     uintptr_t v;
     assert(xco_event_poll(&op.done.base, &v, NULL));
     assert(v == 0);                               /* delivered=false */
     xco_chan_send_op_deinit(&op);
 }
 
 static void test_chan_send_op_close_drain(void) {
     xco_chan_t c; xco_chan_init(&c);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xABBA);
     assert(!op.done.set);
 
     xco_chan_close(&c);
     assert(op.done.set);
     uintptr_t v;
     assert(xco_event_poll(&op.done.base, &v, NULL));
     assert(v == 0);
     xco_chan_send_op_deinit(&op);
 }
 
 static void test_chan_send_op_delivered_payload(void) {
     xco_chan_t c; xco_chan_init(&c);
 
     xco_chan_send_op_t op;
     xco_chan_send_op_init(&op, &c, 0xFEED);
     assert(!op.done.set);
 
     uintptr_t v;
     assert(xco_chan_recv(&c, &v) == XCO_RECV_GOT);
     assert(v == 0xFEED);
     assert(op.done.set);
     uintptr_t pv;
     assert(xco_event_poll(&op.done.base, &pv, NULL));
     assert(pv == 1);
 
     xco_chan_send_op_deinit(&op);
 }
 
 /* ---- Queue close --------------------------------------------------- */
 
 typedef struct {
     xco_mach_t       base;
     xco_queue_t      *q;
     xco_runtime_t    *rt;
     xco_waker_t  sw;
     int           phase;
     xco_recv_status_t status;
     uintptr_t     value;
 } queue_rx_t;
 
 static xco_step_result_t queue_rx_step(xco_mach_t *s, uintptr_t v) {
     queue_rx_t *r = (queue_rx_t *)s;
     (void)v;
     switch (r->phase) {
     case 0: {
         xco_recv_status_t st = xco_queue_recv(r->q, &r->value);
         if (st != XCO_RECV_EMPTY) {
             r->status = st;
             r->phase  = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         }
         xco_waker_init(&r->sw, r->rt, &r->base);
         xco_event_poll(xco_queue_recv_event(r->q), NULL, &r->sw.base);
         r->phase = 1;
         return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
     }
     case 1:
         r->status = xco_queue_recv(r->q, &r->value);
         r->phase  = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void queue_rx_init(queue_rx_t *r, xco_runtime_t *rt, xco_queue_t *q) {
     r->base   = (xco_mach_t){.step = queue_rx_step, .status = XCO_STEP_INIT};
     r->q      = q;
     r->rt     = rt;
     r->phase  = 0;
     r->status = XCO_RECV_EMPTY;
     r->value  = 0;
 }
 
 static void test_queue_close_drains_buffered_then_eof(void) {
     uintptr_t buf[3];
     xco_queue_t   q; xco_queue_init(&q, buf, 3, XCO_QUEUE_BLOCK);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
     assert(xco_queue_send_poll(&q, 2, NULL) == XCO_QSEND_ACCEPTED);
 
     xco_queue_close(&q);
     assert(xco_queue_is_closed(&q));
 
     uintptr_t v;
     assert(xco_queue_recv(&q, &v) == XCO_RECV_GOT && v == 1);
     assert(xco_queue_recv(&q, &v) == XCO_RECV_GOT && v == 2);
     assert(xco_queue_recv(&q, &v) == XCO_RECV_CLOSED);
     xco_queue_close(&q);                              /* idempotent */
     assert(xco_queue_recv(&q, &v) == XCO_RECV_CLOSED);
 }
 
 static void test_queue_close_wakes_receiver(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
 
     queue_rx_t r; queue_rx_init(&r, &rt, &q);
     xco_step(&r.base, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_SUSPENDED);
 
     xco_queue_close(&q);
     xco_rt_run(&rt, 0);
     assert(xco_mach_status(&r.base) == XCO_STEP_DEAD);
     assert(r.status == XCO_RECV_CLOSED);
 }
 
 typedef struct {
     xco_mach_t            base;
     xco_queue_t           *q;
     xco_runtime_t         *rt;
     xco_queue_send_waiter_t qsw;
     uintptr_t          value;
     int                phase;
     bool               done;
     bool               delivered;
 } queue_tx_t;
 
 static xco_step_result_t queue_tx_step(xco_mach_t *s, uintptr_t v) {
     queue_tx_t *snd = (queue_tx_t *)s;
     (void)v;
     switch (snd->phase) {
     case 0:
         xco_queue_send_waiter_init(&snd->qsw, snd->rt, &snd->base);
         switch (xco_queue_send_poll(snd->q, snd->value, &snd->qsw)) {
         case XCO_QSEND_ACCEPTED:
             snd->done      = true;
             snd->delivered = true;
             snd->phase     = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_QSEND_CLOSED:
             snd->done      = true;
             snd->delivered = false;
             snd->phase     = 2;
             return (xco_step_result_t){0, XCO_STEP_DEAD};
         case XCO_QSEND_BLOCKED:
             snd->phase = 1;
             return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
         }
         __builtin_unreachable();
     case 1:
         snd->done      = true;
         snd->delivered = snd->qsw.delivered;
         snd->phase     = 2;
         return (xco_step_result_t){0, XCO_STEP_DEAD};
     }
     __builtin_unreachable();
 }
 
 static void queue_tx_init(queue_tx_t *s, xco_runtime_t *rt, xco_queue_t *q, uintptr_t v) {
     s->base      = (xco_mach_t){.step = queue_tx_step, .status = XCO_STEP_INIT};
     s->q         = q;
     s->rt        = rt;
     s->value     = v;
     s->phase     = 0;
     s->done      = false;
     s->delivered = false;
 }
 
 static void test_queue_close_drains_senders(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
 
     queue_tx_t s[2];
     for (int i = 0; i < 2; i++) {
         queue_tx_init(&s[i], &rt, &q, (uintptr_t)(100 + i));
         xco_step(&s[i].base, 0);
         assert(xco_mach_status(&s[i].base) == XCO_STEP_SUSPENDED);
     }
 
     xco_queue_close(&q);
     xco_rt_run(&rt, 0);
     for (int i = 0; i < 2; i++) {
         assert(xco_mach_status(&s[i].base) == XCO_STEP_DEAD);
         assert(!s[i].delivered);
     }
 }
 
 static void test_queue_send_try_after_close_block(void) {
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     xco_queue_close(&q);
     assert(xco_queue_send_poll(&q, 42, NULL) == XCO_QSEND_CLOSED);
 }
 
 static void test_queue_send_try_after_close_drop(void) {
     /* Closed is closed regardless of policy: DROP_* also reports CLOSED
      * (behavior change from the old try API, which returned "accepted"
      * after close under DROP_*). */
     uintptr_t buf[1];
     xco_queue_t   q1; xco_queue_init(&q1, buf, 1, XCO_QUEUE_DROP_NEWEST);
     xco_queue_close(&q1);
     assert(xco_queue_send_poll(&q1, 42, NULL) == XCO_QSEND_CLOSED);
 
     uintptr_t buf2[1];
     xco_queue_t   q2; xco_queue_init(&q2, buf2, 1, XCO_QUEUE_DROP_OLDEST);
     xco_queue_close(&q2);
     assert(xco_queue_send_poll(&q2, 42, NULL) == XCO_QSEND_CLOSED);
 }
 
 static void test_queue_send_op_close_drain(void) {
     uintptr_t buf[1];
     xco_queue_t   q; xco_queue_init(&q, buf, 1, XCO_QUEUE_BLOCK);
     assert(xco_queue_send_poll(&q, 1, NULL) == XCO_QSEND_ACCEPTED);
 
     xco_queue_send_op_t op;
     xco_queue_send_op_init(&op, &q, 2);
     assert(!op.done.set);
 
     xco_queue_close(&q);
     assert(op.done.set);
     uintptr_t v;
     assert(xco_event_poll(&op.done.base, &v, NULL));
     assert(v == 0);
     xco_queue_send_op_deinit(&op);
 }
 
 /* ---- Ticker -------------------------------------------------------- */
 
 typedef struct {
     xco_mach_t      base;
     xco_ticker_t    *t;
     xco_runtime_t   *rt;
     xco_waker_t sw;
     int          n_seen;
     int          target;
     uint64_t     last;
 } ticker_sub_t;
 
 static xco_step_result_t ticker_sub_step(xco_mach_t *s, uintptr_t v) {
     ticker_sub_t *sub = (ticker_sub_t *)s;
     if (sub->n_seen > 0) sub->last = (uint64_t)v;
     if (sub->n_seen >= sub->target) return (xco_step_result_t){v, XCO_STEP_DEAD};
     xco_event_poll(xco_ticker_event(sub->t), NULL, &sub->sw.base);
     sub->n_seen++;
     return (xco_step_result_t){0, XCO_STEP_SUSPENDED};
 }
 
 static void ticker_sub_init(ticker_sub_t *s, xco_runtime_t *rt, xco_ticker_t *t, int target) {
     s->base   = (xco_mach_t){.step = ticker_sub_step, .status = XCO_STEP_INIT};
     s->t      = t;
     s->rt     = rt;
     s->n_seen = 0;
     s->target = target;
     s->last   = 0;
     xco_waker_init(&s->sw, rt, &s->base);
 }
 
 static void test_ticker_single_tick(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_ticker_t ti;
     xco_ticker_init(&ti, &h.base, 100, 100);
 
     ticker_sub_t s; ticker_sub_init(&s, &rt, &ti, 1);
     xco_step(&s.base, 0);
     assert(xco_mach_status(&s.base) == XCO_STEP_SUSPENDED);
 
     xco_rt_run(&rt, 100);
     assert(xco_mach_status(&s.base) == XCO_STEP_DEAD);
     assert(s.last == 100);
 
     xco_ticker_deinit(&ti);
 }
 
 static void test_ticker_multiple_ticks(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_ticker_t ti;
     xco_ticker_init(&ti, &h.base, 50, 50);
 
     ticker_sub_t s; ticker_sub_init(&s, &rt, &ti, 2);
     xco_step(&s.base, 0);
 
     xco_rt_run(&rt, 50);
     assert(xco_mach_status(&s.base) == XCO_STEP_SUSPENDED);
     assert(s.last == 50);
 
     xco_rt_run(&rt, 100);
     assert(xco_mach_status(&s.base) == XCO_STEP_DEAD);
     assert(s.last == 100);                    /* fired = 50 + period */
 
     xco_ticker_deinit(&ti);
 }
 
 static void test_ticker_deinit_before_fire(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_ticker_t ti;
     xco_ticker_init(&ti, &h.base, 100, 100);
     assert(ti.timer.in_heap);
 
     xco_ticker_deinit(&ti);
     assert(!ti.timer.in_heap);
 
     assert(xco_timers_peek(&h.base) == UINT64_MAX);
     xco_rt_run(&rt, 10000);
 }
 
 static void test_ticker_select(void) {
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_rt_attach_timers(&rt, &h.base);
 
     xco_ticker_t ti; xco_ticker_init(&ti, &h.base, 100, 100);
     xco_cancel_t c;  xco_cancel_init(&c);
 
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, xco_ticker_event(&ti), &c);
 
     waiter_t w; waiter_init(&w, &rt, &sel.done.base);
     xco_step(&w.base, 0);
 
     xco_rt_run(&rt, 100);
     assert(xco_mach_status(&w.base) == XCO_STEP_DEAD);
     assert(w.got == XCO_WAIT_OK);
     assert(inputs[XCO_WAIT_OK].value == 100);
 
     xco_select_event_deinit(&sel);
     xco_ticker_deinit(&ti);
 }
 
 static void test_ticker_try_always_false(void) {
     xco_pairing_heap_t h; xco_pairing_heap_init(&h);
     xco_ticker_t ti; xco_ticker_init(&ti, &h.base, 100, 100);
     assert(!xco_event_poll(xco_ticker_event(&ti), NULL, NULL));
     xco_ticker_deinit(&ti);
 }
 
 /* ---- Task group ---------------------------------------------------- */
 
 static void test_task_group_empty_join(void) {
     /* An empty (no-attach) group's join is not fired — fan-in semantics
      * require at least one attach. */
     xco_task_group_t g; xco_task_group_init(&g);
     assert(!xco_event_poll(xco_task_group_join_event(&g), NULL, NULL));
 }
 
 static void test_task_group_join_after_all_finish(void) {
     xco_runtime_t rt; xco_rt_init(&rt); (void)rt;
 
     taskbody_t a_body, b_body;
     xco_task_t     a, b;
     taskbody_init(&a_body, &a, 1, 0xA);
     xco_task_init(&a, &a_body.base);
     taskbody_init(&b_body, &b, 1, 0xB);
     xco_task_init(&b, &b_body.base);
 
     xco_task_group_t g; xco_task_group_init(&g);
     xco_group_attach_t sa, sb;
     xco_task_group_attach(&g, &a, &sa);
     xco_task_group_attach(&g, &b, &sb);
     assert(!xco_event_poll(xco_task_group_join_event(&g), NULL, NULL));
 
     while (xco_mach_status(&a_body.base) != XCO_STEP_DEAD) xco_step(&a_body.base, 0);
     assert(xco_task_finished(&a));
     assert(!xco_event_poll(xco_task_group_join_event(&g), NULL, NULL));
 
     while (xco_mach_status(&b_body.base) != XCO_STEP_DEAD) xco_step(&b_body.base, 0);
     assert(xco_task_finished(&b));
     uintptr_t v;
     assert(xco_event_poll(xco_task_group_join_event(&g), &v, NULL));
     assert(v == 0);
 }
 
 static void test_task_group_cancel_fanout(void) {
     xco_runtime_t rt; xco_rt_init(&rt); (void)rt;
 
     taskbody_t a_body, b_body;
     xco_task_t     a, b;
     taskbody_init(&a_body, &a, 100, 0xA);
     xco_task_init(&a, &a_body.base);
     taskbody_init(&b_body, &b, 100, 0xB);
     xco_task_init(&b, &b_body.base);
 
     xco_task_group_t g; xco_task_group_init(&g);
     xco_group_attach_t sa, sb;
     xco_task_group_attach(&g, &a, &sa);
     xco_task_group_attach(&g, &b, &sb);
 
     xco_task_group_cancel(&g);
     assert(xco_task_is_cancelled(&a));
     assert(xco_task_is_cancelled(&b));
     assert(xco_cancel_is_set(xco_task_group_cancel_handle(&g)));
 
     while (xco_mach_status(&a_body.base) != XCO_STEP_DEAD) xco_step(&a_body.base, 0);
     while (xco_mach_status(&b_body.base) != XCO_STEP_DEAD) xco_step(&b_body.base, 0);
 
     assert(xco_event_poll(xco_task_group_join_event(&g), NULL, NULL));
 }
 
 static void test_task_group_finished_detaches(void) {
     taskbody_t a_body, b_body;
     xco_task_t     a, b;
     taskbody_init(&a_body, &a, 1, 0xA);
     xco_task_init(&a, &a_body.base);
     taskbody_init(&b_body, &b, 100, 0xB);
     xco_task_init(&b, &b_body.base);
 
     xco_task_group_t g; xco_task_group_init(&g);
     xco_group_attach_t sa, sb;
     xco_task_group_attach(&g, &a, &sa);
     xco_task_group_attach(&g, &b, &sb);
     assert(g.head == &sa && g.tail == &sb);
 
     while (xco_mach_status(&a_body.base) != XCO_STEP_DEAD) xco_step(&a_body.base, 0);
     /* a's bridge has fired → sa is detached. */
     assert(g.head == &sb && g.tail == &sb);
     assert(sa.next == NULL && sa.prev == NULL);
 
     /* Cancel only touches still-attached b. */
     xco_task_group_cancel(&g);
     assert(xco_task_is_cancelled(&b));
 }
 
 /* ---- Runtime test -------------------------------------------------- */
 
 static void test_runtime_drains(void) {
     /* xco_rt_run keeps going until quiescent — including steps queued
      * during another step's resumption. Three waiters on one latch
      * all reach DEAD in a single xco_rt_run. */
     xco_runtime_t rt; xco_rt_init(&rt);
     xco_latch_t   l;  xco_latch_init(&l);
 
     waiter_t w[3];
     for (int i = 0; i < 3; i++) waiter_init(&w[i], &rt, &l.base);
     for (int i = 0; i < 3; i++) xco_step(&w[i].base, 0);
 
     xco_latch_set(&l, 0xC0DE);
     xco_rt_run(&rt, 0);
 
     assert(rt.head == NULL);
     for (int i = 0; i < 3; i++) {
         assert(xco_mach_status(&w[i].base) == XCO_STEP_DEAD);
         assert(w[i].got == 0xC0DE);
     }
 }
 
 int main(void) {
     test_latch_wake();
     test_latch_already_set();
     test_latch_multi_waiter();
     test_latch_set_idempotent();
     test_latch_unpark();
 
     test_select_winner();
     test_select_fast_path();
     test_select_deinit_unparks();
     test_select_compose();
 
     test_allof_basic();
     test_allof_fast_path_partial();
     test_allof_fast_path_all();
     test_allof_empty();
     test_allof_deinit_partial();
     test_allof_compose_with_select();
 
     test_chan_send_blocks_until_recv();
     test_chan_recv_blocks_until_send();
     test_chan_send_try_no_recv();
     test_chan_fifo();
     test_chan_recv_fifo();
     test_chan_send_unpark();
     test_chan_select_recv();
     test_chan_select_recv_fast_path();
     test_chan_send_op_inline();
     test_chan_send_op_blocks();
     test_chan_select_send();
     test_chan_select_send_loses();
 
     test_cancel_basic();
     test_wait_or_cancel_ev_wins();
     test_wait_or_cancel_cancel_wins();
     test_wait_or_cancel_already_cancelled();
     test_wait_or_cancel_ev_precedes_cancel();
     test_wait_or_cancel_chan_recv();
     test_wait_or_cancel_send_op();
 
     test_timer_basic();
     test_timer_peek();
     test_timer_cancel();
     test_timer_park_wake();
     test_timer_select();
     test_timer_pairing_heap_order();
     test_timer_cancel_stress();
     test_timer_deinit_idempotent();
 
     test_timeout_fires();
     test_timeout_deinit_before_fire();
     test_timeout_with_wait_or_cancel();
     test_timeout_ev_wins();
 
     test_rt_run_advances_timers();
     test_rt_run_no_timers();
 
     test_runtime_drains();
 
     test_semaphore_inline_acquire();
     test_semaphore_park_then_release();
     test_semaphore_fifo();
     test_semaphore_release_overflow_to_count();
     test_semaphore_select_acquire();
     test_semaphore_binary_mutex();
 
     test_queue_block_buffered();
     test_queue_block_sender_parks();
     test_queue_drop_newest();
     test_queue_drop_oldest();
     test_queue_direct_handoff();
     test_queue_recv_blocks_then_drains_buffered();
     test_queue_select_recv();
     test_queue_send_unpark();
 
     test_broadcast_try_always_false();
     test_broadcast_publish_wakes_all();
     test_broadcast_rearm();
     test_broadcast_missed_publish();
     test_broadcast_unpark();
 
     test_task_state_machine_join();
     test_task_state_machine_cancel();
 
     test_countdown_basic();
     test_countdown_zero_init();
     test_countdown_add();
     test_countdown_park_wake();
 
     test_notify_try_always_false();
     test_notify_one_fifo();
     test_notify_all();
     test_notify_select();
     test_notify_unpark();
 
     test_mutex_basic();
 
     test_queue_send_op_inline();
     test_queue_send_op_blocks();
     test_queue_send_op_drop_inline();
     test_queue_send_op_select_loses();
 
     test_chan_close_empty();
     test_chan_close_wakes_receiver();
     test_chan_close_drains_senders();
     test_chan_send_try_after_close();
     test_chan_send_op_close_inline();
     test_chan_send_op_close_drain();
     test_chan_send_op_delivered_payload();
 
     test_queue_close_drains_buffered_then_eof();
     test_queue_close_wakes_receiver();
     test_queue_close_drains_senders();
     test_queue_send_try_after_close_block();
     test_queue_send_try_after_close_drop();
     test_queue_send_op_close_drain();
 
     test_ticker_single_tick();
     test_ticker_multiple_ticks();
     test_ticker_deinit_before_fire();
     test_ticker_select();
     test_ticker_try_always_false();
 
     test_task_group_empty_join();
     test_task_group_join_after_all_finish();
     test_task_group_cancel_fanout();
     test_task_group_finished_detaches();
 
     printf("ok\n");
     return 0;
 }