hatchery_node/
session_meta.rs

1//! Node-local session registry fields that HQ also tracks: origin, parent,
2//! gone, last_line. Survives in enroll state; HQ learns gone when the node
3//! omits a session after resume, or when status is exited/failed (HQ marks
4//! tombstone). Origin/parent are set from spawn options when present.
5
6use serde::{Deserialize, Serialize};
7use std::collections::BTreeMap;
8use std::path::{Path, PathBuf};
9use std::sync::Mutex;
10
11#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
12pub struct Meta {
13    #[serde(default, skip_serializing_if = "Option::is_none")]
14    pub origin: Option<String>,
15    #[serde(default, skip_serializing_if = "Option::is_none")]
16    pub parent_session_id: Option<String>,
17    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
18    pub gone: bool,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub last_line: Option<String>,
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub provider: Option<String>,
23    #[serde(default)]
24    pub updated_ms: u64,
25}
26
27static LOCK: Mutex<()> = Mutex::new(());
28
29pub fn path(state_dir: &Path) -> PathBuf {
30    state_dir.join("session-meta.json")
31}
32
33fn load_map(file: &Path) -> BTreeMap<String, Meta> {
34    std::fs::read(file)
35        .ok()
36        .and_then(|b| serde_json::from_slice(&b).ok())
37        .unwrap_or_default()
38}
39
40fn save_map(file: &Path, map: &BTreeMap<String, Meta>) {
41    let tmp = file.with_extension("json.tmp");
42    if let Ok(b) = serde_json::to_vec_pretty(map) {
43        let _ = std::fs::write(&tmp, b).and_then(|_| std::fs::rename(&tmp, file));
44    }
45}
46
47fn key(instance_id: u64) -> String {
48    instance_id.to_string()
49}
50
51fn now_ms() -> u64 {
52    std::time::SystemTime::now()
53        .duration_since(std::time::UNIX_EPOCH)
54        .map(|d| d.as_millis() as u64)
55        .unwrap_or(0)
56}
57
58pub fn upsert(state_dir: &Path, instance_id: u64, patch: Meta) {
59    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
60    let file = path(state_dir);
61    let mut m = load_map(&file);
62    let e = m.entry(key(instance_id)).or_default();
63    if let Some(o) = patch.origin {
64        e.origin = Some(o);
65    }
66    if let Some(p) = patch.parent_session_id {
67        e.parent_session_id = Some(p);
68    }
69    if patch.gone {
70        e.gone = true;
71    }
72    if let Some(l) = patch.last_line {
73        e.last_line = Some(l);
74    }
75    if let Some(p) = patch.provider {
76        e.provider = Some(p);
77    }
78    e.updated_ms = now_ms();
79    save_map(&file, &m);
80}
81
82pub fn mark_gone(state_dir: &Path, instance_id: u64) {
83    upsert(
84        state_dir,
85        instance_id,
86        Meta {
87            gone: true,
88            updated_ms: now_ms(),
89            ..Meta::default()
90        },
91    );
92}
93
94pub fn revive(state_dir: &Path, instance_id: u64) {
95    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
96    let file = path(state_dir);
97    let mut m = load_map(&file);
98    if let Some(e) = m.get_mut(&key(instance_id)) {
99        e.gone = false;
100        e.updated_ms = now_ms();
101        save_map(&file, &m);
102    }
103}
104
105pub fn get(state_dir: &Path, instance_id: u64) -> Option<Meta> {
106    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
107    load_map(&path(state_dir)).remove(&key(instance_id))
108}
109
110/// Instance ids the node has marked gone (still in the journal for HQ sync).
111pub fn gone_ids(state_dir: &Path) -> Vec<u64> {
112    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
113    load_map(&path(state_dir))
114        .into_iter()
115        .filter(|(_, v)| v.gone)
116        .filter_map(|(k, _)| k.parse().ok())
117        .collect()
118}
119
120pub fn set_last_line(state_dir: &Path, instance_id: u64, line: &str) {
121    let line = line.trim();
122    if line.is_empty() {
123        return;
124    }
125    let clipped: String = line.chars().take(160).collect();
126    upsert(
127        state_dir,
128        instance_id,
129        Meta {
130            last_line: Some(clipped),
131            ..Meta::default()
132        },
133    );
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use std::time::{SystemTime, UNIX_EPOCH};
140
141    #[test]
142    fn gone_and_origin_roundtrip() {
143        let dir = std::env::temp_dir().join(format!(
144            "session-meta-test-{}",
145            SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
146        ));
147        let _ = std::fs::create_dir_all(&dir);
148        upsert(
149            &dir,
150            7,
151            Meta {
152                origin: Some("human".into()),
153                parent_session_id: Some("claude#3".into()),
154                provider: Some("claude".into()),
155                ..Meta::default()
156            },
157        );
158        let m = get(&dir, 7).unwrap();
159        assert_eq!(m.origin.as_deref(), Some("human"));
160        assert_eq!(m.parent_session_id.as_deref(), Some("claude#3"));
161        mark_gone(&dir, 7);
162        assert!(get(&dir, 7).unwrap().gone);
163        assert!(gone_ids(&dir).contains(&7));
164        revive(&dir, 7);
165        assert!(!get(&dir, 7).unwrap().gone);
166        let _ = std::fs::remove_dir_all(&dir);
167    }
168}
hatchery_node/session_meta.rs

hatchery_node/
session_meta.rs
