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 = early.join().await;
51
52    super::shutdown::drain(&ctx, scheduler_handle).await;
53
54    serve_result
55}
56
57async fn run_agents_phase(ctx: &AppContext, events: Option<&StartupEventSender>) -> Result<()> {
58    if let Some(tx) = events {
59        tx.phase_started(Phase::Agents);
60    }
61    match reconcile_agents(ctx, events).await {
62        Ok(started_count) => {
63            if let Some(tx) = events {
64                send_startup_event(
65                    tx,
66                    StartupEvent::AgentReconciliationComplete {
67                        running: started_count,
68                        total: started_count,
69                    },
70                );
71                tx.phase_completed(Phase::Agents);
72            }
73            Ok(())
74        },
75        Err(e) => Err(fail_phase(
76            events,
77            Phase::Agents,
78            format!("Agent reconciliation failed: {e}"),
79            e,
80        )),
81    }
82}
83
84/// A scheduler that cannot start is fatal: `run_bootstrap_jobs` is reached only
85/// through this phase, so continuing would serve a process whose boot-time jobs
86/// silently never ran. Disabling the scheduler is done via
87/// `scheduler.enabled: false`, which succeeds here with no handle.
88async fn run_scheduler_phase(
89    ctx: &AppContext,
90    events: Option<&StartupEventSender>,
91) -> Result<Option<SchedulerHandle>> {
92    if let Some(tx) = events {
93        tx.phase_started(Phase::Scheduler);
94    }
95    match initialize_scheduler(ctx, events).await {
96        Ok(handle) => {
97            if let Some(tx) = events {
98                tx.phase_completed(Phase::Scheduler);
99            }
100            Ok(handle)
101        },
102        Err(e) => Err(fail_phase(
103            events,
104            Phase::Scheduler,
105            format!("Scheduler initialization failed: {e}"),
106            e,
107        )),
108    }
109}
110
111fn fail_phase(
112    events: Option<&StartupEventSender>,
113    phase: Phase,
114    message: String,
115    error: anyhow::Error,
116) -> anyhow::Error {
117    if let Some(tx) = events {
118        tx.phase_failed(phase, error.to_string());
119        send_startup_event(
120            tx,
121            StartupEvent::Error {
122                message,
123                fatal: true,
124            },
125        );
126    }
127    error
128}
129
130fn send_startup_event(tx: &StartupEventSender, event: StartupEvent) {
131    if tx.unbounded_send(event).is_err() {
132        tracing::debug!("Startup event receiver dropped");
133    }
134}
135
136fn create_mcp_orchestrator(
137    ctx: &AppContext,
138) -> Result<Arc<systemprompt_mcp::services::McpOrchestrator>> {
139    use systemprompt_mcp::services::McpOrchestrator;
140    let manager = McpOrchestrator::new(
141        Arc::clone(ctx.db_pool()),
142        Arc::clone(ctx.app_paths_arc()),
143        ctx.mcp_registry().clone(),
144    )?;
145    Ok(Arc::new(manager))
146}
systemprompt_api/services/server/runner.rs

systemprompt_api/services/server/
runner.rs
