glass/
daemon.rs

1//! Local Unix-socket daemon lifecycle and client bridge.
2//!
3//! The daemon is deliberately local-only. Each socket client receives an
4//! isolated MCP child session, so a daemon restart or client disconnect cannot
5//! silently transfer a browser session or workflow lease to another client.
6
7use clap::Parser;
8use futures_util::FutureExt;
9use serde::{Deserialize, Serialize};
10use std::collections::BTreeMap;
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13use std::time::Duration;
14
15use crate::cli::args::Cli;
16
17/// Version of the local daemon status and lifecycle contract.
18pub const DAEMON_PROTOCOL_VERSION: u32 = 1;
19/// Version of the persisted interrupted-run recovery record.
20pub const DAEMON_RECOVERY_SCHEMA_VERSION: u32 = 1;
21/// Maximum number of isolated MCP child sessions per daemon.
22pub const MAX_DAEMON_CLIENT_SESSIONS: u32 = 4;
23/// Maximum number of in-flight MCP operations across all daemon clients.
24pub const MAX_DAEMON_CONCURRENT_REQUESTS: usize = 16;
25/// Maximum number of in-flight MCP operations from one daemon client.
26pub const MAX_DAEMON_CLIENT_CONCURRENT_REQUESTS: usize = 4;
27/// Maximum number of workflow requests retained in the daemon status record.
28pub const MAX_DAEMON_ACTIVE_RUNS: usize = 16;
29/// Prefix for the per-client session namespaces owned by the daemon.
30pub const DAEMON_SESSION_ID_PREFIX: &str = "daemon-session-";
31const MIN_LEASE_TTL_MS: u64 = 100;
32const MAX_LEASE_TTL_MS: u64 = 15 * 60 * 1_000;
33
34/// A single mutation lease held by one local client owner.
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(rename_all = "camelCase", deny_unknown_fields)]
37pub struct MutationLease {
38    pub session_id: String,
39    pub owner_id: String,
40    pub token: String,
41    pub expires_at_ms: u64,
42}
43
44/// Typed lease failure used by daemon/session integrations.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub enum LeaseError {
47    InvalidInput(String),
48    AlreadyHeld,
49    NotFound,
50    NotOwner,
51    Expired,
52}
53
54impl std::fmt::Display for LeaseError {
55    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        match self {
57            Self::InvalidInput(detail) => write!(formatter, "invalid lease input: {detail}"),
58            Self::AlreadyHeld => formatter.write_str("mutation lease is already held"),
59            Self::NotFound => formatter.write_str("mutation lease was not found"),
60            Self::NotOwner => formatter.write_str("mutation lease is held by another owner"),
61            Self::Expired => formatter.write_str("mutation lease has expired"),
62        }
63    }
64}
65
66impl std::error::Error for LeaseError {}
67
68/// In-memory lease authority for one daemon session namespace.
69#[derive(Debug, Default)]
70pub struct MutationLeaseManager {
71    leases: BTreeMap<String, MutationLease>,
72    next_token: u64,
73}
74
75/// Lease authority and owner identity for one daemon socket connection.
76#[derive(Debug, Clone)]
77pub struct DaemonLeaseContext {
78    pub manager: Arc<tokio::sync::Mutex<MutationLeaseManager>>,
79    pub session_id: String,
80    pub owner_id: String,
81    pub request_permits: Arc<tokio::sync::Semaphore>,
82    pub client_request_permits: Arc<tokio::sync::Semaphore>,
83    pub status: Arc<DaemonStatusState>,
84}
85
86impl MutationLeaseManager {
87    /// Acquire the only mutation lease for a session.
88    pub fn acquire(
89        &mut self,
90        session_id: &str,
91        owner_id: &str,
92        now_ms: u64,
93        ttl_ms: u64,
94    ) -> Result<MutationLease, LeaseError> {
95        validate_lease_identity(session_id, "sessionId")?;
96        validate_lease_identity(owner_id, "ownerId")?;
97        if !(MIN_LEASE_TTL_MS..=MAX_LEASE_TTL_MS).contains(&ttl_ms) {
98            return Err(LeaseError::InvalidInput(format!(
99                "ttlMs must be {MIN_LEASE_TTL_MS}..={MAX_LEASE_TTL_MS}"
100            )));
101        }
102        if let Some(existing) = self.leases.get(session_id)
103            && existing.expires_at_ms > now_ms
104        {
105            return Err(LeaseError::AlreadyHeld);
106        }
107        self.next_token = self.next_token.saturating_add(1);
108        let lease = MutationLease {
109            session_id: session_id.into(),
110            owner_id: owner_id.into(),
111            token: format!("lease-{}", self.next_token),
112            expires_at_ms: now_ms.saturating_add(ttl_ms),
113        };
114        self.leases.insert(session_id.into(), lease.clone());
115        Ok(lease)
116    }
117
118    /// Renew a lease without allowing a different owner to take it over.
119    pub fn renew(
120        &mut self,
121        session_id: &str,
122        owner_id: &str,
123        token: &str,
124        now_ms: u64,
125        ttl_ms: u64,
126    ) -> Result<MutationLease, LeaseError> {
127        if !(MIN_LEASE_TTL_MS..=MAX_LEASE_TTL_MS).contains(&ttl_ms) {
128            return Err(LeaseError::InvalidInput(format!(
129                "ttlMs must be {MIN_LEASE_TTL_MS}..={MAX_LEASE_TTL_MS}"
130            )));
131        }
132        let lease = self
133            .leases
134            .get_mut(session_id)
135            .ok_or(LeaseError::NotFound)?;
136        if lease.owner_id != owner_id || lease.token != token {
137            return Err(LeaseError::NotOwner);
138        }
139        if lease.expires_at_ms <= now_ms {
140            return Err(LeaseError::Expired);
141        }
142        lease.expires_at_ms = now_ms.saturating_add(ttl_ms);
143        Ok(lease.clone())
144    }
145
146    /// Release a lease only when the owner and token match.
147    pub fn release(
148        &mut self,
149        session_id: &str,
150        owner_id: &str,
151        token: &str,
152    ) -> Result<(), LeaseError> {
153        let lease = self.leases.get(session_id).ok_or(LeaseError::NotFound)?;
154        if lease.owner_id != owner_id || lease.token != token {
155            return Err(LeaseError::NotOwner);
156        }
157        self.leases.remove(session_id);
158        Ok(())
159    }
160
161    /// Verify that a mutation request is still owned by the given client.
162    pub fn validate(
163        &self,
164        session_id: &str,
165        owner_id: &str,
166        token: &str,
167        now_ms: u64,
168    ) -> Result<(), LeaseError> {
169        let lease = self.leases.get(session_id).ok_or(LeaseError::NotFound)?;
170        if lease.owner_id != owner_id || lease.token != token {
171            return Err(LeaseError::NotOwner);
172        }
173        if lease.expires_at_ms <= now_ms {
174            return Err(LeaseError::Expired);
175        }
176        Ok(())
177    }
178
179    /// Release every lease held by a disconnected client owner.
180    pub fn release_owner(&mut self, owner_id: &str) {
181        self.leases.retain(|_, lease| lease.owner_id != owner_id);
182    }
183
184    /// Return the current owner for a session without exposing its token.
185    pub fn current_owner(&self, session_id: &str, now_ms: u64) -> Option<String> {
186        self.leases
187            .get(session_id)
188            .filter(|lease| lease.expires_at_ms > now_ms)
189            .map(|lease| lease.owner_id.clone())
190    }
191}
192
193fn validate_lease_identity(value: &str, field: &str) -> Result<(), LeaseError> {
194    if value.is_empty() || value.len() > 128 || value.chars().any(char::is_whitespace) {
195        return Err(LeaseError::InvalidInput(format!(
196            "{field} must be a bounded non-whitespace identifier"
197        )));
198    }
199    Ok(())
200}
201
202/// One workflow request active in one daemon client session.
203#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
204#[serde(rename_all = "camelCase", deny_unknown_fields)]
205pub struct DaemonActiveRun {
206    pub request_id: String,
207    pub owner_id: String,
208    pub started_at: String,
209}
210
211/// Persisted evidence that workflow requests need checkpoint reconciliation.
212#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
213#[serde(rename_all = "camelCase", deny_unknown_fields)]
214pub struct DaemonRecovery {
215    pub schema_version: u32,
216    pub state: String,
217    pub recovered_at: String,
218    pub runs: Vec<DaemonActiveRun>,
219}
220
221/// Stable daemon status written beside the Unix socket.
222#[derive(Debug, Clone, Serialize, Deserialize)]
223#[serde(rename_all = "camelCase", deny_unknown_fields)]
224pub struct DaemonStatus {
225    pub protocol_version: u32,
226    pub state: String,
227    pub pid: u32,
228    pub socket: PathBuf,
229    pub status_path: PathBuf,
230    #[serde(default, skip_serializing_if = "Option::is_none")]
231    pub log_path: Option<PathBuf>,
232    #[serde(default, skip_serializing_if = "Option::is_none")]
233    pub recovery_path: Option<PathBuf>,
234    pub started_at: String,
235    pub transport: String,
236    pub client_sessions: u32,
237    #[serde(default, skip_serializing_if = "Option::is_none")]
238    pub mutation_lease_owner: Option<String>,
239    #[serde(default)]
240    pub active_runs: Vec<DaemonActiveRun>,
241}
242
243/// Serialized status coordinator shared by daemon clients.
244#[derive(Debug)]
245pub struct DaemonStatusState {
246    path: PathBuf,
247    status: tokio::sync::Mutex<DaemonStatus>,
248}
249
250impl DaemonStatusState {
251    pub fn new(path: &Path, status: DaemonStatus) -> Self {
252        Self {
253            path: path.to_path_buf(),
254            status: tokio::sync::Mutex::new(status),
255        }
256    }
257
258    pub async fn update_client_sessions(
259        &self,
260        client_sessions: u32,
261    ) -> Result<(), Box<dyn std::error::Error>> {
262        let mut status = self.status.lock().await;
263        status.client_sessions = client_sessions;
264        write_status(&self.path, &status)
265    }
266
267    pub async fn update_mutation_lease_owner(
268        &self,
269        owner: Option<String>,
270    ) -> Result<(), Box<dyn std::error::Error>> {
271        let mut status = self.status.lock().await;
272        status.mutation_lease_owner = owner;
273        write_status(&self.path, &status)
274    }
275
276    pub async fn begin_workflow(
277        &self,
278        request_id: &str,
279        owner_id: &str,
280    ) -> Result<(), Box<dyn std::error::Error>> {
281        if request_id.is_empty() || request_id.len() > 128 {
282            return Err("workflow request id exceeds the daemon status bound".into());
283        }
284        let mut status = self.status.lock().await;
285        if status.active_runs.len() >= MAX_DAEMON_ACTIVE_RUNS {
286            return Err("daemon active workflow limit reached".into());
287        }
288        if status
289            .active_runs
290            .iter()
291            .any(|run| run.request_id == request_id)
292        {
293            return Err("daemon workflow request id is already active".into());
294        }
295        status.active_runs.push(DaemonActiveRun {
296            request_id: request_id.into(),
297            owner_id: owner_id.into(),
298            started_at: chrono::Utc::now().to_rfc3339(),
299        });
300        write_status(&self.path, &status)
301    }
302
303    pub async fn finish_workflow(
304        &self,
305        request_id: &str,
306        owner_id: &str,
307    ) -> Result<(), Box<dyn std::error::Error>> {
308        let mut status = self.status.lock().await;
309        status
310            .active_runs
311            .retain(|run| run.request_id != request_id || run.owner_id != owner_id);
312        write_status(&self.path, &status)
313    }
314
315    pub async fn record_interrupted_workflows(&self) -> Result<usize, Box<dyn std::error::Error>> {
316        let status = self.status.lock().await;
317        write_recovery_report(&self.path, &status.active_runs)?;
318        for run in &status.active_runs {
319            append_daemon_log(
320                &self.path,
321                &format!(
322                    "interrupted workflow request {} owned by {}; checkpoint reconciliation is required",
323                    run.request_id, run.owner_id
324                ),
325            )?;
326        }
327        Ok(status.active_runs.len())
328    }
329}
330
331/// Return the default local daemon paths.
332pub fn default_paths() -> (PathBuf, PathBuf) {
333    let root = dirs::data_local_dir()
334        .unwrap_or_else(|| PathBuf::from("."))
335        .join("glass");
336    (root.join("glass.sock"), root.join("daemon.json"))
337}
338
339fn log_path_for(status_path: &Path) -> PathBuf {
340    status_path.with_extension("log")
341}
342
343fn recovery_path_for(status_path: &Path) -> PathBuf {
344    status_path.with_extension("recovery.json")
345}
346
347fn append_daemon_log(status_path: &Path, message: &str) -> Result<(), std::io::Error> {
348    use std::io::Write;
349
350    let path = log_path_for(status_path);
351    let mut log = std::fs::OpenOptions::new()
352        .create(true)
353        .append(true)
354        .open(path)?;
355    writeln!(log, "{message}")
356}
357
358fn write_recovery_report(
359    status_path: &Path,
360    runs: &[DaemonActiveRun],
361) -> Result<(), Box<dyn std::error::Error>> {
362    if runs.is_empty() {
363        return Ok(());
364    }
365    let report = DaemonRecovery {
366        schema_version: DAEMON_RECOVERY_SCHEMA_VERSION,
367        state: "reconciliation_required".into(),
368        recovered_at: chrono::Utc::now().to_rfc3339(),
369        runs: runs.to_vec(),
370    };
371    std::fs::write(
372        recovery_path_for(status_path),
373        serde_json::to_vec_pretty(&report)?,
374    )?;
375    Ok(())
376}
377
378fn read_recovery(status_path: &Path) -> Result<Option<DaemonRecovery>, Box<dyn std::error::Error>> {
379    let path = recovery_path_for(status_path);
380    if !path.is_file() {
381        return Ok(None);
382    }
383    let report: DaemonRecovery = serde_json::from_slice(&std::fs::read(path)?)?;
384    if report.schema_version != DAEMON_RECOVERY_SCHEMA_VERSION {
385        return Err("unsupported daemon recovery schema".into());
386    }
387    Ok(Some(report))
388}
389
390/// Start one background daemon and return its status contract.
391pub async fn start(
392    socket: Option<&Path>,
393    status_path: Option<&Path>,
394) -> Result<DaemonStatus, Box<dyn std::error::Error>> {
395    #[cfg(not(unix))]
396    {
397        let _ = (socket, status_path);
398        return Err("the local daemon supports Linux and macOS only".into());
399    }
400    #[cfg(unix)]
401    {
402        let (default_socket, default_status) = default_paths();
403        let socket = socket.unwrap_or(&default_socket);
404        let status_path = status_path.unwrap_or(&default_status);
405        if let Some(existing) = read_status(status_path)? {
406            if process_is_alive(existing.pid) {
407                return Err(format!("daemon is already running as pid {}", existing.pid).into());
408            }
409            write_recovery_report(status_path, &existing.active_runs)?;
410            append_daemon_log(
411                status_path,
412                &format!(
413                    "recovered stale daemon pid {}; active workflows are indeterminate and require checkpoint reconciliation",
414                    existing.pid
415                ),
416            )?;
417            for run in &existing.active_runs {
418                append_daemon_log(
419                    status_path,
420                    &format!(
421                        "recovered interrupted workflow request {} owned by {}; checkpoint reconciliation is required",
422                        run.request_id, run.owner_id
423                    ),
424                )?;
425            }
426            let _ = std::fs::remove_file(status_path);
427            let _ = remove_socket_if_safe(&existing.socket);
428        }
429        remove_socket_if_safe(socket)?;
430        std::fs::create_dir_all(socket.parent().unwrap_or_else(|| Path::new(".")))?;
431        std::fs::create_dir_all(status_path.parent().unwrap_or_else(|| Path::new(".")))?;
432        let log_path = log_path_for(status_path);
433        let log = std::fs::OpenOptions::new()
434            .create(true)
435            .append(true)
436            .open(&log_path)?;
437        let log_stderr = log.try_clone()?;
438        let executable = std::env::current_exe()?;
439        std::process::Command::new(executable)
440            .args(["daemon", "serve"])
441            .arg("--socket")
442            .arg(socket)
443            .arg("--status")
444            .arg(status_path)
445            .stdin(std::process::Stdio::null())
446            .stdout(std::process::Stdio::from(log))
447            .stderr(std::process::Stdio::from(log_stderr))
448            .spawn()?;
449        let deadline = std::time::Instant::now() + Duration::from_secs(2);
450        loop {
451            if let Some(status) = read_status(status_path)?
452                && status.state == "running"
453                && process_is_alive(status.pid)
454            {
455                return Ok(status);
456            }
457            if std::time::Instant::now() >= deadline {
458                return Err("daemon did not become ready within two seconds".into());
459            }
460            tokio::time::sleep(Duration::from_millis(25)).await;
461        }
462    }
463}
464
465/// Read and validate the current daemon status.
466pub fn status(
467    socket: Option<&Path>,
468    status_path: Option<&Path>,
469) -> Result<DaemonStatus, Box<dyn std::error::Error>> {
470    let (default_socket, default_status) = default_paths();
471    let socket = socket.unwrap_or(&default_socket);
472    let status_path = status_path.unwrap_or(&default_status);
473    let Some(mut status) = read_status(status_path)? else {
474        return Err("daemon is not running".into());
475    };
476    if status.socket != socket {
477        return Err("daemon status socket does not match the requested socket".into());
478    }
479    if !process_is_alive(status.pid) {
480        status.state = "stopped".into();
481    }
482    Ok(status)
483}
484
485/// Stop a daemon by its recorded PID and remove its local status artifacts.
486pub fn stop(
487    socket: Option<&Path>,
488    status_path: Option<&Path>,
489) -> Result<(), Box<dyn std::error::Error>> {
490    #[cfg(not(unix))]
491    {
492        let _ = (socket, status_path);
493        return Err("the local daemon supports Linux and macOS only".into());
494    }
495    #[cfg(unix)]
496    {
497        let (default_socket, default_status) = default_paths();
498        let socket = socket.unwrap_or(&default_socket);
499        let status_path = status_path.unwrap_or(&default_status);
500        let Some(status) = read_status(status_path)? else {
501            return Err("daemon is not running".into());
502        };
503        if status.socket != socket {
504            return Err("daemon status socket does not match the requested socket".into());
505        }
506        if process_is_alive(status.pid) {
507            let result = unsafe { libc::kill(status.pid as libc::pid_t, libc::SIGTERM) };
508            if result != 0 {
509                return Err(std::io::Error::last_os_error().into());
510            }
511        }
512        let _ = std::fs::remove_file(status_path);
513        let _ = remove_socket_if_safe(socket);
514        Ok(())
515    }
516}
517
518/// Validate the daemon process, status file, and socket without connecting.
519pub fn doctor(
520    socket: Option<&Path>,
521    status_path: Option<&Path>,
522) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
523    match status(socket, status_path) {
524        Ok(status) => Ok(serde_json::json!({
525            "status": "healthy",
526            "daemon": status,
527            "recovery": read_recovery(&status.status_path)?,
528            "socketExists": status.socket.exists(),
529            "pidAlive": process_is_alive(status.pid),
530        })),
531        Err(error) => Ok(serde_json::json!({
532            "status": "unavailable",
533            "detail": error.to_string(),
534        })),
535    }
536}
537
538/// Read the persisted interrupted-workflow recovery state.
539pub fn recovery(
540    status_path: Option<&Path>,
541) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
542    let (_, default_status) = default_paths();
543    let status_path = status_path.unwrap_or(&default_status);
544    let path = recovery_path_for(status_path);
545    match read_recovery(status_path)? {
546        Some(report) => Ok(serde_json::json!({
547            "status": "reconciliation_required",
548            "path": path,
549            "recovery": report,
550        })),
551        None => Ok(serde_json::json!({
552            "status": "clear",
553            "path": path,
554        })),
555    }
556}
557
558/// Clear the recovery requirement after the caller reconciles every listed run.
559pub fn acknowledge_recovery(
560    status_path: Option<&Path>,
561    reconciled_request_ids: &[String],
562) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
563    let (_, default_status) = default_paths();
564    let status_path = status_path.unwrap_or(&default_status);
565    let path = recovery_path_for(status_path);
566    let Some(report) = read_recovery(status_path)? else {
567        return Ok(serde_json::json!({
568            "status": "clear",
569            "path": path,
570            "acknowledged": false,
571        }));
572    };
573    let expected: std::collections::BTreeSet<&str> = report
574        .runs
575        .iter()
576        .map(|run| run.request_id.as_str())
577        .collect();
578    let provided: std::collections::BTreeSet<&str> =
579        reconciled_request_ids.iter().map(String::as_str).collect();
580    if reconciled_request_ids.len() != provided.len() || expected != provided {
581        return Err(format!(
582            "recovery acknowledgement must name exactly these request IDs: {}",
583            report
584                .runs
585                .iter()
586                .map(|run| run.request_id.as_str())
587                .collect::<Vec<_>>()
588                .join(", ")
589        )
590        .into());
591    }
592    std::fs::remove_file(&path)?;
593    Ok(serde_json::json!({
594        "status": "acknowledged",
595        "path": path,
596        "runs": report.runs.len(),
597        "reconciledRequestIds": report
598            .runs
599            .iter()
600            .map(|run| run.request_id.as_str())
601            .collect::<Vec<_>>(),
602        "acknowledged": true,
603    }))
604}
605
606/// Read a bounded tail of the daemon log without exposing an unbounded file.
607pub fn logs(status_path: Option<&Path>) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
608    const MAX_LOG_BYTES: usize = 64 * 1024;
609    let (_, default_status) = default_paths();
610    let status_path = status_path.unwrap_or(&default_status);
611    let path = log_path_for(status_path);
612    if !path.is_file() {
613        return Ok(serde_json::json!({
614            "status": "unavailable",
615            "path": path,
616            "content": "",
617            "truncated": false,
618        }));
619    }
620    let bytes = std::fs::read(&path)?;
621    let truncated = bytes.len() > MAX_LOG_BYTES;
622    let start = bytes.len().saturating_sub(MAX_LOG_BYTES);
623    Ok(serde_json::json!({
624        "status": "available",
625        "path": path,
626        "content": String::from_utf8_lossy(&bytes[start..]),
627        "truncated": truncated,
628    }))
629}
630
631/// Run the foreground Unix-socket server used by `daemon start`.
632pub async fn serve(socket: &Path, status_path: &Path) -> Result<(), Box<dyn std::error::Error>> {
633    #[cfg(not(unix))]
634    {
635        let _ = (socket, status_path);
636        return Err("the local daemon supports Linux and macOS only".into());
637    }
638    #[cfg(unix)]
639    {
640        let local = tokio::task::LocalSet::new();
641        local.run_until(serve_local(socket, status_path)).await
642    }
643}
644
645#[cfg(unix)]
646async fn serve_local(socket: &Path, status_path: &Path) -> Result<(), Box<dyn std::error::Error>> {
647    use std::os::unix::fs::PermissionsExt;
648    use tokio::net::UnixListener;
649    use tokio::signal::unix::{SignalKind, signal};
650
651    remove_socket_if_safe(socket)?;
652    std::fs::create_dir_all(socket.parent().unwrap_or_else(|| Path::new(".")))?;
653    std::fs::create_dir_all(status_path.parent().unwrap_or_else(|| Path::new(".")))?;
654    let listener = UnixListener::bind(socket)?;
655    std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o600))?;
656    let status = DaemonStatus {
657        protocol_version: DAEMON_PROTOCOL_VERSION,
658        state: "running".into(),
659        pid: std::process::id(),
660        socket: socket.to_path_buf(),
661        status_path: status_path.to_path_buf(),
662        log_path: Some(log_path_for(status_path)),
663        recovery_path: Some(recovery_path_for(status_path)),
664        started_at: chrono::Utc::now().to_rfc3339(),
665        transport: "unix-mcp-isolated-sessions".into(),
666        client_sessions: 0,
667        mutation_lease_owner: None,
668        active_runs: Vec::new(),
669    };
670    std::fs::write(status_path, serde_json::to_vec_pretty(&status)?)?;
671    let status_state = Arc::new(DaemonStatusState::new(status_path, status));
672    let client_sessions = Arc::new(std::sync::atomic::AtomicU32::new(0));
673    let lease_manager = Arc::new(tokio::sync::Mutex::new(MutationLeaseManager::default()));
674    let request_permits = Arc::new(tokio::sync::Semaphore::new(MAX_DAEMON_CONCURRENT_REQUESTS));
675    let next_owner_id = std::sync::atomic::AtomicU64::new(0);
676    let mut terminate = signal(SignalKind::terminate())?;
677    let mut clients = Vec::new();
678    loop {
679        let (stream, _) = tokio::select! {
680            result = listener.accept() => result?,
681            _ = terminate.recv() => break,
682        };
683        if let Err(error) = authorize_local_peer(&stream) {
684            tracing::warn!(%error, "daemon rejected unauthorized local peer");
685            drop(stream);
686            continue;
687        }
688        let client_sessions = Arc::clone(&client_sessions);
689        if client_sessions.load(std::sync::atomic::Ordering::Relaxed) >= MAX_DAEMON_CLIENT_SESSIONS
690        {
691            tracing::warn!("daemon client session limit reached");
692            drop(stream);
693            continue;
694        }
695        let active = client_sessions.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
696        status_state.update_client_sessions(active).await?;
697        let status_state = Arc::clone(&status_state);
698        let cli = Cli::parse_from(["glass", "--mcp"]);
699        let client_number = next_owner_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
700        let owner_id = format!("daemon-client-{client_number}");
701        let session_id = format!("{DAEMON_SESSION_ID_PREFIX}{client_number}");
702        let lease_context = Arc::new(DaemonLeaseContext {
703            manager: Arc::clone(&lease_manager),
704            session_id,
705            owner_id,
706            request_permits: Arc::clone(&request_permits),
707            client_request_permits: Arc::new(tokio::sync::Semaphore::new(
708                MAX_DAEMON_CLIENT_CONCURRENT_REQUESTS,
709            )),
710            status: Arc::clone(&status_state),
711        });
712        clients.push(tokio::task::spawn_local(async move {
713            let (socket_read, socket_write) = stream.into_split();
714            let result = std::panic::AssertUnwindSafe(crate::mcp::server::run_mcp_stream(
715                tokio::io::BufReader::new(socket_read),
716                socket_write,
717                &cli,
718                Arc::new(tokio::sync::Mutex::new(None)),
719                false,
720                true,
721                Some(lease_context),
722            ))
723            .catch_unwind()
724            .await;
725            match result {
726                Ok(Ok(())) => {}
727                Ok(Err(error)) => tracing::warn!(%error, "daemon client bridge stopped"),
728                Err(panic) => tracing::error!(?panic, "daemon client bridge panicked"),
729            }
730            let active = client_sessions
731                .fetch_sub(1, std::sync::atomic::Ordering::Relaxed)
732                .saturating_sub(1);
733            let _ = status_state.update_client_sessions(active).await;
734        }));
735    }
736    let _ = status_state.record_interrupted_workflows().await;
737    for client in clients {
738        client.abort();
739    }
740    let _ = std::fs::remove_file(status_path);
741    let _ = remove_socket_if_safe(socket);
742    Ok(())
743}
744
745#[cfg(unix)]
746fn authorize_local_peer(stream: &tokio::net::UnixStream) -> Result<(), Box<dyn std::error::Error>> {
747    #[cfg(target_os = "linux")]
748    {
749        use std::mem::size_of;
750        use std::os::fd::AsRawFd;
751
752        let mut credentials = libc::ucred {
753            pid: 0,
754            uid: 0,
755            gid: 0,
756        };
757        let mut length = size_of::<libc::ucred>() as libc::socklen_t;
758        let result = unsafe {
759            libc::getsockopt(
760                stream.as_raw_fd(),
761                libc::SOL_SOCKET,
762                libc::SO_PEERCRED,
763                (&mut credentials as *mut libc::ucred).cast(),
764                &mut length,
765            )
766        };
767        if result != 0 {
768            return Err(std::io::Error::last_os_error().into());
769        }
770        if length as usize != size_of::<libc::ucred>() {
771            return Err("unexpected Unix peer credential size".into());
772        }
773        let current_uid = unsafe { libc::geteuid() };
774        if credentials.uid != current_uid {
775            return Err(format!(
776                "peer uid {} does not match daemon uid {}",
777                credentials.uid, current_uid
778            )
779            .into());
780        }
781    }
782    #[cfg(not(target_os = "linux"))]
783    {
784        let _ = stream;
785    }
786    Ok(())
787}
788
789fn read_status(path: &Path) -> Result<Option<DaemonStatus>, Box<dyn std::error::Error>> {
790    if !path.is_file() {
791        return Ok(None);
792    }
793    let status: DaemonStatus = serde_json::from_slice(&std::fs::read(path)?)?;
794    if status.protocol_version != DAEMON_PROTOCOL_VERSION {
795        return Err("unsupported daemon status protocol".into());
796    }
797    Ok(Some(status))
798}
799
800#[cfg(unix)]
801fn remove_socket_if_safe(path: &Path) -> Result<(), Box<dyn std::error::Error>> {
802    use std::os::unix::fs::FileTypeExt;
803
804    let metadata = match std::fs::symlink_metadata(path) {
805        Ok(metadata) => metadata,
806        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
807        Err(error) => return Err(error.into()),
808    };
809    if !metadata.file_type().is_socket() {
810        return Err(format!(
811            "refusing to replace non-socket daemon path {}",
812            path.display()
813        )
814        .into());
815    }
816    std::fs::remove_file(path)?;
817    Ok(())
818}
819
820fn write_status(path: &Path, status: &DaemonStatus) -> Result<(), Box<dyn std::error::Error>> {
821    std::fs::write(path, serde_json::to_vec_pretty(status)?)?;
822    Ok(())
823}
824
825#[cfg(unix)]
826fn process_is_alive(pid: u32) -> bool {
827    unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
828}
829
830#[cfg(not(unix))]
831fn process_is_alive(_pid: u32) -> bool {
832    false
833}
834
835#[cfg(test)]
836mod tests {
837    use super::*;
838
839    #[test]
840    fn daemon_status_serialization_is_versioned_and_local_only() {
841        let status = DaemonStatus {
842            protocol_version: DAEMON_PROTOCOL_VERSION,
843            state: "running".into(),
844            pid: 42,
845            socket: PathBuf::from("/tmp/glass.sock"),
846            status_path: PathBuf::from("/tmp/glass.json"),
847            log_path: None,
848            recovery_path: None,
849            started_at: "2026-07-28T00:00:00Z".into(),
850            transport: "unix-mcp-isolated-sessions".into(),
851            client_sessions: 0,
852            mutation_lease_owner: None,
853            active_runs: Vec::new(),
854        };
855        let value = serde_json::to_value(&status).unwrap();
856
857        assert_eq!(value["protocolVersion"], 1);
858        assert_eq!(value["transport"], "unix-mcp-isolated-sessions");
859        assert!(value["socket"].as_str().unwrap().starts_with('/'));
860    }
861
862    #[test]
863    fn mutation_leases_are_exclusive_and_owner_bound() {
864        let mut manager = MutationLeaseManager::default();
865        let lease = manager
866            .acquire("session-1", "owner-a", 1_000, 1_000)
867            .unwrap();
868        assert_eq!(lease.token, "lease-1");
869        assert_eq!(
870            manager.acquire("session-1", "owner-b", 1_100, 1_000),
871            Err(LeaseError::AlreadyHeld)
872        );
873        assert_eq!(
874            manager.renew("session-1", "owner-b", &lease.token, 1_200, 1_000),
875            Err(LeaseError::NotOwner)
876        );
877        manager
878            .renew("session-1", "owner-a", &lease.token, 1_200, 1_000)
879            .unwrap();
880        manager
881            .release("session-1", "owner-a", &lease.token)
882            .unwrap();
883        assert_eq!(
884            manager.release("session-1", "owner-a", &lease.token),
885            Err(LeaseError::NotFound)
886        );
887
888        let lease = manager
889            .acquire("session-1", "owner-a", 2_000, 1_000)
890            .unwrap();
891        manager
892            .validate("session-1", "owner-a", &lease.token, 2_500)
893            .unwrap();
894        manager.release_owner("owner-a");
895        assert_eq!(
896            manager.validate("session-1", "owner-a", &lease.token, 2_500),
897            Err(LeaseError::NotFound)
898        );
899    }
900
901    #[test]
902    fn independent_client_sessions_have_independent_mutation_leases() {
903        let mut manager = MutationLeaseManager::default();
904        let first = manager
905            .acquire("daemon-session-1", "daemon-client-1", 1_000, 1_000)
906            .unwrap();
907        let second = manager
908            .acquire("daemon-session-2", "daemon-client-2", 1_000, 1_000)
909            .unwrap();
910
911        assert_ne!(first.session_id, second.session_id);
912        assert_eq!(
913            manager.current_owner("daemon-session-1", 1_100).as_deref(),
914            Some("daemon-client-1")
915        );
916        assert_eq!(
917            manager.current_owner("daemon-session-2", 1_100).as_deref(),
918            Some("daemon-client-2")
919        );
920    }
921
922    #[tokio::test]
923    async fn per_client_request_budget_is_bounded() {
924        let permits = Arc::new(tokio::sync::Semaphore::new(
925            MAX_DAEMON_CLIENT_CONCURRENT_REQUESTS,
926        ));
927        let mut held = Vec::new();
928        for _ in 0..MAX_DAEMON_CLIENT_CONCURRENT_REQUESTS {
929            held.push(Arc::clone(&permits).try_acquire_owned().unwrap());
930        }
931        assert!(Arc::clone(&permits).try_acquire_owned().is_err());
932        drop(held.pop());
933        assert!(Arc::clone(&permits).try_acquire_owned().is_ok());
934    }
935
936    #[cfg(unix)]
937    #[test]
938    fn daemon_cleanup_refuses_to_replace_a_regular_file() {
939        let path =
940            std::env::temp_dir().join(format!("glass-daemon-regular-{}", std::process::id()));
941        std::fs::write(&path, b"not a socket").unwrap();
942        assert!(remove_socket_if_safe(&path).is_err());
943        std::fs::remove_file(path).unwrap();
944    }
945
946    #[cfg(unix)]
947    #[tokio::test]
948    async fn local_peer_authorization_accepts_same_user_socket() {
949        let (stream, _peer) = tokio::net::UnixStream::pair().unwrap();
950        authorize_local_peer(&stream).unwrap();
951    }
952
953    #[tokio::test]
954    async fn active_workflow_status_is_added_and_removed_atomically() {
955        let root = std::env::temp_dir().join(format!(
956            "glass-daemon-active-run-{}-{}",
957            std::process::id(),
958            chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default()
959        ));
960        std::fs::create_dir_all(&root).unwrap();
961        let status_path = root.join("daemon.json");
962        let status = DaemonStatus {
963            protocol_version: DAEMON_PROTOCOL_VERSION,
964            state: "running".into(),
965            pid: 42,
966            socket: root.join("glass.sock"),
967            status_path: status_path.clone(),
968            log_path: None,
969            recovery_path: None,
970            started_at: "2026-07-28T00:00:00Z".into(),
971            transport: "unix-mcp-isolated-sessions".into(),
972            client_sessions: 0,
973            mutation_lease_owner: None,
974            active_runs: Vec::new(),
975        };
976        write_status(&status_path, &status).unwrap();
977        let state = DaemonStatusState::new(&status_path, status);
978
979        state
980            .begin_workflow("workflow-1", "daemon-client-1")
981            .await
982            .unwrap();
983        let active: DaemonStatus =
984            serde_json::from_slice(&std::fs::read(&status_path).unwrap()).unwrap();
985        assert_eq!(active.active_runs[0].request_id, "workflow-1");
986        assert_eq!(state.record_interrupted_workflows().await.unwrap(), 1);
987        let recovery_log = std::fs::read_to_string(log_path_for(&status_path)).unwrap();
988        assert!(recovery_log.contains("workflow-1"));
989
990        state
991            .finish_workflow("workflow-1", "daemon-client-1")
992            .await
993            .unwrap();
994        let finished: DaemonStatus =
995            serde_json::from_slice(&std::fs::read(&status_path).unwrap()).unwrap();
996        assert!(finished.active_runs.is_empty());
997
998        std::fs::remove_file(status_path).unwrap();
999        std::fs::remove_file(log_path_for(&root.join("daemon.json"))).unwrap();
1000        std::fs::remove_file(recovery_path_for(&root.join("daemon.json"))).unwrap();
1001        std::fs::remove_dir(root).unwrap();
1002    }
1003
1004    #[cfg(unix)]
1005    #[tokio::test]
1006    async fn stale_status_cleanup_refuses_a_regular_recorded_socket() {
1007        let root = std::env::temp_dir().join(format!("glass-daemon-stale-{}", std::process::id()));
1008        std::fs::create_dir_all(&root).unwrap();
1009        let socket = root.join("glass.sock");
1010        let status_path = root.join("daemon.json");
1011        std::fs::write(&socket, b"not a socket").unwrap();
1012        let status = DaemonStatus {
1013            protocol_version: DAEMON_PROTOCOL_VERSION,
1014            state: "running".into(),
1015            pid: std::process::id().saturating_add(1_000_000),
1016            socket: socket.clone(),
1017            status_path: status_path.clone(),
1018            log_path: None,
1019            recovery_path: None,
1020            started_at: "2026-07-28T00:00:00Z".into(),
1021            transport: "unix-mcp-isolated-sessions".into(),
1022            client_sessions: 0,
1023            mutation_lease_owner: None,
1024            active_runs: Vec::new(),
1025        };
1026        std::fs::write(&status_path, serde_json::to_vec(&status).unwrap()).unwrap();
1027
1028        let result = start(Some(&socket), Some(&status_path)).await;
1029
1030        assert!(result.is_err());
1031        assert!(socket.is_file());
1032        assert!(!status_path.exists());
1033        std::fs::remove_file(socket).unwrap();
1034        std::fs::remove_file(log_path_for(&status_path)).unwrap();
1035        std::fs::remove_dir(root).unwrap();
1036    }
1037
1038    #[test]
1039    fn daemon_logs_return_only_a_bounded_tail() {
1040        let root = std::env::temp_dir().join(format!("glass-daemon-logs-{}", std::process::id()));
1041        std::fs::create_dir_all(&root).unwrap();
1042        let status_path = root.join("daemon.json");
1043        let log_path = log_path_for(&status_path);
1044        std::fs::write(&log_path, vec![b'x'; 70_000]).unwrap();
1045
1046        let value = logs(Some(&status_path)).unwrap();
1047
1048        assert_eq!(value["status"], "available");
1049        assert_eq!(value["truncated"], true);
1050        assert_eq!(value["content"].as_str().unwrap().len(), 64 * 1024);
1051        std::fs::remove_file(log_path).unwrap();
1052        std::fs::remove_dir(root).unwrap();
1053    }
1054}
glass/daemon.rs

glass/
daemon.rs
