Sync OSS release

symbibot · symbibot · commit a5eeb9ab07ed · 2026-02-24T14:50:58.000-08:00
diff --git a/crates/runtime/src/api/server.rs b/crates/runtime/src/api/server.rs
@@ -353,24 +353,21 @@ impl HttpApiServer {
                 .layer(middleware::from_fn(auth_middleware))
                 .with_state(provider.clone());
 
-            // Protected routes (workflows + metrics) with authentication
+            // Protected routes (workflows + metrics + scheduler health) with authentication.
+            // Note: /api/v1/health (basic) remains public for load-balancer probes;
+            // /api/v1/health/scheduler exposes job counts and run stats so it requires auth.
             let protected_router = Router::new()
                 .route("/api/v1/workflows/execute", post(execute_workflow))
                 .route("/api/v1/metrics", get(get_metrics))
-                .layer(middleware::from_fn(auth_middleware))
-                .with_state(provider.clone());
-
-            // Health routes — no auth so load-balancer probes work without credentials
-            let health_router = Router::new()
                 .route("/api/v1/health/scheduler", get(get_scheduler_health))
+                .layer(middleware::from_fn(auth_middleware))
                 .with_state(provider.clone());
 
             router = router
                 .merge(agent_router)
                 .merge(schedule_router)
                 .merge(channel_router)
-                .merge(protected_router)
-                .merge(health_router);
+                .merge(protected_router);
         }
 
         // Conditionally serve AGENTS.md at well-known paths (auth-gated, no provider state needed)
diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs
@@ -609,7 +609,10 @@ impl RuntimeApiProvider for AgentRuntime {
     }
 
     async fn delete_agent(&self, agent_id: AgentId) -> Result<DeleteAgentResponse, RuntimeError> {
-        // Placeholder implementation - validate input and return success
+        self.scheduler
+            .delete_agent(agent_id)
+            .await
+            .map_err(RuntimeError::Scheduler)?;
 
         Ok(DeleteAgentResponse {
             id: agent_id.to_string(),
@@ -622,8 +625,24 @@ impl RuntimeApiProvider for AgentRuntime {
         agent_id: AgentId,
         request: ExecuteAgentRequest,
     ) -> Result<ExecuteAgentResponse, RuntimeError> {
+        // Ensure the agent exists in the registry
+        if !self.scheduler.has_agent(agent_id) {
+            return Err(RuntimeError::Internal(format!(
+                "Agent {} not found",
+                agent_id
+            )));
+        }
+
+        // Re-schedule from stored config if the agent isn't currently active
         let status = self.get_agent_status(agent_id).await?;
-        if status.state != AgentState::Running {
+        if status.state == AgentState::Completed {
+            if let Some(config) = self.scheduler.get_agent_config(agent_id) {
+                self.scheduler
+                    .schedule_agent(config)
+                    .await
+                    .map_err(RuntimeError::Scheduler)?;
+            }
+        } else if status.state != AgentState::Running {
             self.lifecycle
                 .start_agent(agent_id)
                 .await
@@ -654,7 +673,9 @@ impl RuntimeApiProvider for AgentRuntime {
         &self,
         _agent_id: AgentId,
     ) -> Result<GetAgentHistoryResponse, RuntimeError> {
-        // For now, return empty history as the model logger API is not yet implemented
+        // TODO(T-59): Requires a persistent execution-log storage layer (SQLite
+        // or in-memory ring buffer). The per-schedule history
+        // (/schedules/{id}/history) already works via CronScheduler.
         let history = vec![];
         Ok(GetAgentHistoryResponse { history })
     }
diff --git a/crates/runtime/src/scheduler/mod.rs b/crates/runtime/src/scheduler/mod.rs
@@ -92,6 +92,15 @@ pub trait AgentScheduler {
         agent_id: AgentId,
         request: crate::api::types::UpdateAgentRequest,
     ) -> Result<(), SchedulerError>;
+
+    /// Check whether an agent is registered (regardless of run state)
+    fn has_agent(&self, agent_id: AgentId) -> bool;
+
+    /// Retrieve the stored config for a registered agent
+    fn get_agent_config(&self, agent_id: AgentId) -> Option<AgentConfig>;
+
+    /// Remove an agent from the registry entirely
+    async fn delete_agent(&self, agent_id: AgentId) -> Result<(), SchedulerError>;
 }
 
 /// Scheduler configuration
@@ -251,6 +260,10 @@ pub struct DefaultAgentScheduler {
     task_manager: Arc<TaskManager>,
     running_agents: Arc<DashMap<AgentId, ScheduledTask>>,
     suspended_agents: Arc<DashMap<AgentId, AgentSuspensionInfo>>,
+    /// Persistent registry of all agents that have been scheduled. Agents
+    /// remain here after being dequeued so that status/execute/list continue
+    /// to work even after completion.
+    registered_agents: Arc<DashMap<AgentId, AgentConfig>>,
     system_metrics: Arc<RwLock<SystemMetrics>>,
     shutdown_notify: Arc<Notify>,
     is_running: Arc<RwLock<bool>>,
@@ -274,6 +287,7 @@ impl DefaultAgentScheduler {
         let task_manager = Arc::new(TaskManager::new(config.task_timeout));
         let running_agents = Arc::new(DashMap::new());
         let suspended_agents = Arc::new(DashMap::new());
+        let registered_agents = Arc::new(DashMap::new());
         let system_metrics = Arc::new(RwLock::new(SystemMetrics::new()));
         let shutdown_notify = Arc::new(Notify::new());
         let is_running = Arc::new(RwLock::new(true));
@@ -305,6 +319,7 @@ impl DefaultAgentScheduler {
             task_manager,
             running_agents,
             suspended_agents,
+            registered_agents,
             system_metrics,
             shutdown_notify,
             is_running,
@@ -456,9 +471,12 @@ impl AgentScheduler for DefaultAgentScheduler {
             return Err(SchedulerError::ShuttingDown);
         }
 
-        let task = ScheduledTask::new(config);
+        let task = ScheduledTask::new(config.clone());
         let agent_id = task.agent_id;
 
+        // Persist in the registry so the agent survives dequeue
+        self.registered_agents.insert(agent_id, config);
+
         // Add to priority queue
         self.priority_queue.write().push(task);
 
@@ -503,13 +521,16 @@ impl AgentScheduler for DefaultAgentScheduler {
                     reason: format!("Failed to terminate task: {}", e),
                 })?;
 
+            self.registered_agents.remove(&agent_id);
             tracing::info!("Terminated agent {}", agent_id);
             return Ok(());
         }
 
         // Remove from queue
         let mut queue = self.priority_queue.write();
         if queue.remove(&agent_id).is_some() {
+            drop(queue);
+            self.registered_agents.remove(&agent_id);
             tracing::info!("Removed agent {} from queue", agent_id);
             return Ok(());
         }
@@ -623,6 +644,17 @@ impl AgentScheduler for DefaultAgentScheduler {
                     active_tasks: 0,
                     scheduled_at: task.scheduled_at,
                 })
+            } else if self.registered_agents.contains_key(&agent_id) {
+                // Agent was registered but already ran and was dequeued
+                Ok(AgentStatus {
+                    agent_id,
+                    state: AgentState::Completed,
+                    last_activity: SystemTime::now(),
+                    memory_usage: 0,
+                    cpu_usage: 0.0,
+                    active_tasks: 0,
+                    scheduled_at: SystemTime::now(),
+                })
             } else {
                 // Agent not found anywhere
                 Err(SchedulerError::AgentNotFound { agent_id })
@@ -776,21 +808,11 @@ impl AgentScheduler for DefaultAgentScheduler {
     }
 
     async fn list_agents(&self) -> Vec<AgentId> {
-        let mut agent_ids = Vec::new();
-
-        // Collect running agents
-        for entry in self.running_agents.iter() {
-            agent_ids.push(*entry.key());
-        }
-
-        // Collect queued agents
-        let queue = self.priority_queue.read();
-        let queued_tasks = queue.to_vec();
-        for task in queued_tasks {
-            agent_ids.push(task.agent_id);
-        }
-
-        agent_ids
+        // Return all registered agents (running, queued, and completed)
+        self.registered_agents
+            .iter()
+            .map(|entry| *entry.key())
+            .collect()
     }
 
     #[cfg(feature = "http-api")]
@@ -840,6 +862,35 @@ impl AgentScheduler for DefaultAgentScheduler {
 
         Err(SchedulerError::AgentNotFound { agent_id })
     }
+
+    fn has_agent(&self, agent_id: AgentId) -> bool {
+        self.registered_agents.contains_key(&agent_id)
+    }
+
+    fn get_agent_config(&self, agent_id: AgentId) -> Option<AgentConfig> {
+        self.registered_agents.get(&agent_id).map(|r| r.clone())
+    }
+
+    async fn delete_agent(&self, agent_id: AgentId) -> Result<(), SchedulerError> {
+        // Remove from running agents if present
+        if let Some((_, _)) = self.running_agents.remove(&agent_id) {
+            let _ = self.task_manager.terminate_task(agent_id).await;
+        }
+
+        // Remove from queue if present
+        {
+            let mut queue = self.priority_queue.write();
+            queue.remove(&agent_id);
+        }
+
+        // Remove from registry
+        if self.registered_agents.remove(&agent_id).is_some() {
+            tracing::info!("Deleted agent {} from registry", agent_id);
+            Ok(())
+        } else {
+            Err(SchedulerError::AgentNotFound { agent_id })
+        }
+    }
 }
 
 impl DefaultAgentScheduler {
diff --git a/src/commands/up.rs b/src/commands/up.rs
@@ -143,14 +143,21 @@ pub async fn run(matches: &ArgMatches) {
         }
     };
 
-    let agent_id = if let Some(_agent) = agents_found.first() {
-        // For now, just create a new agent ID
-        // TODO: Implement proper agent creation when runtime API is stable
-        AgentId::new()
+    // Load agents from DSL files into the scheduler registry
+    let first_agent_id = if let Some(ref rt) = runtime {
+        let loaded = load_agents_into_registry(rt).await;
+        if loaded.is_empty() {
+            None
+        } else {
+            println!("✓ {} agent(s) loaded from agents/", loaded.len());
+            Some(loaded[0])
+        }
     } else {
-        AgentId::new()
+        None
     };
 
+    let agent_id = first_agent_id.unwrap_or_else(AgentId::new);
+
     let http_port_num = match http_port.parse::<u16>() {
         Ok(p) => p,
         Err(e) => {
@@ -660,6 +667,55 @@ async fn load_dsl_schedules(cron: &symbi_runtime::CronScheduler) -> usize {
     count
 }
 
+/// Scan DSL files in the agents directory, parse each one, create an
+/// `AgentConfig`, and register it with the runtime scheduler so that
+/// `/api/v1/agents` lists them and `/api/v1/agents/:id/execute` works.
+async fn load_agents_into_registry(runtime: &AgentRuntime) -> Vec<AgentId> {
+    let agents_dir = Path::new("agents");
+    if !agents_dir.exists() {
+        return vec![];
+    }
+
+    let mut ids = Vec::new();
+    if let Ok(entries) = fs::read_dir(agents_dir) {
+        for entry in entries.flatten() {
+            if entry.path().extension().is_some_and(|ext| ext == "dsl") {
+                if let Ok(source) = fs::read_to_string(entry.path()) {
+                    let name = entry
+                        .path()
+                        .file_stem()
+                        .map(|s| s.to_string_lossy().to_string())
+                        .unwrap_or_default();
+
+                    let agent_config = symbi_runtime::types::AgentConfig {
+                        id: symbi_runtime::types::AgentId::new(),
+                        name: name.clone(),
+                        dsl_source: source,
+                        execution_mode: symbi_runtime::types::ExecutionMode::Ephemeral,
+                        security_tier: symbi_runtime::types::SecurityTier::Tier1,
+                        resource_limits: symbi_runtime::types::ResourceLimits::default(),
+                        capabilities: vec![symbi_runtime::types::Capability::Computation],
+                        policies: vec![],
+                        metadata: std::collections::HashMap::new(),
+                        priority: symbi_runtime::types::Priority::Normal,
+                    };
+
+                    match runtime.scheduler.schedule_agent(agent_config).await {
+                        Ok(id) => {
+                            println!("  → {} [{}]", name, id);
+                            ids.push(id);
+                        }
+                        Err(e) => {
+                            eprintln!("  ⚠ Failed to register agent '{}': {}", name, e);
+                        }
+                    }
+                }
+            }
+        }
+    }
+    ids
+}
+
 fn scan_agents_directory() -> Vec<String> {
     let agents_dir = Path::new("agents");
     let mut agents = Vec::new();
