Function take_while 

pub fn take_while<F>(
    core: &Core,
    binding: &Arc<dyn OperatorBinding>,
    source: NodeId,
    predicate: F,
) -> OperatorRegistration
where
    F: Fn(HandleId) -> bool + Send + Sync + 'static,

take_while(source, predicate) — emits while predicate(input) holds; on the first false, emits any preceding passes from the same batch then self-completes. Reuses BindingBoundary::predicate_each (D029).