Function take_until 

pub fn take_until(
    core: &Core,
    binding: &Arc<dyn ProducerBinding>,
    source: NodeId,
    notifier: NodeId,
) -> NodeId

take_until(source, notifier) — forward DATA from source until notifier emits its first DATA, then terminate the producer with COMPLETE. Errors from either source cascade. Source COMPLETE terminates the producer.

Notifier DATA is consumed but never forwarded (zero-FFI on the notifier path — we don’t dereference its payload, just use the emission as a signal).