systemprompt-api 0.72.1

Axum-based HTTP server and API gateway for systemprompt.io AI governance infrastructure. Exposes governed agents, MCP, A2A, and admin endpoints with rate limiting and RBAC.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

//! Scheduler start/stop wiring in the server lifecycle.
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use anyhow::{Result, bail};
use std::sync::Arc;
use systemprompt_loader::ConfigLoader;
use systemprompt_runtime::AppContext;
use systemprompt_scheduler::SchedulerConfig;
use systemprompt_scheduler::services::{
    ConfigReconcileService, RECONCILE_LOCK_WAIT, ReconcileTrigger, SchedulerHandle,
    SchedulerService,
};
use systemprompt_traits::{StartupEventExt, StartupEventSender};

use crate::services::server::scheduler_health;

pub(in crate::services::server) async fn initialize_scheduler(
    ctx: &AppContext,
    events: Option<&StartupEventSender>,
) -> Result<Option<SchedulerHandle>> {
    events.scheduler_initializing();
    reconcile_config_planes(ctx).await?;

    let config = ConfigLoader::load().map_or_else(
        |e| {
            tracing::warn!(error = %e, "Failed to load scheduler config, using defaults");
            SchedulerConfig::with_system_admin()
        },
        |c| {
            c.scheduler
                .unwrap_or_else(SchedulerConfig::with_system_admin)
        },
    );

    let scheduler = SchedulerService::new(config, Arc::clone(ctx.db_pool()), Arc::new(ctx.clone()));

    let available_jobs = scheduler.run_bootstrap_jobs(events).await?;
    let startup = scheduler.start().await?;

    if !startup.degraded.is_empty() {
        tracing::error!(
            count = startup.degraded.len(),
            "scheduler started degraded: jobs skipped due to unresolved owners"
        );
    }
    scheduler_health::record(startup.degraded);

    events.scheduler_ready(startup.scheduled, available_jobs);
    Ok(startup.handle)
}

async fn reconcile_config_planes(ctx: &AppContext) -> Result<()> {
    let service = ConfigReconcileService::from_app_context(
        &Arc::new(ctx.clone()),
        (*crate::repository::config_plane_state(ctx.db_pool())).clone(),
    )?;
    let report = service
        .reconcile_exclusive(ReconcileTrigger::Boot, RECONCILE_LOCK_WAIT)
        .await?;
    let failures: Vec<String> = report
        .failures()
        .iter()
        .map(|p| format!("{}: {}", p.plane, p.error.as_deref().unwrap_or("failed")))
        .collect();
    if !failures.is_empty() {
        bail!(
            "config planes failed to reconcile at boot: {}",
            failures.join("; ")
        );
    }
    tracing::info!(
        planes = report.planes.len(),
        changes = report.changes(),
        "config planes reconciled at boot"
    );
    Ok(())
}