hatchery_node/
kit.rs

1//! Standard node kit cores: dig2browser (`d2b`), mail4agent (`m4a`), the four
2//! session-restore crates (`session-restore`) and kernel WireGuard
3//! (`wireguard`). Each core is compiled only with its own feature; the default
4//! build (`kit`) enables all of them.
5
6use std::sync::OnceLock;
7
8/// Names of the cores linked into this build, in owner order.
9pub fn linked_core_names() -> &'static [&'static str] {
10    static NAMES: OnceLock<Vec<&'static str>> = OnceLock::new();
11    NAMES.get_or_init(|| {
12        let mut names: Vec<&'static str> = Vec::new();
13        #[cfg(feature = "d2b")]
14        names.push("dig2browser");
15        #[cfg(feature = "m4a")]
16        names.push("mail4agent");
17        #[cfg(feature = "session-restore")]
18        names.push("session-restore");
19        #[cfg(feature = "wireguard")]
20        names.push("wireguard");
21        names
22    })
23}
24
25/// Touches a real symbol in each enabled core so the node binary links them.
26/// Does not start a browser, a provider CLI, or a tunnel, and does not read
27/// a secret. The mailbox is started separately by [`start_mailbox`].
28pub fn force_link() -> usize {
29    let mut acc: usize = 0;
30    #[cfg(feature = "d2b")]
31    {
32        acc = acc.wrapping_add(dig2browser::browser_stream::LOCAL_PAGE_HTML.len());
33    }
34    #[cfg(feature = "m4a")]
35    {
36        let mail: fn() -> Result<(), mail4agent::MainError> = mail4agent::run;
37        acc = acc.wrapping_add(mail as usize);
38    }
39    #[cfg(feature = "session-restore")]
40    {
41        let claude = std::mem::size_of::<claude_session_restore::transcript::SessionEvent>();
42        let codex = codex_session_restore::MAX_HEAD_BYTES;
43        let grok: fn() = grok_session_restore::main;
44        let kimi: fn() = kimi_session_restore::main;
45        acc = acc
46            .wrapping_add(claude)
47            .wrapping_add(codex)
48            .wrapping_add(grok as usize)
49            .wrapping_add(kimi as usize);
50    }
51    #[cfg(feature = "wireguard")]
52    {
53        let wireguard: fn(
54            &crate::kit_wireguard::NodeWgPeerConfig,
55        ) -> Result<std::net::IpAddr, crate::kit_wireguard::NodeWgPeerError> =
56            crate::kit_wireguard::bring_up_node_wireguard;
57        acc = acc
58            .wrapping_add(wireguard as usize)
59            .wrapping_add(crate::kit_wireguard::LINK_TAG.len());
60    }
61    acc
62}
63
64/// Start the mail4agent mailbox on a background thread (`m4a` only).
65/// Same daemon the standalone `mail4agent` binary runs.
66#[cfg(feature = "m4a")]
67///
68/// The daemon's own failure (for example a bootstrap refusal) is logged and
69/// visible through [`MailboxHandle::is_finished`]; it is never swallowed.
70pub fn start_mailbox() -> Result<MailboxHandle, mail4agent::MainError> {
71    let join = std::thread::Builder::new()
72        .name("mail4agent".to_owned())
73        .spawn(|| {
74            let result = mail4agent::run();
75            if let Err(error) = &result {
76                tracing::error!(%error, "mail4agent mailbox stopped");
77            }
78            result
79        })
80        .map_err(mail4agent::MainError::Runtime)?;
81    Ok(MailboxHandle { join })
82}
83
84/// Env var mail4agent reads for its bind address (also set by `--m4a-bind`).
85pub const COURIER_BIND_ENV: &str = "MAIL4AGENT_BIND";
86/// mail4agent's default bind.
87pub const COURIER_DEFAULT_BIND: &str = "127.0.0.1:18301";
88
89/// What the node does about the courier port at start.
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub enum CourierPlan {
92    /// The port is free (or ephemeral): run our own mailbox there.
93    Own(std::net::SocketAddr),
94    /// A mail4agent courier already answers there: use it, start none.
95    Attach { addr: std::net::SocketAddr, version: Option<String> },
96    /// Something else holds the port: run without a mailbox.
97    Foreign { addr: std::net::SocketAddr, detail: String },
98}
99
100/// The configured courier address (`MAIL4AGENT_BIND`, else the default).
101pub fn courier_bind() -> Result<std::net::SocketAddr, String> {
102    let raw = std::env::var(COURIER_BIND_ENV).unwrap_or_else(|_| COURIER_DEFAULT_BIND.to_owned());
103    raw.parse().map_err(|error| format!("{COURIER_BIND_ENV}={raw:?} is invalid: {error}"))
104}
105
106/// Oldest courier version the node treats as current (the linked mail4agent).
107pub const COURIER_MIN_VERSION: &str = "0.4.17";
108
109/// Decide [`CourierPlan`] for `addr` via [`crate::local_infra::probe`].
110pub fn plan_courier(addr: std::net::SocketAddr, probe_timeout: std::time::Duration) -> CourierPlan {
111    match crate::local_infra::probe(addr, "mail4agent", probe_timeout) {
112        crate::local_infra::Probe::Free => CourierPlan::Own(addr),
113        crate::local_infra::Probe::Ours(info) => CourierPlan::Attach { addr, version: info.version },
114        crate::local_infra::Probe::Foreign(detail) => CourierPlan::Foreign { addr, detail },
115    }
116}
117
118#[cfg(feature = "m4a")]
119/// Start our mailbox, attach to a running courier, or run without one.
120/// Never exits the process. Returns the handle when our own mailbox runs.
121pub fn start_or_attach_mailbox() -> Option<MailboxHandle> {
122    let addr = match courier_bind() {
123        Ok(addr) => addr,
124        Err(error) => {
125            tracing::error!(%error, "mail4agent courier address; running without a mailbox");
126            return None;
127        }
128    };
129    match plan_courier(addr, std::time::Duration::from_secs(2)) {
130        CourierPlan::Attach { addr, version } => {
131            tracing::info!(%addr, version = version.as_deref().unwrap_or("?"), "mail4agent courier already running; attached, not starting our own");
132            let _ = version;
133            let crate::local_infra::Probe::Ours(info) = crate::local_infra::probe(addr, "mail4agent", std::time::Duration::from_secs(2)) else {
134                return None;
135            };
136            match crate::local_infra::upgrade_plan(&info, COURIER_MIN_VERSION, false) {
137                crate::local_infra::Upgrade::NotNeeded => None,
138                crate::local_infra::Upgrade::ManualOnly => {
139                    tracing::warn!(%addr, minimum = COURIER_MIN_VERSION, "running courier is older and has no control channel; kept as is");
140                    None
141                }
142                crate::local_infra::Upgrade::Handoff if !crate::local_infra::auto_upgrade_allowed() => {
143                    tracing::warn!(%addr, "running courier is older; HATCHERY_LOCAL_UPGRADE=off, kept as is");
144                    None
145                }
146                crate::local_infra::Upgrade::Handoff => {
147                    match crate::local_infra::handoff(addr, &info, 300, std::time::Duration::from_secs(15)) {
148                        Ok(()) => {
149                            tracing::info!(%addr, "older courier handed over; starting ours");
150                            start_own_mailbox(addr)
151                        }
152                        Err(error) => {
153                            tracing::warn!(%addr, %error, "courier handoff failed; keeping the running one");
154                            None
155                        }
156                    }
157                }
158            }
159        }
160        CourierPlan::Foreign { addr, detail } => {
161            tracing::warn!(%addr, %detail, "courier port held by something that is not mail4agent; running without a mailbox (set --m4a-bind)");
162            None
163        }
164        CourierPlan::Own(addr) => start_own_mailbox(addr),
165    }
166}
167
168#[cfg(feature = "m4a")]
169fn start_own_mailbox(addr: std::net::SocketAddr) -> Option<MailboxHandle> {
170    {
171        {
172            let mailbox = match start_mailbox() {
173                Ok(mailbox) => mailbox,
174                Err(error) => {
175                    tracing::error!(%error, "mail4agent mailbox did not start; node continues without it");
176                    return None;
177                }
178            };
179            std::thread::sleep(std::time::Duration::from_millis(750));
180            if mailbox.is_finished() {
181                let error = mailbox.join().err().map_or_else(|| "mailbox exited".to_owned(), |e| e.to_string());
182                tracing::error!(%addr, %error, "mail4agent mailbox stopped at start; node continues without it");
183                return None;
184            }
185            tracing::info!(%addr, "node kit mailbox started");
186            supervise_mailbox(addr, mailbox);
187            None
188        }
189    }
190}
191
192/// The node supervises its own courier: if the mailbox thread ends (crash,
193/// or a handoff to a successor), restart it with backoff once the port is
194/// free again; if someone else now serves a newer courier there, attach.
195#[cfg(feature = "m4a")]
196fn supervise_mailbox(addr: std::net::SocketAddr, first: MailboxHandle) {
197    let _ = std::thread::Builder::new().name("mail4agent-supervisor".to_owned()).spawn(move || {
198        let mut current = first;
199        let mut backoff = std::time::Duration::from_secs(1);
200        loop {
201            let result = current.join();
202            tracing::warn!(%addr, ?result, "node-owned mailbox ended; supervising");
203            loop {
204                std::thread::sleep(backoff);
205                backoff = (backoff * 2).min(std::time::Duration::from_secs(30));
206                match plan_courier(addr, std::time::Duration::from_secs(2)) {
207                    CourierPlan::Attach { version, .. } => {
208                        tracing::info!(%addr, version = version.as_deref().unwrap_or("?"), "a courier took over the port; attached, supervisor stops");
209                        return;
210                    }
211                    CourierPlan::Foreign { detail, .. } => {
212                        tracing::warn!(%addr, %detail, "courier port held by something else; retrying");
213                    }
214                    CourierPlan::Own(_) => match start_mailbox() {
215                        Ok(next) => {
216                            std::thread::sleep(std::time::Duration::from_millis(750));
217                            if next.is_finished() {
218                                let _ = next.join();
219                                continue;
220                            }
221                            tracing::info!(%addr, "node-owned mailbox restarted");
222                            backoff = std::time::Duration::from_secs(1);
223                            current = next;
224                            break;
225                        }
226                        Err(error) => tracing::error!(%error, "mailbox restart failed"),
227                    },
228                }
229            }
230        }
231    });
232}
233
234/// Handle of the background mailbox thread.
235#[cfg(feature = "m4a")]
236pub struct MailboxHandle {
237    join: std::thread::JoinHandle<Result<(), mail4agent::MainError>>,
238}
239
240#[cfg(feature = "m4a")]
241impl MailboxHandle {
242    /// True once the mailbox thread has returned (it only returns on failure).
243    pub fn is_finished(&self) -> bool {
244        self.join.is_finished()
245    }
246
247    /// Block until the mailbox thread ends and return its result.
248    pub fn join(self) -> Result<(), mail4agent::MainError> {
249        self.join.join().unwrap_or_else(|_| {
250            Err(mail4agent::MainError::Runtime(std::io::Error::other("mail4agent thread panicked")))
251        })
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[cfg(all(feature = "d2b", feature = "m4a", feature = "session-restore", feature = "wireguard"))]
260    #[test]
261    fn kit_names_the_four_cores() {
262        assert_eq!(
263            linked_core_names(),
264            ["dig2browser", "mail4agent", "session-restore", "wireguard"]
265        );
266        assert!(force_link() > 0);
267    }
268
269    #[test]
270    fn linked_names_match_enabled_features() {
271        let expected = cfg!(feature = "d2b") as usize
272            + cfg!(feature = "m4a") as usize
273            + cfg!(feature = "session-restore") as usize
274            + cfg!(feature = "wireguard") as usize;
275        assert_eq!(linked_core_names().len(), expected);
276    }
277
278    #[cfg(feature = "m4a")]
279    #[test]
280    fn kit_exposes_mailbox_automaton_entry() {
281        let start: fn() -> Result<MailboxHandle, mail4agent::MainError> = start_mailbox;
282        assert!(start as usize != 0);
283    }
284
285    fn serve_once(body: &'static str, status: &'static str) -> std::net::SocketAddr {
286        use std::io::{Read, Write};
287        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
288        let addr = listener.local_addr().unwrap();
289        std::thread::spawn(move || {
290            if let Ok((mut s, _)) = listener.accept() {
291                let mut buf = [0u8; 1024];
292                let _ = s.read(&mut buf);
293                let _ = write!(s, "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len());
294            }
295        });
296        addr
297    }
298
299    #[test]
300    fn courier_free_port_is_own() {
301        let probe = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
302        let addr = probe.local_addr().unwrap();
303        drop(probe);
304        assert_eq!(plan_courier(addr, std::time::Duration::from_secs(1)), CourierPlan::Own(addr));
305    }
306
307    #[test]
308    fn courier_port_zero_is_own() {
309        let addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
310        assert_eq!(plan_courier(addr, std::time::Duration::from_secs(1)), CourierPlan::Own(addr));
311    }
312
313    #[test]
314    fn courier_held_by_mail4agent_attaches() {
315        let addr = serve_once(r#"{"ok":true,"service":"mail4agent","version":"9.9.9"}"#, "200 OK");
316        assert_eq!(
317            plan_courier(addr, std::time::Duration::from_secs(2)),
318            CourierPlan::Attach { addr, version: Some("9.9.9".to_owned()) }
319        );
320    }
321
322    #[test]
323    fn courier_held_by_other_service_is_foreign() {
324        let addr = serve_once(r#"{"ok":true,"service":"something-else"}"#, "200 OK");
325        assert!(matches!(plan_courier(addr, std::time::Duration::from_secs(2)), CourierPlan::Foreign { .. }));
326    }
327
328    #[test]
329    fn courier_held_by_silent_socket_is_foreign() {
330        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
331        let addr = listener.local_addr().unwrap();
332        assert!(matches!(plan_courier(addr, std::time::Duration::from_millis(300)), CourierPlan::Foreign { .. }));
333        drop(listener);
334    }
335}
hatchery_node/kit.rs

hatchery_node/
kit.rs
