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
100static COURIER_BIND_ARG: std::sync::OnceLock<String> = std::sync::OnceLock::new();
101
102/// The node's `--m4a-bind` argument. It wins over the environment, so a supervisor or service
103/// that does not pass env (WMI, schtasks) still gets the right port; it is also exported to
104/// the in-process mail4agent, which reads [`COURIER_BIND_ENV`].
105pub fn set_courier_bind_arg(value: &str) {
106    let _ = COURIER_BIND_ARG.set(value.to_owned());
107    std::env::set_var(COURIER_BIND_ENV, value);
108}
109
110/// The configured courier address: `--m4a-bind`, else `MAIL4AGENT_BIND`, else the default.
111pub fn courier_bind() -> Result<std::net::SocketAddr, String> {
112    let raw = COURIER_BIND_ARG
113        .get()
114        .cloned()
115        .or_else(|| std::env::var(COURIER_BIND_ENV).ok().filter(|v| !v.is_empty()))
116        .unwrap_or_else(|| COURIER_DEFAULT_BIND.to_owned());
117    if let Some(arg) = COURIER_BIND_ARG.get() {
118        // Keep the in-process mail4agent on the same address as the argument.
119        std::env::set_var(COURIER_BIND_ENV, arg);
120    }
121    raw.parse().map_err(|error| format!("courier bind {raw:?} is invalid: {error}"))
122}
123
124/// Oldest courier version the node treats as current (the linked mail4agent).
125pub const COURIER_MIN_VERSION: &str = "0.4.17";
126
127/// Decide [`CourierPlan`] for `addr` via [`crate::local_infra::probe`].
128pub fn plan_courier(addr: std::net::SocketAddr, probe_timeout: std::time::Duration) -> CourierPlan {
129    match crate::local_infra::probe(addr, "mail4agent", probe_timeout) {
130        crate::local_infra::Probe::Free => CourierPlan::Own(addr),
131        crate::local_infra::Probe::Ours(info) => CourierPlan::Attach { addr, version: info.version },
132        crate::local_infra::Probe::Foreign(detail) => CourierPlan::Foreign { addr, detail },
133    }
134}
135
136#[cfg(feature = "m4a")]
137/// Start our mailbox, attach to a running courier, or run without one.
138/// Never exits the process. Returns the handle when our own mailbox runs.
139pub fn start_or_attach_mailbox() -> Option<MailboxHandle> {
140    let addr = match courier_bind() {
141        Ok(addr) => addr,
142        Err(error) => {
143            tracing::error!(%error, "mail4agent courier address; running without a mailbox");
144            return None;
145        }
146    };
147    match plan_courier(addr, std::time::Duration::from_secs(2)) {
148        CourierPlan::Attach { addr, version } => {
149            tracing::info!(%addr, version = version.as_deref().unwrap_or("?"), "mail4agent courier already running; attached, not starting our own");
150            let _ = version;
151            let crate::local_infra::Probe::Ours(info) = crate::local_infra::probe(addr, "mail4agent", std::time::Duration::from_secs(2)) else {
152                return None;
153            };
154            match crate::local_infra::upgrade_plan(&info, COURIER_MIN_VERSION, false) {
155                crate::local_infra::Upgrade::NotNeeded => None,
156                crate::local_infra::Upgrade::ManualOnly => {
157                    tracing::warn!(%addr, minimum = COURIER_MIN_VERSION, "running courier is older and has no control channel; kept as is");
158                    None
159                }
160                crate::local_infra::Upgrade::Handoff if !crate::local_infra::auto_upgrade_allowed() => {
161                    tracing::warn!(%addr, "running courier is older; HATCHERY_LOCAL_UPGRADE=off, kept as is");
162                    None
163                }
164                crate::local_infra::Upgrade::Handoff => {
165                    match crate::local_infra::handoff(addr, &info, 300, std::time::Duration::from_secs(15)) {
166                        Ok(()) => {
167                            tracing::info!(%addr, "older courier handed over; starting ours");
168                            start_own_mailbox(addr)
169                        }
170                        Err(error) => {
171                            tracing::warn!(%addr, %error, "courier handoff failed; keeping the running one");
172                            None
173                        }
174                    }
175                }
176            }
177        }
178        CourierPlan::Foreign { addr, detail } => {
179            tracing::warn!(%addr, %detail, "courier port held by something that is not mail4agent; running without a mailbox (set --m4a-bind)");
180            None
181        }
182        CourierPlan::Own(addr) => start_own_mailbox(addr),
183    }
184}
185
186#[cfg(feature = "m4a")]
187fn start_own_mailbox(addr: std::net::SocketAddr) -> Option<MailboxHandle> {
188    {
189        {
190            let mailbox = match start_mailbox() {
191                Ok(mailbox) => mailbox,
192                Err(error) => {
193                    tracing::error!(%error, "mail4agent mailbox did not start; node continues without it");
194                    return None;
195                }
196            };
197            std::thread::sleep(std::time::Duration::from_millis(750));
198            if mailbox.is_finished() {
199                let error = mailbox.join().err().map_or_else(|| "mailbox exited".to_owned(), |e| e.to_string());
200                tracing::error!(%addr, %error, "mail4agent mailbox stopped at start; node continues without it");
201                return None;
202            }
203            tracing::info!(%addr, "node kit mailbox started");
204            supervise_mailbox(addr, mailbox);
205            None
206        }
207    }
208}
209
210/// The node supervises its own courier: if the mailbox thread ends (crash,
211/// or a handoff to a successor), restart it with backoff once the port is
212/// free again; if someone else now serves a newer courier there, attach.
213#[cfg(feature = "m4a")]
214fn supervise_mailbox(addr: std::net::SocketAddr, first: MailboxHandle) {
215    let _ = std::thread::Builder::new().name("mail4agent-supervisor".to_owned()).spawn(move || {
216        let mut current = first;
217        let mut backoff = std::time::Duration::from_secs(1);
218        loop {
219            let result = current.join();
220            tracing::warn!(%addr, ?result, "node-owned mailbox ended; supervising");
221            loop {
222                std::thread::sleep(backoff);
223                backoff = (backoff * 2).min(std::time::Duration::from_secs(30));
224                match plan_courier(addr, std::time::Duration::from_secs(2)) {
225                    CourierPlan::Attach { version, .. } => {
226                        tracing::info!(%addr, version = version.as_deref().unwrap_or("?"), "a courier took over the port; attached, supervisor stops");
227                        return;
228                    }
229                    CourierPlan::Foreign { detail, .. } => {
230                        tracing::warn!(%addr, %detail, "courier port held by something else; retrying");
231                    }
232                    CourierPlan::Own(_) => match start_mailbox() {
233                        Ok(next) => {
234                            std::thread::sleep(std::time::Duration::from_millis(750));
235                            if next.is_finished() {
236                                let _ = next.join();
237                                continue;
238                            }
239                            tracing::info!(%addr, "node-owned mailbox restarted");
240                            backoff = std::time::Duration::from_secs(1);
241                            current = next;
242                            break;
243                        }
244                        Err(error) => tracing::error!(%error, "mailbox restart failed"),
245                    },
246                }
247            }
248        }
249    });
250}
251
252/// Handle of the background mailbox thread.
253#[cfg(feature = "m4a")]
254pub struct MailboxHandle {
255    join: std::thread::JoinHandle<Result<(), mail4agent::MainError>>,
256}
257
258#[cfg(feature = "m4a")]
259impl MailboxHandle {
260    /// True once the mailbox thread has returned (it only returns on failure).
261    pub fn is_finished(&self) -> bool {
262        self.join.is_finished()
263    }
264
265    /// Block until the mailbox thread ends and return its result.
266    pub fn join(self) -> Result<(), mail4agent::MainError> {
267        self.join.join().unwrap_or_else(|_| {
268            Err(mail4agent::MainError::Runtime(std::io::Error::other("mail4agent thread panicked")))
269        })
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276
277    #[cfg(all(feature = "d2b", feature = "m4a", feature = "session-restore", feature = "wireguard"))]
278    #[test]
279    fn kit_names_the_four_cores() {
280        assert_eq!(
281            linked_core_names(),
282            ["dig2browser", "mail4agent", "session-restore", "wireguard"]
283        );
284        assert!(force_link() > 0);
285    }
286
287    #[test]
288    fn linked_names_match_enabled_features() {
289        let expected = cfg!(feature = "d2b") as usize
290            + cfg!(feature = "m4a") as usize
291            + cfg!(feature = "session-restore") as usize
292            + cfg!(feature = "wireguard") as usize;
293        assert_eq!(linked_core_names().len(), expected);
294    }
295
296    #[cfg(feature = "m4a")]
297    #[test]
298    fn kit_exposes_mailbox_automaton_entry() {
299        let start: fn() -> Result<MailboxHandle, mail4agent::MainError> = start_mailbox;
300        assert!(start as usize != 0);
301    }
302
303    fn serve_once(body: &'static str, status: &'static str) -> std::net::SocketAddr {
304        use std::io::{Read, Write};
305        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
306        let addr = listener.local_addr().unwrap();
307        std::thread::spawn(move || {
308            if let Ok((mut s, _)) = listener.accept() {
309                let mut buf = [0u8; 1024];
310                let _ = s.read(&mut buf);
311                let _ = write!(s, "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len());
312            }
313        });
314        addr
315    }
316
317    #[test]
318    fn courier_free_port_is_own() {
319        let probe = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
320        let addr = probe.local_addr().unwrap();
321        drop(probe);
322        assert_eq!(plan_courier(addr, std::time::Duration::from_secs(1)), CourierPlan::Own(addr));
323    }
324
325    #[test]
326    fn courier_port_zero_is_own() {
327        let addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
328        assert_eq!(plan_courier(addr, std::time::Duration::from_secs(1)), CourierPlan::Own(addr));
329    }
330
331    #[test]
332    fn courier_held_by_mail4agent_attaches() {
333        let addr = serve_once(r#"{"ok":true,"service":"mail4agent","version":"9.9.9"}"#, "200 OK");
334        assert_eq!(
335            plan_courier(addr, std::time::Duration::from_secs(2)),
336            CourierPlan::Attach { addr, version: Some("9.9.9".to_owned()) }
337        );
338    }
339
340    #[test]
341    fn courier_held_by_other_service_is_foreign() {
342        let addr = serve_once(r#"{"ok":true,"service":"something-else"}"#, "200 OK");
343        assert!(matches!(plan_courier(addr, std::time::Duration::from_secs(2)), CourierPlan::Foreign { .. }));
344    }
345
346    #[test]
347    fn courier_held_by_silent_socket_is_foreign() {
348        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
349        let addr = listener.local_addr().unwrap();
350        assert!(matches!(plan_courier(addr, std::time::Duration::from_millis(300)), CourierPlan::Foreign { .. }));
351        drop(listener);
352    }
353}
hatchery_node/kit.rs

hatchery_node/
kit.rs
