symbi-runtime 1.18.0

//! Action executor with parallel dispatch
//!
//! Executes approved actions concurrently using `FuturesUnordered`,
//! with per-tool timeouts, circuit breaker integration, and barrier
//! sync before returning observations.

use crate::reasoning::circuit_breaker::CircuitBreakerRegistry;
use crate::reasoning::inference::ToolDefinition;
use crate::reasoning::loop_types::{LoopConfig, Observation, ProposedAction};
use async_trait::async_trait;
use futures::stream::{FuturesUnordered, StreamExt};
use std::time::Duration;

/// Trait for executing proposed actions and producing observations.
#[async_trait]
pub trait ActionExecutor: Send + Sync {
    /// Execute a batch of approved actions, potentially in parallel.
    ///
    /// Returns observations from all action results. Circuit breakers
    /// are checked before each dispatch.
    async fn execute_actions(
        &self,
        actions: &[ProposedAction],
        config: &LoopConfig,
        circuit_breakers: &CircuitBreakerRegistry,
    ) -> Vec<Observation>;

    /// Return tool definitions this executor can handle.
    ///
    /// The runner auto-populates `LoopConfig.tool_definitions` from this
    /// when the config's list is empty. Override in executors that discover
    /// tools dynamically.
    fn tool_definitions(&self) -> Vec<ToolDefinition> {
        Vec::new()
    }
}

/// Default executor that dispatches tool calls in parallel.
pub struct DefaultActionExecutor {
    tool_timeout: Duration,
}

impl DefaultActionExecutor {
    pub fn new(tool_timeout: Duration) -> Self {
        Self { tool_timeout }
    }
}

impl Default for DefaultActionExecutor {
    fn default() -> Self {
        Self::new(Duration::from_secs(30))
    }
}

#[async_trait]
impl ActionExecutor for DefaultActionExecutor {
    async fn execute_actions(
        &self,
        actions: &[ProposedAction],
        config: &LoopConfig,
        circuit_breakers: &CircuitBreakerRegistry,
    ) -> Vec<Observation> {
        let tool_calls: Vec<&ProposedAction> = actions
            .iter()
            .filter(|a| matches!(a, ProposedAction::ToolCall { .. }))
            .collect();

        if tool_calls.is_empty() {
            return Vec::new();
        }

        let timeout = self.tool_timeout.min(config.tool_timeout);

        // Dispatch tool calls concurrently
        let mut futures = FuturesUnordered::new();

        for action in &tool_calls {
            if let ProposedAction::ToolCall {
                call_id,
                name,
                arguments,
            } = action
            {
                let name = name.clone();
                let arguments = arguments.clone();
                let call_id = call_id.clone();

                // Check circuit breaker first
                let cb_result = circuit_breakers.check(&name).await;

                futures.push(async move {
                    if let Err(cb_err) = cb_result {
                        return Observation {
                            source: name,
                            content: format!(
                                "Tool circuit is open: {}. The tool endpoint has been failing and is temporarily disabled.",
                                cb_err
                            ),
                            is_error: true,
                            call_id: Some(call_id),
                            metadata: {
                                let mut m = std::collections::HashMap::new();
                                m.insert("error_type".into(), "circuit_open".into());
                                m
                            },
                        };
                    }

                    // Execute the tool call with timeout
                    let result = tokio::time::timeout(timeout, async {
                        // In production, this would call the ToolInvocationEnforcer.
                        // For now, produce an observation indicating the tool was called.
                        execute_tool_call(&name, &arguments).await
                    })
                    .await;

                    match result {
                        Ok(Ok(content)) => {
                            Observation::tool_result(&name, content).with_call_id(call_id)
                        }
                        Ok(Err(err)) => {
                            Observation::tool_error(&name, err).with_call_id(call_id)
                        }
                        Err(_) => Observation {
                            source: name.clone(),
                            content: format!(
                                "Tool '{}' timed out after {:?}",
                                name, timeout
                            ),
                            is_error: true,
                            call_id: Some(call_id),
                            metadata: {
                                let mut m = std::collections::HashMap::new();
                                m.insert("error_type".into(), "timeout".into());
                                m
                            },
                        },
                    }
                });
            }
        }

        // Barrier sync: wait for all tool calls to complete
        let mut observations = Vec::with_capacity(tool_calls.len());
        while let Some(obs) = futures.next().await {
            // Record success/failure in circuit breaker
            let tool_name = obs
                .metadata
                .get("tool_name")
                .cloned()
                .unwrap_or_else(|| obs.source.clone());
            if obs.is_error {
                circuit_breakers.record_failure(&tool_name).await;
            } else {
                circuit_breakers.record_success(&tool_name).await;
            }
            observations.push(obs);
        }

        observations
    }
}

/// Execute a single tool call. In production, this delegates to the
/// ToolInvocationEnforcer → MCP client pipeline. This default implementation
/// returns the arguments as the "result" for testing purposes.
async fn execute_tool_call(name: &str, arguments: &str) -> Result<String, String> {
    tracing::debug!("Executing tool '{}' with arguments: {}", name, arguments);
    // Production implementation would call through ToolInvocationEnforcer here.
    // For the reasoning loop infrastructure, we return a placeholder.
    Ok(format!(
        "Tool '{}' executed successfully with arguments: {}",
        name, arguments
    ))
}

/// An [`ActionExecutor`] for runners that have **no tool backend wired**.
///
/// It advertises no tools (so the model is not offered any) and, if a tool
/// call is nonetheless proposed, returns a clear `is_error` observation
/// instead of fabricating a success. It is the fallback for entry points like
/// `symbi run` and the DSL `reason()` builtin when no `tools/` ToolClad
/// manifests are present, so those paths never tell the model a tool
/// "executed successfully" when nothing ran.
///
/// Contrast: [`DefaultActionExecutor`] is a parallel-dispatch executor whose
/// per-call result is an echo placeholder (a test double, not for production
/// tool execution); [`crate::toolclad::executor::ToolCladExecutor`] performs
/// real, SchemaPin-verified execution when `tools/` manifests are configured.
#[derive(Default)]
pub struct UnavailableToolExecutor;

#[async_trait]
impl ActionExecutor for UnavailableToolExecutor {
    async fn execute_actions(
        &self,
        actions: &[ProposedAction],
        _config: &LoopConfig,
        _circuit_breakers: &CircuitBreakerRegistry,
    ) -> Vec<Observation> {
        actions
            .iter()
            .filter_map(|action| match action {
                ProposedAction::ToolCall { call_id, name, .. } => Some(Observation {
                    source: name.clone(),
                    content: format!(
                        "Tool '{}' was not executed: this runner has no tool backend \
                         configured (MCP-backed tool execution is not yet available). \
                         Reason about the task and respond without tool results.",
                        name
                    ),
                    is_error: true,
                    call_id: Some(call_id.clone()),
                    metadata: Default::default(),
                }),
                _ => None,
            })
            .collect()
    }

    // tool_definitions() falls back to the trait default (empty) — advertise
    // no tools, so the model isn't offered capabilities the runner can't run.
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_default_executor_no_actions() {
        let executor = DefaultActionExecutor::default();
        let config = LoopConfig::default();
        let circuit_breakers = CircuitBreakerRegistry::default();

        let obs = executor
            .execute_actions(&[], &config, &circuit_breakers)
            .await;
        assert!(obs.is_empty());
    }

    #[tokio::test]
    async fn test_default_executor_single_tool() {
        let executor = DefaultActionExecutor::default();
        let config = LoopConfig::default();
        let circuit_breakers = CircuitBreakerRegistry::default();

        let actions = vec![ProposedAction::ToolCall {
            call_id: "c1".into(),
            name: "search".into(),
            arguments: r#"{"q": "test"}"#.into(),
        }];

        let obs = executor
            .execute_actions(&actions, &config, &circuit_breakers)
            .await;
        assert_eq!(obs.len(), 1);
        assert!(!obs[0].is_error);
        assert_eq!(obs[0].source, "search");
        assert_eq!(obs[0].call_id.as_deref(), Some("c1"));
    }

    #[tokio::test]
    async fn test_unavailable_tool_executor_reports_error_not_fake_success() {
        let executor = UnavailableToolExecutor;
        let config = LoopConfig::default();
        let circuit_breakers = CircuitBreakerRegistry::default();

        // Advertises no tools.
        assert!(executor.tool_definitions().is_empty());

        let actions = vec![ProposedAction::ToolCall {
            call_id: "c1".into(),
            name: "search".into(),
            arguments: r#"{"q": "test"}"#.into(),
        }];
        let obs = executor
            .execute_actions(&actions, &config, &circuit_breakers)
            .await;

        assert_eq!(obs.len(), 1);
        // The key property: a call is surfaced as an error, never as a
        // fabricated success.
        assert!(obs[0].is_error);
        assert_eq!(obs[0].source, "search");
        assert_eq!(obs[0].call_id.as_deref(), Some("c1"));
        assert!(obs[0].content.contains("not executed"));

        // Non-tool actions produce no observations.
        let non_tool = vec![ProposedAction::Respond {
            content: "hi".into(),
        }];
        assert!(executor
            .execute_actions(&non_tool, &config, &circuit_breakers)
            .await
            .is_empty());
    }

    #[tokio::test]
    async fn test_default_executor_parallel_dispatch() {
        let executor = DefaultActionExecutor::default();
        let config = LoopConfig::default();
        let circuit_breakers = CircuitBreakerRegistry::default();

        let actions: Vec<ProposedAction> = (0..3)
            .map(|i| ProposedAction::ToolCall {
                call_id: format!("c{}", i),
                name: format!("tool_{}", i),
                arguments: "{}".into(),
            })
            .collect();

        let start = std::time::Instant::now();
        let obs = executor
            .execute_actions(&actions, &config, &circuit_breakers)
            .await;
        let elapsed = start.elapsed();

        assert_eq!(obs.len(), 3);
        // All should succeed
        assert!(obs.iter().all(|o| !o.is_error));
        // Parallel dispatch means wall-clock ≈ max(individual), not sum
        // Individual calls are near-instant in the default executor,
        // so elapsed should be well under 100ms
        assert!(
            elapsed.as_millis() < 100,
            "Parallel dispatch took {}ms, expected <100ms",
            elapsed.as_millis()
        );
    }

    #[tokio::test]
    async fn test_executor_skips_non_tool_actions() {
        let executor = DefaultActionExecutor::default();
        let config = LoopConfig::default();
        let circuit_breakers = CircuitBreakerRegistry::default();

        let actions = vec![
            ProposedAction::Respond {
                content: "done".into(),
            },
            ProposedAction::Delegate {
                target: "other".into(),
                message: "hi".into(),
            },
        ];

        let obs = executor
            .execute_actions(&actions, &config, &circuit_breakers)
            .await;
        assert!(obs.is_empty());
    }

    #[test]
    fn test_default_executor_has_empty_tool_definitions() {
        let executor = DefaultActionExecutor::default();
        assert!(executor.tool_definitions().is_empty());
    }

    #[tokio::test]
    async fn test_executor_circuit_breaker_integration() {
        let executor = DefaultActionExecutor::default();
        let config = LoopConfig::default();
        let circuit_breakers =
            CircuitBreakerRegistry::new(crate::reasoning::circuit_breaker::CircuitBreakerConfig {
                failure_threshold: 2,
                recovery_timeout: std::time::Duration::from_secs(30),
                half_open_max_calls: 1,
            });

        // Trip the circuit breaker for "failing_tool"
        circuit_breakers.record_failure("failing_tool").await;
        circuit_breakers.record_failure("failing_tool").await;

        let actions = vec![ProposedAction::ToolCall {
            call_id: "c1".into(),
            name: "failing_tool".into(),
            arguments: "{}".into(),
        }];

        let obs = executor
            .execute_actions(&actions, &config, &circuit_breakers)
            .await;
        assert_eq!(obs.len(), 1);
        assert!(obs[0].is_error);
        assert!(obs[0].content.contains("circuit is open"));
    }
}