systemprompt_api/services/server/
runner.rs

1//! Server run loop: MCP orchestrator wiring and lifecycle supervision.
2//!
3//! The startup phases a node runs follow its `server.role`
4//! ([`super::routes::role::lifecycle_plan`]): a gateway node spawns no MCP
5//! servers or agents and runs no scheduler; a runner node does neither and
6//! instead starts every extension's [`systemprompt_traits::RunnerService`].
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11use anyhow::{Context, Result};
12use std::sync::Arc;
13use systemprompt_identifiers::Actor;
14use systemprompt_runtime::AppContext;
15use systemprompt_scheduler::services::SchedulerHandle;
16use systemprompt_traits::{
17    Dependencies, JobContext, OwnedTask, Phase, StartupEvent, StartupEventExt, StartupEventSender,
18};
19
20use super::lifecycle::{
21    initialize_scheduler, reconcile_agents, reconcile_system_services, start_event_bridge,
22    start_registry_heartbeat,
23};
24
25pub async fn run_server(
26    ctx: AppContext,
27    events: Option<StartupEventSender>,
28    early: super::startup::EarlyServer,
29) -> Result<()> {
30    let start_time = std::time::Instant::now();
31
32    let instance_claim = claim_replica_identity(&ctx)
33        .await
34        .context("replica identity")?;
35    tracing::info!(instance_id = %instance_claim.instance_id(), "replica identity claimed");
36
37    let plan = super::routes::role::lifecycle_plan(ctx.config().role);
38    tracing::info!(role = %ctx.config().role, "node role");
39
40    start_event_bridge(&ctx);
41    start_registry_heartbeat(&ctx);
42    if plan.reconcile_mcp {
43        let mcp_orchestrator = create_mcp_orchestrator(&ctx)?;
44        reconcile_system_services(&ctx, &mcp_orchestrator, events.as_ref()).await?;
45    }
46    if plan.reconcile_agents {
47        run_agents_phase(&ctx, events.as_ref()).await?;
48    }
49    let scheduler_handle = if plan.scheduler {
50        run_scheduler_phase(&ctx, events.as_ref()).await?
51    } else {
52        None
53    };
54    if plan.runner_services {
55        start_runner_services(&ctx);
56    }
57
58    if let Some(ref tx) = events {
59        tx.phase_started(Phase::ApiServer);
60    }
61    let router = crate::services::server::setup_api_server(&ctx, events.as_ref())?;
62    start_accounting_recovery(&ctx).await?;
63    let addr = ctx.server_address();
64
65    early.activate(router);
66    let metrics_listener = start_metrics_listener(&ctx).await?;
67    super::readiness::signal_ready();
68
69    if let Some(ref tx) = events {
70        tx.phase_completed(Phase::ApiServer);
71    }
72
73    if let Some(ref tx) = events {
74        tx.startup_complete(start_time.elapsed(), format!("http://{}", addr), vec![]);
75    }
76
77    systemprompt_logging::set_startup_mode(false);
78
79    let restart = ctx.shutdown_request().clone();
80    let serve_result = super::shutdown::join_within_drain_grace(early.join(), &restart).await;
81
82    let forced_exit = super::shutdown::arm_forced_exit(restart);
83    if let Some(listener) = metrics_listener
84        && listener.abort_and_join().await.is_some()
85    {
86        tracing::debug!("Metrics listener had already stopped on the shutdown signal");
87    }
88    super::shutdown::drain(&ctx, scheduler_handle).await;
89    instance_claim.release().await;
90    forced_exit.abort();
91
92    serve_result
93}
94
95fn start_runner_services(ctx: &AppContext) {
96    let shared = Arc::new(ctx.clone());
97    let mut started = 0usize;
98    for extension in ctx.extension_registry().extensions() {
99        for service in extension.runner_services() {
100            let name = service.name();
101            let actor = Actor::job(ctx.system_admin().id().clone(), name);
102            let dependencies = Dependencies::new()
103                .with(Arc::clone(ctx.app_paths_arc()))
104                .with(Arc::clone(ctx.db_pool()))
105                .with(Arc::clone(&shared));
106            let job_ctx = JobContext::new(actor, dependencies);
107            ctx.background_tasks()
108                .spawn_cancellable(name, move |shutdown| async move {
109                    service.run(job_ctx, shutdown).await;
110                });
111            tracing::info!(
112                extension = extension.id(),
113                service = name,
114                "runner service started"
115            );
116            started += 1;
117        }
118    }
119    tracing::info!(started, "runner services started");
120}
121
122async fn run_agents_phase(ctx: &AppContext, events: Option<&StartupEventSender>) -> Result<()> {
123    if let Some(tx) = events {
124        tx.phase_started(Phase::Agents);
125    }
126    match reconcile_agents(ctx, events).await {
127        Ok(started_count) => {
128            if let Some(tx) = events {
129                send_startup_event(
130                    tx,
131                    StartupEvent::AgentReconciliationComplete {
132                        running: started_count,
133                        total: started_count,
134                    },
135                );
136                tx.phase_completed(Phase::Agents);
137            }
138            Ok(())
139        },
140        Err(e) => Err(fail_phase(
141            events,
142            Phase::Agents,
143            format!("Agent reconciliation failed: {e}"),
144            e,
145        )),
146    }
147}
148
149async fn run_scheduler_phase(
150    ctx: &AppContext,
151    events: Option<&StartupEventSender>,
152) -> Result<Option<SchedulerHandle>> {
153    if let Some(tx) = events {
154        tx.phase_started(Phase::Scheduler);
155    }
156    match initialize_scheduler(ctx, events).await {
157        Ok(handle) => {
158            if let Some(tx) = events {
159                tx.phase_completed(Phase::Scheduler);
160            }
161            Ok(handle)
162        },
163        Err(e) => Err(fail_phase(
164            events,
165            Phase::Scheduler,
166            format!("Scheduler initialization failed: {e}"),
167            e,
168        )),
169    }
170}
171
172fn fail_phase(
173    events: Option<&StartupEventSender>,
174    phase: Phase,
175    message: String,
176    error: anyhow::Error,
177) -> anyhow::Error {
178    if let Some(tx) = events {
179        tx.phase_failed(phase, error.to_string());
180        send_startup_event(
181            tx,
182            StartupEvent::Error {
183                message,
184                fatal: true,
185            },
186        );
187    }
188    error
189}
190
191fn send_startup_event(tx: &StartupEventSender, event: StartupEvent) {
192    if tx.unbounded_send(event).is_err() {
193        tracing::debug!("Startup event receiver dropped");
194    }
195}
196
197fn create_mcp_orchestrator(
198    ctx: &AppContext,
199) -> Result<Arc<systemprompt_mcp::services::McpOrchestrator>> {
200    use systemprompt_mcp::services::McpOrchestrator;
201    let manager = McpOrchestrator::new(
202        (**ctx.service_repository()).clone(),
203        Arc::clone(ctx.app_paths_arc()),
204        ctx.mcp_registry().clone(),
205    )?;
206    Ok(Arc::new(manager))
207}
208
209async fn start_metrics_listener(ctx: &AppContext) -> Result<Option<OwnedTask<()>>> {
210    let Some(port) = ctx.config().metrics_port else {
211        return Ok(None);
212    };
213    let handle = super::metrics::install_recorder(&ctx.config().instance_id)?;
214    super::pool_metrics::spawn_sampler(ctx);
215    let addr = std::net::SocketAddr::new(ctx.config().host.parse()?, port);
216    Ok(Some(
217        super::metrics::serve_metrics_listener(addr, handle).await?,
218    ))
219}
220
221async fn start_accounting_recovery(ctx: &AppContext) -> Result<()> {
222    let settlement = crate::routes::gateway::gateway_repositories(ctx)?.settlement();
223    let settled = systemprompt_gateway::audit::journal::recover(&settlement).await?;
224    if settled > 0 {
225        tracing::info!(settled, "Gateway accounting receipts recovered at startup");
226    }
227    systemprompt_gateway::audit::journal::spawn_recovery(settlement, ctx.background_tasks());
228    Ok(())
229}
230
231// Why: a redeployed host reuses its HOSTNAME, and the session of the process
232// it replaced holds the claim until the server's keepalive probes declare it
233// dead; waiting past that window tells a vanished holder from a live one.
234async fn claim_replica_identity(
235    ctx: &AppContext,
236) -> Result<systemprompt_database::InstanceClaim, systemprompt_database::InstanceClaimError> {
237    let window = std::time::Duration::from_secs(systemprompt_database::CLAIM_DEAD_PEER_SECS * 3);
238    let deadline = std::time::Instant::now() + window;
239    let mut backoff = std::time::Duration::from_secs(1);
240    loop {
241        match ctx.service_repository().claim_instance().await {
242            Err(systemprompt_database::InstanceClaimError::Claimed { instance_id })
243                if std::time::Instant::now() + backoff < deadline =>
244            {
245                tracing::warn!(
246                    instance_id = %instance_id,
247                    retry_in_ms = backoff.as_millis(),
248                    "replica identity held by another session; waiting for a dead holder to be reaped"
249                );
250                tokio::time::sleep(backoff).await;
251                backoff = (backoff * 2).min(std::time::Duration::from_secs(8));
252            },
253            outcome => return outcome,
254        }
255    }
256}
systemprompt_api/services/server/runner.rs

systemprompt_api/services/server/
runner.rs
