feat: add per-service runtime monitor core

Cargo.toml

@@ -114,10 +114,13 @@ anyhow = { workspace = true }
 assert_cmd = { workspace = true }
 camino = { workspace = true }
 camino-tempfile = { workspace = true }
+daemon = { path = "crates/daemon" }
 insta = { workspace = true }
 portable-pty = { workspace = true }
+pv-fake = { path = "crates/pv-fake" }
 serde_json = { workspace = true }
 state = { path = "crates/state" }
+tokio = { workspace = true }
 vt100 = { workspace = true }
 
 [[test]]

crates/cli/src/args.rs

@@ -46,6 +46,20 @@ pub(crate) enum Command {
     #[command(name = "daemon:run", about = "Run the internal PV daemon", hide = true)]
     DaemonRun,
 
+    #[command(
+        name = "monitor:run",
+        about = "Run an internal PV runtime monitor",
+        hide = true
+    )]
+    MonitorRun,
+
+    #[command(
+        name = "monitor:gate",
+        about = "Start a runtime for an internal PV runtime monitor",
+        hide = true
+    )]
+    MonitorGate,
+
     #[command(
         name = "shim:php",
         about = "Run the internal PV PHP shim",

crates/cli/src/commands/daemon.rs

@@ -1,6 +1,7 @@
 use std::process::ExitCode;
 use std::time::Duration;
 
+use ::daemon::MonitorHooks;
 use camino::Utf8PathBuf;
 use platform::{LaunchAgentConfig, LaunchAgentFileState};
 use state::{PvPaths, StateError};
@@ -203,6 +204,21 @@ pub(crate) fn run() -> Result<ExitCode, ExecuteError> {
     Ok(ExitCode::SUCCESS)
 }
 
+pub(crate) fn run_monitor() -> Result<ExitCode, ExecuteError> {
+    let paths = PvPaths::default_home()?;
+
+    ::daemon::run_monitor_blocking(paths, MonitorHooks::default())?;
+
+    Ok(ExitCode::SUCCESS)
+}
+
+/// Returns only when the runtime could not start: on success, the runtime replaces this process.
+pub(crate) fn run_monitor_gate() -> Result<ExitCode, ExecuteError> {
+    ::daemon::run_monitor_gate(&MonitorHooks::default())?;
+
+    Ok(ExitCode::SUCCESS)
+}
+
 /// Returns success when PV may touch the LaunchAgent file; otherwise writes
 /// the refusal to stderr and returns failure, for a file PV does not own or
 /// cannot read.

crates/cli/src/commands/logs.rs

@@ -570,6 +570,54 @@ mod tests {
         })
     }
 
+    #[test]
+    fn follow_sources_continues_after_a_rotation_truncates_the_log() -> anyhow::Result<()> {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?;
+
+        runtime.block_on(async {
+            let tempdir = tempdir()?;
+            let gateway_log = tempdir.path().join("gateway.log");
+            write_test_file(&gateway_log, "")?;
+            let sources = vec![LogSource {
+                label: "gateway".to_string(),
+                active_path: gateway_log.clone(),
+            }];
+            let writer = tokio::spawn(async move {
+                tokio::time::sleep(Duration::from_millis(250)).await;
+                append_test_file(&gateway_log, "before rotation\n")?;
+                tokio::time::sleep(Duration::from_millis(250)).await;
+                // A monitor rotates by truncating the log the runtime keeps appending to.
+                truncate_test_file(&gateway_log)?;
+                append_test_file(&gateway_log, "after\n")?;
+
+                anyhow::Ok(())
+            });
+            let mut stdout = Vec::new();
+
+            follow_sources_async(&sources, false, &mut stdout, Some(2)).await?;
+            writer.await??;
+
+            assert_eq!(String::from_utf8(stdout)?, "before rotation\nafter\n");
+
+            anyhow::Ok(())
+        })
+    }
+
+    #[expect(
+        clippy::disallowed_types,
+        reason = "logs tests truncate fixture logs as rotation does"
+    )]
+    fn truncate_test_file(path: &Utf8Path) -> anyhow::Result<()> {
+        std::fs::OpenOptions::new()
+            .write(true)
+            .open(path)?
+            .set_len(0)?;
+
+        Ok(())
+    }
+
     #[expect(clippy::disallowed_methods, reason = "logs tests create fixture files")]
     fn write_test_file(path: &Utf8Path, contents: &str) -> anyhow::Result<()> {
         std::fs::write(path, contents)?;

crates/cli/src/commands/mod.rs

@@ -65,6 +65,8 @@ where
         Command::DaemonDisable => daemon::disable(environment, streams),
         Command::DaemonRestart => daemon::restart(environment, streams),
         Command::DaemonRun => daemon::run(),
+        Command::MonitorRun => daemon::run_monitor(),
+        Command::MonitorGate => daemon::run_monitor_gate(),
         Command::ShimPhp(args) => php::shim(args, environment, runtime_lifecycle_lock),
         Command::ShimComposer(args) => composer::shim(args, environment, runtime_lifecycle_lock),
         Command::DnsStatus => dns::status(environment, streams),
@@ -145,10 +147,14 @@ fn acquire_runtime_lifecycle(
     command: &Command,
     environment: &impl Environment,
 ) -> Result<Option<state::RuntimeLifecycleLock>, ExecuteError> {
+    // A monitor and its runtime outlive any command, so holding admission for their whole life
+    // would block `daemon:disable`; their controller holds it while starting them.
     if !cfg!(target_os = "macos")
         || matches!(
             command,
             Command::DaemonRun
+                | Command::MonitorRun
+                | Command::MonitorGate
                 | Command::Uninstall(_)
                 | Command::Env(_)
                 | Command::Completions(_)
@@ -198,6 +204,7 @@ fn required_capability(command: &Command) -> Option<PlatformCapability> {
         Command::DaemonRun | Command::Link(_) | Command::Unlink(_) => {
             Some(PlatformCapability::DaemonIpc)
         }
+        Command::MonitorRun | Command::MonitorGate => Some(PlatformCapability::ProcessContainment),
         Command::PortsStatus(_) | Command::PortsInstall | Command::PortsUninstall => {
             Some(PlatformCapability::LowPortFrontend)
         }
@@ -650,6 +657,14 @@ mod tests {
         );
     }
 
+    #[test]
+    fn required_capability_maps_runtime_monitor_commands() {
+        assert_required_capability(
+            &[Command::MonitorRun, Command::MonitorGate],
+            Some(PlatformCapability::ProcessContainment),
+        );
+    }
+
     #[test]
     fn required_capability_maps_internal_update_like_update() {
         assert_required_capability(

crates/daemon/examples/pv-monitor.rs

@@ -0,0 +1,44 @@
+//! Builds a runtime monitor with test hooks along with the daemon's tests; see
+//! `pv_fake::example_binary`. It runs the entry points of `pv monitor:run` and `pv monitor:gate`
+//! with the [`daemon::MonitorHooks`] a test writes to `pv-monitor-hooks.json` in its PV home.
+
+use std::io::{self, Write as _};
+use std::process::ExitCode;
+
+use daemon::{DaemonError, MONITOR_GATE_COMMAND, MONITOR_RUN_COMMAND, MonitorHooks};
+use state::PvPaths;
+
+const HOOKS_FILE: &str = "pv-monitor-hooks.json";
+
+fn main() -> ExitCode {
+    let command = std::env::args().nth(1);
+    let result = PvPaths::default_home()
+        .map_err(DaemonError::from)
+        .and_then(|paths| {
+            let hooks = hooks(&paths);
+            match command.as_deref() {
+                Some(MONITOR_RUN_COMMAND) => daemon::run_monitor_blocking(paths, hooks),
+                Some(MONITOR_GATE_COMMAND) => daemon::run_monitor_gate(&hooks),
+                _ => Err(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    format!("usage: pv-monitor {MONITOR_RUN_COMMAND}|{MONITOR_GATE_COMMAND}"),
+                )
+                .into()),
+            }
+        });
+
+    match result {
+        Ok(()) => ExitCode::SUCCESS,
+        Err(error) => {
+            let _write_result = writeln!(io::stderr(), "pv-monitor: {error}");
+            ExitCode::FAILURE
+        }
+    }
+}
+
+fn hooks(paths: &PvPaths) -> MonitorHooks {
+    state::fs::read_to_string(&paths.home().join(HOOKS_FILE))
+        .ok()
+        .and_then(|hooks| serde_json::from_str(&hooks).ok())
+        .unwrap_or_default()
+}

crates/daemon/src/error.rs

@@ -254,6 +254,30 @@ pub enum DaemonError {
     #[error("cleanup of process {pid} is unproven: {reason}")]
     RuntimeCleanupUnproven { pid: u32, reason: String },
 
+    #[error("monitor socket path {path} is longer than the {limit}-byte Unix socket limit")]
+    MonitorSocketPathTooLong {
+        path: camino::Utf8PathBuf,
+        limit: usize,
+    },
+
+    #[error(
+        "the monitor of `{subject}` left records for instance {instance}; recover them before starting it again"
+    )]
+    MonitorUnresolved { subject: String, instance: String },
+
+    #[error("the monitor of `{subject}` failed to start: {message}")]
+    MonitorStartupFailed { subject: String, message: String },
+
+    #[error("the monitor of `{subject}` is unavailable: {reason}")]
+    MonitorUnavailable { subject: String, reason: String },
+
+    #[error("the monitor of `{subject}` rejected the request ({kind}): {message}")]
+    MonitorRejected {
+        subject: String,
+        kind: crate::monitor::MonitorErrorKind,
+        message: String,
+    },
+
     #[error("runtime record is incomplete or invalid: {path}")]
     InvalidRuntimeRecord { path: camino::Utf8PathBuf },
 

crates/daemon/src/lib.rs

@@ -8,6 +8,7 @@ mod health;
 mod ipc;
 mod jobs;
 mod managed_resources;
+mod monitor;
 mod project_env;
 mod reconciliation;
 mod runtime_stop;
@@ -40,6 +41,13 @@ pub use client::{
 };
 pub use dns::{dns_port_available, response_bytes};
 pub use error::{DaemonError, ManagedResourceProjectFailure, RuntimeReconciliationFailure};
+pub use monitor::{
+    MONITOR_GATE_COMMAND, MONITOR_RUN_COMMAND, MonitorCleanup, MonitorErrorKind, MonitorExit,
+    MonitorHooks, MonitorPause, MonitorStart, MonitorState, MonitorStop, run_monitor_blocking,
+    run_monitor_gate,
+};
+#[cfg(target_os = "macos")]
+pub use monitor::{monitor_state, recover_monitor, release_monitor, start_monitor, stop_monitor};
 pub use protocol::PROTOCOL_VERSION;
 pub use reconciliation::{
     EnqueueResult, QueuedReconciliation, ReconciliationDebouncer, ReconciliationJob,