systemprompt_api/services/server/
runner.rs

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

systemprompt_api/services/server/
runner.rs
