systemprompt_logging/
lib.rs

1//! # systemprompt-logging
2//!
3//! Tracing and audit infrastructure for systemprompt.io. Owns the
4//! structured-event pipeline, the database-backed `tracing` layer,
5//! log/analytics repositories, and the CLI display
6//! sinks. The trace-assembly queries over agent, AI and MCP tables live in
7//! `systemprompt_runtime::trace`.
8//!
9//! ## Feature flags
10//!
11//! | Feature   | Description                                                               |
12//! |-----------|---------------------------------------------------------------------------|
13//! | (default) | Database layer, repositories                                              |
14//! | `cli`     | CLI display helpers (`CliService`, tables, banners) — pulls in `console`, `indicatif` |
15//!
16//! ## Top-level entry points
17//!
18//! - [`init_logging`] / [`init_console_logging`] /
19//!   [`init_console_logging_with_level`] — install the global `tracing`
20//!   subscriber (with optional database sink) and return its [`LogSink`].
21//! - [`LoggingRuntime`] — the sink paired with the platform owner; built once
22//!   at bootstrap and carried by `AppContext`.
23//! - [`OutputState::set_console_format`] — switch stderr between text and JSON
24//!   lines (`observability.log_format`); both redact the same fields.
25//! - [`LogSink::shutdown`] — flush the database sink's buffered and queued
26//!   entries and stop its writer task; awaited on server shutdown.
27//! - [`LoggingExtension`] — schema/extension registration via the `inventory`
28//!   framework.
29//! - [`LoggingRepository`], [`AnalyticsRepository`] — direct repository access.
30//!
31//! ## Process state
32//!
33//! The `tracing` subscriber is the one process singleton: `tracing` installs
34//! its global dispatcher once, so the crate mirrors it with a set-once guard,
35//! and the CLI output-mode flags ride with it. Everything else — the database
36//! sink, the instance label, the drop count, the platform owner — is owned by
37//! the [`LogSink`] / [`LoggingRuntime`] values. A sink is recovered from the
38//! installed dispatcher rather than from a crate-level cell.
39//!
40//! Copyright (c) systemprompt.io — Business Source License 1.1.
41//! See <https://systemprompt.io> for licensing details.
42
43pub mod extension;
44pub mod layer;
45pub mod models;
46pub mod repository;
47pub mod runtime;
48pub mod sanitize;
49pub mod services;
50
51pub use extension::LoggingExtension;
52
53pub use layer::{DatabaseLayer, LogWriterHandle, LogWriterShutdownError};
54pub use models::{LogActor, LogEntry, LogFilter, LogLevel};
55pub use repository::{
56    AnalyticsEvent, AnalyticsRepository, LoggingOwnerReassignment, LoggingRepository,
57};
58pub use runtime::{LogSink, LoggingRuntime};
59#[cfg(feature = "cli")]
60pub use services::CliService;
61pub use services::{
62    BufferedNotice, ConsoleEventFormat, DatabaseLogService, FilterSystemFields, JsonEventFormat,
63    LogThrottle, LoggingMaintenanceService, OutputState, SYSTEM_LOG_OWNER, SystemSpan,
64};
65
66use std::sync::OnceLock;
67
68use layer::{InstanceLabel, ProxyDatabaseLayer};
69use systemprompt_database::DbPool;
70use systemprompt_identifiers::InstanceId;
71use tracing::Level;
72use tracing_subscriber::filter::FilterFn;
73use tracing_subscriber::layer::SubscriberExt;
74use tracing_subscriber::util::SubscriberInitExt;
75use tracing_subscriber::{EnvFilter, Layer};
76
77// Why: tracing's global default dispatcher is process-wide and set-once; this
78// flag mirrors it so a second install returns the existing proxy.
79static SUBSCRIBER_INITIALIZED: OnceLock<()> = OnceLock::new();
80
81const NOISE_FILTERS: &[&str] = &[
82    "tokio_cron_scheduler=warn",
83    "sqlx::postgres::notice=warn",
84    "sqlx::query=warn",
85    "handlebars=warn",
86    "systemprompt_database::lifecycle=info",
87    "systemprompt_templates=info",
88    "systemprompt_extension::registry=info",
89    "systemprompt_api::services::middleware::session=info",
90    "rmcp=warn",
91    "rmcp::transport=warn",
92];
93
94fn build_filter(base: &str) -> EnvFilter {
95    let filter_str = std::iter::once(base.to_owned())
96        .chain(NOISE_FILTERS.iter().map(ToString::to_string))
97        .collect::<Vec<_>>()
98        .join(",");
99    EnvFilter::new(filter_str)
100}
101
102fn installed_proxy() -> Option<ProxyDatabaseLayer> {
103    tracing::dispatcher::get_default(|dispatch| {
104        dispatch.downcast_ref::<ProxyDatabaseLayer>().cloned()
105    })
106}
107
108fn ensure_subscriber(level_override: Option<&str>, output: OutputState) -> LogSink {
109    if SUBSCRIBER_INITIALIZED.set(()).is_err() {
110        return LogSink::from_proxy(installed_proxy().unwrap_or_default());
111    }
112
113    let proxy = ProxyDatabaseLayer::new();
114    let instance: InstanceLabel = proxy.shared().instance.clone();
115
116    let base_filter = level_override.map_or_else(
117        || EnvFilter::try_from_default_env().unwrap_or_else(|_| build_filter("info")),
118        |level| EnvFilter::try_from_default_env().unwrap_or_else(|_| build_filter(level)),
119    );
120
121    let gate_active = level_override.is_none();
122    let startup_output = output.clone();
123    let armed_output = output.clone();
124    let armed_gate = FilterFn::new(move |_| armed_output.is_console_armed());
125    let startup_gate = FilterFn::new(move |meta| {
126        !(gate_active && startup_output.is_startup_mode()) || *meta.level() <= Level::WARN
127    });
128
129    let fmt_layer = tracing_subscriber::fmt::layer()
130        .fmt_fields(FilterSystemFields::labelled(instance.clone()))
131        .event_format(ConsoleEventFormat::labelled(instance, output.clone()))
132        .with_writer(std::io::stderr)
133        .log_internal_errors(true)
134        .with_filter(base_filter)
135        .with_filter(startup_gate)
136        .with_filter(armed_gate);
137
138    let db_layer = proxy.clone().with_filter(build_filter("info"));
139
140    tracing_subscriber::registry()
141        .with(output)
142        .with(fmt_layer)
143        .with(db_layer)
144        .init();
145
146    LogSink::from_proxy(proxy)
147}
148
149pub fn init_logging(db_pool: &DbPool, instance_id: Option<InstanceId>) -> LogSink {
150    let output = OutputState::current();
151    let sink = ensure_subscriber(None, output.clone());
152    output.set_console_armed(true);
153    sink.attach(db_pool, instance_id);
154    sink
155}
156
157pub fn init_console_logging() -> LogSink {
158    init_console_logging_with_level(None, &OutputState::current())
159}
160
161pub fn init_console_logging_with_level(level: Option<&str>, output: &OutputState) -> LogSink {
162    let sink = ensure_subscriber(level, output.clone());
163    output.set_console_armed(true);
164    sink
165}
166
167pub fn init_deferred_console_logging(output: OutputState) -> LogSink {
168    output.set_console_armed(false);
169    ensure_subscriber(None, output)
170}
systemprompt_logging/lib.rs

systemprompt_logging/
lib.rs
