1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
//! Single-watcher lock per (hub, role) on a machine.
//!
//! Two `confer watch` processes for the same role on one machine share the same
//! cursor (`~/.confer/cursor/<hub>/<role>.json`), so a second watcher silently
//! steals events from the first. This happens most often when a Claude Code
//! session is compacted or ends but its background watch keeps running — an
//! orphan that races the new session's watch. The lock makes a live duplicate
//! refuse to start (protection), and lets a fresh watcher reclaim a dead/hung
//! lock automatically or `--replace` a live orphan (cleanup). It's machine-local
//! because the race is machine-local (the cursor lives under `$HOME`).
use crate::config;
use anyhow::{anyhow, Result};
use std::path::{Path, PathBuf};
pub struct WatchLock {
path: PathBuf,
/// When this watcher started (stable across heartbeats) — for `watch-status`.
started_at: String,
/// How this watcher was armed — a self-declared stamp (e.g. `monitor`, `poll`, `background`) so
/// `watch-status` can affirm the watcher actually DELIVERS wake events to the agent vs. just runs.
/// A plain background watcher and a Monitor-hosted one look identical from the outside, but only
/// the latter reaches the AI (Heliosphere field report / design/36). Portable BY DESIGN: whatever
/// harness arms the watch passes its own method string — nothing here is Claude-Code-specific.
delivery: Option<String>,
}
/// A snapshot of a watcher lock, for `watch-status` / healing decisions.
pub struct LockInfo {
pub pid: u32,
pub host: String,
pub version: Option<String>,
pub started_at: Option<String>,
pub age_secs: u64,
/// pid is alive AND on this host.
pub alive: bool,
pub same_host: bool,
/// heartbeat is newer than the stale window.
pub fresh: bool,
/// Self-declared arming method (see [`WatchLock::delivery`]). `None` = not recorded — an older
/// watcher, or one armed without the stamp (possibly a plain background process not delivering).
pub delivery: Option<String>,
}
/// The health of a role's watcher — shared by `watch-status` and `session-heal`
/// so they can never disagree on what "healthy/stale/outdated" means.
#[derive(PartialEq, Eq, Debug, Clone, Copy)]
pub enum WatchState {
Healthy,
NotWatching,
Stale,
Outdated,
OtherHost,
}
/// Classify a lock snapshot against the current binary version.
pub fn classify(info: &Option<LockInfo>, cur_version: &str) -> WatchState {
match info {
None => WatchState::NotWatching,
Some(i) if !i.same_host => WatchState::OtherHost,
Some(i) if !(i.alive && i.fresh) => WatchState::Stale,
Some(i) if i.version.as_deref() != Some(cur_version) => WatchState::Outdated,
Some(_) => WatchState::Healthy,
}
}
/// Read + classify the lock for (hub, role) without touching it. `None` = no lock
/// (not watching). Powers `confer watch-status` and the session-start self-heal.
pub fn inspect(hub: &str, role: &str, stale_secs: u64) -> Option<LockInfo> {
let path = lock_path(hub, role).ok()?;
if !path.exists() {
return None;
}
let v: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&path).ok()?).ok()?;
let pid = v.get("pid")?.as_u64()? as u32;
let host = v.get("host").and_then(|x| x.as_str()).unwrap_or_default().to_string();
let version = v.get("version").and_then(|x| x.as_str()).map(String::from);
let started_at = v.get("started_at").and_then(|x| x.as_str()).map(String::from);
let delivery = v.get("delivery").and_then(|x| x.as_str()).map(String::from);
let same_host = host == config::hostname().unwrap_or_default();
let age = age_secs(&path);
Some(LockInfo {
alive: same_host && process_alive(pid),
same_host,
fresh: age < stale_secs,
age_secs: age,
pid,
host,
version,
started_at,
delivery,
})
}
fn lock_path(hub: &str, role: &str) -> Result<PathBuf> {
let role = if role.is_empty() { "_all" } else { role };
Ok(config::home()?
.join(".confer")
.join("watch")
.join(hub)
.join(format!("{role}.json")))
}
/// Same-host liveness via `kill -0` (signal 0 probes without delivering).
fn process_alive(pid: u32) -> bool {
std::process::Command::new("kill")
.args(["-0", &pid.to_string()])
// Silence "kill: (pid): No such process" on a dead pid — a liveness probe read by
// fleet/where/adopt-clone must not leak stderr for a stale watch pid (a fleet finding).
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
fn read_info(path: &Path) -> Option<(u32, String)> {
let v: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(path).ok()?).ok()?;
Some((v.get("pid")?.as_u64()? as u32, v.get("host")?.as_str()?.to_string()))
}
/// Seconds since the lock file was last refreshed (its heartbeat).
fn age_secs(path: &Path) -> u64 {
std::fs::metadata(path)
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.elapsed().ok())
.map(|d| d.as_secs())
.unwrap_or(u64::MAX)
}
impl WatchLock {
/// Acquire the (hub, role) watch lock. If a *live, fresh* watcher already
/// holds it, refuse — unless `replace`, which kills it and takes over. A
/// dead / hung / other-host lock is reclaimed silently (auto-cleanup).
pub fn acquire(
hub: &str,
role: &str,
stale_secs: u64,
replace: bool,
delivery: Option<String>,
) -> Result<Self> {
let path = lock_path(hub, role)?;
if let Some(p) = path.parent() {
std::fs::create_dir_all(p)?;
}
let label = if role.is_empty() { "<all>" } else { role };
if path.exists() {
let this_host = config::hostname().unwrap_or_default();
let info = read_info(&path);
let same_host = info.as_ref().is_some_and(|(_, h)| *h == this_host);
let pid = info.as_ref().map(|(p, _)| *p);
let alive = same_host && pid.is_some_and(process_alive);
let fresh = age_secs(&path) < stale_secs;
if alive && fresh {
if !replace {
return Err(anyhow!(
"another confer watch for role '{label}' is already running on {this_host} \
(pid {}). It owns the cursor — a second watcher would race it and silently \
drop events. Stop it, or run `confer watch --replace` to take over.",
pid.unwrap_or(0)
));
}
if let Some(p) = pid {
let _ = std::process::Command::new("kill").arg(p.to_string())
.stderr(std::process::Stdio::null()).status();
eprintln!("confer watch: --replace killed the existing watcher (pid {p}) for role '{label}'.");
}
} else {
eprintln!(
"confer watch: reclaimed a stale watch lock for role '{label}' \
(previous watcher not running or unresponsive)."
);
}
}
let lock = WatchLock {
path,
started_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
delivery,
};
lock.write()?;
Ok(lock)
}
fn write(&self) -> Result<()> {
// version + started_at let `watch-status` tell an agent "your watcher is on
// an OLD confer — replace it" (the studio-didn't-adopt-the-fix case). We store
// the build SHA (not CARGO_PKG_VERSION, which never changes between rebuilds)
// so a stale watcher is actually detectable at build granularity.
let info = serde_json::json!({
"pid": std::process::id(),
"host": config::hostname().unwrap_or_default(),
"version": env!("CONFER_GIT_SHA"),
"started_at": self.started_at,
"delivery": self.delivery,
});
std::fs::write(&self.path, serde_json::to_string_pretty(&info)?)?;
Ok(())
}
/// Refresh the lock's mtime so a later watcher can tell we're alive. Call each cycle.
pub fn heartbeat(&self) {
let _ = self.write();
}
}
impl Drop for WatchLock {
fn drop(&mut self) {
// Clean exit removes the lock; a hard kill leaves it, and the next
// watcher reclaims it as stale (dead pid / old heartbeat).
let _ = std::fs::remove_file(&self.path);
}
}