Shared execution engine used by both [`execute`] and [`execute_streaming`]. When `event_tx` is `Some`, node-level [`StreamEvent`]s are emitted at every milestone. When `None` (non-streaming path) no channel operations occur, so performance is identical to the original `execute()`.
(
&self,
workflow: Workflow,
guardrail_enforcer: Option<Arc<Enforcer>>,
event_tx: Option<tokio::sync::mpsc::Sender<crate::stream::StreamEvent>>,
stream_mode: cr
| 389 | /// milestone. When `None` (non-streaming path) no channel operations occur, so |
| 390 | /// performance is identical to the original `execute()`. |
| 391 | async fn execute_internal( |
| 392 | &self, |
| 393 | workflow: Workflow, |
| 394 | guardrail_enforcer: Option<Arc<Enforcer>>, |
| 395 | event_tx: Option<tokio::sync::mpsc::Sender<crate::stream::StreamEvent>>, |
| 396 | stream_mode: crate::stream::StreamMode, |
| 397 | mut context: WorkflowContext, |
| 398 | ) -> GraphBitResult<WorkflowContext> { |
| 399 | use crate::stream::{StreamEvent, error_type_from_graphbit_error, error_type_from_string}; |
| 400 | |
| 401 | let start_time = std::time::Instant::now(); |
| 402 | |
| 403 | // Set initial workflow state |
| 404 | context.state = WorkflowState::Running { |
| 405 | current_node: NodeId::new(), |
| 406 | }; |
| 407 | |
| 408 | // Validate workflow before execution |
| 409 | if let Err(e) = workflow.validate() { |
| 410 | if let Some(ref tx) = event_tx { |
| 411 | let _ = tx |
| 412 | .send(StreamEvent::WorkflowFailed { |
| 413 | error: e.to_string(), |
| 414 | error_type: error_type_from_graphbit_error(&e), |
| 415 | }) |
| 416 | .await; |
| 417 | } |
| 418 | return Err(e); |
| 419 | } |
| 420 | |
| 421 | // Validate unique LLM configurations once per workflow (deduped by config fingerprint). |
| 422 | // This prevents validating the same key/provider repeatedly during agent creation. |
| 423 | { |
| 424 | // If we are re-entering execution (e.g. tool-resolution downstream reruns), avoid |
| 425 | // re-validating LLM configs again. This flag is workflow-run scoped and does not |
| 426 | // persist outside the current execution context. |
| 427 | let already_validated = context |
| 428 | .metadata |
| 429 | .get("workflow_llm_configs_validated") |
| 430 | .and_then(|v| v.as_bool()) |
| 431 | .unwrap_or(false); |
| 432 | if already_validated { |
| 433 | tracing::info!("Workflow LLM validation: already validated; skipping"); |
| 434 | } else { |
| 435 | use crate::llm::{LlmProvider, LlmProviderFactory, LlmRequest}; |
| 436 | use std::collections::HashMap; |
| 437 | |
| 438 | let mut unique: HashMap<String, crate::llm::LlmConfig> = HashMap::with_capacity(4); |
| 439 | for node in workflow.graph.get_nodes().values() { |
| 440 | if matches!(node.node_type, NodeType::Agent { .. }) { |
| 441 | let resolved = self.resolve_llm_config_for_node(&node.config); |
| 442 | if let Some(fp) = resolved.validation_fingerprint() { |
| 443 | unique.entry(fp).or_insert(resolved); |
| 444 | } |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | tracing::info!( |
no test coverage detected