feat: Add Slack approval workflow, conversation memory, and multi-tenant routing

Major Features:
- Human-in-the-loop approval workflow with Slack reactions (✅/❌)
- Conversation memory system for contextual follow-up messages
- AgentFlow multi-tenant routing for message-to-agent mapping
- Bot self-approval prevention with auto-detected bot_user_id
- Improved kubectl syntax guidance in agent instructions

Technical Changes:
- New flow registry and router for AgentFlow-based message routing
- Reaction event handling in trigger handler
- Pending approvals storage with DashMap
- Platform-specific approval UI (Slack reactions)
- Conversation memory with channel/thread isolation

Documentation:
- New approval workflow guide
- New conversation memory guide
- Updated documentation index
- Example multi-tenant flow configurations

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: gouravjshah

CLAUDE.md

@@ -502,4 +502,5 @@ NEVER create files unless they're absolutely necessary for achieving your goal.
 ALWAYS prefer editing an existing file to creating a new one.
 NEVER proactively create documentation files (*.md) or README files. Only create documentation files if explicitly requested by the User.
 Never save working files, text/mds and tests to the root folder.
-- Strictly follow kubectl style implementation for aofctl. For example use "aofctl run agent" instead of "aofctl agent run". If you find anything non compliant, correct it.
+- Strictly follow kubectl style implementation for aofctl. For example use "aofctl run agent" instead of "aofctl agent run". If you find anything non compliant, correct it.
+- for every feature added, add/update  docs/ so that we are keeping track of every single feautre the product has and how it works.

crates/aof-core/src/agentflow.rs

@@ -92,11 +92,44 @@ pub struct AgentFlowSpec {
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub triggers: Vec<FlowTrigger>,
 
+    /// Execution context (environment, kubeconfig, etc.)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub context: Option<FlowContext>,
+
     /// Global flow configuration
     #[serde(skip_serializing_if = "Option::is_none")]
     pub config: Option<FlowConfig>,
 }
 
+/// Flow execution context - environment and runtime configuration
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FlowContext {
+    /// Kubeconfig file path (for kubectl tools)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub kubeconfig: Option<String>,
+
+    /// Kubernetes namespace (default context)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
+
+    /// Kubernetes cluster name
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cluster: Option<String>,
+
+    /// Environment variables to set for agent execution
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub env: HashMap<String, String>,
+
+    /// Working directory for tool execution
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub working_dir: Option<String>,
+
+    /// Additional context variables available in templates
+    #[serde(flatten, skip_serializing_if = "HashMap::is_empty")]
+    pub extra: HashMap<String, serde_json::Value>,
+}
+
 /// Flow trigger configuration
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -137,6 +170,18 @@ pub struct TriggerConfig {
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub events: Vec<String>,
 
+    /// Channels to listen on (Slack/Discord channel names or IDs)
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub channels: Vec<String>,
+
+    /// Users to respond to (user IDs or patterns)
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub users: Vec<String>,
+
+    /// Message patterns to match (regex)
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub patterns: Vec<String>,
+
     /// Bot token (or env var reference ${VAR_NAME})
     #[serde(skip_serializing_if = "Option::is_none")]
     pub bot_token: Option<String>,
@@ -227,6 +272,11 @@ pub struct NodeConfig {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub agent: Option<String>,
 
+    /// Inline agent configuration (YAML string)
+    /// Use this to embed agent config directly in the flow
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub agent_config: Option<String>,
+
     /// Input to the agent (can contain ${variable} references)
     #[serde(skip_serializing_if = "Option::is_none")]
     pub input: Option<String>,

crates/aof-core/src/lib.rs

@@ -46,7 +46,7 @@ pub use fleet::{
 };
 pub use agentflow::{
     AgentFlow, AgentFlowMetadata, AgentFlowSpec, AgentFlowState, FlowConfig,
-    FlowConnection, FlowError, FlowExecutionStatus, FlowNode, FlowRetryConfig,
+    FlowConnection, FlowContext, FlowError, FlowExecutionStatus, FlowNode, FlowRetryConfig,
     FlowTrigger, NodeCondition, NodeConfig, NodeExecutionStatus, NodeResult, NodeType,
     TriggerConfig, TriggerType,
 };

crates/aof-runtime/src/executor/agentflow_executor.rs

@@ -8,16 +8,17 @@
 //! - Platform-specific actions (Slack, Discord, etc.)
 
 use std::collections::HashMap;
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use chrono::Utc;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, RwLock};
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
 use aof_core::{
-    AgentFlow, AgentFlowState, AofError, AofResult, FlowError, FlowExecutionStatus, FlowNode,
-    NodeExecutionStatus, NodeResult, NodeType,
+    AgentConfig, AgentFlow, AgentFlowState, AofError, AofResult, FlowError, FlowExecutionStatus,
+    FlowNode, NodeExecutionStatus, NodeResult, NodeType,
 };
 
 use super::Runtime;
@@ -63,23 +64,36 @@ pub enum AgentFlowEvent {
 /// AgentFlow executor
 pub struct AgentFlowExecutor {
     flow: AgentFlow,
-    #[allow(dead_code)]
-    runtime: Arc<Runtime>,
+    runtime: Arc<RwLock<Runtime>>,
     event_tx: Option<mpsc::Sender<AgentFlowEvent>>,
+    /// Directory to search for agent YAML files
+    agents_dir: Option<PathBuf>,
 }
 
 impl AgentFlowExecutor {
     /// Create a new AgentFlow executor
-    pub fn new(flow: AgentFlow, runtime: Arc<Runtime>) -> Self {
+    pub fn new(flow: AgentFlow, runtime: Arc<RwLock<Runtime>>) -> Self {
         Self {
             flow,
             runtime,
             event_tx: None,
+            agents_dir: None,
         }
     }
 
+    /// Create with a non-locked runtime (convenience constructor)
+    pub fn with_runtime(flow: AgentFlow, runtime: Runtime) -> Self {
+        Self::new(flow, Arc::new(RwLock::new(runtime)))
+    }
+
+    /// Set the agents directory for loading agent configs
+    pub fn with_agents_dir(mut self, dir: impl Into<PathBuf>) -> Self {
+        self.agents_dir = Some(dir.into());
+        self
+    }
+
     /// Load AgentFlow from file
-    pub async fn from_file(path: &str, runtime: Arc<Runtime>) -> AofResult<Self> {
+    pub async fn from_file(path: &str, runtime: Arc<RwLock<Runtime>>) -> AofResult<Self> {
         let content = std::fs::read_to_string(path).map_err(|e| {
             AofError::Config(format!("Failed to read AgentFlow config {}: {}", path, e))
         })?;
@@ -388,31 +402,146 @@ impl AgentFlowExecutor {
         Ok(agent_result)
     }
 
-    /// Run an agent (placeholder - needs integration with AgentExecutor)
+    /// Run an agent using the runtime
+    ///
+    /// This method:
+    /// 1. Checks if the agent is already loaded in the runtime
+    /// 2. If not, tries to load it from the agents directory
+    /// 3. Applies flow context (kubeconfig, env vars) to the execution
+    /// 4. Executes the agent and returns the result
     async fn run_agent(
         &self,
         agent_name: &str,
         input: &str,
-        _state: &AgentFlowState,
+        state: &AgentFlowState,
     ) -> AofResult<serde_json::Value> {
-        // This is a simplified implementation
-        // In production, this would:
-        // 1. Load agent config from registry or file
-        // 2. Create AgentExecutor
-        // 3. Run the agent with input
-        // 4. Return the response
-
-        // For now, try to use the runtime to check if agent exists
-        info!("Agent {} called with input: {}", agent_name, input);
-
-        // Return a placeholder response
-        // The real implementation would integrate with aof_runtime::AgentExecutor
-        Ok(serde_json::json!({
-            "agent": agent_name,
-            "input": input,
-            "output": format!("Agent {} processed: {}", agent_name, input),
-            "requires_approval": false
-        }))
+        info!("Executing agent '{}' with input: {}", agent_name, input);
+
+        // First, try to execute with the agent already loaded
+        {
+            let runtime = self.runtime.read().await;
+            if runtime.has_agent(agent_name) {
+                // Apply flow context if specified
+                self.apply_flow_context().await?;
+
+                let result = runtime.execute(agent_name, input).await?;
+                return Ok(serde_json::json!({
+                    "agent": agent_name,
+                    "input": input,
+                    "output": result,
+                    "requires_approval": false
+                }));
+            }
+        }
+
+        // Agent not loaded - try to load from agents directory
+        if let Some(ref agents_dir) = self.agents_dir {
+            // Try common naming patterns
+            let possible_paths = vec![
+                agents_dir.join(format!("{}.yaml", agent_name)),
+                agents_dir.join(format!("{}.yml", agent_name)),
+                agents_dir.join(format!("{}-agent.yaml", agent_name)),
+                agents_dir.join(format!("{}-agent.yml", agent_name)),
+            ];
+
+            for path in possible_paths {
+                if path.exists() {
+                    info!("Loading agent '{}' from {}", agent_name, path.display());
+
+                    let mut runtime = self.runtime.write().await;
+                    runtime.load_agent_from_file(path.to_string_lossy().as_ref()).await?;
+
+                    // Apply flow context
+                    drop(runtime); // Release write lock
+                    self.apply_flow_context().await?;
+
+                    let runtime = self.runtime.read().await;
+                    let result = runtime.execute(agent_name, input).await?;
+
+                    return Ok(serde_json::json!({
+                        "agent": agent_name,
+                        "input": input,
+                        "output": result,
+                        "requires_approval": false
+                    }));
+                }
+            }
+        }
+
+        // Agent config might also be embedded in the flow
+        // Check if there's a node config with agent config
+        if let Some(node) = self.flow.spec.nodes.iter().find(|n| {
+            n.config.agent.as_ref() == Some(&agent_name.to_string())
+        }) {
+            if let Some(ref config_yaml) = node.config.agent_config {
+                info!("Loading agent '{}' from inline config", agent_name);
+
+                let agent_config: AgentConfig = serde_yaml::from_str(config_yaml).map_err(|e| {
+                    AofError::Config(format!("Failed to parse inline agent config: {}", e))
+                })?;
+
+                let mut runtime = self.runtime.write().await;
+                runtime.load_agent_from_config(agent_config).await?;
+
+                drop(runtime);
+                self.apply_flow_context().await?;
+
+                let runtime = self.runtime.read().await;
+                let result = runtime.execute(agent_name, input).await?;
+
+                return Ok(serde_json::json!({
+                    "agent": agent_name,
+                    "input": input,
+                    "output": result,
+                    "requires_approval": false
+                }));
+            }
+        }
+
+        // Could not find or load the agent
+        Err(AofError::Config(format!(
+            "Agent '{}' not found. Ensure it's loaded or available in the agents directory.",
+            agent_name
+        )))
+    }
+
+    /// Apply flow context to the environment
+    async fn apply_flow_context(&self) -> AofResult<()> {
+        if let Some(ref context) = self.flow.spec.context {
+            // Set KUBECONFIG if specified
+            if let Some(ref kubeconfig) = context.kubeconfig {
+                std::env::set_var("KUBECONFIG", kubeconfig);
+                info!("Set KUBECONFIG to {}", kubeconfig);
+            }
+
+            // Set namespace as env var if specified
+            if let Some(ref namespace) = context.namespace {
+                std::env::set_var("K8S_NAMESPACE", namespace);
+                info!("Set K8S_NAMESPACE to {}", namespace);
+            }
+
+            // Set cluster name if specified
+            if let Some(ref cluster) = context.cluster {
+                std::env::set_var("K8S_CLUSTER", cluster);
+                info!("Set K8S_CLUSTER to {}", cluster);
+            }
+
+            // Set working directory if specified
+            if let Some(ref working_dir) = context.working_dir {
+                std::env::set_current_dir(working_dir).map_err(|e| {
+                    AofError::Config(format!("Failed to change working directory: {}", e))
+                })?;
+                info!("Changed working directory to {}", working_dir);
+            }
+
+            // Set additional environment variables
+            for (key, value) in &context.env {
+                std::env::set_var(key, value);
+                debug!("Set env var {}={}", key, value);
+            }
+        }
+
+        Ok(())
     }
 
     /// Execute a Conditional node
@@ -874,7 +1003,7 @@ mod tests {
 
     #[test]
     fn test_evaluate_condition() {
-        let runtime = Arc::new(Runtime::new());
+        let runtime = Arc::new(RwLock::new(Runtime::new()));
         let flow: AgentFlow = serde_yaml::from_str(
             r#"
 apiVersion: aof.dev/v1

crates/aof-runtime/src/executor/runtime.rs

@@ -328,6 +328,11 @@ impl Runtime {
         self.agents.keys().cloned().collect()
     }
 
+    /// Check if an agent is loaded
+    pub fn has_agent(&self, name: &str) -> bool {
+        self.agents.contains_key(name)
+    }
+
     /// Get agent executor by name
     pub fn get_agent(&self, name: &str) -> Option<Arc<AgentExecutor>> {
         self.agents.get(name).cloned()

crates/aof-triggers/Cargo.toml

@@ -13,6 +13,7 @@ aof-core.workspace = true
 aof-runtime.workspace = true
 aof-llm.workspace = true
 aof-memory.workspace = true
+aof-tools = { workspace = true, features = ["all"] }
 
 # Async runtime
 tokio.workspace = true
@@ -22,6 +23,7 @@ futures.workspace = true
 # Serialization
 serde.workspace = true
 serde_json.workspace = true
+serde_yaml.workspace = true
 
 # Error handling
 thiserror.workspace = true

crates/aof-triggers/src/flow/mod.rs

@@ -0,0 +1,11 @@
+//! Flow module - FlowRegistry and FlowRouter for AgentFlow management
+//!
+//! This module provides:
+//! - `FlowRegistry` - Loads and manages AgentFlow configurations
+//! - `FlowRouter` - Routes incoming trigger events to matching flows
+
+pub mod registry;
+pub mod router;
+
+pub use registry::FlowRegistry;
+pub use router::{FlowMatch, FlowRouter, MatchReason};