Function window 

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

Splits source into sub-nodes triggered by notifier. Each “window” is a fresh state node created via Core::register_state(). The operator emits the inner node’s NodeId as a handle via binding.intern_node(inner_id).

• On activation: create first inner window node, emit it.
• Source DATA: forward to current inner window via core.emit_or_defer(inner_id, handle).
• Notifier DATA: complete current window, create new window, emit it.
• Source COMPLETE: complete current window, then complete self.
• Either ERROR: error current window + error self.
• Notifier COMPLETE: do NOT auto-complete.