systemprompt_api/services/server/lifecycle/
agents.rs

1//! Agent reconciliation during server startup.
2//!
3//! [`reconcile_agents`] starts every enabled agent into a known-clean state,
4//! retrying failed agents once after cleanup, and fails server startup if any
5//! required agent cannot be brought up — agents are a hard dependency of the
6//! API.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11use anyhow::Result;
12use futures_util::future::join_all;
13use std::sync::Arc;
14use std::time::Duration;
15use systemprompt_agent::AgentState;
16use systemprompt_agent::services::a2a_server::streaming::webhook_client::HttpWebhookBroadcaster;
17use systemprompt_agent::services::agent_orchestration::AgentOrchestrator;
18use systemprompt_agent::services::registry::AgentRegistry;
19use systemprompt_loader::subprocess::{self, ChildKind, StopOutcome};
20use systemprompt_manifest::AgentConfig;
21use systemprompt_oauth::JwtValidationProviderImpl;
22use systemprompt_runtime::AppContext;
23use systemprompt_traits::{StartupEventExt, StartupEventSender};
24
25const AGENT_STOP_GRACE: Duration = Duration::from_secs(5);
26
27pub async fn reconcile_agents(
28    ctx: &AppContext,
29    events: Option<&StartupEventSender>,
30) -> Result<usize> {
31    let orchestrator = build_orchestrator(ctx, events).await?;
32
33    let agent_registry = match AgentRegistry::new() {
34        Ok(registry) => registry,
35        Err(e) => {
36            events.error(format!("Failed to load agent registry: {e}"), true);
37            return Err(e.into());
38        },
39    };
40
41    let enabled_agents = match agent_registry.list_enabled_agents().await {
42        Ok(agents) => agents,
43        Err(e) => {
44            events.error(format!("Failed to list enabled agents: {e}"), true);
45            return Err(e.into());
46        },
47    };
48
49    let required_count = enabled_agents.len();
50    let (started, failed_agents) =
51        start_enabled_agents(&orchestrator, &enabled_agents, events).await;
52
53    let started = if failed_agents.is_empty() {
54        started
55    } else {
56        handle_failed_agents(
57            started,
58            &failed_agents,
59            &agent_registry,
60            &orchestrator,
61            events,
62        )
63        .await?
64    };
65
66    if started < required_count {
67        return Err(anyhow::anyhow!(
68            "FATAL: Only {}/{} required agents started successfully\n\nAll enabled agents must be \
69             running for API to start.",
70            started,
71            required_count
72        ));
73    }
74
75    Ok(started)
76}
77
78async fn build_orchestrator(
79    ctx: &AppContext,
80    events: Option<&StartupEventSender>,
81) -> Result<AgentOrchestrator> {
82    let jwt_provider = Arc::new(JwtValidationProviderImpl::from_config()?);
83    let agent_state = Arc::new(AgentState::new(
84        Arc::clone(ctx.db_pool()),
85        Arc::new(ctx.config().clone()),
86        jwt_provider,
87        Arc::clone(ctx.a2a_repositories()),
88        Arc::new(HttpWebhookBroadcaster::from_config(ctx.config())?),
89    ));
90
91    match AgentOrchestrator::new(agent_state, Arc::clone(ctx.app_paths_arc()), events).await {
92        Ok(orch) => Ok(orch),
93        Err(e) => {
94            events.error(
95                format!("Failed to initialize agent orchestrator: {e}"),
96                true,
97            );
98            Err(e.into())
99        },
100    }
101}
102
103async fn start_enabled_agents(
104    orchestrator: &AgentOrchestrator,
105    enabled_agents: &[AgentConfig],
106    events: Option<&StartupEventSender>,
107) -> (usize, Vec<(String, anyhow::Error)>) {
108    let start_futures: Vec<_> = enabled_agents
109        .iter()
110        .map(|agent_config| {
111            let name = agent_config.name.clone();
112            let port = agent_config.port;
113            async move {
114                enforce_clean_agent_state(orchestrator, &name, port, events)
115                    .await
116                    .map(|_| name.clone())
117                    .map_err(|e| (name.clone(), e))
118            }
119        })
120        .collect();
121
122    let results = join_all(start_futures).await;
123
124    let (succeeded, failed): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_ok);
125    let failed_agents: Vec<(String, anyhow::Error)> =
126        failed.into_iter().filter_map(Result::err).collect();
127
128    (succeeded.len(), failed_agents)
129}
130
131async fn handle_failed_agents(
132    mut started: usize,
133    failed_agents: &[(String, anyhow::Error)],
134    agent_registry: &AgentRegistry,
135    orchestrator: &AgentOrchestrator,
136    events: Option<&StartupEventSender>,
137) -> Result<usize> {
138    events.warning_with_context(
139        format!(
140            "{} agent(s) failed to start on first attempt",
141            failed_agents.len()
142        ),
143        "Attempting cleanup and retry",
144    );
145
146    let mut retry_failed: Vec<(String, String)> = Vec::new();
147
148    for (agent_name, first_error) in failed_agents {
149        tracing::warn!(agent = %agent_name, error = ?first_error, "Agent failed on first start attempt");
150        let agent_config = match agent_registry.get_agent(agent_name).await {
151            Ok(config) => config,
152            Err(e) => {
153                events.agent_failed(agent_name.clone(), format!("Agent config not found: {e}"));
154                retry_failed.push((agent_name.clone(), format!("Agent config not found: {e}")));
155                continue;
156            },
157        };
158
159        match enforce_clean_agent_state(orchestrator, agent_name, agent_config.port, events).await {
160            Ok(_) => {
161                started += 1;
162            },
163            Err(e) => {
164                retry_failed.push((agent_name.clone(), e.to_string()));
165            },
166        }
167    }
168
169    if !retry_failed.is_empty() {
170        let agent_names: Vec<String> = retry_failed.iter().map(|(name, _)| name.clone()).collect();
171        return Err(anyhow::anyhow!(
172            "FATAL: {} required agent(s) failed to start after retry: {}\n\nsystemprompt.io OS \
173             cannot operate without all enabled agents.\nAgents are the core service \
174             layer.\n\nFailures:\n{}\n\nPossible causes:\n  - Agent binaries not built (run: \
175             cargo build)\n  - Ports occupied by non-agent processes (check with: lsof -i:PORT)\n  \
176             - Missing environment variables (check .env file)\n  - File permission \
177             issues\n\nBuild agents with: cargo build",
178            retry_failed.len(),
179            agent_names.join(", "),
180            retry_failed
181                .iter()
182                .map(|(name, err)| format!("  - {name}: {err}"))
183                .collect::<Vec<_>>()
184                .join("\n")
185        ));
186    }
187
188    Ok(started)
189}
190
191async fn enforce_clean_agent_state(
192    orchestrator: &AgentOrchestrator,
193    agent: &str,
194    desired_port: u16,
195    events: Option<&StartupEventSender>,
196) -> Result<bool> {
197    use systemprompt_agent::services::agent_orchestration::{AgentStatus, PortService};
198
199    let agent_name = systemprompt_identifiers::AgentName::new(agent);
200    if let Ok(status) = orchestrator.get_status(&agent_name).await {
201        match status {
202            AgentStatus::Running { pid, port } => {
203                let reason = if port == desired_port {
204                    format!("Restarting agent to ensure fresh state (pid {pid})")
205                } else {
206                    format!(
207                        "On wrong port {port} (expected {desired_port}), killing and restarting"
208                    )
209                };
210                events.agent_cleanup(agent.to_owned(), reason);
211                stop_recorded_agent(pid, &agent_name, events).await?;
212                if let Err(e) = orchestrator.delete_agent(&agent_name).await {
213                    tracing::warn!(error = %e, agent = %agent, "Failed to delete agent during cleanup");
214                }
215            },
216            AgentStatus::Failed { .. } => {
217                events.agent_cleanup(agent.to_owned(), "Previously failed, restarting");
218            },
219        }
220    }
221
222    let port_manager = PortService::new();
223    if let Err(e) = port_manager
224        .cleanup_port_if_needed(desired_port, &agent_name)
225        .await
226    {
227        events.error(
228            format!("Failed to cleanup port {desired_port} for agent {agent}: {e}"),
229            false,
230        );
231        return Err(e.into());
232    }
233
234    match orchestrator.start_agent(&agent_name, events).await {
235        Ok(_) => Ok(true),
236        Err(e) => Err(e.into()),
237    }
238}
239
240async fn stop_recorded_agent(
241    pid: u32,
242    agent_name: &systemprompt_identifiers::AgentName,
243    events: Option<&StartupEventSender>,
244) -> Result<()> {
245    let service = systemprompt_identifiers::ServiceName::of_agent(agent_name);
246    match subprocess::stop_owned(pid, ChildKind::Agent, &service, AGENT_STOP_GRACE).await {
247        Ok(StopOutcome::NotRunning | StopOutcome::Stopped(_)) => Ok(()),
248        Ok(StopOutcome::NotOurs) => {
249            tracing::warn!(
250                pid,
251                agent = %agent_name,
252                "Recorded PID is alive but is not our child (recycled/stale); skipping signal"
253            );
254            Ok(())
255        },
256        Err(e) => {
257            if let Some(tx) = events {
258                tx.error(
259                    format!("Failed to stop agent {agent_name} (pid {pid}): {e}"),
260                    false,
261                );
262            }
263            Err(e.into())
264        },
265    }
266}
systemprompt_api/services/server/lifecycle/agents.rs

systemprompt_api/services/server/lifecycle/
agents.rs
