From c89db8e72188a5ca84166cbea91b6c7ba5bff637 Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Mon, 9 Nov 2026 01:17:00 +0800 Subject: [PATCH 1/2] feat(server): add explicit native daemon lifecycle and durable schedule recovery --- Cargo.lock | 3 + crates/cli/Cargo.toml | 3 + crates/cli/src/daemon_command.rs | 139 +++++++ crates/cli/src/main.rs | 11 + crates/cli/tests/daemon_process.rs | 382 ++++++++++++++++++ crates/server/AGENTS.md | 2 +- crates/server/src/bootstrap.rs | 142 +++++-- crates/server/src/daemon.rs | 84 ++++ crates/server/src/lib.rs | 3 + crates/server/src/persistence.rs | 10 +- crates/server/src/runtime.rs | 2 + .../src/runtime/daemon_lifecycle_tests.rs | 108 +++++ .../server/src/runtime/handlers/schedule.rs | 24 ++ .../src/runtime/handlers/schedule_tests.rs | 40 ++ crates/server/src/runtime/lifecycle.rs | 3 + crates/server/src/schedule_store.rs | 20 +- crates/server/src/schedule_store/outbox.rs | 46 +++ crates/server/src/singleton.rs | 225 ++++++++--- crates/server/src/singleton_files.rs | 150 +++++++ crates/server/src/transport.rs | 166 ++++++-- docs/native-background-daemon.md | 109 +++++ 21 files changed, 1550 insertions(+), 122 deletions(-) create mode 100644 crates/cli/src/daemon_command.rs create mode 100644 crates/cli/tests/daemon_process.rs create mode 100644 crates/server/src/daemon.rs create mode 100644 crates/server/src/runtime/daemon_lifecycle_tests.rs create mode 100644 crates/server/src/singleton_files.rs create mode 100644 docs/native-background-daemon.md diff --git a/Cargo.lock b/Cargo.lock index 7a4c0167..f48034a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1399,6 +1399,7 @@ version = "0.2.16" dependencies = [ "anyhow", "async-trait", + "chrono", "clap", "colored", "devo-arg0", @@ -1414,11 +1415,13 @@ dependencies = [ "devo-util-paths", "devo-windows-sandbox", "futures", + "libc", "pretty_assertions", "serde", "serde_json", "tempfile", "tokio", + "tokio-tungstenite", "toml", "tracing", "tracing-subscriber", diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index 04c99539..1426bb58 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -24,6 +24,7 @@ devo-windows-sandbox = { workspace = true } devo-arg0 = { workspace = true } anyhow = { workspace = true } +libc = { workspace = true } async-trait = { workspace = true } clap = { workspace = true } colored = { workspace = true } @@ -39,5 +40,7 @@ tracing-subscriber = { workspace = true } tokio-console = ["devo-arg0/tokio-console"] [dev-dependencies] +tokio-tungstenite = { workspace = true } +chrono = { workspace = true } tempfile = { workspace = true } pretty_assertions = "1" diff --git a/crates/cli/src/daemon_command.rs b/crates/cli/src/daemon_command.rs new file mode 100644 index 00000000..1a0bd915 --- /dev/null +++ b/crates/cli/src/daemon_command.rs @@ -0,0 +1,139 @@ +//! Opt-in process orchestration; never installs a service or changes permissions. +use std::process::{Command, Stdio}; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use clap::Subcommand; +use devo_server::{ProtocolSet, ServerProcessArgs, ServerProcessRunOptions, ServerTransportMode}; + +#[derive(Debug, Subcommand)] +pub(crate) enum DaemonCommand { + /// Start a detached, user-owned Native runtime (Unix). + Start, + /// Run the independent runtime in the foreground for an external supervisor. + Run, + /// Show authenticated runtime status. + Status, + /// Explicitly stop the independent runtime and wait for ownership release. + Stop, + /// Attach an NDJSON Native client; fail if no daemon is running. + Attach, +} + +pub(crate) async fn run_daemon(command: &DaemonCommand) -> Result<()> { + let home = devo_server::validate_daemon_home(&devo_util_paths::find_devo_home()?)?; + match command { + DaemonCommand::Run | DaemonCommand::Attach => { + devo_server::run_server_process( + ServerProcessArgs { + transport: if matches!(command, DaemonCommand::Run) { + ServerTransportMode::Daemon + } else { + ServerTransportMode::Attach + }, + protocols: ProtocolSet::default(), + status: false, + shutdown: false, + }, + ServerProcessRunOptions::default(), + ) + .await + } + DaemonCommand::Status => { + match devo_server::daemon_status(&home).await? { + Some(status) => println!( + "devo daemon running\npid: {}\nstarted_at: {}", + status.pid, status.started_at + ), + None => println!("devo daemon is not running"), + } + Ok(()) + } + DaemonCommand::Stop => { + devo_server::stop_daemon(&home).await?; + println!("devo daemon stopped"); + Ok(()) + } + DaemonCommand::Start => { + // A concurrent launcher may own the lock before its metadata or + // authenticated listener is ready. Never spawn a replacement based + // on that uncertainty; wait for a confirmed owner or absence. + let existing = tokio::time::timeout(Duration::from_secs(45), async { + loop { + match devo_server::daemon_status(&home).await { + Ok(status) => return status, + Err(_) => tokio::time::sleep(Duration::from_millis(100)).await, + } + } + }).await.context("existing owner could not be confirmed as a daemon; run devo daemon status or devo daemon run to inspect the error")?; + if let Some(status) = existing { + println!("devo daemon already running (pid {})", status.pid); + return Ok(()); + } + let mut command = + Command::new(std::env::current_exe().context("locate devo executable")?); + command + .args(["daemon", "run"]) + .env("DEVO_HOME", &home) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + // No stdio inherited from the caller and no controlling terminal. + // Only the child changes its session; the user's shell is untouched. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + } + let mut child = command.spawn().context("launch detached daemon")?; + let deadline = tokio::time::Instant::now() + Duration::from_secs(45); + loop { + if let Ok(Some(status)) = devo_server::daemon_status(&home).await { + println!( + "devo daemon ready (pid {}). Closing clients will not stop it. Use devo daemon stop to stop it.", + status.pid + ); + return Ok(()); + } + if let Some(status) = child.try_wait()? { + bail!( + "daemon child exited before readiness: {status}; run devo daemon run to inspect startup errors" + ); + } + if tokio::time::Instant::now() >= deadline { + // Readiness timeout does not prove failure. Do not kill an + // unverified owner or spawn a replacement behind the user's back. + bail!( + "daemon readiness unconfirmed after 45 seconds; it may still be starting. Check devo daemon status before retrying" + ); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + #[derive(Parser)] + struct Arguments { + #[command(subcommand)] + command: DaemonCommand, + } + #[test] + fn all_explicit_lifecycle_commands_parse() { + for command in ["start", "run", "status", "stop", "attach"] { + assert!(Arguments::try_parse_from(["daemon", command]).is_ok()); + } + assert!(Arguments::try_parse_from(["daemon", "install"]).is_err()); + } +} diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index af062875..d0137f3d 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -30,6 +30,7 @@ use tracing_subscriber::filter::LevelFilter; mod agent_command; mod app_exit; +mod daemon_command; mod doctor_command; mod mcp_command; mod prompt_command; @@ -186,6 +187,10 @@ async fn run_cli() -> Result<()> { run_mcp(command) } Some(Command::Upgrade) => run_upgrade(), + Some(Command::Daemon { command }) => { + let _logging = install_server_logging(&cli)?; + daemon_command::run_daemon(command).await + } Some(Command::Resume { session_id }) => { maybe_print_startup_update(&cli).await; let _logging = install_logging(&cli)?; @@ -302,12 +307,18 @@ fn server_process_args_from_cli(cli: &Cli) -> Option { | Some(Command::Doctor) | Some(Command::Mcp { .. }) | Some(Command::Upgrade) + | Some(Command::Daemon { .. }) | None => None, } } #[derive(Debug, Subcommand)] enum Command { + /// Manage an explicit independent Native runtime (does not install a service). + Daemon { + #[command(subcommand)] + command: daemon_command::DaemonCommand, + }, /// Resume a saved interactive session by id. Resume { /// Session identifier printed by Devo at exit time. diff --git a/crates/cli/tests/daemon_process.rs b/crates/cli/tests/daemon_process.rs new file mode 100644 index 00000000..4e36c410 --- /dev/null +++ b/crates/cli/tests/daemon_process.rs @@ -0,0 +1,382 @@ +//! Real process ownership/attach/recovery contracts; no provider credentials. +#![cfg(unix)] +use std::path::Path; +use std::process::Stdio; +use std::time::Duration; + +use anyhow::{Context, Result}; +use futures::{SinkExt, StreamExt}; +use pretty_assertions::assert_eq; +use serde_json::{Value, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; + +fn command(home: &Path) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_devo")); + command + .env("DEVO_HOME", home) + .env("TOKIO_WORKER_THREADS", "2") + .env_remove("OPENAI_API_KEY") + .env_remove("ANTHROPIC_API_KEY") + .kill_on_drop(true); + command +} + +async fn run(home: &Path, args: &[&str]) -> Result { + tokio::time::timeout(Duration::from_secs(60), command(home).args(args).output()) + .await + .context("command deadline")? + .context("run command") +} + +async fn ready(home: &Path) -> Result { + tokio::time::timeout(Duration::from_secs(45), async { + loop { + if let Ok(Some(status)) = devo_server::daemon_status(home).await { + return status; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .context("daemon readiness") +} + +struct Attached { + child: Child, + input: ChildStdin, + output: Lines>, + next_id: u64, +} + +impl Attached { + async fn open(home: &Path) -> Result { + let mut child = command(home) + .args(["daemon", "attach"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn()?; + let input = child.stdin.take().context("stdin")?; + let output = BufReader::new(child.stdout.take().context("stdout")?).lines(); + let mut attached = Self { + child, + input, + output, + next_id: 0, + }; + attached + .request( + "initialize", + json!({ + "protocolVersion": 1, "clientCapabilities": {"terminal": false}, + "_meta": { "devo": { "protocol": "native" } }, + }), + ) + .await?; + Ok(attached) + } + + async fn request(&mut self, method: &str, params: Value) -> Result { + self.next_id += 1; + let id = self.next_id; + self.input + .write_all( + format!( + "{}\n", + json!({"jsonrpc":"2.0", "id":id, "method":method, "params":params}) + ) + .as_bytes(), + ) + .await?; + self.input.flush().await?; + tokio::time::timeout(Duration::from_secs(10), async { + while let Some(line) = self.output.next_line().await? { + let value: Value = serde_json::from_str(&line)?; + if value["id"] == id { + anyhow::ensure!(value.get("error").is_none(), "RPC failed: {value}"); + return Ok(value["result"].clone()); + } + } + anyhow::bail!("attach closed before response") + }) + .await + .context("RPC deadline")? + } + + async fn disconnect(mut self) -> Result<()> { + drop(self.input); + tokio::time::timeout(Duration::from_secs(5), self.child.wait()).await??; + Ok(()) + } +} + +fn home() -> Result { + let home = tempfile::tempdir()?; + std::fs::write( + home.path().join("config.toml"), + "[catalog]\nrefresh_on_startup = false\n[server]\nlisten = [\"invalid-public-listener\"]\n", + )?; + Ok(home) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn daemon_survives_attach_disconnect_and_refuses_legacy_shutdown() -> Result<()> { + let home = home()?; + let absent = run(home.path(), &["daemon", "attach"]).await?; + assert!(!absent.status.success()); + assert!(!home.path().join("server.lock.json").exists()); + let mut owner = command(home.path()) + .args(["daemon", "run"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + let status = ready(home.path()).await?; + let mut client = Attached::open(home.path()).await?; + client.request("runtime/ping", json!({})).await?; + client.disconnect().await?; + assert_eq!(ready(home.path()).await?, status); + let legacy = run(home.path(), &["server", "--shutdown"]).await?; + assert!(!legacy.status.success()); + assert_eq!(ready(home.path()).await?, status); + // Old Desktop bundles issue this wire method without the new CLI guard. + let metadata: Value = + serde_json::from_slice(&std::fs::read(home.path().join("server.lock.json"))?)?; + let (mut socket, _) = + tokio_tungstenite::connect_async(metadata["endpoint"].as_str().context("endpoint")?) + .await?; + use tokio_tungstenite::tungstenite::Message; + socket + .send(Message::Text( + metadata["token"] + .as_str() + .context("token")? + .to_string() + .into(), + )) + .await?; + socket + .send(Message::Text( + json!({"jsonrpc":"2.0","id":99,"method":"server/shutdown"}) + .to_string() + .into(), + )) + .await?; + let response = tokio::time::timeout(Duration::from_secs(5), socket.next()) + .await? + .context("shutdown response")??; + let response: Value = serde_json::from_str(response.to_text()?)?; + assert!(response.get("error").is_some()); + assert_eq!(ready(home.path()).await?, status); + let mut attached = Vec::new(); + for _ in 0..16 { + attached.push(Attached::open(home.path()).await?); + } + // Full long-lived data-client capacity still leaves status/stop slots. + assert_eq!(ready(home.path()).await?, status); + assert!(Attached::open(home.path()).await.is_err()); + devo_server::stop_daemon(home.path()).await?; + assert!( + attached[0] + .request("runtime/ping", json!({})) + .await + .is_err() + ); + assert!( + tokio::time::timeout(Duration::from_secs(10), owner.wait()) + .await?? + .success() + ); + assert_eq!(devo_server::daemon_status(home.path()).await?, None); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn crash_restart_rebuilds_index_before_claiming_persisted_schedule() -> Result<()> { + let home = home()?; + let mut owner = command(home.path()) + .args(["daemon", "run"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + ready(home.path()).await?; + let mut client = Attached::open(home.path()).await?; + let session = client + .request( + "session/new", + json!({"cwd":home.path(), "idempotencyKey":"daemon-fixture"}), + ) + .await?; + let session_id = session["session"]["id"].clone(); + let job = client + .request( + "session/schedule/upsert", + json!({ + "kind":"cron", "sessionId":session_id, "intervalMs":3_600_000, + "prompt":"isolated recovery fixture", "deliveryMode":"followUp", + }), + ) + .await?; + let job_id = job["job"]["jobId"].clone(); + client.disconnect().await?; + // Force process loss, leaving a stale metadata file and durable journal. + owner.kill().await?; + let schedules = home.path().join("schedules.json"); + let mut persisted: Value = serde_json::from_slice(&std::fs::read(&schedules)?)?; + persisted["jobs"][0]["nextRunAt"] = json!("2026-01-01T00:00:00Z"); + std::fs::write(&schedules, serde_json::to_vec(&persisted)?)?; + for name in ["devo.db", "devo.db-wal", "devo.db-shm"] { + let _ = std::fs::remove_file(home.path().join(name)); + } + let mut owner = command(home.path()) + .args(["daemon", "run"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + ready(home.path()).await?; + let mut client = Attached::open(home.path()).await?; + tokio::time::timeout(Duration::from_secs(15), async { + loop { + let jobs = client + .request("session/schedule/list", json!({"sessionId":session_id})) + .await?; + assert_eq!(jobs["jobs"].as_array().context("jobs")?.len(), 1); + assert_eq!(jobs["jobs"][0]["jobId"], job_id); + assert_eq!(jobs["jobs"][0]["status"], "active"); + if jobs["jobs"][0]["runCount"].as_u64().context("run count")? >= 1 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + Ok::<(), anyhow::Error>(()) + }) + .await??; + client.disconnect().await?; + let scheduled_at = chrono::DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?; + let occurrence_id = format!( + "schedule:{}:{}", + job_id.as_str().context("job id")?, + scheduled_at.timestamp_micros() + ); + let rollout = home.path().join("sessions").join(format!( + "{}.jsonl", + session_id.as_str().context("session id")? + )); + tokio::time::timeout(Duration::from_secs(15), async { + loop { + if materialized_count(&rollout, &occurrence_id) == 1 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .context("schedule must materialize without an attached client")?; + devo_server::stop_daemon(home.path()).await?; + assert!( + tokio::time::timeout(Duration::from_secs(10), owner.wait()) + .await?? + .success() + ); + let mut restarted = command(home.path()) + .args(["daemon", "run"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + ready(home.path()).await?; + // Observe reconciliation, rather than stopping before the first tick. + tokio::time::timeout(Duration::from_secs(15), async { + loop { + let outbox: Value = + serde_json::from_slice(&std::fs::read(home.path().join("schedule-outbox.json"))?)?; + let pending = outbox["occurrences"] + .as_array() + .context("occurrences")? + .iter() + .any(|item| item["occurrence_id"] == occurrence_id); + if !pending { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + Ok::<(), anyhow::Error>(()) + }) + .await + .context("restarted dispatcher must acknowledge the materialized occurrence")??; + devo_server::stop_daemon(home.path()).await?; + assert!( + tokio::time::timeout(Duration::from_secs(10), restarted.wait()) + .await?? + .success() + ); + assert_eq!(materialized_count(&rollout, &occurrence_id), 1); + Ok(()) +} + +fn materialized_count(rollout: &Path, occurrence_id: &str) -> usize { + devo_core::read_canonical_history(rollout).map(|history| history.items.iter().filter(|envelope| { + matches!(&envelope.item, devo_protocol::native::item::Item::UserMessage { client_user_message_id: Some(id), .. } if id == occurrence_id) + }).count()).unwrap_or(0) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn detached_start_is_idempotent_and_remains_live_after_launchers_exit() -> Result<()> { + let home = home()?; + let result = async { + let (first, second) = tokio::join!( + run(home.path(), &["daemon", "start"]), + run(home.path(), &["daemon", "start"]), + ); + let first = first?; + let second = second?; + anyhow::ensure!( + first.status.success(), + "first start: {}", + String::from_utf8_lossy(&first.stderr) + ); + anyhow::ensure!( + second.status.success(), + "second start: {}", + String::from_utf8_lossy(&second.stderr) + ); + let status = ready(home.path()).await?; + let mut client = Attached::open(home.path()).await?; + client.request("runtime/ping", json!({})).await?; + client.disconnect().await?; + assert_eq!(ready(home.path()).await?, status); + Ok::<(), anyhow::Error>(()) + } + .await; + let cleanup = devo_server::stop_daemon(home.path()).await; + result?; + cleanup?; + assert_eq!(devo_server::daemon_status(home.path()).await?, None); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn supervisor_sigterm_releases_daemon_ownership() -> Result<()> { + let home = home()?; + let mut owner = command(home.path()) + .args(["daemon", "run"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + ready(home.path()).await?; + // Signal only this test's still-owned child, never a PID from metadata. + let pid = i32::try_from(owner.id().context("child id")?)?; + assert_eq!(unsafe { libc::kill(pid, libc::SIGTERM) }, 0); + assert!( + tokio::time::timeout(Duration::from_secs(10), owner.wait()) + .await?? + .success() + ); + assert_eq!(devo_server::daemon_status(home.path()).await?, None); + assert!(!home.path().join("server.lock.json").exists()); + Ok(()) +} diff --git a/crates/server/AGENTS.md b/crates/server/AGENTS.md index 8b764776..90ece2cc 100644 --- a/crates/server/AGENTS.md +++ b/crates/server/AGENTS.md @@ -52,7 +52,7 @@ The server runtime uses **one session actor per session**. Durable session state - **In-memory session actors** are loaded on demand via `get_or_load_parent_session`; root sessions are LRU-evicted (capacity 16) when unpinned. - **`session/list`** returns durable user-visible sessions only (non-ephemeral, no `agent_path`; includes forks with `parent_session_id`); subagent rows are indexed but hidden from list. - **`session/resume`** loads sessions lazily from rollout files (roots and subagents). Subagent actors stay off the root LRU. Missing rollout files fail with an explicit restore error. -- **Startup** runs `index_rollout_metadata` in the background instead of replaying every rollout into memory. +- **Startup** finishes `index_rollout_metadata` before constructing the schedule wake loop, so durable jobs cannot be retired against an incomplete index. It indexes metadata without replaying every rollout into memory. ### Session title generation diff --git a/crates/server/src/bootstrap.rs b/crates/server/src/bootstrap.rs index e5099d90..86cf2653 100644 --- a/crates/server/src/bootstrap.rs +++ b/crates/server/src/bootstrap.rs @@ -36,6 +36,10 @@ pub enum ServerTransportMode { Stdio, #[value(name = "websocket")] WebSocket, + /// Independent authenticated loopback runtime, with no public listener. + Daemon, + /// Attach to an existing daemon only; never start a runtime. + Attach, } /// Command-line arguments accepted by the standalone server process entrypoint. @@ -115,14 +119,29 @@ pub async fn run_server_process( args: ServerProcessArgs, options: ServerProcessRunOptions, ) -> Result<()> { - let resolver = FileSystemConfigPathResolver::from_env()?; + let mut resolver = FileSystemConfigPathResolver::from_env()?; let action = args.action()?; // Decide whether this process is the one true server or a lightweight proxy/ // control client. Lock file lives under DEVO_HOME (see singleton.rs). + if matches!( + args.transport, + ServerTransportMode::Daemon | ServerTransportMode::Attach + ) { + resolver = FileSystemConfigPathResolver::new(crate::singleton_files::daemon_home( + &resolver.user_config_dir(), + )?); + } let singleton_role = acquire_singleton_role(&resolver.user_config_dir())?; let real_server_guard = match singleton_role { SingletonRole::Real(guard) => match action { - ServerProcessAction::Run => guard, + ServerProcessAction::Run => { + if args.transport == ServerTransportMode::Attach { + anyhow::bail!( + "no daemon is running; start one explicitly with devo daemon start" + ); + } + guard + } ServerProcessAction::Status | ServerProcessAction::Shutdown => { // We hold the lock but were not asked to run — no metadata file yet. println!("devo server is not running"); @@ -131,12 +150,26 @@ pub async fn run_server_process( }, SingletonRole::Proxy(metadata) => match action { ServerProcessAction::Run => { + if matches!( + args.transport, + ServerTransportMode::Daemon | ServerTransportMode::Attach + ) { + let status = run_server_control(&metadata, ServerControlAction::Status).await?; + if status.lifetime != crate::singleton::ServerLifetime::Daemon { + anyhow::bail!( + "a client-owned server is running; stop it explicitly before starting a daemon" + ); + } + } let result = run_server_control( &metadata, ServerControlAction::EnableProtocols(args.protocols.clone()), ) .await?; - if args.transport == ServerTransportMode::Stdio { + if matches!( + args.transport, + ServerTransportMode::Stdio | ServerTransportMode::Attach + ) { tracing::info!( pid = metadata.pid, endpoint = %metadata.endpoint, @@ -166,6 +199,12 @@ pub async fn run_server_process( return Ok(()); } ServerProcessAction::Shutdown => { + let status = run_server_control(&metadata, ServerControlAction::Status).await?; + if status.lifetime == crate::singleton::ServerLifetime::Daemon { + anyhow::bail!( + "refusing legacy shutdown of an independent daemon; use devo daemon stop explicitly" + ); + } let result = run_server_control(&metadata, ServerControlAction::Shutdown).await?; print_existing_server_status( &metadata, @@ -180,8 +219,14 @@ pub async fn run_server_process( let internal_proxy = InternalProxyEndpoint::bind().await?; // Persist ws://127.0.0.1: + random token into server.lock.json so // proxy/control processes know where and how to connect. - let singleton_metadata = - real_server_guard.publish_endpoint(internal_proxy.endpoint().to_string())?; + let singleton_metadata = real_server_guard.publish_endpoint( + internal_proxy.endpoint().to_string(), + if args.transport == ServerTransportMode::Daemon { + crate::singleton::ServerLifetime::Daemon + } else { + crate::singleton::ServerLifetime::Client + }, + )?; if let Err(error) = devo_core::migrate_session_defaults_to_config_toml(&resolver.user_config_dir()) @@ -212,13 +257,18 @@ pub async fn run_server_process( .effective_config() .clone(); let effective_listen = match args.transport { + ServerTransportMode::Daemon | ServerTransportMode::Attach => Vec::new(), ServerTransportMode::Config => config.server.listen.clone(), ServerTransportMode::Stdio => vec!["stdio://".to_string()], ServerTransportMode::WebSocket => { vec![format!("ws://{DEFAULT_WEBSOCKET_BIND_ADDRESS}")] } }; - let listen_targets = resolve_listen_targets(&effective_listen)?; + let listen_targets = if args.transport == ServerTransportMode::Daemon { + Vec::new() + } else { + resolve_listen_targets(&effective_listen)? + }; let effective_listen = listen_targets .iter() .map(|target| match target { @@ -275,6 +325,13 @@ pub async fn run_server_process( tracing::info!(db_path = %db_path.display(), "opening database"); let db = Arc::new(Database::open(db_path)?); + // A due job must not be retired as missing while the durable journal index + // is still rebuilding. Finish recovery before constructing the wake loop. + let index_store = + crate::persistence::RolloutStore::new(resolver.user_config_dir(), Some(Arc::clone(&db))); + let index_db = Arc::clone(&db); + tokio::task::spawn_blocking(move || index_store.index_rollout_metadata(&index_db)).await??; + let registry = Arc::new(registry); let provider_router = Arc::clone(&provider.provider_router); let process_context = Arc::new(crate::session_context::SessionRuntimeContext::from_parts( @@ -313,19 +370,6 @@ pub async fn run_server_process( serde_json::Map::from_iter([("trigger".to_string(), serde_json::json!("init"))]), ) .await; - // Rebuild SQLite session index from on-disk SessionMeta headers. Always - // refresh: empty DBs with pre-seeded rollouts (restore / stress corpora) - // otherwise stay invisible to session/list and session/resume. - { - let rollout_store = runtime.rollout_store(); - let db = runtime.deps_db(); - tokio::task::spawn_blocking(move || match rollout_store.index_rollout_metadata(&db) { - Ok(()) => tracing::info!("rollout metadata index refresh completed"), - Err(error) => { - tracing::warn!(%error, "rollout metadata index refresh failed"); - } - }); - } // Delivery-log reconciliation (08 §7): backfill event_log rows a crash // prevented the append path from writing. Runs in the background; // session/list correctness never depends on it. @@ -351,38 +395,50 @@ pub async fn run_server_process( } let shutdown_signal = tokio_util::sync::CancellationToken::new(); - let internal_proxy_control = InternalProxyControl::new(shutdown_signal.clone()); + let internal_proxy_control = if args.transport == ServerTransportMode::Daemon { + InternalProxyControl::daemon(shutdown_signal.clone()) + } else { + InternalProxyControl::new(shutdown_signal.clone()) + }; let external_shutdown = options.external_shutdown.clone(); - // Concurrent listeners: configured stdio/ws targets + internal proxy task. - // Returns when any listener exits; shutdown also via Ctrl+C, external token, - // or internal-proxy `server/shutdown` (cancels shutdown_signal). - tokio::select! { - result = run_listeners_with_internal_proxy( - runtime.clone(), - &effective_listen, - internal_proxy, - singleton_metadata.token.clone(), - internal_proxy_control, - ) => { - result?; - } - result = tokio::signal::ctrl_c() => { - result?; - tracing::info!("server shutdown requested"); - } - _ = wait_for_external_shutdown(external_shutdown.as_ref()) => { - tracing::info!("server shutdown requested from external process controller"); - } - _ = shutdown_signal.cancelled() => { - tracing::info!("server shutdown requested from singleton control"); + let listener_runtime = runtime.clone(); + let mut listeners = tokio::spawn(async move { + if args.transport == ServerTransportMode::Daemon { + crate::transport::run_daemon_listener( + listener_runtime, + internal_proxy, + singleton_metadata.token, + internal_proxy_control, + ) + .await + } else { + run_listeners_with_internal_proxy( + listener_runtime, + &effective_listen, + internal_proxy, + singleton_metadata.token, + internal_proxy_control, + ) + .await } - } + }); + let finished = tokio::select! { + result = &mut listeners => Some(result), + result = crate::daemon::wait_for_shutdown_signal() => { result?; None }, + _ = wait_for_external_shutdown(external_shutdown.as_ref()) => None, + }; + shutdown_signal.cancel(); + let listener_result = match finished { + Some(result) => result, + None => listeners.await, + }; tracing::info!("terminating unified exec processes"); registry.terminate_unified_exec_processes().await; tracing::info!("completing deferred items for active turns"); runtime.shutdown().await; + listener_result??; Ok(()) } diff --git a/crates/server/src/daemon.rs b/crates/server/src/daemon.rs new file mode 100644 index 00000000..dde8ea57 --- /dev/null +++ b/crates/server/src/daemon.rs @@ -0,0 +1,84 @@ +//! Explicit control of the independent, user-owned Native runtime. +use std::path::Path; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use serde::Serialize; + +use crate::singleton::{ + ServerControlAction, ServerLifetime, owner_lock_held, probe_singleton, run_server_control, +}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct DaemonStatus { + pub pid: u32, + pub started_at: String, +} + +/// Probe an authenticated daemon. Never promote a missing server to an owner. +pub async fn daemon_status(home: &Path) -> Result> { + let home = crate::singleton_files::daemon_home(home)?; + match probe_singleton(&home)? { + None => Ok(None), + Some(metadata) => { + let result = run_server_control(&metadata, ServerControlAction::Status).await?; + if result.lifetime != ServerLifetime::Daemon { + bail!( + "a client-owned server is running; stop it explicitly before starting a daemon" + ); + } + Ok(Some(DaemonStatus { + pid: metadata.pid, + started_at: metadata.started_at, + })) + } + } +} + +/// Request graceful shutdown and confirm the singleton lock has been released. +/// Never signal a PID read from disk or silently stop a client-owned server. +pub async fn stop_daemon(home: &Path) -> Result<()> { + let home = crate::singleton_files::daemon_home(home)?; + match probe_singleton(&home)? { + None => return Ok(()), + Some(metadata) => { + let status = run_server_control(&metadata, ServerControlAction::Status).await?; + if status.lifetime != ServerLifetime::Daemon { + bail!("refusing to stop a client-owned server"); + } + run_server_control(&metadata, ServerControlAction::ShutdownDaemon).await?; + } + } + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if !owner_lock_held(&home)? { + return Ok::<(), anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .context("daemon acknowledged shutdown but did not release ownership within 30 seconds")??; + Ok(()) +} + +/// Check the directory boundary before a detached child is launched. +pub fn validate_daemon_home(home: &Path) -> Result { + crate::singleton_files::daemon_home(home) +} + +/// Service managers normally send SIGTERM; preserve the graceful shutdown path. +pub(crate) async fn wait_for_shutdown_signal() -> Result<()> { + #[cfg(unix)] + { + let mut terminate = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + tokio::select! { result = tokio::signal::ctrl_c() => result?, _ = terminate.recv() => {} } + Ok(()) + } + #[cfg(not(unix))] + { + tokio::signal::ctrl_c().await?; + Ok(()) + } +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 166e7071..6bf42df7 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -3,6 +3,7 @@ mod approval_reviewer; mod bootstrap; mod client; mod connection; +mod daemon; pub mod db; mod event; mod event_reconcile; @@ -31,6 +32,7 @@ mod schedule_store; mod session; mod session_context; mod singleton; +mod singleton_files; pub mod subagent; pub mod test_support; mod titles; @@ -43,6 +45,7 @@ pub use approval::*; pub use bootstrap::*; pub use client::*; pub use connection::*; +pub use daemon::*; pub use event::*; pub use execution::ServerRuntimeDependencies; pub use execution::empty_mcp_manager; diff --git a/crates/server/src/persistence.rs b/crates/server/src/persistence.rs index 45ba781e..3e1f4bdb 100644 --- a/crates/server/src/persistence.rs +++ b/crates/server/src/persistence.rs @@ -1493,11 +1493,17 @@ impl RolloutStore { pub(crate) fn rollout_paths(&self) -> Result> { let mut files = Vec::new(); let sessions = self.data_root.join("sessions"); - if sessions.exists() { + if sessions + .try_exists() + .context("inspect sessions directory")? + { collect_rollout_files(&sessions, &mut files)?; } let artifacts = self.data_root.join("session-artifacts"); - if artifacts.exists() { + if artifacts + .try_exists() + .context("inspect session artifacts directory")? + { collect_rollout_files(&artifacts, &mut files)?; } files.sort(); diff --git a/crates/server/src/runtime.rs b/crates/server/src/runtime.rs index 5a7e58fd..099153fc 100644 --- a/crates/server/src/runtime.rs +++ b/crates/server/src/runtime.rs @@ -108,6 +108,8 @@ pub(crate) mod context_occupancy; mod context_usage; mod control_requests; mod credential_binding; +#[cfg(test)] +mod daemon_lifecycle_tests; mod goal_accounting; mod goal_background; mod goal_continuation; diff --git a/crates/server/src/runtime/daemon_lifecycle_tests.rs b/crates/server/src/runtime/daemon_lifecycle_tests.rs new file mode 100644 index 00000000..fa788ed1 --- /dev/null +++ b/crates/server/src/runtime/daemon_lifecycle_tests.rs @@ -0,0 +1,108 @@ +//! Admission must remain tracked after the initiating client disappears. +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result}; +use futures::{SinkExt, StreamExt}; +use pretty_assertions::assert_eq; +use serde_json::{Value, json}; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async, tungstenite::Message}; +use tokio_util::sync::CancellationToken; + +use crate::transport::{InternalProxyControl, InternalProxyEndpoint, run_daemon_listener}; + +type Socket = WebSocketStream>; + +async fn response(socket: &mut Socket, id: u64) -> Result { + tokio::time::timeout(Duration::from_secs(5), async { + while let Some(frame) = socket.next().await { + let frame = frame?; + if let Message::Text(text) = frame { + let value: Value = serde_json::from_str(&text)?; + if value["id"] == id { + anyhow::ensure!(value.get("error").is_none(), "RPC failed: {value}"); + return Ok(value["result"].clone()); + } + } + } + anyhow::bail!("socket closed") + }) + .await? +} + +#[tokio::test] +async fn daemon_stop_waits_for_rpc_admitted_by_already_disconnected_client() -> Result<()> { + let home = tempfile::tempdir()?; + let runtime = crate::TestRuntime::noop().runtime(home.path()); + let endpoint = InternalProxyEndpoint::bind().await?; + let url = endpoint.endpoint().to_string(); + let cancel = CancellationToken::new(); + let mut listener = tokio::spawn(run_daemon_listener( + Arc::clone(&runtime), + endpoint, + "test-token".into(), + InternalProxyControl::daemon(cancel.clone()), + )); + let (mut socket, _) = connect_async(url).await?; + socket.send(Message::Text("test-token".into())).await?; + socket.send(Message::Text(json!({"jsonrpc":"2.0","id":1,"method":"initialize", + "params":{"protocolVersion":1,"clientCapabilities":{},"_meta":{"devo":{"protocol":"native"}}} + }).to_string().into())).await?; + response(&mut socket, 1).await?; + socket + .send(Message::Text( + json!({"jsonrpc":"2.0","id":2,"method":"session/new", + "params":{"cwd":home.path(),"idempotencyKey":"drain-fixture"} + }) + .to_string() + .into(), + )) + .await?; + let created = response(&mut socket, 2).await?; + let session_id = serde_json::from_value(created["session"]["id"].clone())?; + let gate = runtime + .session_metadata_write_gate + .acquire(session_id) + .await; + socket.send(Message::Text(json!({"jsonrpc":"2.0","id":3,"method":"session/metadata/update", + "params":{"sessionId":session_id,"expectedVersion":0,"title":"admitted before disconnect"} + }).to_string().into())).await?; + // The following response proves the reader admitted request 3 first. Its + // handler remains blocked at the metadata gate while ping stays responsive. + socket + .send(Message::Text( + json!({"jsonrpc":"2.0","id":4,"method":"runtime/ping","params":{}}) + .to_string() + .into(), + )) + .await?; + response(&mut socket, 4).await?; + socket.close(None).await?; + tokio::time::timeout(Duration::from_secs(5), async { + while !runtime.connections.lock().await.is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .context("client unregister")?; + cancel.cancel(); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut listener) + .await + .is_err() + ); + drop(gate); + tokio::time::timeout(Duration::from_secs(5), &mut listener).await???; + assert_eq!( + runtime + .deps + .db + .get_session_index(&session_id)? + .context("index")? + .session + .title, + Some("admitted before disconnect".into()) + ); + runtime.shutdown().await; + Ok(()) +} diff --git a/crates/server/src/runtime/handlers/schedule.rs b/crates/server/src/runtime/handlers/schedule.rs index ef9ab991..2b1c02b2 100644 --- a/crates/server/src/runtime/handlers/schedule.rs +++ b/crates/server/src/runtime/handlers/schedule.rs @@ -328,6 +328,30 @@ impl ServerRuntime { match error { crate::runtime::session_cache::LoadSessionError::SessionNotFound | crate::runtime::session_cache::LoadSessionError::RolloutMissing => { + // SQLite is only a cache. A failed/partial startup scan + // must never turn durable work into a deleted schedule. + let runtime = Arc::clone(self); + let missing = + tokio::task::spawn_blocking(move || -> anyhow::Result { + let found = runtime + .rollout_store + .find_journal_containing_session_strict( + &session_id, + crate::persistence::JournalVisibility::IncludeDeleted, + )?; + if found.is_some() { + runtime + .rollout_store + .index_rollout_metadata(&runtime.deps.db)?; + } + Ok(found.is_none()) + }) + .await; + if !matches!(missing, Ok(Ok(true))) { + tracing::warn!(%session_id, "schedule journal is present or uncertain; retaining occurrence for retry"); + self.retry_schedule_occurrence(occurrence_id); + continue; + } tracing::warn!( job_id = job.job_id.as_str(), session_id = %session_id, diff --git a/crates/server/src/runtime/handlers/schedule_tests.rs b/crates/server/src/runtime/handlers/schedule_tests.rs index 1268b0fc..77905338 100644 --- a/crates/server/src/runtime/handlers/schedule_tests.rs +++ b/crates/server/src/runtime/handlers/schedule_tests.rs @@ -685,3 +685,43 @@ async fn inactive_schedules_discard_claimed_but_undelivered_occurrences() -> Res runtime.shutdown().await; Ok(()) } + +#[tokio::test] +async fn schedule_retains_occurrence_when_missing_index_has_unreadable_journal() -> Result<()> { + let home = TempDir::new()?; + let runtime = TestRuntime::noop().runtime(home.path()); + let session_id = devo_protocol::native::ids::SessionId::new(); + std::fs::create_dir_all(home.path().join("sessions"))?; + std::fs::write( + home.path() + .join("sessions") + .join(format!("{session_id}.jsonl")), + b"{invalid journal\n", + )?; + let job = runtime.schedule_store.upsert( + devo_protocol::native::rpc_schedule::SessionScheduleUpsertParams { + kind: devo_protocol::native::rpc_schedule::ScheduleJobKind::Cron, + session_id, + job_id: None, + cwd: None, + schedule: None, + interval_ms: Some(1), + prompt: Some("retain uncertain recovery".into()), + instruction: None, + delivery_mode: devo_protocol::native::rpc_schedule::ScheduleDeliveryMode::FollowUp, + label: None, + }, + )?; + runtime + .schedule_store + .claim_due_with_outbox(job.next_run_at.context("due")?)?; + runtime.dispatch_due_schedule_jobs().await?; + assert!(runtime.schedule_store.get_job(&job.job_id)?.is_some()); + let pending = runtime + .schedule_store + .pending_occurrences(chrono::Utc::now() + chrono::Duration::hours(1))?; + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].job.job_id, job.job_id); + runtime.shutdown().await; + Ok(()) +} diff --git a/crates/server/src/runtime/lifecycle.rs b/crates/server/src/runtime/lifecycle.rs index 14624b17..52b55b69 100644 --- a/crates/server/src/runtime/lifecycle.rs +++ b/crates/server/src/runtime/lifecycle.rs @@ -460,6 +460,9 @@ impl ServerRuntime { /// persists interrupted turn records. Called on graceful shutdown. pub async fn shutdown(self: &Arc) { self.schedule_wake_cancel.cancel(); + // Wait for an already admitted durable dispatch before snapshotting sessions. + let schedule_drain = self.schedule_store.lock_dispatch().await; + drop(schedule_drain); self.command_exec_manager.terminate_all().await; let session_handles = self.list_session_handles().await; diff --git a/crates/server/src/schedule_store.rs b/crates/server/src/schedule_store.rs index cc2db47d..92e8b9e8 100644 --- a/crates/server/src/schedule_store.rs +++ b/crates/server/src/schedule_store.rs @@ -294,6 +294,11 @@ impl ScheduleStore { let mut file = self.read_unlocked()?; let mut due = Vec::new(); let mut occurrences = Vec::new(); + let pending_jobs = if matches!(mode, DueClaimMode::WithOutbox) { + self.pending_job_times_unlocked()? + } else { + HashMap::new() + }; for job in &mut file.jobs { if job.status != ScheduleJobStatus::Active { continue; @@ -304,9 +309,22 @@ impl ScheduleStore { if scheduled_at > now { continue; } + // One undelivered occurrence per job bounds approval/offline backlogs. + // The same timestamp is allowed so a crash between outbox and job + // writes can reconcile its persisted phase without a second delivery. + if pending_jobs + .get(job.job_id.as_str()) + .is_some_and(|pending| *pending != scheduled_at) + { + continue; + } job.last_run_at = Some(now); job.run_count = job.run_count.saturating_add(1); - job.next_run_at = next_run_from_interval(job.interval_ms, now); + job.next_run_at = job.interval_ms.and_then(|interval| { + let elapsed = u64::try_from((now - scheduled_at).num_milliseconds()).ok()?; + let steps = elapsed.checked_div(interval)?.checked_add(1)?; + next_run_from_interval(interval.checked_mul(steps), scheduled_at) + }); job.updated_at = now; let claimed = job.clone(); if matches!(mode, DueClaimMode::WithOutbox) { diff --git a/crates/server/src/schedule_store/outbox.rs b/crates/server/src/schedule_store/outbox.rs index d6741305..e0db5290 100644 --- a/crates/server/src/schedule_store/outbox.rs +++ b/crates/server/src/schedule_store/outbox.rs @@ -43,6 +43,18 @@ struct ScheduleOutboxFile { } impl ScheduleStore { + pub(super) fn pending_job_times_unlocked( + &self, + ) -> Result>> { + let mut jobs = std::collections::HashMap::new(); + for occurrence in self.read_outbox_unlocked()?.occurrences { + jobs.entry(occurrence.job.job_id.as_str().to_string()) + .and_modify(|at: &mut DateTime| *at = (*at).min(occurrence.scheduled_at)) + .or_insert(occurrence.scheduled_at); + } + Ok(jobs) + } + pub(crate) fn claim_due_with_outbox(&self, now: DateTime) -> Result> { self.claim_due_inner(now, DueClaimMode::WithOutbox) } @@ -340,6 +352,40 @@ mod tests { assert_eq!(occurrences[0], occurrence); } + #[test] + fn restart_keeps_interval_phase_and_bounds_pending_delivery() { + let (dir, store, scheduled_at) = store_with_due_job(); + let late = scheduled_at + Duration::minutes(17); + let due = store + .claim_due_with_outbox(late) + .expect("coalesce missed slots"); + assert_eq!( + due[0].next_run_at, + Some(scheduled_at + Duration::minutes(20)) + ); + drop(store); + let reopened = ScheduleStore::new(dir.path()); + assert!( + reopened + .claim_due_with_outbox(late + Duration::hours(24)) + .expect("blocked delivery") + .is_empty() + ); + let pending = reopened.pending_occurrences(late).expect("pending"); + assert_eq!(pending.len(), 1); + reopened + .acknowledge_occurrence(&pending[0].occurrence_id) + .expect("acknowledge"); + let due = reopened + .claim_due_with_outbox(late + Duration::hours(24)) + .expect("coalesce after acknowledgement"); + assert_eq!(due.len(), 1); + assert_eq!( + due[0].next_run_at, + Some(scheduled_at + Duration::hours(24) + Duration::minutes(20)) + ); + } + #[test] fn outbox_retry_and_acknowledgement_are_durable() { let (dir, store, _) = store_with_due_job(); diff --git a/crates/server/src/singleton.rs b/crates/server/src/singleton.rs index facad142..adff937a 100644 --- a/crates/server/src/singleton.rs +++ b/crates/server/src/singleton.rs @@ -14,7 +14,6 @@ use std::fs; use std::fs::File; -use std::fs::OpenOptions; use std::io::ErrorKind; use std::path::Path; use std::path::PathBuf; @@ -47,6 +46,7 @@ const METADATA_READ_RETRIES: usize = 100; const METADATA_READ_RETRY_DELAY: Duration = Duration::from_millis(50); pub(crate) const SERVER_CONTROL_STATUS_METHOD: &str = "server/status"; pub(crate) const SERVER_CONTROL_SHUTDOWN_METHOD: &str = "server/shutdown"; +pub(crate) const DAEMON_CONTROL_SHUTDOWN_METHOD: &str = "daemon/shutdown"; pub(crate) const SERVER_CONTROL_ENABLE_PROTOCOLS_METHOD: &str = "server/protocols/enable"; const SERVER_CONTROL_REQUEST_ID: u64 = 1; @@ -61,8 +61,16 @@ pub(crate) enum SingletonRole { Proxy(ServerLockMetadata), } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ServerLifetime { + #[default] + Client, + Daemon, +} + /// Published in `server.lock.json` while the real server is running. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct ServerLockMetadata { pub(crate) version: u32, pub(crate) pid: u32, @@ -71,12 +79,29 @@ pub(crate) struct ServerLockMetadata { /// Must be sent as the first WebSocket text frame after connect. pub(crate) token: String, pub(crate) started_at: String, + #[serde(default)] + pub(crate) lifetime: ServerLifetime, +} + +impl std::fmt::Debug for ServerLockMetadata { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ServerLockMetadata") + .field("version", &self.version) + .field("pid", &self.pid) + .field("endpoint", &self.endpoint) + .field("token", &"[redacted]") + .field("started_at", &self.started_at) + .field("lifetime", &self.lifetime) + .finish() + } } #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum ServerControlAction { Status, Shutdown, + ShutdownDaemon, EnableProtocols(ProtocolSet), } @@ -85,13 +110,14 @@ impl ServerControlAction { match self { ServerControlAction::Status => SERVER_CONTROL_STATUS_METHOD, ServerControlAction::Shutdown => SERVER_CONTROL_SHUTDOWN_METHOD, + ServerControlAction::ShutdownDaemon => DAEMON_CONTROL_SHUTDOWN_METHOD, ServerControlAction::EnableProtocols(_) => SERVER_CONTROL_ENABLE_PROTOCOLS_METHOD, } } fn params(&self) -> Option { match self { - Self::Status | Self::Shutdown => None, + Self::Status | Self::Shutdown | Self::ShutdownDaemon => None, Self::EnableProtocols(protocols) => Some(serde_json::json!({ "protocols": protocols.names(), })), @@ -102,6 +128,7 @@ impl ServerControlAction { #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct ServerControlResult { pub(crate) status: String, + pub(crate) lifetime: ServerLifetime, pub(crate) enabled_protocols: Option, } @@ -113,6 +140,7 @@ impl ServerLockMetadata { endpoint, token, started_at: Utc::now().to_rfc3339(), + lifetime: ServerLifetime::Client, } } } @@ -122,6 +150,18 @@ impl ServerLockMetadata { pub(crate) async fn run_server_control( metadata: &ServerLockMetadata, action: ServerControlAction, +) -> Result { + tokio::time::timeout( + Duration::from_secs(5), + run_server_control_inner(metadata, action), + ) + .await + .context("singleton control request timed out")? +} + +async fn run_server_control_inner( + metadata: &ServerLockMetadata, + action: ServerControlAction, ) -> Result { let (socket, _) = connect_async(metadata.endpoint.as_str()) .await @@ -166,15 +206,22 @@ pub(crate) struct RealServerGuard { impl RealServerGuard { /// Writes `server.lock.json` so proxy/control clients can find this server. - pub(crate) fn publish_endpoint(&self, endpoint: String) -> Result { - let metadata = ServerLockMetadata::new(endpoint, Uuid::new_v4().to_string()); + pub(crate) fn publish_endpoint( + &self, + endpoint: String, + lifetime: ServerLifetime, + ) -> Result { + let mut metadata = ServerLockMetadata::new(endpoint, Uuid::new_v4().to_string()); + metadata.lifetime = lifetime; let encoded = serde_json::to_vec_pretty(&metadata).context("serialize server metadata")?; - fs::write(&self.metadata_path, encoded).with_context(|| { - format!( - "write server singleton metadata {}", - self.metadata_path.display() - ) - })?; + crate::singleton_files::publish_metadata(&self.metadata_path, &encoded).with_context( + || { + format!( + "write server singleton metadata {}", + self.metadata_path.display() + ) + }, + )?; Ok(metadata) } } @@ -197,26 +244,47 @@ pub(crate) fn acquire_singleton_role(devo_home: &Path) -> Result fs::create_dir_all(devo_home) .with_context(|| format!("create DEVO_HOME {}", devo_home.display()))?; let lock_path = lock_path(devo_home); - let lock_file = OpenOptions::new() - .create(true) - .truncate(false) - .read(true) - .write(true) - .open(&lock_path) + let lock_file = crate::singleton_files::open_lock(&lock_path) .with_context(|| format!("open server singleton lock {}", lock_path.display()))?; - match lock_file.try_lock_exclusive() { - Ok(()) => Ok(SingletonRole::Real(RealServerGuard { - lock_file, - metadata_path: metadata_path(devo_home), - })), - Err(error) if is_lock_temporarily_unavailable(&error) => { - // Real server may be mid-startup; retry until metadata appears. - Ok(SingletonRole::Proxy(read_metadata_with_retry(devo_home)?)) + for _ in 0..METADATA_READ_RETRIES { + match lock_file.try_lock_exclusive() { + Ok(()) => { + let _ = fs::remove_file(metadata_path(devo_home)); + return Ok(SingletonRole::Real(RealServerGuard { + lock_file, + metadata_path: metadata_path(devo_home), + })); + } + Err(error) if is_lock_temporarily_unavailable(&error) => { + if let Ok(metadata) = read_metadata(devo_home) { + return Ok(SingletonRole::Proxy(metadata)); + } + std::thread::sleep(METADATA_READ_RETRY_DELAY); + } + Err(error) => return Err(error).context("acquire singleton owner lock"), } - Err(error) => { - Err(error).with_context(|| format!("lock server singleton {}", lock_path.display())) + } + bail!("singleton owner has not published valid metadata; retry after startup completes") +} + +/// A status probe never publishes/removes metadata or keeps an owner guard. +pub(crate) fn probe_singleton(home: &Path) -> Result> { + if !owner_lock_held(home)? { + return Ok(None); + } + read_metadata(home).map(Some) +} + +pub(crate) fn owner_lock_held(home: &Path) -> Result { + let file = crate::singleton_files::open_lock(&lock_path(home))?; + match file.try_lock_exclusive() { + Ok(()) => { + file.unlock()?; + Ok(false) } + Err(error) if is_lock_temporarily_unavailable(&error) => Ok(true), + Err(error) => Err(error).context("probe singleton owner lock"), } } @@ -316,6 +384,12 @@ fn decode_server_control_response(text: &str) -> Result { }) .transpose()?; Ok(ServerControlResult { + lifetime: value + .pointer("/result/lifetime") + .cloned() + .map(serde_json::from_value) + .transpose()? + .unwrap_or_default(), status, enabled_protocols, }) @@ -339,24 +413,9 @@ fn is_lock_temporarily_unavailable(error: &std::io::Error) -> bool { } } -/// Polls until the real server writes metadata (startup race with lock holder). -fn read_metadata_with_retry(devo_home: &Path) -> Result { - let mut last_error = None; - for _ in 0..METADATA_READ_RETRIES { - match read_metadata(devo_home) { - Ok(metadata) => return Ok(metadata), - Err(error) => { - last_error = Some(error); - std::thread::sleep(METADATA_READ_RETRY_DELAY); - } - } - } - Err(last_error.expect("metadata read should have failed at least once")) -} - fn read_metadata(devo_home: &Path) -> Result { let path = metadata_path(devo_home); - let encoded = fs::read(&path) + let encoded = crate::singleton_files::read_metadata(&path) .with_context(|| format!("read server singleton metadata {}", path.display()))?; let metadata: ServerLockMetadata = serde_json::from_slice(&encoded).context("decode server singleton metadata")?; @@ -366,6 +425,20 @@ fn read_metadata(devo_home: &Path) -> Result { metadata.version ); } + let uri: tokio_tungstenite::tungstenite::http::Uri = metadata + .endpoint + .parse() + .context("invalid singleton endpoint")?; + if uri.scheme_str() != Some("ws") + || !matches!(uri.host(), Some("127.0.0.1" | "[::1]")) + || uri.port_u16().is_none() + || uri.path() != "/" + || uri.query().is_some() + || metadata.endpoint.contains('@') + || metadata.endpoint.contains('#') + { + bail!("singleton endpoint must be an authenticated numeric loopback WebSocket"); + } Ok(metadata) } @@ -421,6 +494,7 @@ mod tests { .expect("decode response"), ServerControlResult { status: "running".to_string(), + lifetime: ServerLifetime::Client, enabled_protocols: None, } ); @@ -435,6 +509,7 @@ mod tests { .expect("decode response"), ServerControlResult { status: "running".to_string(), + lifetime: ServerLifetime::Client, enabled_protocols: Some(ProtocolSet::all()), } ); @@ -463,7 +538,7 @@ mod tests { }; let metadata = guard - .publish_endpoint("ws://127.0.0.1:0".to_string()) + .publish_endpoint("ws://127.0.0.1:0".to_string(), ServerLifetime::Client) .expect("publish endpoint"); let encoded = fs::read_to_string(metadata_path(temp_dir.path())).expect("metadata file"); let persisted: ServerLockMetadata = serde_json::from_str(&encoded).expect("metadata json"); @@ -472,6 +547,66 @@ mod tests { assert!(!encoded.contains("working_root")); } + #[test] + fn metadata_only_accepts_numeric_loopback_and_redacts_debug_token() { + let home = TempDir::new().unwrap(); + for endpoint in ["ws://127.0.0.1:3210", "ws://[::1]:3210/"] { + let metadata = ServerLockMetadata::new(endpoint.into(), "secret-test-token".into()); + crate::singleton_files::publish_metadata( + &metadata_path(home.path()), + &serde_json::to_vec(&metadata).unwrap(), + ) + .unwrap(); + assert_eq!(read_metadata(home.path()).unwrap(), metadata); + assert!(!format!("{metadata:?}").contains("secret-test-token")); + } + for endpoint in [ + "ws://example.com:3210", + "ws://localhost:3210", + "wss://127.0.0.1:3210", + "ws://127.0.0.1:3210/other", + "ws://127.0.0.1:3210/?query", + "ws://token@127.0.0.1:3210/", + "ws://127.0.0.1:3210/#fragment", + ] { + let metadata = ServerLockMetadata::new(endpoint.into(), "secret-test-token".into()); + crate::singleton_files::publish_metadata( + &metadata_path(home.path()), + &serde_json::to_vec(&metadata).unwrap(), + ) + .unwrap(); + assert!(read_metadata(home.path()).is_err(), "{endpoint}"); + } + } + + #[test] + fn status_probe_preserves_metadata_and_never_keeps_ownership() { + let home = TempDir::new().unwrap(); + let SingletonRole::Real(guard) = acquire_singleton_role(home.path()).unwrap() else { + panic!("owner"); + }; + let metadata = guard + .publish_endpoint("ws://127.0.0.1:3210".into(), ServerLifetime::Daemon) + .unwrap(); + assert_eq!( + probe_singleton(home.path()).unwrap(), + Some(metadata.clone()) + ); + drop(guard); + // Simulate a crash residue: an absent owner must not remove it on status. + crate::singleton_files::publish_metadata( + &metadata_path(home.path()), + &serde_json::to_vec(&metadata).unwrap(), + ) + .unwrap(); + assert_eq!(probe_singleton(home.path()).unwrap(), None); + assert!(metadata_path(home.path()).exists()); + assert!(matches!( + acquire_singleton_role(home.path()).unwrap(), + SingletonRole::Real(_) + )); + } + #[test] fn dropping_real_guard_removes_metadata() { let temp_dir = TempDir::new().expect("temp dir"); @@ -480,7 +615,7 @@ mod tests { panic!("expected real server role"); }; guard - .publish_endpoint("ws://127.0.0.1:0".to_string()) + .publish_endpoint("ws://127.0.0.1:0".to_string(), ServerLifetime::Client) .expect("publish endpoint"); let metadata_path = metadata_path(temp_dir.path()); diff --git a/crates/server/src/singleton_files.rs b/crates/server/src/singleton_files.rs new file mode 100644 index 00000000..33f2a56d --- /dev/null +++ b/crates/server/src/singleton_files.rs @@ -0,0 +1,150 @@ +//! Owner-only singleton files. Daemon startup fails closed on unsupported ACLs. +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, bail}; + +pub(crate) fn daemon_home(home: &Path) -> Result { + #[cfg(not(unix))] + { + let _ = home; + bail!( + "background daemon requires Unix owner-only file protection; Windows ACL support is not implemented" + ) + } + #[cfg(unix)] + { + use std::os::unix::fs::{DirBuilderExt, MetadataExt, PermissionsExt}; + fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(home)?; + if fs::symlink_metadata(home)?.file_type().is_symlink() { + bail!("daemon home must not be a symbolic link"); + } + let canonical = home.canonicalize()?; + let uid = unsafe { libc::geteuid() }; + for (index, ancestor) in canonical.ancestors().enumerate() { + let metadata = fs::metadata(ancestor)?; + let mode = metadata.permissions().mode(); + if (index == 0 && metadata.uid() != uid) + || (metadata.uid() != uid && metadata.uid() != 0) + || (mode & 0o022 != 0 && !(index > 0 && metadata.uid() == 0 && mode & 0o1000 != 0)) + { + bail!( + "daemon home has an unsafe owner or writable ancestor: {}", + ancestor.display() + ); + } + } + Ok(canonical) + } +} + +fn options() -> OpenOptions { + let mut options = OpenOptions::new(); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options + .mode(0o600) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC); + } + options +} + +fn validate(file: &File, private: bool) -> Result<()> { + let metadata = file.metadata()?; + if !metadata.is_file() { + bail!("singleton path is not a regular file"); + } + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + if metadata.uid() != unsafe { libc::geteuid() } + || (private && metadata.permissions().mode() & 0o077 != 0) + { + bail!( + "singleton file must be owned by this user with owner-only permissions; stop the old server before restarting" + ); + } + } + #[cfg(not(unix))] + let _ = private; + Ok(()) +} + +pub(crate) fn open_lock(path: &Path) -> Result { + let file = options() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(path)?; + validate(&file, /*private*/ false)?; + Ok(file) +} + +pub(crate) fn read_metadata(path: &Path) -> Result> { + let mut file = options().read(true).open(path)?; + validate(&file, /*private*/ true)?; + if file.metadata()?.len() > 16 * 1024 { + bail!("singleton metadata exceeds size limit"); + } + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes)?; + Ok(bytes) +} + +pub(crate) fn publish_metadata(path: &Path, bytes: &[u8]) -> Result<()> { + let temporary = path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4())); + let result = (|| -> Result<()> { + let mut file = options().create_new(true).write(true).open(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + drop(file); + fs::rename(&temporary, path).context("publish singleton metadata")?; + #[cfg(unix)] + File::open(path.parent().context("metadata parent")?)?.sync_all()?; + Ok(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use std::os::unix::fs::{PermissionsExt, symlink}; + + #[test] + fn metadata_is_private_and_symlinks_are_not_followed() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("server.json"); + publish_metadata(&path, b"private").unwrap(); + assert_eq!( + fs::metadata(&path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + assert_eq!(read_metadata(&path).unwrap(), b"private"); + let link = dir.path().join("link"); + symlink(&path, &link).unwrap(); + assert!(read_metadata(&link).is_err()); + assert!(open_lock(&link).is_err()); + } + + #[test] + fn daemon_rejects_writable_parent_even_when_home_is_private() { + let dir = tempfile::tempdir().unwrap(); + let home = dir.path().join("home"); + fs::create_dir(&home).unwrap(); + fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o777)).unwrap(); + assert!(daemon_home(&home).is_err()); + fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!(daemon_home(&home).unwrap(), home); + } +} diff --git a/crates/server/src/transport.rs b/crates/server/src/transport.rs index 8108757a..a1e3eb2b 100644 --- a/crates/server/src/transport.rs +++ b/crates/server/src/transport.rs @@ -35,9 +35,11 @@ use crate::runtime::OutboundFrame; use crate::runtime::enqueue_outbound; use crate::runtime::log_outbound_frame; use crate::runtime::outbound_frame_to_value; +use crate::singleton::DAEMON_CONTROL_SHUTDOWN_METHOD; use crate::singleton::SERVER_CONTROL_ENABLE_PROTOCOLS_METHOD; use crate::singleton::SERVER_CONTROL_SHUTDOWN_METHOD; use crate::singleton::SERVER_CONTROL_STATUS_METHOD; +use crate::singleton::ServerLifetime; /// Per-connection transport state shared between the read loop and handler tasks. struct ConnectionTransport { @@ -240,11 +242,22 @@ pub struct InternalProxyEndpoint { #[derive(Clone)] pub struct InternalProxyControl { shutdown_token: CancellationToken, + lifetime: ServerLifetime, } impl InternalProxyControl { pub fn new(shutdown_token: CancellationToken) -> Self { - Self { shutdown_token } + Self { + shutdown_token, + lifetime: ServerLifetime::Client, + } + } + + pub(crate) fn daemon(shutdown_token: CancellationToken) -> Self { + Self { + shutdown_token, + lifetime: ServerLifetime::Daemon, + } } fn request_shutdown(&self) { @@ -334,6 +347,15 @@ pub async fn run_listeners_with_internal_proxy( run_listener_tasks(runtime, targets, Some((internal_proxy, token, control))).await } +pub(crate) async fn run_daemon_listener( + runtime: Arc, + endpoint: InternalProxyEndpoint, + token: String, + control: InternalProxyControl, +) -> Result<()> { + run_internal_proxy(runtime, endpoint, token, control).await +} + async fn run_listener_tasks( runtime: Arc, targets: Vec, @@ -509,21 +531,37 @@ async fn run_internal_proxy( ) -> Result<()> { let InternalProxyEndpoint { listener, endpoint } = internal_proxy; tracing::info!(endpoint = %endpoint, "internal stdio proxy listener bound"); - loop { - let (stream, remote_addr) = listener.accept().await?; + // Long-lived clients cannot consume the slots reserved for controls/auth. + let capacity = Arc::new(Semaphore::new(32)); + let client_capacity = Arc::new(Semaphore::new(16)); + let mut connections = JoinSet::new(); + let listener_result: Result<()> = loop { + let accepted = tokio::select! { + biased; + _ = control.shutdown_token.cancelled() => break Ok(()), + joined = connections.join_next(), if !connections.is_empty() => { let _ = joined; continue; }, + accepted = listener.accept() => match accepted { Ok(value) => value, Err(error) => break Err(error.into()) }, + }; + let (stream, remote_addr) = accepted; + let Ok(permit) = Arc::clone(&capacity).try_acquire_owned() else { + continue; + }; let runtime = Arc::clone(&runtime); let token = token.clone(); let control = control.clone(); - tokio::spawn(async move { - tracing::info!(remote_addr = %remote_addr, "internal stdio proxy connected"); - if let Err(error) = - handle_internal_proxy_connection(runtime, stream, &token, control).await - { - tracing::warn!(remote_addr = %remote_addr, error = %error, "internal stdio proxy closed with error"); + let client_capacity = Arc::clone(&client_capacity); + connections.spawn(async move { + let _permit = permit; + if let Err(error) = handle_internal_proxy_connection(runtime, stream, &token, control, client_capacity).await { + tracing::debug!(remote_addr = %remote_addr, error = %error, "internal proxy closed"); } - tracing::info!(remote_addr = %remote_addr, "internal stdio proxy disconnected"); }); - } + }; + control.request_shutdown(); + // Do not let the process release its owner lock while admitted RPCs are + // still changing durable state. Each connection stops reads and drains. + while connections.join_next().await.is_some() {} + listener_result } async fn handle_internal_proxy_connection( @@ -531,34 +569,80 @@ async fn handle_internal_proxy_connection( stream: tokio::net::TcpStream, expected_token: &str, control: InternalProxyControl, + client_capacity: Arc, ) -> Result<()> { - let websocket = accept_async(stream).await?; - let (mut writer, mut reader) = websocket.split(); - match reader.next().await { + let mut websocket = tokio::time::timeout( + std::time::Duration::from_secs(5), + tokio_tungstenite::accept_async_with_config( + stream, + Some( + tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default() + .max_message_size(Some(16 * 1024 * 1024)) + .max_frame_size(Some(16 * 1024 * 1024)), + ), + ), + ) + .await + .context("internal proxy handshake timed out")??; + match tokio::time::timeout(std::time::Duration::from_secs(5), websocket.next()) + .await + .context("internal proxy authentication timed out")? + { Some(Ok(Message::Text(token))) if token.as_str() == expected_token => {} Some(Ok(Message::Close(_))) | None => return Ok(()), Some(Ok(_)) => bail!("internal stdio proxy did not send auth token"), Some(Err(error)) => return Err(error.into()), } + let (mut writer, mut reader) = websocket.split(); + let initialization_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); let first_value = loop { - match reader.next().await { + let frame = tokio::select! { + biased; + _ = control.shutdown_token.cancelled() => return Ok(()), + frame = tokio::time::timeout_at(initialization_deadline, reader.next()) => frame.context("internal proxy initialization timed out")?, + }; + match frame { Some(Ok(Message::Text(text))) => { let value: serde_json::Value = serde_json::from_str(&text)?; if let Some(request) = parse_internal_proxy_control_request(&value) { + let mismatched_shutdown = matches!( + (&request.action, control.lifetime), + (InternalProxyControlAction::Shutdown, ServerLifetime::Daemon) + | ( + InternalProxyControlAction::ShutdownDaemon, + ServerLifetime::Client + ) + ); + if mismatched_shutdown { + writer.send(Message::Text(serde_json::json!({ + "jsonrpc":"2.0", "id":request.id, + "error":{"code":-32000,"message":"shutdown owner lifetime mismatch; use the explicit lifecycle command"}, + }).to_string().into())).await?; + return Ok(()); + } let enabled_protocols = match &request.action { InternalProxyControlAction::EnableProtocols(requested) => { runtime.enable_protocols(requested).await } InternalProxyControlAction::Status - | InternalProxyControlAction::Shutdown => runtime.enabled_protocols().await, + | InternalProxyControlAction::Shutdown + | InternalProxyControlAction::ShutdownDaemon => { + runtime.enabled_protocols().await + } }; - let response = internal_proxy_control_response(&request, &enabled_protocols); + let mut response = + internal_proxy_control_response(&request, &enabled_protocols); + response["result"]["lifetime"] = serde_json::to_value(control.lifetime)?; writer .send(Message::Text(response.to_string().into())) .await .context("send internal proxy control response")?; - if matches!(request.action, InternalProxyControlAction::Shutdown) { + if matches!( + request.action, + InternalProxyControlAction::Shutdown + | InternalProxyControlAction::ShutdownDaemon + ) { control.request_shutdown(); } return Ok(()); @@ -571,6 +655,9 @@ async fn handle_internal_proxy_connection( } }; + let _client_permit = client_capacity + .try_acquire_owned() + .context("internal proxy client capacity reached")?; let (transport, outbound_rx) = ConnectionTransport::new(); let outbound_tx = transport.outbound_tx.clone(); let connection_id = runtime @@ -601,18 +688,26 @@ async fn handle_internal_proxy_connection( return Ok(()); } - while let Some(frame) = reader.next().await { + loop { + let frame = tokio::select! { + biased; + _ = control.shutdown_token.cancelled() => break, + frame = reader.next() => frame, + }; + let Some(frame) = frame else { + break; + }; match frame { Ok(Message::Text(text)) => { - accept_incoming_client_message( - Arc::clone(&runtime), - connection_id, - outbound_tx.clone(), - Arc::clone(&transport.inbound_semaphore), - text.as_str(), - "internal_proxy_notifications", - ) - .await; + tokio::select! { + biased; + _ = control.shutdown_token.cancelled() => break, + _ = accept_incoming_client_message( + Arc::clone(&runtime), connection_id, outbound_tx.clone(), + Arc::clone(&transport.inbound_semaphore), text.as_str(), + "internal_proxy_notifications", + ) => {}, + } } Ok(Message::Close(_)) => break, Ok(_) => {} @@ -624,8 +719,15 @@ async fn handle_internal_proxy_connection( .await; runtime.unregister_connection(connection_id).await; - tracing::info!(connection_id, "internal stdio proxy connection closed"); outbound_writer.abort(); + // Keep this connection in the listener JoinSet until every admitted RPC + // finishes, even if it disconnected before shutdown began. Otherwise the + // next stop could miss detached handlers still mutating durable state. + let _drain = transport + .inbound_semaphore + .acquire_many(INBOUND_CONCURRENCY_LIMIT as u32) + .await; + tracing::info!(connection_id, "internal stdio proxy connection closed"); result } @@ -633,6 +735,7 @@ async fn handle_internal_proxy_connection( enum InternalProxyControlAction { Status, Shutdown, + ShutdownDaemon, EnableProtocols(ProtocolSet), } @@ -640,7 +743,9 @@ impl InternalProxyControlAction { fn response_status(&self) -> &'static str { match self { InternalProxyControlAction::Status => "running", - InternalProxyControlAction::Shutdown => "shutting down", + InternalProxyControlAction::Shutdown | InternalProxyControlAction::ShutdownDaemon => { + "shutting down" + } InternalProxyControlAction::EnableProtocols(_) => "running", } } @@ -659,6 +764,7 @@ fn parse_internal_proxy_control_request( let action = match method { SERVER_CONTROL_STATUS_METHOD => InternalProxyControlAction::Status, SERVER_CONTROL_SHUTDOWN_METHOD => InternalProxyControlAction::Shutdown, + DAEMON_CONTROL_SHUTDOWN_METHOD => InternalProxyControlAction::ShutdownDaemon, SERVER_CONTROL_ENABLE_PROTOCOLS_METHOD => { let protocols = serde_json::from_value::>( value.pointer("/params/protocols")?.clone(), diff --git a/docs/native-background-daemon.md b/docs/native-background-daemon.md new file mode 100644 index 00000000..f30dcdf2 --- /dev/null +++ b/docs/native-background-daemon.md @@ -0,0 +1,109 @@ +# Independent Native runtime (Unix) + +This adds an explicit process owner for Devo's existing Native runtime. It reuses +the same session actors, singleton lock, authenticated proxy, durable journals, +interval schedule store and schedule outbox. It does not create a second agent +engine or install an operating-system service. + +## Commands and ownership + +- `devo daemon start`: launch a detached child, without inherited stdio or a + controlling terminal. Report ready only after an authenticated status reply. +- `devo daemon run`: foreground owner for an operator-managed supervisor. +- `devo daemon status`: authenticated live status; no owner is created. +- `devo daemon attach`: NDJSON Native connection to an existing daemon. Missing + owner is an error, never an implicit server launch. +- `devo daemon stop`: authenticated graceful stop; wait for the singleton lock to + release before reporting stopped. A 30-second drain timeout is an uncertain + outcome, not permission to kill a PID or start a replacement. + +Start the daemon before opening Desktop/TUI. Existing `devo server --transport +stdio` clients attach through their proxy processes. Closing those clients does +not end the daemon. A client-owned server is never silently promoted, killed or +replaced: explicitly stop it first. Legacy `devo server --shutdown`, including +Desktop protocol-recovery attempts, refuses to stop a daemon. The server itself +rejects the old shutdown wire method for a daemon, including from older clients. +Use `daemon stop`, which sends the explicit daemon lifecycle control. + +A start readiness timeout means startup may still be in progress. Check status +before retrying. There is no automatic restart, login startup, upgrade handover, +or service installation in this change. OS power, suspend, network access, +provider availability and approvals still constrain execution; this is not a +24/7 availability promise. Windows daemon commands fail closed until owner-only +ACL handling is implemented and tested; existing client-owned Windows behavior +is retained. + +## Security and shutdown + +The daemon uses only the random internal loopback listener. It does not expand an +empty listener list, honor configured public WebSocket listeners, or read stdin. +The proxy requires a random token before Native/control traffic. Metadata is +published atomically with owner-only Unix permissions. Lock/metadata paths reject +symlink following, non-regular files and wrong owners. Startup checks directory +ancestors for cross-user replacement risk. Discovery accepts only numeric loopback +`ws` endpoints, with no credentials, query, fragment or external hostname. + +The proxy limits concurrent connections to 32, with at most 16 long-lived clients +so controls retain capacity, message/frame size to 16 MiB and +handshake, token and initial request waiting to five seconds. Stop closes admission +and drains existing requests before snapshotting runtime shutdown. SIGTERM and +SIGINT follow the same graceful owner teardown. A blocked drain remains visible as +a stop timeout rather than claiming completion. No permission preset, approval +answer, unattended tool privilege or credential grant is changed. + +**Same-user local processes and agent tools are not isolated by this token.** +The existing sandbox permits access to DEVO_HOME, and same-user code can read the +control token or edit runtime files. This architecture is not a security boundary +against hostile local tools or prompt-injected code. Enforcing that boundary +requires a separate protected control-credential store plus sandbox/IPC capability +work. Do not describe owner-only file permissions as agent-tool isolation. + +## Persistent Native schedule recovery + +Native interval schedules and their outbox survive owner restarts. Startup must +finish rebuilding the session index from durable journals before the wake loop can +decide that a schedule's session is missing. This prevents an empty/rebuilt SQLite +index from deleting valid scheduled work. Missing-index recovery also checks the +durable journals strictly: present, unreadable or damaged journals preserve the +job/outbox for retry instead of interpreting an index failure as deletion. + +The next occurrence keeps its persisted interval phase after delayed wake or +restart. Missed slots coalesce into one delivery rather than a catch-up storm. At +most one undelivered occurrence per job is added to the outbox, so offline, +busy or approval-blocked sessions cannot accumulate an unbounded interval backlog. +The outbox is written before advancing the job; recovery reuses the same occurrence +identity if a crash separates those writes. Pending queue/journal evidence is +reconciled before redelivery. + +These are durable admission and deduplication properties, not exactly-once +external side effects. Interrupted or unknown tool outcomes still require review. +A model/tool call can have reached a remote system before any local completion was +persisted. Approval and question waits must never be automatically approved to +make background execution appear healthy. + +Desktop RRULE automations remain separate Electron-owned jobs. PR #269 hardens +that layer; this Native daemon slice does not silently migrate or double-run those +jobs. Converging RRULE/timezone configuration and Desktop run history onto Native +scheduling is a later explicit migration with one authoritative owner. + +## Relation to Prime and remaining work + +Prime separates resident workers from attaching clients, tracks ownership and +session leases, and persists recovery/scheduling evidence. This change adopts the +independent-owner and explicit-attach lifecycle using Devo's existing singleton +instead of copying Prime's protocol or worker engine. + +The comparison used Prime snapshot `74591cddef28d11249c3171112114fc4755375d7`: +`crates/pa-daemon/src/ownership.rs` separates client ownership and promotion; +`lease.rs` fences canonical session files with owner identity and process-start +evidence; `scheduled_jobs.rs` connects persistent cron/heartbeat storage to queue +hooks and recovery journals. Devo's existing singleton already supplies one owner +per DEVO_HOME, so this change makes that owner's lifetime explicit instead of +adding Prime's per-worker lease/supervisor model. Automatic worker retry, promotion +and multi-process per-session hosting are not implemented by these commands. + +Still required for complete background parity: Windows ACL/service support, +Desktop schedule migration, explicit detach UI and daemon health, reconnectable +approval controllers, protected capabilities against untrusted same-user tools, +operator-approved restart policy, version/upgrade handover, resource quotas/log +retention, and multi-day OS suspend/reboot/fault-injection validation. From b89bc44d9ab1e26380808d7a59cb114b6b7473c0 Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Mon, 9 Nov 2026 01:20:21 +0800 Subject: [PATCH 2/2] fix(server): keep singleton file options warning-free on Windows --- crates/server/src/singleton_files.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/server/src/singleton_files.rs b/crates/server/src/singleton_files.rs index 33f2a56d..3eeffe04 100644 --- a/crates/server/src/singleton_files.rs +++ b/crates/server/src/singleton_files.rs @@ -43,15 +43,19 @@ pub(crate) fn daemon_home(home: &Path) -> Result { } fn options() -> OpenOptions { - let mut options = OpenOptions::new(); #[cfg(unix)] { use std::os::unix::fs::OpenOptionsExt; + let mut options = OpenOptions::new(); options .mode(0o600) .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC); + options + } + #[cfg(not(unix))] + { + OpenOptions::new() } - options } fn validate(file: &File, private: bool) -> Result<()> {