systemprompt_cli/commands/infrastructure/services/
serve.rs

1//! `infra services serve` command: port reclamation and early bind.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use crate::cli_settings::CliConfig;
7use crate::interactive::{Prompter, confirm_optional};
8use anyhow::{Context, Result};
9use std::sync::Arc;
10use systemprompt_logging::CliService;
11use systemprompt_runtime::{AppContext, ServiceCategory, validate_system};
12use systemprompt_scheduler::ProcessCleanup;
13use systemprompt_traits::{ModuleInfo, Phase, StartupEvent, StartupEventExt, StartupEventSender};
14
15use super::{get_api_addr, get_api_port};
16
17#[derive(Debug, Clone, Copy)]
18pub struct ServeOptions {
19    pub foreground: bool,
20    pub kill_port_process: bool,
21    pub run_migrations: bool,
22}
23
24pub async fn execute_with_events(
25    prompter: &dyn Prompter,
26    options: ServeOptions,
27    config: &CliConfig,
28    events: Option<&StartupEventSender>,
29) -> Result<String> {
30    let ServeOptions {
31        foreground,
32        kill_port_process,
33        run_migrations,
34    } = options;
35    let port = get_api_port();
36
37    if events.is_none() {
38        CliService::startup_banner(Some("Starting services..."));
39    }
40
41    ensure_port_free(prompter, port, kill_port_process, config, events).await?;
42
43    register_modules(events);
44
45    let early = bind_early(foreground, events).await?;
46
47    let ctx = Arc::new(
48        AppContext::builder()
49            .with_startup_warnings(true)
50            .with_migrations(run_migrations)
51            .build()
52            .await
53            .context("Failed to initialize application context")?,
54    );
55
56    if events.is_none() {
57        CliService::phase_success("Database schemas installed", None);
58    }
59
60    if let Some(tx) = events {
61        tx.phase_started(Phase::Database);
62        if let Err(e) = tx.unbounded_send(StartupEvent::DatabaseValidated) {
63            tracing::debug!(error = %e, "startup event channel closed: DatabaseValidated");
64        }
65        tx.phase_completed(Phase::Database);
66    } else {
67        CliService::phase("Validation");
68        CliService::phase_info("Running system validation...", None);
69    }
70
71    validate_system(&ctx)
72        .await
73        .context("System validation failed")?;
74
75    if events.is_none() {
76        CliService::phase_success("System validation complete", None);
77    }
78
79    if events.is_none() {
80        CliService::phase("Server");
81        if !foreground {
82            CliService::phase_warning("Daemon mode not supported", Some("running in foreground"));
83        }
84    } else if let Some(tx) = events {
85        tx.phase_started(Phase::ApiServer);
86        if !foreground {
87            tx.warning("Daemon mode not supported, running in foreground");
88        }
89    }
90
91    if let Some(early) = early {
92        systemprompt_api::services::server::run_server(
93            Arc::unwrap_or_clone(ctx),
94            events.cloned(),
95            early,
96        )
97        .await?;
98    }
99
100    Ok(format!("http://127.0.0.1:{}", port))
101}
102
103pub async fn execute(
104    prompter: &dyn Prompter,
105    foreground: bool,
106    kill_port_process: bool,
107    config: &CliConfig,
108) -> Result<()> {
109    execute_with_events(
110        prompter,
111        ServeOptions {
112            foreground,
113            kill_port_process,
114            run_migrations: true,
115        },
116        config,
117        None,
118    )
119    .await
120    .map(|_| ())
121}
122
123async fn ensure_port_free(
124    prompter: &dyn Prompter,
125    port: u16,
126    kill_port_process: bool,
127    config: &CliConfig,
128    events: Option<&StartupEventSender>,
129) -> Result<()> {
130    if let Some(pid) = check_port_available(port) {
131        if let Some(tx) = events
132            && let Err(e) = tx.unbounded_send(StartupEvent::PortConflict { port, pid })
133        {
134            tracing::debug!(error = %e, "startup event channel closed: PortConflict");
135        }
136        handle_port_conflict(prompter, port, pid, kill_port_process, config, events).await?;
137        if let Some(tx) = events
138            && let Err(e) = tx.unbounded_send(StartupEvent::PortConflictResolved { port })
139        {
140            tracing::debug!(error = %e, "startup event channel closed: PortConflictResolved");
141        }
142    } else if let Some(tx) = events {
143        tx.port_available(port);
144    } else {
145        CliService::phase_success(&format!("Port {} available", port), None);
146    }
147    Ok(())
148}
149
150async fn bind_early(
151    foreground: bool,
152    events: Option<&StartupEventSender>,
153) -> Result<Option<systemprompt_api::services::server::EarlyServer>> {
154    if !foreground {
155        return Ok(None);
156    }
157    let addr = get_api_addr().context("Profile not initialized; cannot determine bind address")?;
158    let early = systemprompt_api::services::server::bind_and_serve(&addr, events.cloned())
159        .await
160        .context("Failed to bind API listener")?;
161    Ok(Some(early))
162}
163
164fn check_port_available(port: u16) -> Option<u32> {
165    ProcessCleanup::check_port(port)
166}
167
168fn kill_process(pid: u32) {
169    ProcessCleanup::kill_process(pid);
170}
171
172#[expect(
173    clippy::too_many_arguments,
174    reason = "port-conflict handling threads discrete CLI flags plus the prompt seam"
175)]
176async fn handle_port_conflict(
177    prompter: &dyn Prompter,
178    port: u16,
179    pid: u32,
180    kill_port_process: bool,
181    config: &CliConfig,
182    events: Option<&StartupEventSender>,
183) -> Result<()> {
184    if events.is_none() {
185        CliService::warning(&format!("Port {} is already in use by PID {}", port, pid));
186    }
187
188    let should_kill = kill_port_process
189        || confirm_optional(
190            prompter,
191            &format!("Kill process {} and restart?", pid),
192            false,
193            config,
194        )?;
195
196    if should_kill {
197        if events.is_none() {
198            CliService::info(&format!("Killing process {}...", pid));
199        }
200        kill_process(pid);
201        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
202
203        if check_port_available(port).is_some() {
204            return Err(anyhow::anyhow!(
205                "Failed to free port {} after killing PID {}",
206                port,
207                pid
208            ));
209        }
210        if events.is_none() {
211            CliService::success(&format!("Port {} is now available", port));
212        }
213        return Ok(());
214    }
215
216    if config.is_interactive() {
217        return Err(anyhow::anyhow!(
218            "Port {} is occupied by PID {}. Aborted by user.",
219            port,
220            pid
221        ));
222    }
223
224    if events.is_none() {
225        CliService::error(&format!("Port {} is already in use by PID {}", port, pid));
226        CliService::info("Use --kill-port-process to terminate the process, or:");
227        CliService::info("   - just api-rebuild    (rebuild and restart)");
228        CliService::info("   - just api-nuke       (nuclear option - kill everything)");
229        CliService::info(&format!(
230            "   - kill {}             (manually kill the process)",
231            pid
232        ));
233    }
234    Err(anyhow::anyhow!(
235        "Port {} is occupied by PID {}. Use --kill-port-process to terminate.",
236        port,
237        pid
238    ))
239}
240
241fn register_modules(events: Option<&StartupEventSender>) {
242    let api_registrations: Vec<_> =
243        inventory::iter::<systemprompt_runtime::ModuleApiRegistration>().collect();
244
245    if let Some(tx) = events {
246        let modules: Vec<_> = api_registrations
247            .iter()
248            .map(|r| ModuleInfo {
249                name: r.module_name.to_owned(),
250                category: format!("{:?}", r.category),
251            })
252            .collect();
253        tx.modules_loaded(modules.len(), modules);
254    } else {
255        CliService::phase_info(
256            &format!("Loading {} route modules", api_registrations.len()),
257            None,
258        );
259
260        for registration in &api_registrations {
261            let category_name = match registration.category {
262                ServiceCategory::Core => "Core",
263                ServiceCategory::Agent => "Agent",
264                ServiceCategory::Mcp => "Mcp",
265                ServiceCategory::Meta => "Meta",
266            };
267            CliService::phase_success(
268                registration.module_name,
269                Some(&format!("{} routes", category_name)),
270            );
271        }
272    }
273}
systemprompt_cli/commands/infrastructure/services/serve.rs

systemprompt_cli/commands/infrastructure/services/
serve.rs
