agent_core/core/
session_lock.rs

1//! Per-session journal ownership lock.
2//!
3//! Whoever holds a `Runtime` for session X holds an exclusive advisory
4//! `flock` on `<sessions_dir>/<id>.lock`. Two runtimes for the same
5//! journal are never allowed — the second caller gets an actionable error
6//! pointing at the holder's pid.
7//!
8//! The lock is released on drop (or when the process dies — `flock`
9//! semantics). Stale lock files from dead processes never block.
10
11use std::fs::OpenOptions;
12use std::io;
13use std::path::{Path, PathBuf};
14
15/// Body written into the lock file so the error message can name the holder.
16#[derive(Debug, Clone)]
17pub struct LockHolder {
18    pub pid: u32,
19    pub kind: String, // "daemon", "tui", "chat", …
20}
21
22impl LockHolder {
23    fn serialize(&self) -> String {
24        format!("{}\n{}\n", self.pid, self.kind)
25    }
26
27    fn deserialize(s: &str) -> Option<Self> {
28        let mut lines = s.lines();
29        let pid: u32 = lines.next()?.parse().ok()?;
30        let kind = lines.next().unwrap_or("unknown").to_string();
31        Some(Self { pid, kind })
32    }
33}
34
35/// Held for the runtime's lifetime; dropping releases the `flock`, even
36/// while a forked child still holds an inherited descriptor.
37#[derive(Debug)]
38pub struct SessionLock {
39    _lock: crate::core::lock_file::ExclusiveLock,
40    path: PathBuf,
41}
42
43impl SessionLock {
44    /// Release the lock AND remove the lock file. For a session that never
45    /// journaled (ended `Idle`, F18) the `.lock` next to no `.json` is pure
46    /// litter; a session WITH a journal keeps its file (harmless, reaped on
47    /// the next acquire). Best effort: a failed unlink is not an error.
48    pub fn release_and_remove(self) {
49        let path = self.path.clone();
50        drop(self);
51        let _ = std::fs::remove_file(path);
52    }
53
54    /// Try to acquire an exclusive `flock` on `<dir>/<id>.lock`.
55    ///
56    /// Returns `Ok(lock)` on success; `Err` with an actionable message if
57    /// another process holds it (or on I/O error).
58    pub fn try_acquire(dir: &Path, id: &str, holder: LockHolder) -> Result<Self, SessionLockError> {
59        // F24: refuse to lock a session that was compacted into a successor.
60        if let Ok(Some(successor)) = crate::session::read_compacted_into_in_dir(dir, id) {
61            return Err(SessionLockError::CompactedInto {
62                session_id: id.to_string(),
63                successor_id: successor,
64            });
65        }
66        let path = dir.join(format!("{}.lock", id));
67        std::fs::create_dir_all(dir).map_err(|e| SessionLockError::Io(e, path.clone()))?;
68        let file = OpenOptions::new()
69            .create(true)
70            .truncate(false)
71            .read(true)
72            .write(true)
73            .open(&path)
74            .map_err(|e| SessionLockError::Io(e, path.clone()))?;
75        #[cfg(unix)]
76        {
77            use std::os::unix::fs::PermissionsExt;
78            let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
79        }
80        if let Some(lock) =
81            crate::core::lock_file::try_lock(file).map_err(|e| SessionLockError::Io(e, path.clone()))?
82        {
83            // We hold it — write our identity.
84            use std::io::Write;
85            let _ = lock.file().set_len(0);
86            let _ = lock.file().write_all(holder.serialize().as_bytes());
87            Ok(Self { _lock: lock, path })
88        } else {
89            // Someone else holds it — read their identity.
90            let body = std::fs::read_to_string(&path).unwrap_or_default();
91            let existing = LockHolder::deserialize(&body);
92            Err(SessionLockError::Held {
93                session_id: id.to_string(),
94                holder: existing,
95            })
96        }
97    }
98
99    /// Path of the lock file (for diagnostics).
100    pub fn path(&self) -> &Path {
101        &self.path
102    }
103}
104
105/// Lock-file path for a session (for cleanup).
106pub fn lock_path(dir: &Path, id: &str) -> PathBuf {
107    dir.join(format!("{}.lock", id))
108}
109
110/// The sessions directory (delegates to config).
111pub fn sessions_dir() -> PathBuf {
112    crate::config::get_active_config_dir().join("sessions")
113}
114
115#[derive(Debug)]
116pub enum SessionLockError {
117    Io(io::Error, PathBuf),
118    Held {
119        session_id: String,
120        holder: Option<LockHolder>,
121    },
122    /// The session was compacted into a successor — locking the predecessor
123    /// would fork pre-compaction history (F24).
124    CompactedInto {
125        session_id: String,
126        successor_id: String,
127    },
128}
129
130impl std::fmt::Display for SessionLockError {
131    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        match self {
133            Self::Io(e, path) => write!(f, "session lock I/O error on {}: {}", path.display(), e),
134            Self::Held {
135                session_id,
136                holder: Some(h),
137            } => write!(
138                f,
139                "session {} is live in another process (pid {}, {}) \
140                 — use `synaps --attach {}`, or `synaps daemon sessions`",
141                session_id, h.pid, h.kind, session_id
142            ),
143            Self::Held {
144                session_id,
145                holder: None,
146            } => write!(
147                f,
148                "session {} is live in another process \
149                 — use `synaps --attach {}`, or `synaps daemon sessions`",
150                session_id, session_id
151            ),
152            Self::CompactedInto {
153                session_id,
154                successor_id,
155            } => write!(
156                f,
157                "session {} was compacted into {} \
158                 — continue {} instead",
159                session_id, successor_id, successor_id
160            ),
161        }
162    }
163}
164
165impl std::error::Error for SessionLockError {}
166
167// Dropping a `SessionLock` unlocks explicitly, so these tests can run beside
168// concurrent process spawns (see `fork_tests`).
169#[cfg(test)]
170mod tests {
171    use super::*;
172
173    fn holder(kind: &str) -> LockHolder {
174        LockHolder {
175            pid: std::process::id(),
176            kind: kind.to_string(),
177        }
178    }
179
180    #[test]
181    fn take_and_refuse() {
182        let dir = tempfile::tempdir().unwrap();
183        let lock = SessionLock::try_acquire(dir.path(), "abc123", holder("tui")).unwrap();
184        // Second acquire must fail.
185        let err = SessionLock::try_acquire(dir.path(), "abc123", holder("tui")).unwrap_err();
186        let msg = err.to_string();
187        assert!(msg.contains("abc123"), "error names the session: {msg}");
188        assert!(
189            msg.contains(&format!("pid {}", std::process::id())),
190            "error names the pid: {msg}"
191        );
192        // Drop releases.
193        drop(lock);
194        SessionLock::try_acquire(dir.path(), "abc123", holder("tui")).unwrap();
195    }
196
197    #[test]
198    fn different_sessions_independent() {
199        let dir = tempfile::tempdir().unwrap();
200        let _a = SessionLock::try_acquire(dir.path(), "aaa", holder("tui")).unwrap();
201        let _b = SessionLock::try_acquire(dir.path(), "bbb", holder("tui")).unwrap();
202    }
203
204    #[test]
205    fn stale_lock_from_dead_process_does_not_block() {
206        // flock is released when the fd is closed (process death does that).
207        // We simulate by writing a lock file body without holding the flock.
208        let dir = tempfile::tempdir().unwrap();
209        let path = dir.path().join("stale.lock");
210        std::fs::write(&path, "99999\ndaemon\n").unwrap();
211        // Must succeed — no flock held.
212        let _lock = SessionLock::try_acquire(dir.path(), "stale", holder("tui")).unwrap();
213    }
214}
215
216#[cfg(all(test, unix))]
217mod fork_tests {
218    use super::*;
219
220    /// A runtime that drops its session releases the journal at once, even
221    /// while a child that another thread forked (a tool, an MCP server)
222    /// still holds a copy of the lock's descriptor.
223    #[test]
224    #[serial_test::serial(process_spawn)]
225    fn a_dropped_lock_is_free_while_a_forked_child_holds_its_descriptor() {
226        let dir = tempfile::tempdir().unwrap();
227        let holder = |kind: &str| LockHolder { pid: std::process::id(), kind: kind.to_string() };
228        let first = SessionLock::try_acquire(dir.path(), "s", holder("daemon")).unwrap();
229        let child = crate::core::lock_file::ParkedChild::park();
230        drop(first);
231        let second = SessionLock::try_acquire(dir.path(), "s", holder("tui"));
232        child.release();
233        if let Err(e) = second {
234            panic!("dropped session lock still held via a forked child's copy: {e}");
235        }
236    }
237}
agent_core/core/session_lock.rs

agent_core/core/
session_lock.rs
