Method execute_workflow_internal

python/src/workflow/executor.rs:1662–1782

Internal workflow execution with mode-specific optimizations and tool call handling. When `guardrail_enforcer` is `Some`, the core encodes before LLM and decodes after LLM; we decode before tool usage only (no encode after tool).

(
        llm_config: graphbit_core::llm::LlmConfig,
        workflow: graphbit_core::workflow::Workflow,
        config: ExecutionConfig,
        guardrail_enforcer: Option<Arc<Enforcer>>,
    )

    /// When `guardrail_enforcer` is `Some`, the core encodes before LLM and decodes after LLM;
    /// we decode before tool usage only (no encode after tool).
    async fn execute_workflow_internal(
        llm_config: graphbit_core::llm::LlmConfig,
        workflow: graphbit_core::workflow::Workflow,
        config: ExecutionConfig,
        guardrail_enforcer: Option<Arc<Enforcer>>,
    ) -> Result<graphbit_core::types::WorkflowContext, graphbit_core::errors::GraphBitError> {
        let conditional_handlers =
            crate::workflow::node::build_core_conditional_handlers(&workflow)?;
        let executor = match config.mode {
            ExecutionMode::Balanced => CoreWorkflowExecutor::new()
                .with_default_llm_config(llm_config.clone())
                .with_conditional_handlers(conditional_handlers),
        };

        // Execute the workflow (core applies encode before LLM, decode after LLM when enforcer is Some)
        let mut context = executor
            .execute(workflow.clone(), guardrail_enforcer.clone())
            .await?;

        // Store LLM config in context metadata for tool call handling
        if let Ok(llm_config_json) = serde_json::to_value(&llm_config) {
            context
                .metadata
                .insert("llm_config".to_string(), llm_config_json);
        }

        // Store workflow name in context metadata for result schema
        context.metadata.insert(
            "workflow_name".to_string(),
            serde_json::Value::String(workflow.name.clone()),
        );

        // Check if any node outputs contain tool_calls_required responses and handle them
        let mut context = context;
        let mut rerun_attempts = 0;
        loop {
            let (ctx, nodes_with_tool_calls) = Self::handle_tool_calls_in_context(
                context,
                &workflow,
                guardrail_enforcer.as_ref().map(|arc| arc.as_ref()),
            )
            .await?;
            context = ctx;

            if nodes_with_tool_calls.is_empty() {
                break;
            }

            // Identify downstream nodes that depend on tool-resolved outputs and need rerun.
            // This includes nodes that themselves may have tool calls, as they can still depend on
            // upstream nodes whose outputs were just resolved.
            let mut downstream_nodes: HashSet<String> = HashSet::new();
            if let Some(deps_obj) = context.metadata.get("node_dependencies").and_then(|v| v.as_object()) {
                let mut queue: Vec<String> = nodes_with_tool_calls.clone();
                while let Some(parent_id) = queue.pop() {
                    for (node_id, parents) in deps_obj.iter() {
                        if downstream_nodes.contains(node_id) {
                            continue;