systemprompt_api/services/server/
runner.rs

1//! Server run loop: MCP orchestrator wiring and lifecycle supervision.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use anyhow::Result;
7use std::sync::Arc;
8use systemprompt_runtime::AppContext;
9use systemprompt_traits::{Phase, StartupEvent, StartupEventExt, StartupEventSender};
10
11use super::lifecycle::{
12    initialize_scheduler, reconcile_agents, reconcile_system_services, start_event_bridge,
13};
14
15pub async fn run_server(
16    ctx: AppContext,
17    events: Option<StartupEventSender>,
18    early: super::startup::EarlyServer,
19) -> Result<()> {
20    let start_time = std::time::Instant::now();
21
22    let mcp_orchestrator = create_mcp_orchestrator(&ctx)?;
23
24    start_event_bridge(&ctx);
25    reconcile_system_services(&ctx, &mcp_orchestrator, events.as_ref()).await?;
26
27    if let Some(ref tx) = events {
28        tx.phase_started(Phase::Agents);
29    }
30    match reconcile_agents(&ctx, events.as_ref()).await {
31        Ok(started_count) => {
32            if let Some(ref tx) = events {
33                if tx
34                    .unbounded_send(StartupEvent::AgentReconciliationComplete {
35                        running: started_count,
36                        total: started_count,
37                    })
38                    .is_err()
39                {
40                    tracing::debug!("Startup event receiver dropped");
41                }
42                tx.phase_completed(Phase::Agents);
43            }
44        },
45        Err(e) => {
46            if let Some(ref tx) = events {
47                tx.phase_failed(Phase::Agents, e.to_string());
48                if tx
49                    .unbounded_send(StartupEvent::Error {
50                        message: format!("Agent reconciliation failed: {e}"),
51                        fatal: true,
52                    })
53                    .is_err()
54                {
55                    tracing::debug!("Startup event receiver dropped");
56                }
57            }
58            return Err(e);
59        },
60    }
61
62    if let Some(ref tx) = events {
63        tx.phase_started(Phase::Scheduler);
64    }
65    let scheduler_handle = match initialize_scheduler(&ctx, events.as_ref()).await {
66        Ok(handle) => {
67            if let Some(ref tx) = events {
68                tx.phase_completed(Phase::Scheduler);
69            }
70            handle
71        },
72        Err(e) => {
73            if let Some(ref tx) = events {
74                tx.phase_failed(Phase::Scheduler, e.to_string());
75            }
76            None
77        },
78    };
79
80    if let Some(ref tx) = events {
81        tx.phase_started(Phase::ApiServer);
82    }
83    let router = crate::services::server::setup_api_server(&ctx, events.as_ref())?;
84    let addr = ctx.server_address();
85
86    early.activate(router);
87    super::readiness::signal_ready();
88
89    if let Some(ref tx) = events {
90        tx.phase_completed(Phase::ApiServer);
91    }
92
93    if let Some(ref tx) = events {
94        tx.startup_complete(start_time.elapsed(), format!("http://{}", addr), vec![]);
95    }
96
97    systemprompt_logging::set_startup_mode(false);
98
99    let serve_result = early.join().await;
100
101    super::shutdown::drain(&ctx, scheduler_handle).await;
102
103    serve_result
104}
105
106fn create_mcp_orchestrator(
107    ctx: &AppContext,
108) -> Result<Arc<systemprompt_mcp::services::McpOrchestrator>> {
109    use systemprompt_mcp::services::McpOrchestrator;
110    let manager = McpOrchestrator::new(
111        Arc::clone(ctx.db_pool()),
112        Arc::clone(ctx.app_paths_arc()),
113        ctx.mcp_registry().clone(),
114    )?;
115    Ok(Arc::new(manager))
116}
systemprompt_api/services/server/runner.rs

systemprompt_api/services/server/
runner.rs
