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_scheduler::services::SchedulerHandle;
10use systemprompt_traits::{Phase, StartupEvent, StartupEventExt, StartupEventSender};
11
12use super::lifecycle::{
13    initialize_scheduler, reconcile_agents, reconcile_system_services, start_event_bridge,
14};
15
16pub async fn run_server(
17    ctx: AppContext,
18    events: Option<StartupEventSender>,
19    early: super::startup::EarlyServer,
20) -> Result<()> {
21    let start_time = std::time::Instant::now();
22
23    let mcp_orchestrator = create_mcp_orchestrator(&ctx)?;
24
25    start_event_bridge(&ctx);
26    reconcile_system_services(&ctx, &mcp_orchestrator, events.as_ref()).await?;
27
28    run_agents_phase(&ctx, events.as_ref()).await?;
29    let scheduler_handle = run_scheduler_phase(&ctx, events.as_ref()).await?;
30
31    if let Some(ref tx) = events {
32        tx.phase_started(Phase::ApiServer);
33    }
34    let router = crate::services::server::setup_api_server(&ctx, events.as_ref())?;
35    let addr = ctx.server_address();
36
37    early.activate(router);
38    super::readiness::signal_ready();
39
40    if let Some(ref tx) = events {
41        tx.phase_completed(Phase::ApiServer);
42    }
43
44    if let Some(ref tx) = events {
45        tx.startup_complete(start_time.elapsed(), format!("http://{}", addr), vec![]);
46    }
47
48    systemprompt_logging::set_startup_mode(false);
49
50    let serve_result = super::shutdown::join_within_drain_grace(early.join()).await;
51
52    super::shutdown::arm_forced_exit();
53    super::shutdown::drain(&ctx, scheduler_handle).await;
54
55    serve_result
56}
57
58async fn run_agents_phase(ctx: &AppContext, events: Option<&StartupEventSender>) -> Result<()> {
59    if let Some(tx) = events {
60        tx.phase_started(Phase::Agents);
61    }
62    match reconcile_agents(ctx, events).await {
63        Ok(started_count) => {
64            if let Some(tx) = events {
65                send_startup_event(
66                    tx,
67                    StartupEvent::AgentReconciliationComplete {
68                        running: started_count,
69                        total: started_count,
70                    },
71                );
72                tx.phase_completed(Phase::Agents);
73            }
74            Ok(())
75        },
76        Err(e) => Err(fail_phase(
77            events,
78            Phase::Agents,
79            format!("Agent reconciliation failed: {e}"),
80            e,
81        )),
82    }
83}
84
85/// A scheduler that cannot start is fatal: `run_bootstrap_jobs` is reached only
86/// through this phase, so continuing would serve a process whose boot-time jobs
87/// silently never ran. Disabling the scheduler is done via
88/// `scheduler.enabled: false`, which succeeds here with no handle.
89async fn run_scheduler_phase(
90    ctx: &AppContext,
91    events: Option<&StartupEventSender>,
92) -> Result<Option<SchedulerHandle>> {
93    if let Some(tx) = events {
94        tx.phase_started(Phase::Scheduler);
95    }
96    match initialize_scheduler(ctx, events).await {
97        Ok(handle) => {
98            if let Some(tx) = events {
99                tx.phase_completed(Phase::Scheduler);
100            }
101            Ok(handle)
102        },
103        Err(e) => Err(fail_phase(
104            events,
105            Phase::Scheduler,
106            format!("Scheduler initialization failed: {e}"),
107            e,
108        )),
109    }
110}
111
112fn fail_phase(
113    events: Option<&StartupEventSender>,
114    phase: Phase,
115    message: String,
116    error: anyhow::Error,
117) -> anyhow::Error {
118    if let Some(tx) = events {
119        tx.phase_failed(phase, error.to_string());
120        send_startup_event(
121            tx,
122            StartupEvent::Error {
123                message,
124                fatal: true,
125            },
126        );
127    }
128    error
129}
130
131fn send_startup_event(tx: &StartupEventSender, event: StartupEvent) {
132    if tx.unbounded_send(event).is_err() {
133        tracing::debug!("Startup event receiver dropped");
134    }
135}
136
137fn create_mcp_orchestrator(
138    ctx: &AppContext,
139) -> Result<Arc<systemprompt_mcp::services::McpOrchestrator>> {
140    use systemprompt_mcp::services::McpOrchestrator;
141    let manager = McpOrchestrator::new(
142        Arc::clone(ctx.db_pool()),
143        Arc::clone(ctx.app_paths_arc()),
144        ctx.mcp_registry().clone(),
145    )?;
146    Ok(Arc::new(manager))
147}
systemprompt_api/services/server/runner.rs

systemprompt_api/services/server/
runner.rs
