Method execute_streaming

python/src/workflow/executor.rs:450–720

(
        &mut self,
        workflow: &Workflow,
        policy: Option<&Bound<'_, GuardRailPolicyConfig>>,
        stream_mode: Option<&str>,
    )

    /// `Executor.get_stream_event_schema()`.
    #[pyo3(signature = (workflow, policy=None, stream_mode=None))]
    fn execute_streaming(
        &mut self,
        workflow: &Workflow,
        policy: Option<&Bound<'_, GuardRailPolicyConfig>>,
        stream_mode: Option<&str>,
    ) -> PyResult<WorkflowStreamIterator> {
        // Validate workflow
        if workflow.inner.graph.node_count() == 0 {
            return Err(validation_error(
                "workflow",
                None,
                "Workflow cannot be empty",
            ));
        }
        if let Err(e) = workflow.inner.validate() {
            return Err(validation_error(
                "workflow",
                None,
                &format!("Invalid workflow: {}", e),
            ));
        }

        // Parse stream mode (default: Updates)
        let mode = stream_mode
            .and_then(StreamMode::from_str_opt)
            .unwrap_or(StreamMode::Updates);

        let workflow_clone = workflow.inner.clone();
        let llm_config = self.llm_config.inner.clone();
        let config = self.config.clone();
        let timeout_duration = config.timeout;
        let guardrail_enforcer = policy.map(|p| {
            let cfg = p.borrow().get_inner();
            Arc::new(GuardRail::enforcer_for(cfg, workflow_clone.id.to_string()))
        });
        let guardrail_for_iterator = guardrail_enforcer.clone();

        // Internal channel from core executor -> processor
        let (core_event_tx, mut core_event_rx) = tokio::sync::mpsc::channel::<StreamEvent>(256);
        // Public channel from processor -> Python iterator
        let (event_tx, event_rx) = tokio::sync::mpsc::channel::<TimedStreamEvent>(256);

        // Stream mode for follow-up LLM calls in the live tool loop (Messages/All → token stream).
        let tool_loop_stream_mode = mode;

        // Spawn the streaming executor on the shared Tokio runtime
        get_runtime().spawn(async move {
            let conditional_handlers =
                match crate::workflow::node::build_core_conditional_handlers(&workflow_clone) {
                    Ok(h) => h,
                    Err(e) => {
                        let err_msg = e.to_string();
                        let _ = core_event_tx
                            .send(StreamEvent::WorkflowFailed {
                                error: err_msg.clone(),
                                error_type: graphbit_core::error_type_from_string(&err_msg),
                            })
                            .await;