Skip to main content

capsule_runtime/
formation_launch.rs

1//! Launching a formation: every member of an admitted roster as a `mur run` subprocess of its own,
2//! peers first and the entry member last, and every member stopped once the entry member's process
3//! has ended.
4//!
5//! **The launcher supervises; it runs no session.** Each member, the entry member included, is a
6//! `mur run --capsule <capsule> --capsule-version <version>` process this module starts and owns.
7//! The launcher waits on the entry member's process, so the peers are stopped however the entry
8//! member ends — its task finishing, its own `exit(143)`, or a `SIGKILL` of its process.
9//!
10//! **Every member holds a lifeline.** Each member is handed the read end of a pipe of its own
11//! ([`crate::lifeline`]), whose write end only the launcher holds and never writes to. A member
12//! that reads EOF winds down as a first `SIGTERM` would wind it down. The launcher ends a
13//! formation by closing lifelines; when the launcher itself dies — `SIGKILL` and the OOM killer
14//! included — the kernel closes them, so no member outlives its formation.
15//!
16//! **Ready means the door answers as the session that reported it.** A peer is ready when the door
17//! at the URL its readiness line reported serves an agent card naming the session id that line
18//! reported, read through [`crate::running`]'s door probe. A live pid proves nothing, and a door
19//! some other process holds on the reported port names another session. Every member has one
20//! deadline, counted from its spawn.
21//!
22//! **All or nothing.** A peer that fails to come up refuses the whole launch: every member already
23//! started is stopped and reaped before the refusal is returned, and the entry member is never
24//! started.
25//!
26//! **The launcher is the formation's one principal.** It generates the formation's signing key
27//! ([`FormationAuthority`]) after the id and holds it in memory alone. Each member gets a pipe of
28//! its own, whose read end only that member's `mur run` inherits, named by
29//! [`FORMATION_CHANNEL_ENV`]. Its first line — written before the member is spawned — carries the
30//! formation id, the member's name, the verification key and one token per member it may call.
31//! Once every peer's door answers, each peer that calls other peers is written one address line
32//! naming their doors; the entry member's first line and address line are both written before it is
33//! spawned. Nothing here writes a token to stdout, stderr, a file, argv or an environment
34//! variable. Callees never include the entry member: it runs the formation's own task from launch
35//! until the formation ends and serves no peer, and admission never admits an edge into it. The
36//! key and the write ends are dropped once every member is reaped.
37//!
38//! **Teardown.**
39//!
40//! * One owner holds every started member, and every return path stops and reaps them. Its `Drop`
41//!   does the same, so an early return or an unwinding panic leaves nothing running.
42//! * Stopping the members is closing every lifeline, up to [`MEMBER_STOP_GRACE`] for each to end
43//!   in order, `SIGKILL` to the process group of any still running, and a reap. The launcher sends
44//!   no member `SIGTERM`. Members are waited on concurrently.
45//! * Every lifeline is created before the first member is spawned, and the launcher spawns
46//!   nothing but members, so no process but the launcher ever holds a write end.
47//! * Peers lead process groups of their own, so a terminal's `^C` reaches only the launcher and
48//!   the entry member, and the launcher stops the peers in order. The entry member stays in the
49//!   launcher's group and behaves at a terminal as a hand-run `mur run` does.
50//! * Under `panic = "abort"` no destructor runs, so a panic hook — installed once per process —
51//!   `SIGKILL`s every registered member's process group before the abort. It reads a lock-free
52//!   registry of member pids, never a mutex a panicking thread might hold.
53//! * [`catch_launcher_signals`] turns `SIGINT`, `SIGTERM` and `SIGHUP` into a flag every wait here
54//!   polls, so a signalled launcher stops its members — the entry member through its lifeline,
55//!   like every other — before it exits.
56
57use std::io::Write;
58use std::path::{Path, PathBuf};
59use std::process::{Child, ExitStatus};
60use std::sync::atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering};
61use std::sync::{mpsc, Once};
62use std::time::{Duration, Instant};
63
64use crate::child_launch::{self, AfterReadiness, ProcessEnv, ProcessLaunch, StartedProcess};
65use crate::delegation::SPAWNER_ENV;
66use crate::door_auth::DoorToken;
67use crate::formation::{FormationId, FormationPeer, FORMATION_PEERS_ENV};
68use crate::formation_credentials::{
69    render_address_line, FormationAuthority, FORMATION_CHANNEL_ENV,
70};
71use crate::lifeline::SPAWNER_LIFELINE_ENV;
72use crate::roster::AdmittedRoster;
73use crate::running::ProcessIdentity;
74use serde::{Deserialize, Serialize};
75
76/// How long a member has, from its spawn, to report itself and have its door answer.
77///
78/// The bound a delegated child gets for its readiness line, for the same reason: it bounds one
79/// `mur run` subprocess becoming usable on a cold or loaded host, where compiling the driver
80/// component is the slowest step.
81pub const MEMBER_READY_TIMEOUT: Duration = Duration::from_secs(180);
82
83/// How long a member has, after its lifeline is closed, to end in order before its process group
84/// is killed.
85///
86/// Longer than the runtime's own teardown deadline, so a member whose session is ending in order
87/// finishes before it is killed and its trace stays readable.
88pub const MEMBER_STOP_GRACE: Duration = Duration::from_secs(25);
89
90/// How often a peer's door is probed while it is not yet ready.
91const READY_PROBE_INTERVAL: Duration = Duration::from_millis(100);
92
93/// How often a wait here looks at a member's process, at the launcher's signal flag, and at
94/// whether another peer has already failed.
95const POLL_INTERVAL: Duration = Duration::from_millis(50);
96
97/// How long a member whose stdout has closed is given to exit on its own before it is ended.
98const EXIT_AFTER_EOF_WAIT: Duration = Duration::from_secs(2);
99
100/// The address every peer binds its door on. Members of one formation share a host.
101pub const PEER_BIND_ADDR: &str = "127.0.0.1";
102
103/// The hidden `mur run` flag carrying the sha256 admission bound a member to.
104pub const CAPSULE_SHA256_FLAG: &str = "--capsule-sha256";
105
106/// The hidden `mur run` flag naming the directory a peer resolves its artifacts, its lock and its
107/// workspace from, apart from its `--workdir`.
108pub const STORE_ROOT_FLAG: &str = "--store-root";
109
110/// The directory under the murmur home that holds every formation's peer directories.
111pub const FORMATIONS_DIR: &str = "formations";
112
113/// The ownership marker in `<murmur home>/formations/<frm_id>/`, a [`FormationMarker`]. Member
114/// names start with a letter and hold no `.`, so it never collides with a member's directory.
115pub const FORMATION_MARKER_FILE: &str = "formation.json";
116
117/// What a formation launch is given besides its roster.
118#[derive(Debug, Clone)]
119pub struct FormationLaunchOptions {
120    /// The id every member is launched with, minted once for this launch.
121    pub formation_id: FormationId,
122    /// The roster's project directory: the entry member's `--workdir`, and every peer's
123    /// `--store-root`, so every member resolves from the stores admission read.
124    pub project_dir: PathBuf,
125    /// `--task`, passed to the entry member only.
126    pub task: Option<String>,
127    /// `--json`, passed to the entry member only; peers always run with it.
128    pub json: bool,
129    /// `--verbose`, passed to the entry member only.
130    pub verbose: bool,
131    /// `--no-env-file`, passed to every member.
132    pub no_env_file: bool,
133    /// `--containment <class>`, passed to every member.
134    pub containment: Option<String>,
135    /// Each member's deadline to become ready, from its spawn. [`MEMBER_READY_TIMEOUT`] in
136    /// production.
137    pub ready_timeout: Duration,
138    /// How long a stopped member has after its lifeline is closed. [`MEMBER_STOP_GRACE`] in
139    /// production.
140    pub stop_grace: Duration,
141    /// Where peers' directories are made in place of `<murmur home>/formations`. `None` in
142    /// production; set by tests that must not write under the developer's home.
143    pub(crate) formations_dir: Option<PathBuf>,
144}
145
146impl FormationLaunchOptions {
147    /// Options with the production deadline and grace, and nothing passed through.
148    pub fn new(formation_id: FormationId, project_dir: PathBuf) -> Self {
149        Self {
150            formation_id,
151            project_dir,
152            task: None,
153            json: false,
154            verbose: false,
155            no_env_file: false,
156            containment: None,
157            ready_timeout: MEMBER_READY_TIMEOUT,
158            stop_grace: MEMBER_STOP_GRACE,
159            formations_dir: None,
160        }
161    }
162}
163
164/// One member as the launcher starts it.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub(crate) struct PlannedMember {
167    pub(crate) name: String,
168    pub(crate) capsule: String,
169    pub(crate) version: String,
170    /// The sha256 admission bound the member to, handed to its `mur run` to check.
171    pub(crate) sha256: String,
172}
173
174/// What a launch starts: the peers in roster order, the entry member, and the members each member
175/// may call.
176#[derive(Debug, Clone)]
177pub(crate) struct FormationPlan {
178    pub(crate) peers: Vec<PlannedMember>,
179    pub(crate) entry: PlannedMember,
180    /// Every member's callees in roster order. Admission never admits an edge into the entry
181    /// member, so it is never among them. A member with none is absent.
182    pub(crate) callees: Vec<(String, Vec<String>)>,
183}
184
185impl FormationPlan {
186    pub(crate) fn from_roster(roster: &AdmittedRoster) -> Self {
187        let planned = |member: &crate::roster::AdmittedMember| PlannedMember {
188            name: member.name.clone(),
189            capsule: member.capsule.clone(),
190            version: member.version.clone(),
191            sha256: member.sha256.clone(),
192        };
193        let entry = roster.entry();
194        Self {
195            peers: roster
196                .members()
197                .iter()
198                .filter(|member| !member.entry)
199                .map(planned)
200                .collect(),
201            entry: planned(entry),
202            callees: roster
203                .members()
204                .iter()
205                .map(|member| {
206                    let callees: Vec<String> = roster
207                        .callees(&member.name)
208                        .map(|callee| callee.name.clone())
209                        .collect();
210                    (member.name.clone(), callees)
211                })
212                .filter(|(_, callees)| !callees.is_empty())
213                .collect(),
214        }
215    }
216
217    /// The members `member` may call, in roster order.
218    pub(crate) fn callees_of(&self, member: &str) -> &[String] {
219        self.callees
220            .iter()
221            .find(|(name, _)| name == member)
222            .map(|(_, callees)| callees.as_slice())
223            .unwrap_or_default()
224    }
225
226    /// `member`'s channel first line, signed by `authority` and naming `launcher` as the process
227    /// that started it.
228    fn first_line(
229        &self,
230        authority: &FormationAuthority,
231        launcher: Option<&ProcessIdentity>,
232        member: &str,
233    ) -> String {
234        let callees: Vec<&str> = self.callees_of(member).iter().map(String::as_str).collect();
235        let mut bundle = authority.member_bundle(member, &callees);
236        bundle.launcher = launcher.cloned();
237        bundle.render_line()
238    }
239}
240
241/// A peer whose door answered as the session its readiness line named.
242pub struct ReadyPeer {
243    pub name: String,
244    pub capsule: String,
245    pub version: String,
246    pub pid: u32,
247    pub session_id: String,
248    /// `http://host:port`.
249    pub url: String,
250    /// The peer's accessible directory, `<murmur home>/formations/<frm_id>/<member>`: its
251    /// `--workdir`, where its `task.md` is written and its sessions nest.
252    pub workdir: PathBuf,
253    /// The operator token from the peer's readiness line, present when the peer declares
254    /// `network.authentication`. Held for the readiness probe and never printed.
255    door_token: Option<DoorToken>,
256}
257
258impl ReadyPeer {
259    /// The operator token for this peer's door, or `None` for a public door.
260    pub fn door_token(&self) -> Option<&DoorToken> {
261        self.door_token.as_ref()
262    }
263}
264
265impl std::fmt::Debug for ReadyPeer {
266    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
267        f.debug_struct("ReadyPeer")
268            .field("name", &self.name)
269            .field("capsule", &self.capsule)
270            .field("version", &self.version)
271            .field("pid", &self.pid)
272            .field("session_id", &self.session_id)
273            .field("url", &self.url)
274            .field("workdir", &self.workdir)
275            .field(
276                "door_token",
277                &self.door_token.as_ref().map(|_| "<redacted>"),
278            )
279            .finish()
280    }
281}
282
283/// Why one member did not come up.
284#[derive(Debug, Clone, PartialEq, Eq)]
285pub enum MemberFailureReason {
286    /// Its process could not be started.
287    NotStarted { error: String },
288    /// Its process ended before it printed a readiness line.
289    ExitedBeforeReporting {
290        status: String,
291        stderr_tail: Vec<String>,
292    },
293    /// Its process reported and then ended before its door answered.
294    ExitedBeforeDoorAnswered {
295        status: String,
296        stderr_tail: Vec<String>,
297    },
298    /// No readiness line within the deadline.
299    DidNotReport { deadline: Duration },
300    /// The first line it printed is not a readiness line.
301    UnreadableReport { why: String },
302    /// Its readiness line names no door to probe.
303    ReportedNoDoor,
304    /// Its readiness line names another formation, or none.
305    WrongFormation {
306        reported: Option<String>,
307        expected: String,
308    },
309    /// Its door never answered as the session it reported, within the deadline.
310    DoorDidNotAnswer {
311        url: String,
312        session_id: String,
313        deadline: Duration,
314        last_error: String,
315    },
316}
317
318impl std::fmt::Display for MemberFailureReason {
319    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
320        let tail = |f: &mut std::fmt::Formatter<'_>, lines: &[String]| -> std::fmt::Result {
321            if lines.is_empty() {
322                return f.write_str("; it wrote nothing to stderr");
323            }
324            f.write_str("; its last stderr lines were:")?;
325            for line in lines {
326                write!(f, "\n      {line}")?;
327            }
328            Ok(())
329        };
330        match self {
331            Self::NotStarted { error } => write!(f, "its process could not be started: {error}"),
332            Self::ExitedBeforeReporting {
333                status,
334                stderr_tail,
335            } => {
336                write!(f, "its process exited with {status} before reporting")?;
337                tail(f, stderr_tail)
338            }
339            Self::ExitedBeforeDoorAnswered {
340                status,
341                stderr_tail,
342            } => {
343                write!(
344                    f,
345                    "its process reported and then exited with {status} before its door answered"
346                )?;
347                tail(f, stderr_tail)
348            }
349            Self::DidNotReport { deadline } => {
350                write!(f, "it did not report within {}s", deadline.as_secs_f32())
351            }
352            Self::UnreadableReport { why } => {
353                write!(f, "its first line is not a readiness line: {why}")
354            }
355            Self::ReportedNoDoor => f.write_str("it reported no door"),
356            Self::WrongFormation { reported, expected } => match reported {
357                Some(reported) => {
358                    write!(f, "it reported formation {reported} instead of {expected}")
359                }
360                None => write!(f, "it reported no formation instead of {expected}"),
361            },
362            Self::DoorDidNotAnswer {
363                url,
364                session_id,
365                deadline,
366                last_error,
367            } => write!(
368                f,
369                "its door at {url} did not answer as session {session_id} within {}s; the last \
370                 probe said: {last_error}",
371                deadline.as_secs_f32()
372            ),
373        }
374    }
375}
376
377/// One member that did not come up, and why.
378#[derive(Debug, Clone, PartialEq, Eq)]
379pub struct MemberFailure {
380    pub name: String,
381    pub capsule: String,
382    pub version: String,
383    pub reason: MemberFailureReason,
384}
385
386/// A member whose process was still running when the launcher stopped it.
387#[derive(Debug, Clone, PartialEq, Eq)]
388pub struct StoppedMember {
389    pub name: String,
390    pub pid: u32,
391}
392
393/// A launch that did not complete. By the time this is returned, nothing it started is running.
394#[derive(Debug, Clone, PartialEq, Eq)]
395pub struct MemberLaunchFailure {
396    /// Every member known to have failed, in roster order. Empty when the launch was interrupted
397    /// by a signal before any member failed.
398    pub failed: Vec<MemberFailure>,
399    /// Every member that had started and was stopped, in roster order.
400    pub stopped: Vec<StoppedMember>,
401    /// The signal the launcher caught while members were coming up, when that is what ended the
402    /// launch.
403    pub interrupted_by: Option<i32>,
404}
405
406impl std::fmt::Display for MemberLaunchFailure {
407    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
408        match self.interrupted_by {
409            Some(signal) if self.failed.is_empty() => write!(
410                f,
411                "the formation launch was interrupted by {} before every member was ready",
412                signal_name(signal)
413            )?,
414            _ => f.write_str("the formation could not be launched")?,
415        }
416        for failure in &self.failed {
417            write!(
418                f,
419                "\n  '{}' ({}@{}): {}",
420                failure.name, failure.capsule, failure.version, failure.reason
421            )?;
422        }
423        write!(f, "\n  stopped: {}", render_stopped(&self.stopped))
424    }
425}
426
427/// `name (pid N), …`, or `none`.
428pub fn render_stopped(stopped: &[StoppedMember]) -> String {
429    if stopped.is_empty() {
430        return "none".to_string();
431    }
432    stopped
433        .iter()
434        .map(|member| format!("{} (pid {})", member.name, member.pid))
435        .collect::<Vec<_>>()
436        .join(", ")
437}
438
439/// The conventional name of the three signals a launcher catches.
440pub fn signal_name(signal: i32) -> String {
441    match signal {
442        libc::SIGINT => "SIGINT".to_string(),
443        libc::SIGTERM => "SIGTERM".to_string(),
444        libc::SIGHUP => "SIGHUP".to_string(),
445        other => format!("signal {other}"),
446    }
447}
448
449/// How the entry member's process, and with it the formation, ended.
450#[derive(Debug, Clone, PartialEq, Eq)]
451pub enum FormationEnding {
452    /// The entry member's process ended on its own, or by a signal the launcher did not send.
453    EntryExited(ExitStatus),
454    /// The launcher caught this signal and stopped the formation, the entry member included.
455    Signalled(i32),
456}
457
458/// A formation that has ended, every member stopped and reaped.
459#[derive(Debug, Clone)]
460pub struct FormationExit {
461    pub ending: FormationEnding,
462    /// Every member still running when the formation was stopped, in roster order.
463    pub stopped: Vec<StoppedMember>,
464}
465
466impl FormationExit {
467    /// The launcher's exit status: the entry member's code, or 128 plus the signal that ended it,
468    /// or 128 plus the signal the launcher caught.
469    pub fn exit_code(&self) -> i32 {
470        match &self.ending {
471            FormationEnding::EntryExited(status) => crate::shell::exit_code_of(status),
472            FormationEnding::Signalled(signal) => 128 + signal,
473        }
474    }
475}
476
477/// `status N` or `signal N`, for a refusal.
478fn describe_status(status: ExitStatus) -> String {
479    if let Some(code) = status.code() {
480        return format!("status {code}");
481    }
482    #[cfg(unix)]
483    {
484        use std::os::unix::process::ExitStatusExt;
485        if let Some(signal) = status.signal() {
486            return format!("signal {signal}");
487        }
488    }
489    status.to_string()
490}
491
492// ── The launcher's signals ────────────────────────────────────────────────────
493
494/// The signal the launcher caught, or 0. Written only by [`record_signal`].
495static CAUGHT_SIGNAL: AtomicI32 = AtomicI32::new(0);
496
497#[cfg(unix)]
498extern "C" fn record_signal(signal: libc::c_int) {
499    // An atomic store is async-signal-safe; nothing else happens in the handler.
500    CAUGHT_SIGNAL.store(signal, Ordering::SeqCst);
501}
502
503/// Catch `SIGINT`, `SIGTERM` and `SIGHUP` for the rest of this process's life, so a formation
504/// launcher stops its members before it exits rather than dying with them running.
505///
506/// Every wait in this module polls the flag the handler sets. Called by the launcher only: a test
507/// harness that launched formations in-process would otherwise stop answering `^C`.
508#[cfg(unix)]
509#[allow(unsafe_code)]
510pub fn catch_launcher_signals() -> std::io::Result<()> {
511    for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
512        // SAFETY: `sigaction` is given a zeroed, then fully initialised, struct whose handler is
513        // an `extern "C"` function that only stores to an atomic, which is async-signal-safe. The
514        // old-action pointer is null, so nothing is written back.
515        let installed = unsafe {
516            let mut action: libc::sigaction = std::mem::zeroed();
517            action.sa_sigaction = record_signal as extern "C" fn(libc::c_int) as libc::sighandler_t;
518            action.sa_flags = libc::SA_RESTART;
519            libc::sigemptyset(&mut action.sa_mask);
520            libc::sigaction(signal, &action, std::ptr::null_mut())
521        };
522        if installed != 0 {
523            return Err(std::io::Error::last_os_error());
524        }
525    }
526    Ok(())
527}
528
529#[cfg(not(unix))]
530pub fn catch_launcher_signals() -> std::io::Result<()> {
531    Ok(())
532}
533
534/// The signal [`catch_launcher_signals`] caught, if any.
535pub fn caught_signal() -> Option<i32> {
536    match CAUGHT_SIGNAL.load(Ordering::SeqCst) {
537        0 => None,
538        signal => Some(signal),
539    }
540}
541
542// ── The member registry the panic hook reads ──────────────────────────────────
543
544/// How many members one process can have registered at once. A slot that cannot be had leaves
545/// that member to the owner's `Drop` alone; nothing else depends on the registry.
546const REGISTRY_SLOTS: usize = 1024;
547
548/// Every running member's pid: positive for a member leading its own process group, which the
549/// hook kills as a group; negative for the entry member, which shares the launcher's group and is
550/// killed alone; 0 for a free slot.
551static MEMBER_REGISTRY: [AtomicI64; REGISTRY_SLOTS] = [const { AtomicI64::new(0) }; REGISTRY_SLOTS];
552
553/// One registered member. Releasing it — on reap, or on drop — frees the slot.
554struct RegistrySlot(Option<usize>);
555
556impl RegistrySlot {
557    fn register(pid: u32, own_group: bool) -> Self {
558        let pid = i64::from(pid);
559        let value = if own_group { pid } else { -pid };
560        let slot = MEMBER_REGISTRY.iter().position(|slot| {
561            slot.compare_exchange(0, value, Ordering::SeqCst, Ordering::SeqCst)
562                .is_ok()
563        });
564        Self(slot)
565    }
566
567    fn release(&mut self) {
568        if let Some(index) = self.0.take() {
569            MEMBER_REGISTRY[index].store(0, Ordering::SeqCst);
570        }
571    }
572}
573
574impl Drop for RegistrySlot {
575    fn drop(&mut self) {
576        self.release();
577    }
578}
579
580/// `SIGKILL` every registered member: a member leading its own group as a group, the entry member
581/// alone. Reads atomics only, so a panic hook can call it whatever lock the panicking thread holds.
582pub(crate) fn kill_registered_members() {
583    for slot in &MEMBER_REGISTRY {
584        match slot.load(Ordering::SeqCst) {
585            0 => {}
586            pid if pid > 0 => signal_group(pid, libc::SIGKILL),
587            pid => signal_pid(-pid, libc::SIGKILL),
588        }
589    }
590}
591
592/// Install, once per process, the hook that kills every registered member before a
593/// `panic = "abort"` build aborts. An unwinding build runs the owner's `Drop` instead, which stops
594/// members in order.
595fn install_panic_hook() {
596    static INSTALLED: Once = Once::new();
597    INSTALLED.call_once(|| {
598        let previous = std::panic::take_hook();
599        std::panic::set_hook(Box::new(move |info| {
600            previous(info);
601            if cfg!(panic = "abort") {
602                kill_registered_members();
603            }
604        }));
605    });
606}
607
608#[cfg(unix)]
609#[allow(unsafe_code)]
610fn signal_pid(pid: i64, signal: libc::c_int) {
611    let Ok(pid) = libc::pid_t::try_from(pid) else {
612        return;
613    };
614    if pid <= 0 {
615        return;
616    }
617    // SAFETY: `kill` takes two integers and dereferences nothing. `pid` is positive, so it names
618    // one process, never a group or every process.
619    unsafe {
620        libc::kill(pid, signal);
621    }
622}
623
624#[cfg(unix)]
625#[allow(unsafe_code)]
626fn signal_group(pgid: i64, signal: libc::c_int) {
627    let Ok(pgid) = libc::pid_t::try_from(pgid) else {
628        return;
629    };
630    if pgid <= 1 {
631        return;
632    }
633    // SAFETY: `killpg` takes two integers and dereferences nothing. `pgid` is greater than 1, so
634    // it names one process group, never the caller's own or every process.
635    unsafe {
636        libc::killpg(pgid, signal);
637    }
638}
639
640#[cfg(not(unix))]
641fn signal_pid(_pid: i64, _signal: i32) {}
642
643#[cfg(not(unix))]
644fn signal_group(_pgid: i64, _signal: i32) {}
645
646// ── One member's process ──────────────────────────────────────────────────────
647
648/// A started member's process, owned here and nowhere else.
649struct MemberProcess {
650    name: String,
651    pid: u32,
652    child: Child,
653    own_group: bool,
654    /// The launcher's side of the member's lifeline: the write end, open until the member is
655    /// stopped.
656    #[cfg(unix)]
657    lifeline: Option<crate::lifeline::MemberLifeline>,
658    /// The exit status, once reaped. A reaped pid may already belong to another process, so
659    /// nothing signals or probes it after this is set.
660    status: Option<ExitStatus>,
661    slot: RegistrySlot,
662    grace: Duration,
663    /// The write end of the member's formation channel, kept open while the member runs so an
664    /// address line can follow its first line, and closed once it is reaped.
665    channel: Option<std::io::PipeWriter>,
666}
667
668impl MemberProcess {
669    fn new(
670        name: &str,
671        child: Child,
672        own_group: bool,
673        grace: Duration,
674        channel: std::io::PipeWriter,
675        #[cfg(unix)] lifeline: Option<crate::lifeline::MemberLifeline>,
676    ) -> Self {
677        let pid = child.id();
678        Self {
679            name: name.to_string(),
680            pid,
681            child,
682            own_group,
683            #[cfg(unix)]
684            lifeline,
685            status: None,
686            slot: RegistrySlot::register(pid, own_group),
687            grace,
688            channel: Some(channel),
689        }
690    }
691
692    /// Write one line to the member's channel. A member that has already gone has closed its end;
693    /// the line then reaches nobody, which is what it is for.
694    fn send_line(&mut self, line: &str) {
695        if let Some(channel) = self.channel.as_mut() {
696            let _ = channel.write_all(format!("{line}\n").as_bytes());
697        }
698    }
699
700    /// Whether the process has ended, reaped or not.
701    fn has_exited(&self) -> bool {
702        self.status.is_some() || !crate::running::pid_is_alive(self.pid)
703    }
704
705    /// The exit status, reaping the process if it has ended. `None` while it runs.
706    ///
707    /// A member leading its own group has the group killed first, while its unreaped process
708    /// still holds the group id, so nothing it started outlives it and no recycled id is signalled.
709    fn exit_status_now(&mut self) -> Option<ExitStatus> {
710        if self.status.is_none() && self.has_exited() {
711            self.reap();
712        }
713        self.status
714    }
715
716    fn reap(&mut self) {
717        if self.status.is_some() {
718            return;
719        }
720        if self.own_group {
721            signal_group(i64::from(self.pid), libc::SIGKILL);
722        } else if !self.has_exited() {
723            signal_pid(i64::from(self.pid), libc::SIGKILL);
724        }
725        self.status = self.child.wait().ok();
726        if self.status.is_none() {
727            // `wait` failing leaves nothing further to ask; the process is not ours to probe.
728            self.status = Some(ExitStatus::default());
729        }
730        self.slot.release();
731        self.channel = None;
732    }
733
734    /// Close the member's lifeline: it reads EOF and winds down. Idempotent.
735    fn close_lifeline(&mut self) {
736        #[cfg(unix)]
737        if let Some(lifeline) = &mut self.lifeline {
738            lifeline.close();
739        }
740    }
741
742    /// Whether the process is running and not yet reaped.
743    fn is_running(&self) -> bool {
744        self.status.is_none() && !self.has_exited()
745    }
746
747    /// Close its lifeline, up to the grace for it to end, `SIGKILL` to its group, and a reap.
748    /// Idempotent. `Some` when the process was still running when the stop began.
749    fn stop(&mut self) -> Option<StoppedMember> {
750        let was_running = self.is_running();
751        self.stop_from(was_running)
752    }
753
754    /// [`Self::stop`], with whether the process was running when the stop began read by the
755    /// caller: a member can end on its lifeline's EOF the moment that lifeline closes.
756    fn stop_from(&mut self, was_running: bool) -> Option<StoppedMember> {
757        if self.status.is_some() {
758            return None;
759        }
760        self.close_lifeline();
761        if was_running {
762            let deadline = Instant::now() + self.grace;
763            while !self.has_exited() && Instant::now() < deadline {
764                std::thread::sleep(POLL_INTERVAL);
765            }
766        }
767        self.reap();
768        was_running.then(|| StoppedMember {
769            name: self.name.clone(),
770            pid: self.pid,
771        })
772    }
773}
774
775impl Drop for MemberProcess {
776    fn drop(&mut self) {
777        let _ = self.stop();
778    }
779}
780
781/// Every started member, in roster order: the one owner, and what every exit path stops.
782struct MemberSet(Vec<MemberProcess>);
783
784impl MemberSet {
785    /// Close every member's lifeline, then wait on, kill if need be, and reap every member
786    /// concurrently. Idempotent.
787    fn stop_all(&mut self) -> Vec<StoppedMember> {
788        // Read before any lifeline closes: a member that ends on its EOF while another lifeline is
789        // still being closed was running when the stop began, and is reported as stopped.
790        let running: Vec<bool> = self.0.iter().map(MemberProcess::is_running).collect();
791        for member in &mut self.0 {
792            member.close_lifeline();
793        }
794        std::thread::scope(|scope| {
795            let stopping: Vec<_> = self
796                .0
797                .iter_mut()
798                .zip(running)
799                .map(|(member, was_running)| scope.spawn(move || member.stop_from(was_running)))
800                .collect();
801            stopping
802                .into_iter()
803                .filter_map(|handle| handle.join().ok().flatten())
804                .collect()
805        })
806    }
807}
808
809impl Drop for MemberSet {
810    fn drop(&mut self) {
811        let _ = self.stop_all();
812    }
813}
814
815// ── Starting members ──────────────────────────────────────────────────────────
816
817/// The arguments every member's `mur run` starts with, running in `workdir`.
818fn member_args(member: &PlannedMember, workdir: &Path) -> Vec<String> {
819    vec![
820        "run".to_string(),
821        "--capsule".to_string(),
822        member.capsule.clone(),
823        "--capsule-version".to_string(),
824        member.version.clone(),
825        CAPSULE_SHA256_FLAG.to_string(),
826        member.sha256.clone(),
827        "--workdir".to_string(),
828        workdir.display().to_string(),
829    ]
830}
831
832/// The pass-through flags every member gets.
833fn pass_through(args: &mut Vec<String>, options: &FormationLaunchOptions) {
834    if options.no_env_file {
835        args.push("--no-env-file".to_string());
836    }
837    if let Some(class) = &options.containment {
838        args.push("--containment".to_string());
839        args.push(class.clone());
840    }
841}
842
843/// A peer's arguments after the binary: it runs in `workdir`, a directory of its own, and resolves
844/// from the project's stores. Its readiness line carries its tokens, which this launcher reads on
845/// the peer's stdout pipe for the door probe.
846pub(crate) fn peer_args(
847    member: &PlannedMember,
848    options: &FormationLaunchOptions,
849    workdir: &Path,
850) -> Vec<String> {
851    let mut args = member_args(member, workdir);
852    args.extend([
853        STORE_ROOT_FLAG.to_string(),
854        options.project_dir.display().to_string(),
855        "--json".to_string(),
856        child_launch::READINESS_TOKENS_FLAG.to_string(),
857        "--bind".to_string(),
858        PEER_BIND_ADDR.to_string(),
859    ]);
860    pass_through(&mut args, options);
861    args
862}
863
864/// The entry member's arguments after the binary. It runs in the project directory, where its
865/// `--task` is written. Never [`child_launch::READINESS_TOKENS_FLAG`]: the entry's output is the
866/// operator's.
867pub(crate) fn entry_args(member: &PlannedMember, options: &FormationLaunchOptions) -> Vec<String> {
868    let mut args = member_args(member, &options.project_dir);
869    args.extend([
870        "--lifecycle-task-acceptance".to_string(),
871        "single".to_string(),
872        "--lifecycle-after-task".to_string(),
873        "exit".to_string(),
874    ]);
875    if let Some(task) = &options.task {
876        args.push("--task".to_string());
877        args.push(task.clone());
878    }
879    if options.json {
880        args.push("--json".to_string());
881    }
882    if options.verbose {
883        args.push("--verbose".to_string());
884    }
885    pass_through(&mut args, options);
886    args
887}
888
889/// `<murmur home>/formations`, resolved without creating anything.
890fn formations_dir() -> Result<PathBuf, String> {
891    Ok(crate::state_store::murmur_home_dir()?.join(FORMATIONS_DIR))
892}
893
894/// Every session root a formation's peers record under: `<dir>/.murmur` for each directory under
895/// `<murmur home>/formations/<frm_id>/`, sorted.
896///
897/// Reads the directory and creates nothing. A formation with no directory, or a `HOME` that does
898/// not resolve, has no roots. A formation directory that exists and cannot be listed is returned
899/// as a root itself, so the caller's read of it fails and is reported as an unreadable root.
900pub fn formation_member_roots(formation_id: &FormationId) -> Vec<PathBuf> {
901    match formations_dir() {
902        Ok(dir) => formation_member_roots_in(&dir, formation_id),
903        Err(_) => Vec::new(),
904    }
905}
906
907/// [`formation_member_roots`] under `formations_dir` in place of `<murmur home>/formations`.
908pub fn formation_member_roots_in(
909    formations_dir: &Path,
910    formation_id: &FormationId,
911) -> Vec<PathBuf> {
912    let dir = formations_dir.join(formation_id.as_str());
913    let entries = match std::fs::read_dir(&dir) {
914        Ok(entries) => entries,
915        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
916        Err(_) => return vec![dir],
917    };
918    let mut roots: Vec<PathBuf> = entries
919        .filter_map(Result::ok)
920        .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir()))
921        .map(|entry| entry.path().join(".murmur"))
922        .filter(|root| root.exists())
923        .collect();
924    roots.sort();
925    roots
926}
927
928/// Who launched a formation, from which project: the contents of [`FORMATION_MARKER_FILE`].
929///
930/// The launcher writes it once, before the first peer's directory is made, and nothing rewrites
931/// it. Retention reads it to decide whether a formation directory is this project's to remove and
932/// whether the launcher that made it is still running; a directory without one is never removed.
933#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
934pub struct FormationMarker {
935    /// The roster's project directory, canonicalized when the launcher could, as given otherwise.
936    pub project_dir: PathBuf,
937    /// The launcher's pid.
938    pub launcher_pid: u32,
939    /// The launcher's start token, read as [`crate::running::process_start_token`] reads it, or
940    /// `""` when it could not be read.
941    pub launcher_start: String,
942}
943
944impl FormationMarker {
945    /// The marker for a formation this process launches from `project_dir`.
946    pub fn for_this_launcher(project_dir: &Path) -> Self {
947        let pid = std::process::id();
948        Self {
949            project_dir: canonical_project_dir(project_dir),
950            launcher_pid: pid,
951            launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
952        }
953    }
954
955    /// The marker in `formation_dir`, or `None` when it is missing or does not parse.
956    pub fn read(formation_dir: &Path) -> Option<Self> {
957        let raw = std::fs::read_to_string(formation_dir.join(FORMATION_MARKER_FILE)).ok()?;
958        serde_json::from_str(&raw).ok()
959    }
960}
961
962/// `project_dir` canonicalized, or as given when it cannot be: the one form a marker records and
963/// the one ownership is compared in.
964pub(crate) fn canonical_project_dir(project_dir: &Path) -> PathBuf {
965    std::fs::canonicalize(project_dir).unwrap_or_else(|_| project_dir.to_path_buf())
966}
967
968/// Write this launch's [`FormationMarker`] into `<formations dir>/<frm_id>/`, holding both
969/// directories owner-only first. The marker itself is written at `0600`.
970fn write_formation_marker(options: &FormationLaunchOptions) -> Result<(), String> {
971    let formations_dir = launch_formations_dir(options)?;
972    let formation_dir = formations_dir.join(options.formation_id.as_str());
973    for dir in [formations_dir.as_path(), formation_dir.as_path()] {
974        crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
975            .map_err(|reason| format!("{}: {reason}", dir.display()))?;
976    }
977    let marker = FormationMarker::for_this_launcher(&options.project_dir);
978    let json = serde_json::to_string(&marker).map_err(|error| error.to_string())?;
979    crate::murmur_home::write_private_file(
980        &formation_dir.join(FORMATION_MARKER_FILE),
981        format!("{json}\n").as_bytes(),
982    )
983}
984
985/// Remove the ended formation directories of `project_dir` that the entry member's `policy` does
986/// not keep, and return what went.
987///
988/// Called by the launcher once every member of `current` has been stopped and reaped. `current`
989/// and every formation minted after it are never removed; see [`crate::retention::prune_formations`]
990/// for what else never is. A `HOME` that does not resolve removes nothing.
991pub fn prune_ended_formations(
992    current: &FormationId,
993    project_dir: &Path,
994    policy: &murmur_artifact::TraceRetainConfig,
995) -> Vec<crate::retention::PrunedFormation> {
996    let (Ok(formations_dir), Ok(running_dir)) =
997        (formations_dir(), crate::running::running_dir_location())
998    else {
999        return Vec::new();
1000    };
1001    prune_ended_formations_in(&formations_dir, &running_dir, current, project_dir, policy)
1002}
1003
1004/// [`prune_ended_formations`] over `formations_dir` and `running_dir` in place of the murmur
1005/// home's.
1006pub fn prune_ended_formations_in(
1007    formations_dir: &Path,
1008    running_dir: &Path,
1009    current: &FormationId,
1010    project_dir: &Path,
1011    policy: &murmur_artifact::TraceRetainConfig,
1012) -> Vec<crate::retention::PrunedFormation> {
1013    // One scan for the whole pass, and only a scan: nothing here unlinks a record.
1014    let records = crate::running::scan(running_dir)
1015        .ok()
1016        .map(|scan| scan.records);
1017    crate::retention::prune_formations(
1018        formations_dir,
1019        current,
1020        project_dir,
1021        policy,
1022        crate::retention::now_ms(),
1023        |id, marker| formation_is_live(id, marker, records.as_deref()),
1024    )
1025}
1026
1027/// Whether any part of formation `id` may still be running: its launcher is not gone, or a running
1028/// record names `id` and its process is not gone.
1029///
1030/// `records` is `None` when the running directory could not be listed, and then every formation
1031/// is live. [`crate::running::ProcessState::Unverified`] is live: a reading that failed says
1032/// nothing about the process.
1033fn formation_is_live(
1034    id: &FormationId,
1035    marker: &FormationMarker,
1036    records: Option<&[crate::running::RunningRecord]>,
1037) -> bool {
1038    let not_gone = |state| !matches!(state, crate::running::ProcessState::Gone(_));
1039    let Some(records) = records else {
1040        return true;
1041    };
1042    not_gone(crate::running::process_state_of(
1043        marker.launcher_pid,
1044        &marker.launcher_start,
1045    )) || records.iter().any(|record| {
1046        record.formation_id.as_ref() == Some(id) && not_gone(crate::running::process_state(record))
1047    })
1048}
1049
1050/// Make `member`'s directory under `formations_dir`, owner-only, and return it.
1051///
1052/// `formations_dir` and the formation's directory are held at `0700` whether or not this call made
1053/// them. The member's own directory is made with a single non-recursive create at `0700`, so a
1054/// path that already exists — a directory, a file or a symlink — is refused rather than reused:
1055/// nothing another process placed there becomes a member's preopen.
1056fn make_member_workdir(
1057    formations_dir: &Path,
1058    formation_id: &FormationId,
1059    member: &str,
1060) -> Result<PathBuf, String> {
1061    use std::os::unix::fs::DirBuilderExt;
1062
1063    let formation_dir = formations_dir.join(formation_id.as_str());
1064    let workdir = formation_dir.join(member);
1065    let failed = |path: &Path, reason: String| {
1066        format!(
1067            "its directory {} could not be made: {}: {reason}",
1068            workdir.display(),
1069            path.display()
1070        )
1071    };
1072    for dir in [formations_dir, formation_dir.as_path()] {
1073        crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
1074            .map_err(|reason| failed(dir, reason))?;
1075    }
1076    std::fs::DirBuilder::new()
1077        .mode(crate::murmur_home::MURMUR_HOME_DIR_MODE)
1078        .create(&workdir)
1079        .map_err(|error| failed(&workdir, error.to_string()))?;
1080    // The umask may have narrowed the create's mode further; it can never have widened it.
1081    Ok(workdir)
1082}
1083
1084/// The directory peers' directories are made under for this launch: the override, or
1085/// `<murmur home>/formations` with the home held owner-only first.
1086fn launch_formations_dir(options: &FormationLaunchOptions) -> Result<PathBuf, String> {
1087    match &options.formations_dir {
1088        Some(dir) => Ok(dir.clone()),
1089        None => {
1090            crate::murmur_home::ensure_murmur_home()?;
1091            formations_dir()
1092        }
1093    }
1094}
1095
1096/// What became of one peer's start.
1097enum PeerOutcome {
1098    Ready(ReadyPeer),
1099    Failed(MemberFailureReason),
1100    /// Given up on because another peer failed or the launcher was signalled.
1101    Abandoned,
1102}
1103
1104/// A member's formation channel: the write end the launcher keeps, already holding the member's
1105/// first line, and the read end the member's process inherits.
1106struct Channel {
1107    writer: std::io::PipeWriter,
1108    reader: std::io::PipeReader,
1109}
1110
1111impl Channel {
1112    /// A fresh pipe with `first_line` already written into it. Both ends are close-on-exec, so no
1113    /// process inherits either unless [`ProcessLaunch::inherit_fd`] names it.
1114    fn open(first_line: &str) -> Result<Self, MemberFailureReason> {
1115        let (reader, mut writer) = std::io::pipe().map_err(|error| {
1116            not_started(&format!(
1117                "its formation channel could not be created: {error}"
1118            ))
1119        })?;
1120        writer
1121            .write_all(format!("{first_line}\n").as_bytes())
1122            .map_err(|error| {
1123                not_started(&format!(
1124                    "its formation channel could not be written: {error}"
1125                ))
1126            })?;
1127        Ok(Self { writer, reader })
1128    }
1129
1130    /// The descriptor number the member reads its channel from.
1131    fn fd(&self) -> i32 {
1132        use std::os::fd::AsRawFd;
1133        self.reader.as_raw_fd()
1134    }
1135
1136    /// The environment every member is started with: its formation id, its channel, and
1137    /// [`crate::host_warnings::HOST_WARNINGS_REPORTED_ENV`], because the launcher prints the
1138    /// host-level warnings once before any member starts.
1139    fn member_env(&self, formation_id: &FormationId) -> Vec<(String, String)> {
1140        let (id_name, id_value) = formation_id.env_pair();
1141        vec![
1142            (id_name.to_string(), id_value),
1143            (FORMATION_CHANNEL_ENV.to_string(), self.fd().to_string()),
1144            crate::host_warnings::reported_env_pair(),
1145        ]
1146    }
1147}
1148
1149/// Start one peer and wait until its door answers, it fails, or `abort` is raised.
1150///
1151/// The process, when one was started, is returned whatever the outcome, so the caller owns
1152/// stopping it.
1153fn start_peer(
1154    binary: &Path,
1155    member: &PlannedMember,
1156    first_line: &str,
1157    options: &FormationLaunchOptions,
1158    abort: &AtomicBool,
1159    #[cfg(unix)] mut lifeline: crate::lifeline::MemberLifeline,
1160) -> (Option<MemberProcess>, PeerOutcome) {
1161    let workdir = match launch_formations_dir(options)
1162        .and_then(|dir| make_member_workdir(&dir, &options.formation_id, &member.name))
1163    {
1164        Ok(workdir) => workdir,
1165        Err(error) => return (None, PeerOutcome::Failed(not_started(&error))),
1166    };
1167    let args = peer_args(member, options, &workdir);
1168    let prefix = format!("[{}] ", member.name);
1169    let channel = match Channel::open(first_line) {
1170        Ok(channel) => channel,
1171        Err(reason) => return (None, PeerOutcome::Failed(reason)),
1172    };
1173    let set = channel.member_env(&options.formation_id);
1174    let spawned_at = Instant::now();
1175    let deadline = spawned_at + options.ready_timeout;
1176    let started = child_launch::start_process(&ProcessLaunch {
1177        binary,
1178        args: &args,
1179        cwd: None,
1180        env: ProcessEnv::Inherited {
1181            set: &set,
1182            remove: &[FORMATION_PEERS_ENV, SPAWNER_ENV, SPAWNER_LIFELINE_ENV],
1183        },
1184        stdin_line: None,
1185        stderr_prefix: Some(&prefix),
1186        inherit_output: false,
1187        own_process_group: true,
1188        inherit_fd: Some(channel.fd()),
1189        #[cfg(unix)]
1190        lifeline: Some(crate::child_launch::Lifeline::Member(&lifeline)),
1191    });
1192    // The member holds its own copy of the read end now, or never will.
1193    let Channel { writer, reader } = channel;
1194    drop(reader);
1195    let StartedProcess {
1196        mut child,
1197        stderr_tail,
1198        ..
1199    } = match started {
1200        Ok(started) => started,
1201        Err(error) => {
1202            return (
1203                None,
1204                PeerOutcome::Failed(MemberFailureReason::NotStarted {
1205                    error: format!("{}: {error}", binary.display()),
1206                }),
1207            )
1208        }
1209    };
1210    #[cfg(unix)]
1211    lifeline.spawned();
1212    let stdout = child.stdout.take();
1213    let mut process = MemberProcess::new(
1214        &member.name,
1215        child,
1216        true,
1217        options.stop_grace,
1218        writer,
1219        #[cfg(unix)]
1220        Some(lifeline),
1221    );
1222    let Some(stdout) = stdout else {
1223        return (
1224            Some(process),
1225            PeerOutcome::Failed(MemberFailureReason::NotStarted {
1226                error: "the process exposed no standard output".to_string(),
1227            }),
1228        );
1229    };
1230    let lines = child_launch::relay_readiness_line(stdout, AfterReadiness::PrefixedStderr(prefix));
1231
1232    // The readiness line.
1233    let line = loop {
1234        if abort.load(Ordering::SeqCst) {
1235            return (Some(process), PeerOutcome::Abandoned);
1236        }
1237        let now = Instant::now();
1238        if now >= deadline {
1239            return (
1240                Some(process),
1241                PeerOutcome::Failed(MemberFailureReason::DidNotReport {
1242                    deadline: options.ready_timeout,
1243                }),
1244            );
1245        }
1246        match lines.recv_timeout(POLL_INTERVAL.min(deadline - now)) {
1247            Ok(Some(line)) => break line,
1248            Ok(None) | Err(mpsc::RecvTimeoutError::Disconnected) => {
1249                // Its stdout closes as it exits, so the exit itself is at most a moment behind. A
1250                // process that closed stdout and stayed up is as good as gone, and is ended.
1251                let closed_at = Instant::now();
1252                let status = loop {
1253                    if let Some(status) = process.exit_status_now() {
1254                        break status;
1255                    }
1256                    if closed_at.elapsed() >= EXIT_AFTER_EOF_WAIT {
1257                        process.reap();
1258                        break process.status.unwrap_or_default();
1259                    }
1260                    std::thread::sleep(POLL_INTERVAL);
1261                };
1262                return (
1263                    Some(process),
1264                    PeerOutcome::Failed(MemberFailureReason::ExitedBeforeReporting {
1265                        status: describe_status(status),
1266                        stderr_tail: stderr_tail.lines_at_end(),
1267                    }),
1268                );
1269            }
1270            Err(mpsc::RecvTimeoutError::Timeout) => {}
1271        }
1272    };
1273
1274    let report = match read_report(&line, &options.formation_id) {
1275        Ok(report) => report,
1276        Err(reason) => return (Some(process), PeerOutcome::Failed(reason)),
1277    };
1278
1279    // The door.
1280    let mut last_error;
1281    loop {
1282        if abort.load(Ordering::SeqCst) {
1283            return (Some(process), PeerOutcome::Abandoned);
1284        }
1285        if let Some(status) = process.exit_status_now() {
1286            return (
1287                Some(process),
1288                PeerOutcome::Failed(MemberFailureReason::ExitedBeforeDoorAnswered {
1289                    status: describe_status(status),
1290                    stderr_tail: stderr_tail.lines_at_end(),
1291                }),
1292            );
1293        }
1294        match door_answers_as(
1295            crate::running::probe_door_session_id(&report.url, report.door_token.as_ref()),
1296            &report.session_id,
1297        ) {
1298            Ok(()) => {
1299                let pid = process.pid;
1300                return (
1301                    Some(process),
1302                    PeerOutcome::Ready(ReadyPeer {
1303                        name: member.name.clone(),
1304                        capsule: member.capsule.clone(),
1305                        version: member.version.clone(),
1306                        pid,
1307                        session_id: report.session_id,
1308                        url: report.url,
1309                        workdir,
1310                        door_token: report.door_token,
1311                    }),
1312                );
1313            }
1314            Err(why) => last_error = why,
1315        }
1316        let now = Instant::now();
1317        if now >= deadline {
1318            return (
1319                Some(process),
1320                PeerOutcome::Failed(MemberFailureReason::DoorDidNotAnswer {
1321                    url: report.url,
1322                    session_id: report.session_id,
1323                    deadline: options.ready_timeout,
1324                    last_error,
1325                }),
1326            );
1327        }
1328        std::thread::sleep(READY_PROBE_INTERVAL.min(deadline - now));
1329    }
1330}
1331
1332/// Whether a door probe's answer counts as the reported session being ready: `Ok` only when the
1333/// card names `reported`. A card naming another session, or a failed probe, is the `Err` that
1334/// becomes the refusal's `last_error` if the deadline passes.
1335fn door_answers_as(probed: Result<String, String>, reported: &str) -> Result<(), String> {
1336    match probed {
1337        Ok(session_id) if session_id == reported => Ok(()),
1338        Ok(other) => Err(format!("the door answers as session {other}")),
1339        Err(error) => Err(error),
1340    }
1341}
1342
1343/// What a peer's readiness line says about its door.
1344struct Report {
1345    session_id: String,
1346    /// `http://host:port`.
1347    url: String,
1348    door_token: Option<DoorToken>,
1349}
1350
1351/// Read a readiness line, refusing one that names another formation or no door.
1352fn read_report(line: &str, expected: &FormationId) -> Result<Report, MemberFailureReason> {
1353    let report: serde_json::Value =
1354        serde_json::from_str(line).map_err(|error| MemberFailureReason::UnreadableReport {
1355            why: format!("it does not parse as JSON ({error})"),
1356        })?;
1357    let reported = report
1358        .get("formation_id")
1359        .and_then(serde_json::Value::as_str);
1360    if reported != Some(expected.as_str()) {
1361        return Err(MemberFailureReason::WrongFormation {
1362            reported: reported.map(str::to_string),
1363            expected: expected.as_str().to_string(),
1364        });
1365    }
1366    let session_id = report
1367        .get("session_id")
1368        .and_then(serde_json::Value::as_str)
1369        .filter(|session_id| !session_id.is_empty())
1370        .ok_or_else(|| MemberFailureReason::UnreadableReport {
1371            why: "it names no session_id".to_string(),
1372        })?;
1373    let url = report
1374        .get("url")
1375        .and_then(serde_json::Value::as_str)
1376        .unwrap_or_default();
1377    if url.is_empty() {
1378        return Err(MemberFailureReason::ReportedNoDoor);
1379    }
1380    Ok(Report {
1381        session_id: session_id.to_string(),
1382        url: format!("http://{}", url.trim_start_matches("http://")),
1383        door_token: child_launch::readiness_door_token(&report),
1384    })
1385}
1386
1387// ── The launch ────────────────────────────────────────────────────────────────
1388
1389/// Start every peer of `roster` at once and wait for each door to answer.
1390///
1391/// On success the peers are running and ready, and the entry member is not yet started: the caller
1392/// announces the formation, then calls [`RunningFormation::start_entry`]. On failure nothing this
1393/// started is still running.
1394pub fn launch_formation(
1395    roster: &AdmittedRoster,
1396    options: FormationLaunchOptions,
1397) -> Result<RunningFormation, MemberLaunchFailure> {
1398    launch_plan(FormationPlan::from_roster(roster), options)
1399}
1400
1401pub(crate) fn launch_plan(
1402    plan: FormationPlan,
1403    options: FormationLaunchOptions,
1404) -> Result<RunningFormation, MemberLaunchFailure> {
1405    install_panic_hook();
1406    let binary = match child_launch::mur_binary() {
1407        Ok(binary) => binary,
1408        Err(error) => {
1409            return Err(MemberLaunchFailure {
1410                failed: plan
1411                    .peers
1412                    .iter()
1413                    .chain([&plan.entry])
1414                    .map(|member| failure(member, not_started(&error.to_string())))
1415                    .collect(),
1416                stopped: Vec::new(),
1417                interrupted_by: None,
1418            })
1419        }
1420    };
1421
1422    let authority = match FormationAuthority::generate(&options.formation_id) {
1423        Ok(authority) => authority,
1424        Err(error) => {
1425            return Err(MemberLaunchFailure {
1426                failed: plan
1427                    .peers
1428                    .iter()
1429                    .chain([&plan.entry])
1430                    .map(|member| failure(member, not_started(&error)))
1431                    .collect(),
1432                stopped: Vec::new(),
1433                interrupted_by: None,
1434            })
1435        }
1436    };
1437    // One per peer and one for the entry member, all before the first spawn: where the
1438    // close-on-exec flag is set after `pipe` returns, a member forked in between would inherit
1439    // another member's write end and keep that member's formation alive.
1440    #[cfg(unix)]
1441    let (peer_lifelines, entry_lifeline) = match member_lifelines(plan.peers.len() + 1) {
1442        Ok(mut lifelines) => {
1443            let entry = lifelines.pop();
1444            (lifelines, entry)
1445        }
1446        Err(error) => {
1447            return Err(MemberLaunchFailure {
1448                failed: plan
1449                    .peers
1450                    .iter()
1451                    .chain([&plan.entry])
1452                    .map(|member| {
1453                        failure(
1454                            member,
1455                            not_started(&format!("its lifeline could not be created: {error}")),
1456                        )
1457                    })
1458                    .collect(),
1459                stopped: Vec::new(),
1460                interrupted_by: None,
1461            })
1462        }
1463    };
1464    // Read once, so every member, the entry member included, records the same launcher. A
1465    // launcher that cannot read its own start time names none, and `mur stop` then cannot reach
1466    // it through its members.
1467    let launcher = ProcessIdentity::of_this_process();
1468    let first_lines: Vec<String> = plan
1469        .peers
1470        .iter()
1471        .map(|member| plan.first_line(&authority, launcher.as_ref(), &member.name))
1472        .collect();
1473    // Before any peer's directory exists, so no formation directory a launch made is without one.
1474    // A launch with no peers makes no formation directory, and so no marker.
1475    if !plan.peers.is_empty() {
1476        if let Err(reason) = write_formation_marker(&options) {
1477            crate::runtime_err!(
1478                "[mur run] warning: formation {} has no ownership marker ({reason}); its \
1479                 directory will never be removed automatically",
1480                options.formation_id
1481            );
1482        }
1483    }
1484
1485    let abort = AtomicBool::new(false);
1486    let mut outcomes: Vec<Option<(Option<MemberProcess>, PeerOutcome)>> =
1487        plan.peers.iter().map(|_| None).collect();
1488    let mut interrupted_by = caught_signal();
1489    if interrupted_by.is_some() {
1490        abort.store(true, Ordering::SeqCst);
1491    }
1492    std::thread::scope(|scope| {
1493        let (tx, rx) = mpsc::channel();
1494        #[cfg(unix)]
1495        let mut peer_lifelines = peer_lifelines.into_iter();
1496        for (index, member) in plan.peers.iter().enumerate() {
1497            let tx = tx.clone();
1498            let (binary, options, abort) = (&binary, &options, &abort);
1499            let first_line = first_lines[index].as_str();
1500            #[cfg(unix)]
1501            let lifeline = peer_lifelines
1502                .next()
1503                .expect("member_lifelines made one lifeline per peer");
1504            scope.spawn(move || {
1505                let started = start_peer(
1506                    binary,
1507                    member,
1508                    first_line,
1509                    options,
1510                    abort,
1511                    #[cfg(unix)]
1512                    lifeline,
1513                );
1514                let _ = tx.send((index, started));
1515            });
1516        }
1517        drop(tx);
1518        loop {
1519            match rx.recv_timeout(POLL_INTERVAL) {
1520                Ok((index, outcome)) => {
1521                    if matches!(outcome.1, PeerOutcome::Failed(_)) {
1522                        abort.store(true, Ordering::SeqCst);
1523                    }
1524                    outcomes[index] = Some(outcome);
1525                }
1526                Err(mpsc::RecvTimeoutError::Timeout) => {}
1527                Err(mpsc::RecvTimeoutError::Disconnected) => break,
1528            }
1529            if interrupted_by.is_none() {
1530                if let Some(signal) = caught_signal() {
1531                    interrupted_by = Some(signal);
1532                    abort.store(true, Ordering::SeqCst);
1533                }
1534            }
1535        }
1536    });
1537
1538    let mut members = MemberSet(Vec::new());
1539    let mut peers = Vec::new();
1540    let mut failed = Vec::new();
1541    for (member, outcome) in plan.peers.iter().zip(outcomes) {
1542        let Some((process, outcome)) = outcome else {
1543            continue;
1544        };
1545        members.0.extend(process);
1546        match outcome {
1547            PeerOutcome::Ready(peer) => peers.push(peer),
1548            PeerOutcome::Failed(reason) => failed.push(failure(member, reason)),
1549            PeerOutcome::Abandoned => {}
1550        }
1551    }
1552
1553    if !failed.is_empty() || interrupted_by.is_some() || peers.len() != plan.peers.len() {
1554        let stopped = members.stop_all();
1555        return Err(MemberLaunchFailure {
1556            failed,
1557            stopped,
1558            interrupted_by,
1559        });
1560    }
1561
1562    // Every peer's door answers, so every address is known: each peer that calls other peers is
1563    // told where they are.
1564    for member in &mut members.0 {
1565        let addresses = addresses_of(&peers, plan.callees_of(&member.name));
1566        if !addresses.is_empty() {
1567            member.send_line(&render_address_line(&addresses));
1568        }
1569    }
1570
1571    Ok(RunningFormation {
1572        formation_id: options.formation_id.clone(),
1573        entry: plan.entry.clone(),
1574        plan,
1575        peers,
1576        members,
1577        entry_index: None,
1578        #[cfg(unix)]
1579        entry_lifeline,
1580        binary,
1581        options,
1582        authority,
1583        launcher,
1584    })
1585}
1586
1587/// The door of each of `callees` among the ready peers, in roster order.
1588fn addresses_of(ready: &[ReadyPeer], callees: &[String]) -> Vec<FormationPeer> {
1589    ready
1590        .iter()
1591        .filter(|peer| callees.contains(&peer.name))
1592        .map(|peer| FormationPeer {
1593            name: peer.name.clone(),
1594            url: peer.url.clone(),
1595        })
1596        .collect()
1597}
1598
1599/// `count` new lifelines, or the first error creating one.
1600#[cfg(unix)]
1601fn member_lifelines(count: usize) -> std::io::Result<Vec<crate::lifeline::MemberLifeline>> {
1602    (0..count)
1603        .map(|_| crate::lifeline::MemberLifeline::new())
1604        .collect()
1605}
1606
1607fn failure(member: &PlannedMember, reason: MemberFailureReason) -> MemberFailure {
1608    MemberFailure {
1609        name: member.name.clone(),
1610        capsule: member.capsule.clone(),
1611        version: member.version.clone(),
1612        reason,
1613    }
1614}
1615
1616fn not_started(error: &str) -> MemberFailureReason {
1617    MemberFailureReason::NotStarted {
1618        error: error.to_string(),
1619    }
1620}
1621
1622/// A formation whose peers are ready: the one owner of every member it started, and of the
1623/// formation's signing key.
1624///
1625/// Dropping it stops and reaps every member, then drops the key: nothing can mint for this
1626/// formation again.
1627pub struct RunningFormation {
1628    formation_id: FormationId,
1629    entry: PlannedMember,
1630    plan: FormationPlan,
1631    peers: Vec<ReadyPeer>,
1632    /// Declared before `authority`, so the members are stopped and reaped before the key is
1633    /// dropped.
1634    members: MemberSet,
1635    /// The entry member's position in `members`, once started.
1636    entry_index: Option<usize>,
1637    /// The entry member's lifeline, created with the peers' and held here until the entry
1638    /// member is started with it.
1639    #[cfg(unix)]
1640    entry_lifeline: Option<crate::lifeline::MemberLifeline>,
1641    binary: PathBuf,
1642    options: FormationLaunchOptions,
1643    authority: FormationAuthority,
1644    /// This process, as every member's first line names it.
1645    launcher: Option<ProcessIdentity>,
1646}
1647
1648impl std::fmt::Debug for RunningFormation {
1649    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1650        f.debug_struct("RunningFormation")
1651            .field("formation_id", &self.formation_id)
1652            .field("entry", &self.entry.name)
1653            .field("peers", &self.peers)
1654            .finish()
1655    }
1656}
1657
1658impl RunningFormation {
1659    pub fn formation_id(&self) -> &FormationId {
1660        &self.formation_id
1661    }
1662
1663    /// The entry member's roster name.
1664    pub fn entry_name(&self) -> &str {
1665        &self.entry.name
1666    }
1667
1668    /// Every peer, ready, in roster order.
1669    pub fn peers(&self) -> &[ReadyPeer] {
1670        &self.peers
1671    }
1672
1673    /// The members `member` is handed a credential and an address for, in roster order: its
1674    /// roster callees.
1675    pub fn callees_of(&self, member: &str) -> &[String] {
1676        self.plan.callees_of(member)
1677    }
1678
1679    /// The entry member's pid, once started.
1680    pub fn entry_pid(&self) -> Option<u32> {
1681        self.entry_index.map(|index| self.members.0[index].pid)
1682    }
1683
1684    /// Start the entry member, last, with its standard output and error this process's own.
1685    ///
1686    /// On failure every peer has been stopped and reaped.
1687    pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
1688        let args = entry_args(&self.entry, &self.options);
1689        // Its whole channel is written before it starts: every address it may need is known.
1690        let channel = Channel::open(&self.plan.first_line(
1691            &self.authority,
1692            self.launcher.as_ref(),
1693            &self.entry.name,
1694        ))
1695        .and_then(|mut channel| {
1696            let addresses = addresses_of(&self.peers, self.plan.callees_of(&self.entry.name));
1697            if !addresses.is_empty() {
1698                channel
1699                    .writer
1700                    .write_all(format!("{}\n", render_address_line(&addresses)).as_bytes())
1701                    .map_err(|error| {
1702                        not_started(&format!(
1703                            "its formation channel could not be written: {error}"
1704                        ))
1705                    })?;
1706            }
1707            Ok(channel)
1708        });
1709        let channel = match channel {
1710            Ok(channel) => channel,
1711            Err(reason) => {
1712                let stopped = self.members.stop_all();
1713                return Err(MemberLaunchFailure {
1714                    failed: vec![failure(&self.entry, reason)],
1715                    stopped,
1716                    interrupted_by: None,
1717                });
1718            }
1719        };
1720        let set = channel.member_env(&self.formation_id);
1721        let started = child_launch::start_process(&ProcessLaunch {
1722            binary: &self.binary,
1723            args: &args,
1724            cwd: None,
1725            env: ProcessEnv::Inherited {
1726                set: &set,
1727                remove: &[SPAWNER_ENV, FORMATION_PEERS_ENV, SPAWNER_LIFELINE_ENV],
1728            },
1729            stdin_line: None,
1730            stderr_prefix: None,
1731            inherit_output: true,
1732            own_process_group: false,
1733            inherit_fd: Some(channel.fd()),
1734            #[cfg(unix)]
1735            lifeline: self
1736                .entry_lifeline
1737                .as_ref()
1738                .map(crate::child_launch::Lifeline::Member),
1739        });
1740        let Channel { writer, reader } = channel;
1741        drop(reader);
1742        match started {
1743            Ok(started) => {
1744                #[cfg(unix)]
1745                let lifeline = self.entry_lifeline.take().map(|mut lifeline| {
1746                    lifeline.spawned();
1747                    lifeline
1748                });
1749                let process = MemberProcess::new(
1750                    &self.entry.name,
1751                    started.child,
1752                    false,
1753                    self.options.stop_grace,
1754                    writer,
1755                    #[cfg(unix)]
1756                    lifeline,
1757                );
1758                let pid = process.pid;
1759                self.members.0.push(process);
1760                self.entry_index = Some(self.members.0.len() - 1);
1761                Ok(pid)
1762            }
1763            Err(error) => {
1764                let stopped = self.members.stop_all();
1765                Err(MemberLaunchFailure {
1766                    failed: vec![failure(
1767                        &self.entry,
1768                        not_started(&format!("{}: {error}", self.binary.display())),
1769                    )],
1770                    stopped,
1771                    interrupted_by: None,
1772                })
1773            }
1774        }
1775    }
1776
1777    /// Wait for the entry member's process to end, or for the launcher to be signalled, then stop
1778    /// and reap every member: every lifeline closed, the entry member's included.
1779    pub fn wait(mut self) -> FormationExit {
1780        let ending = match self.entry_index {
1781            None => caught_signal()
1782                .map(FormationEnding::Signalled)
1783                .unwrap_or(FormationEnding::EntryExited(ExitStatus::default())),
1784            Some(index) => loop {
1785                if let Some(status) = self.members.0[index].exit_status_now() {
1786                    break FormationEnding::EntryExited(status);
1787                }
1788                if let Some(signal) = caught_signal() {
1789                    break FormationEnding::Signalled(signal);
1790                }
1791                std::thread::sleep(POLL_INTERVAL);
1792            },
1793        };
1794        let stopped = self.members.stop_all();
1795        FormationExit { ending, stopped }
1796    }
1797}
1798
1799#[cfg(test)]
1800mod tests {
1801    use super::*;
1802    use std::io::{Read, Write};
1803    use std::net::TcpListener;
1804    use std::sync::atomic::AtomicUsize;
1805    use std::sync::Arc;
1806
1807    /// How long a launch that is expected to succeed may take before the test gives up on it: a
1808    /// guard against a launch that never completes on a loaded host, not a claim about how fast
1809    /// one does.
1810    const LIVENESS_BOUND: Duration = Duration::from_secs(120);
1811
1812    /// The host's 1-, 5- and 15-minute load averages, for failure messages that may be load.
1813    fn load_average() -> String {
1814        std::fs::read_to_string("/proc/loadavg")
1815            .unwrap_or_default()
1816            .split(' ')
1817            .take(3)
1818            .collect::<Vec<_>>()
1819            .join(" ")
1820    }
1821
1822    /// The production bounds, stated: the readiness deadline a delegated child also gets, and a
1823    /// grace that outlasts the runtime's own teardown deadline.
1824    #[test]
1825    fn the_production_bounds_are_180s_and_a_grace_past_the_runtime_teardown() {
1826        assert_eq!(MEMBER_READY_TIMEOUT, Duration::from_secs(180));
1827        assert_eq!(MEMBER_STOP_GRACE, Duration::from_secs(25));
1828        assert!(MEMBER_STOP_GRACE > crate::runtime::TERMINATE_TEARDOWN_DEADLINE);
1829    }
1830
1831    fn member(name: &str) -> PlannedMember {
1832        PlannedMember {
1833            name: name.to_string(),
1834            capsule: name.to_string(),
1835            version: "0.1.0".to_string(),
1836            sha256: "ab".repeat(32),
1837        }
1838    }
1839
1840    /// Every member, peer or entry, is started told its launcher already printed the host-level
1841    /// warnings.
1842    #[test]
1843    fn every_member_is_told_host_warnings_were_reported() {
1844        let channel = Channel::open("first").expect("a channel opens");
1845        let env = channel.member_env(&FormationId::mint());
1846        let values: Vec<&str> = env
1847            .iter()
1848            .filter(|(key, _)| key == crate::host_warnings::HOST_WARNINGS_REPORTED_ENV)
1849            .map(|(_, value)| value.as_str())
1850            .collect();
1851        assert_eq!(values, vec!["1"], "{env:?}");
1852    }
1853
1854    fn options(deadline: Duration) -> FormationLaunchOptions {
1855        let mut options =
1856            FormationLaunchOptions::new(FormationId::mint(), PathBuf::from("/tmp/project"));
1857        options.ready_timeout = deadline;
1858        options.stop_grace = Duration::from_secs(2);
1859        // Peers' directories go under a scratch directory, never the developer's home.
1860        options.formations_dir = Some(tempfile::tempdir().unwrap().keep());
1861        options
1862    }
1863
1864    #[test]
1865    fn member_argv_is_exactly_the_documented_shape() {
1866        let mut options = options(MEMBER_READY_TIMEOUT);
1867        let coder = member("coder");
1868        let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
1869        assert_eq!(
1870            peer_args(&coder, &options, &workdir),
1871            [
1872                "run",
1873                "--capsule",
1874                "coder",
1875                "--capsule-version",
1876                "0.1.0",
1877                "--capsule-sha256",
1878                &"ab".repeat(32),
1879                "--workdir",
1880                "/home/u/.murmur/formations/frm_x/coder",
1881                "--store-root",
1882                "/tmp/project",
1883                "--json",
1884                "--readiness-tokens",
1885                "--bind",
1886                "127.0.0.1",
1887            ]
1888        );
1889        options.task = Some("probe".to_string());
1890        options.json = true;
1891        options.verbose = true;
1892        options.no_env_file = true;
1893        options.containment = Some("scoped".to_string());
1894        let entry = entry_args(&member("planner"), &options);
1895        assert_eq!(&entry[7..9], ["--workdir", "/tmp/project"]);
1896        assert_eq!(
1897            &entry[9..],
1898            [
1899                "--lifecycle-task-acceptance",
1900                "single",
1901                "--lifecycle-after-task",
1902                "exit",
1903                "--task",
1904                "probe",
1905                "--json",
1906                "--verbose",
1907                "--no-env-file",
1908                "--containment",
1909                "scoped",
1910            ]
1911        );
1912        assert!(
1913            !entry.iter().any(|arg| arg == "--bind"
1914                || arg == STORE_ROOT_FLAG
1915                || arg == child_launch::READINESS_TOKENS_FLAG),
1916            "{entry:?}"
1917        );
1918        let peer = peer_args(&coder, &options, &workdir);
1919        assert!(peer.ends_with(&[
1920            "--no-env-file".to_string(),
1921            "--containment".to_string(),
1922            "scoped".to_string()
1923        ]));
1924        assert!(!peer.iter().any(|arg| arg == "--task" || arg == "--verbose"));
1925    }
1926
1927    /// Serializes the tests that point `MURMUR_MUR_BINARY` at a stand-in; the variable is
1928    /// process-wide.
1929    fn binary_guard() -> std::sync::MutexGuard<'static, ()> {
1930        crate::formation::FORMATION_ENV_LOCK
1931            .lock()
1932            .unwrap_or_else(|poisoned| poisoned.into_inner())
1933    }
1934
1935    /// A stand-in `mur` that prints `line` (when given), records any `SIGTERM` it is sent in
1936    /// `dir/sigterm`, and runs until its lifeline reads EOF, whatever it is asked to run.
1937    fn stand_in(dir: &Path, line: Option<&str>) -> PathBuf {
1938        let script = dir.join("mur");
1939        let print = match line {
1940            Some(line) => {
1941                std::fs::write(dir.join("line"), format!("{line}\n")).unwrap();
1942                format!("cat '{}'\n", dir.join("line").display())
1943            }
1944            None => String::new(),
1945        };
1946        // Through `/dev/fd`: a POSIX shell's `<&` takes a single-digit descriptor only, and a
1947        // lifeline is numbered wherever `pipe` put it.
1948        let body = format!(
1949            "#!/bin/sh\ntrap 'echo TERM >> \"{dir}/sigterm\"' TERM\n{print}\
1950             cat \"/dev/fd/$MURMUR_FORMATION_LIFELINE\" >/dev/null\n",
1951            dir = dir.display()
1952        );
1953        std::fs::write(&script, body).unwrap();
1954        use std::os::unix::fs::PermissionsExt;
1955        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1956        script
1957    }
1958
1959    fn plan_of(peer: &str) -> FormationPlan {
1960        FormationPlan {
1961            peers: vec![member(peer)],
1962            entry: member("planner"),
1963            callees: vec![("planner".to_string(), vec![peer.to_string()])],
1964        }
1965    }
1966
1967    /// Launch one peer from a stand-in printing `line`, returning the refusal and the stand-in's
1968    /// pid as the refusal named it.
1969    fn refused(line: Option<&str>, deadline: Duration) -> (MemberLaunchFailure, FormationId) {
1970        let _guard = binary_guard();
1971        let dir = tempfile::tempdir().unwrap();
1972        let options = options(deadline);
1973        let id = options.formation_id.clone();
1974        let line = line.map(|line| line.replace("{formation}", id.as_str()));
1975        std::env::set_var(
1976            child_launch::MUR_BINARY_ENV,
1977            stand_in(dir.path(), line.as_deref()),
1978        );
1979        let result = launch_plan(plan_of("coder"), options);
1980        std::env::remove_var(child_launch::MUR_BINARY_ENV);
1981        match result {
1982            Ok(_) => panic!("the launch was expected to be refused"),
1983            Err(failure) => (failure, id),
1984        }
1985    }
1986
1987    fn assert_reaped(failure: &MemberLaunchFailure) {
1988        assert_eq!(failure.stopped.len(), 1, "{failure:?}");
1989        let pid = failure.stopped[0].pid;
1990        assert_eq!(failure.stopped[0].name, "coder");
1991        assert!(
1992            !crate::running::pid_is_alive(pid),
1993            "the stand-in {pid} is still running"
1994        );
1995        assert!(
1996            !Path::new(&format!("/proc/{pid}")).exists() || cfg!(not(target_os = "linux")),
1997            "the stand-in {pid} was not reaped"
1998        );
1999    }
2000
2001    fn only_reason(failure: &MemberLaunchFailure) -> &MemberFailureReason {
2002        assert_eq!(failure.failed.len(), 1, "{failure:?}");
2003        assert_eq!(failure.failed[0].name, "coder");
2004        &failure.failed[0].reason
2005    }
2006
2007    /// A port with nothing listening on it: claimed, and never bound.
2008    fn silent_port() -> u16 {
2009        crate::pinned_port::claim()
2010    }
2011
2012    #[test]
2013    fn a_door_nothing_answers_on_is_refused_at_the_deadline() {
2014        let deadline = Duration::from_secs(10);
2015        let port = silent_port();
2016        let started = Instant::now();
2017        let (failure, _) = refused(
2018            Some(&format!(
2019                r#"{{"formation_id":"{{formation}}","session_id":"ses_standin","url":"localhost:{port}"}}"#
2020            )),
2021            deadline,
2022        );
2023        assert!(started.elapsed() >= deadline);
2024        match only_reason(&failure) {
2025            MemberFailureReason::DoorDidNotAnswer {
2026                url, session_id, ..
2027            } => {
2028                assert_eq!(url, &format!("http://localhost:{port}"));
2029                assert_eq!(session_id, "ses_standin");
2030            }
2031            // One deadline covers the readiness line and the door, so a stand-in that is not
2032            // scheduled in time is refused before its door is ever probed.
2033            MemberFailureReason::DidNotReport { .. } => panic!(
2034                "the stand-in's readiness line did not arrive within {deadline:?} (load average {}); \
2035                 host load is the likely cause, since the stand-in prints it at once: {failure:?}",
2036                load_average()
2037            ),
2038            other => panic!("{other:?}"),
2039        }
2040        let message = failure.to_string();
2041        assert!(
2042            message.contains("did not answer as session ses_standin"),
2043            "{message}"
2044        );
2045        assert!(message.contains("'coder' (coder@0.1.0)"), "{message}");
2046        assert_reaped(&failure);
2047    }
2048
2049    /// A one-route door on `127.0.0.1`, served forever: each connection's request is read, then
2050    /// `GET /.well-known/agent-card.json` is answered with a card naming the session `answer`
2051    /// returns for that request, or with `{}` when it returns `None`.
2052    fn card_door(mut answer: impl FnMut() -> Option<&'static str> + Send + 'static) -> u16 {
2053        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2054        let port = listener.local_addr().unwrap().port();
2055        std::thread::spawn(move || {
2056            for stream in listener.incoming().flatten() {
2057                let mut stream = stream;
2058                let mut buffer = [0u8; 4096];
2059                let _ = stream.read(&mut buffer);
2060                let body = match answer() {
2061                    Some(session_id) => serde_json::json!({
2062                        "name": "coder",
2063                        "capabilities": {"extensions": [{
2064                            "uri": crate::identity::CAPSULE_EXTENSION_URI,
2065                            "params": {"sessionId": session_id},
2066                        }]},
2067                    })
2068                    .to_string(),
2069                    None => "{}".to_string(),
2070                };
2071                let _ = write!(
2072                    stream,
2073                    "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\
2074                     connection: close\r\n\r\n{body}",
2075                    body.len()
2076                );
2077            }
2078        });
2079        port
2080    }
2081
2082    /// A door answering with a card naming `session_id` from `answer_after` on, and with a card
2083    /// naming no session before then.
2084    fn door(session_id: &'static str, answer_after: Duration) -> u16 {
2085        let opened = Instant::now();
2086        card_door(move || (opened.elapsed() >= answer_after).then_some(session_id))
2087    }
2088
2089    /// How many card requests [`door_naming_another_session_first`] answers as another session.
2090    const WRONG_ANSWERS: usize = 3;
2091
2092    /// A door that answers as `ses_somebody_else` to its first [`WRONG_ANSWERS`] card requests and
2093    /// as `ses_standin` to every one after, with the count of requests it has read. The count is
2094    /// raised before the answer is written, so it covers every answer a client has seen.
2095    fn door_naming_another_session_first() -> (u16, Arc<AtomicUsize>) {
2096        let served = Arc::new(AtomicUsize::new(0));
2097        let counted = Arc::clone(&served);
2098        let port = card_door(move || {
2099            if counted.fetch_add(1, Ordering::SeqCst) < WRONG_ANSWERS {
2100                Some("ses_somebody_else")
2101            } else {
2102                Some("ses_standin")
2103            }
2104        });
2105        (port, served)
2106    }
2107
2108    /// The launcher probes past every answer naming another session and is ready only on the one
2109    /// naming the reported session. Accepting any answer would stop at the door's first request.
2110    #[test]
2111    fn a_door_naming_another_session_is_not_ready() {
2112        let _guard = binary_guard();
2113        let (port, served) = door_naming_another_session_first();
2114        let dir = tempfile::tempdir().unwrap();
2115        let options = options(LIVENESS_BOUND);
2116        let line = format!(
2117            r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2118            options.formation_id
2119        );
2120        std::env::set_var(
2121            child_launch::MUR_BINARY_ENV,
2122            stand_in(dir.path(), Some(&line)),
2123        );
2124        let result = launch_plan(plan_of("coder"), options);
2125        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2126        let formation = match result {
2127            Ok(formation) => formation,
2128            Err(failure) => match only_reason(&failure) {
2129                MemberFailureReason::DidNotReport { .. } => panic!(
2130                    "the stand-in's readiness line did not arrive within {LIVENESS_BOUND:?} \
2131                     (load average {}); host load is the likely cause: {failure:?}",
2132                    load_average()
2133                ),
2134                _ => panic!(
2135                    "the launch was refused although the door answered as ses_standin from its \
2136                     request {} on: {failure:?}",
2137                    WRONG_ANSWERS + 1
2138                ),
2139            },
2140        };
2141        let peer = &formation.peers()[0];
2142        assert_eq!(peer.session_id, "ses_standin");
2143        let served = served.load(Ordering::SeqCst);
2144        assert!(
2145            served > WRONG_ANSWERS,
2146            "an answer naming another session was counted as ready: the door served {served} \
2147             request(s), and its first {WRONG_ANSWERS} named ses_somebody_else"
2148        );
2149        let pid = peer.pid;
2150        drop(formation);
2151        assert!(
2152            !crate::running::pid_is_alive(pid),
2153            "dropping the owner stops the peer"
2154        );
2155    }
2156
2157    #[test]
2158    fn a_door_counts_as_ready_only_when_it_answers_as_the_reported_session() {
2159        let port = door("ses_somebody_else", Duration::ZERO);
2160        let probed = crate::running::probe_door_session_id(&format!("localhost:{port}"), None);
2161        match door_answers_as(probed, "ses_standin") {
2162            Err(why) => assert!(
2163                why.contains("the door answers as session ses_somebody_else"),
2164                "{why}"
2165            ),
2166            Ok(()) => panic!("a door answering as ses_somebody_else counted as ses_standin"),
2167        }
2168        assert_eq!(
2169            door_answers_as(Ok("ses_standin".to_string()), "ses_standin"),
2170            Ok(())
2171        );
2172        assert_eq!(
2173            door_answers_as(Err("connection refused".to_string()), "ses_standin"),
2174            Err("connection refused".to_string())
2175        );
2176    }
2177
2178    #[test]
2179    fn an_empty_url_is_refused_as_no_door() {
2180        let (failure, _) = refused(
2181            Some(r#"{"formation_id":"{formation}","session_id":"ses_standin","url":""}"#),
2182            Duration::from_secs(10),
2183        );
2184        assert_eq!(only_reason(&failure), &MemberFailureReason::ReportedNoDoor);
2185        assert!(failure.to_string().contains("it reported no door"));
2186        assert_reaped(&failure);
2187    }
2188
2189    #[test]
2190    fn a_wrong_formation_or_none_is_refused_naming_what_was_reported() {
2191        let other = FormationId::mint();
2192        let (failure, id) = refused(
2193            Some(&format!(
2194                r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:1"}}"#,
2195                other.as_str()
2196            )),
2197            Duration::from_secs(10),
2198        );
2199        assert_eq!(
2200            only_reason(&failure),
2201            &MemberFailureReason::WrongFormation {
2202                reported: Some(other.as_str().to_string()),
2203                expected: id.as_str().to_string(),
2204            }
2205        );
2206        assert!(failure
2207            .to_string()
2208            .contains(&format!("it reported formation {other} instead of {id}")));
2209        assert_reaped(&failure);
2210
2211        let (failure, id) = refused(
2212            Some(r#"{"session_id":"ses_standin","url":"localhost:1"}"#),
2213            Duration::from_secs(10),
2214        );
2215        assert!(failure
2216            .to_string()
2217            .contains(&format!("it reported no formation instead of {id}")));
2218        assert_reaped(&failure);
2219    }
2220
2221    #[test]
2222    fn no_line_at_all_is_refused_at_the_deadline() {
2223        let started = Instant::now();
2224        let (failure, _) = refused(None, Duration::from_millis(600));
2225        assert!(started.elapsed() >= Duration::from_millis(600));
2226        assert_eq!(
2227            only_reason(&failure),
2228            &MemberFailureReason::DidNotReport {
2229                deadline: Duration::from_millis(600)
2230            }
2231        );
2232        assert_reaped(&failure);
2233    }
2234
2235    #[test]
2236    fn a_door_that_starts_answering_before_the_deadline_is_ready() {
2237        let _guard = binary_guard();
2238        let port = door("ses_standin", Duration::from_millis(400));
2239        let dir = tempfile::tempdir().unwrap();
2240        let options = options(Duration::from_secs(20));
2241        let line = format!(
2242            r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2243            options.formation_id
2244        );
2245        std::env::set_var(
2246            child_launch::MUR_BINARY_ENV,
2247            stand_in(dir.path(), Some(&line)),
2248        );
2249        let result = launch_plan(plan_of("coder"), options);
2250        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2251        let formation = result.expect("the door answered before the deadline");
2252        assert_eq!(formation.peers().len(), 1);
2253        let peer = &formation.peers()[0];
2254        assert_eq!(peer.session_id, "ses_standin");
2255        assert_eq!(peer.url, format!("http://localhost:{port}"));
2256        assert_eq!(formation.callees_of("planner"), ["coder"]);
2257        assert!(formation.callees_of("coder").is_empty());
2258        let pid = peer.pid;
2259        assert!(crate::running::pid_is_alive(pid));
2260        let dropped = Instant::now();
2261        drop(formation);
2262        assert!(
2263            !crate::running::pid_is_alive(pid),
2264            "dropping the owner stops the peer"
2265        );
2266        assert!(
2267            dropped.elapsed() < Duration::from_secs(2),
2268            "the peer ended on its lifeline's EOF, inside the grace: {:?}",
2269            dropped.elapsed()
2270        );
2271        assert!(
2272            !dir.path().join("sigterm").exists(),
2273            "the launcher sent the peer SIGTERM"
2274        );
2275    }
2276
2277    /// A member that does not end on EOF is killed with its process group once the grace has run.
2278    #[test]
2279    fn a_member_deaf_to_its_lifeline_is_killed_after_the_grace() {
2280        let _guard = binary_guard();
2281        let port = door("ses_standin", Duration::ZERO);
2282        let dir = tempfile::tempdir().unwrap();
2283        let mut options = options(Duration::from_secs(20));
2284        options.stop_grace = Duration::from_millis(600);
2285        let line = format!(
2286            r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2287            options.formation_id
2288        );
2289        let script = dir.path().join("mur");
2290        std::fs::write(dir.path().join("line"), format!("{line}\n")).unwrap();
2291        std::fs::write(
2292            &script,
2293            format!(
2294                "#!/bin/sh\ncat '{}'\nexec sleep 600\n",
2295                dir.path().join("line").display()
2296            ),
2297        )
2298        .unwrap();
2299        use std::os::unix::fs::PermissionsExt;
2300        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2301        std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2302        let result = launch_plan(plan_of("coder"), options);
2303        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2304        let formation = result.expect("the door answers");
2305        let pid = formation.peers()[0].pid;
2306        let dropped = Instant::now();
2307        drop(formation);
2308        assert!(dropped.elapsed() >= Duration::from_millis(600));
2309        assert!(!crate::running::pid_is_alive(pid));
2310    }
2311
2312    /// Every member reads exactly its own bundle on the descriptor `MURMUR_FORMATION_CHANNEL`
2313    /// names: a token per callee and nothing about anyone else, then — once every peer is ready —
2314    /// its callees' doors. The entry member gets both lines before it starts, and an edge into it
2315    /// is served to nobody. No token is in any member's argv or environment.
2316    #[test]
2317    fn each_member_reads_only_its_own_credentials_and_callees_on_its_channel() {
2318        use crate::formation_credentials::{parse_address_line, FormationBundle};
2319        let _guard = binary_guard();
2320        let dir = tempfile::tempdir().unwrap();
2321        let coder_port = door("ses_coder", Duration::ZERO);
2322        let reviewer_port = door("ses_reviewer", Duration::ZERO);
2323        let options = options(Duration::from_secs(20));
2324        let id = options.formation_id.clone();
2325        let line = |session: &str, port: u16| {
2326            format!(
2327                r#"{{"formation_id":"{id}","session_id":"{session}","url":"localhost:{port}"}}"#
2328            )
2329        };
2330        let out = dir.path().display().to_string();
2331        let script = dir.path().join("mur");
2332        std::fs::write(
2333            &script,
2334            format!(
2335                "#!/bin/sh\n\
2336                 case \" $* \" in\n\
2337                 *' --capsule coder '*) name=coder; line='{coder}';;\n\
2338                 *' --capsule reviewer '*) name=reviewer; line='{reviewer}';;\n\
2339                 *) name=planner; line='';;\n\
2340                 esac\n\
2341                 env > '{out}/'$name.env\n\
2342                 printf '%s\\n' \"$*\" > '{out}/'$name.argv\n\
2343                 cat /dev/fd/$MURMUR_FORMATION_CHANNEL >> '{out}/'$name.channel &\n\
2344                 [ -n \"$line\" ] && printf '%s\\n' \"$line\"\n\
2345                 exec sleep 600\n",
2346                coder = line("ses_coder", coder_port),
2347                reviewer = line("ses_reviewer", reviewer_port),
2348            ),
2349        )
2350        .unwrap();
2351        use std::os::unix::fs::PermissionsExt;
2352        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2353        std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2354        let plan = FormationPlan {
2355            peers: vec![member("coder"), member("reviewer")],
2356            entry: member("planner"),
2357            // `reviewer` calls no one, so it is absent; no plan has a callee that is the entry
2358            // member, because admission never admits an edge into it.
2359            callees: vec![
2360                ("planner".to_string(), vec!["coder".to_string()]),
2361                ("coder".to_string(), vec!["reviewer".to_string()]),
2362            ],
2363        };
2364        let result = launch_plan(plan, options);
2365        let mut formation = result.expect("both stand-ins came up");
2366        formation.start_entry().expect("the entry stand-in starts");
2367        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2368
2369        let read_lines = |name: &str, count: usize| -> Vec<String> {
2370            let path = dir.path().join(format!("{name}.channel"));
2371            let deadline = Instant::now() + Duration::from_secs(10);
2372            loop {
2373                let lines: Vec<String> = std::fs::read_to_string(&path)
2374                    .unwrap_or_default()
2375                    .lines()
2376                    .map(str::to_string)
2377                    .collect();
2378                if lines.len() >= count || Instant::now() >= deadline {
2379                    return lines;
2380                }
2381                std::thread::sleep(POLL_INTERVAL);
2382            }
2383        };
2384        let coder = read_lines("coder", 2);
2385        let planner = read_lines("planner", 2);
2386        let reviewer = read_lines("reviewer", 1);
2387        std::thread::sleep(Duration::from_millis(300));
2388        assert_eq!(read_lines("reviewer", 1).len(), 1, "reviewer calls nobody");
2389
2390        let bundle = |line: &str| FormationBundle::parse_line(line).expect("a member's bundle");
2391        let names = |bundle: &FormationBundle| -> Vec<String> {
2392            bundle.calls.iter().map(|(name, _)| name.clone()).collect()
2393        };
2394        let (coder_bundle, planner_bundle, reviewer_bundle) =
2395            (bundle(&coder[0]), bundle(&planner[0]), bundle(&reviewer[0]));
2396        assert_eq!(coder_bundle.member, "coder");
2397        assert_eq!(names(&coder_bundle), ["reviewer"]);
2398        assert_eq!(names(&planner_bundle), ["coder"]);
2399        assert!(names(&reviewer_bundle).is_empty());
2400        // Every member, the entry member included, is told the one launcher that started it:
2401        // this test process.
2402        let launcher = crate::running::ProcessIdentity::of_this_process();
2403        assert!(launcher.is_some());
2404        for bundle in [&coder_bundle, &planner_bundle, &reviewer_bundle] {
2405            assert_eq!(bundle.formation_id, id);
2406            assert_eq!(bundle.verify_key, coder_bundle.verify_key);
2407            assert_eq!(bundle.launcher, launcher, "{}", bundle.member);
2408        }
2409        assert!(!planner.join("\n").contains("reviewer"), "{planner:?}");
2410        let addresses = |line: &str| -> Vec<(String, String)> {
2411            parse_address_line(line)
2412                .unwrap()
2413                .into_iter()
2414                .map(|peer| (peer.name, peer.url))
2415                .collect()
2416        };
2417        assert_eq!(
2418            addresses(&coder[1]),
2419            [(
2420                "reviewer".to_string(),
2421                format!("http://localhost:{reviewer_port}")
2422            )]
2423        );
2424        assert_eq!(
2425            addresses(&planner[1]),
2426            [(
2427                "coder".to_string(),
2428                format!("http://localhost:{coder_port}")
2429            )]
2430        );
2431
2432        for name in ["coder", "reviewer", "planner"] {
2433            let env = std::fs::read_to_string(dir.path().join(format!("{name}.env"))).unwrap();
2434            let argv = std::fs::read_to_string(dir.path().join(format!("{name}.argv"))).unwrap();
2435            assert!(!env.contains("mft1.") && !argv.contains("mft1."), "{name}");
2436            assert!(!env.contains(FORMATION_PEERS_ENV), "{name}: {env}");
2437            assert!(
2438                env.contains(&format!("{FORMATION_CHANNEL_ENV}=")),
2439                "{name}: {env}"
2440            );
2441        }
2442        let pids: Vec<u32> = formation.peers().iter().map(|peer| peer.pid).collect();
2443        drop(formation);
2444        for pid in pids {
2445            assert!(!crate::running::pid_is_alive(pid), "{pid}");
2446        }
2447    }
2448
2449    #[test]
2450    fn a_registered_member_is_killed_by_the_panic_hooks_sweep() {
2451        use std::os::unix::process::CommandExt;
2452        // The sweep kills every member this process registered, so no other launch may be live.
2453        let _guard = binary_guard();
2454        let mut child = std::process::Command::new("sleep")
2455            .arg("600")
2456            .process_group(0)
2457            .spawn()
2458            .unwrap();
2459        let slot = RegistrySlot::register(child.id(), true);
2460        assert!(slot.0.is_some());
2461        kill_registered_members();
2462        let status = child.wait().unwrap();
2463        use std::os::unix::process::ExitStatusExt;
2464        assert_eq!(status.signal(), Some(libc::SIGKILL));
2465        drop(slot);
2466        assert!(MEMBER_REGISTRY
2467            .iter()
2468            .all(|slot| slot.load(Ordering::SeqCst) != i64::from(child.id())));
2469    }
2470
2471    #[test]
2472    fn a_launch_failure_names_each_member_and_every_stopped_one() {
2473        let failure = MemberLaunchFailure {
2474            failed: vec![failure(
2475                &member("broken"),
2476                MemberFailureReason::ExitedBeforeReporting {
2477                    status: "status 1".to_string(),
2478                    stderr_tail: vec!["error[E-RUN-008]: missing".to_string()],
2479                },
2480            )],
2481            stopped: vec![
2482                StoppedMember {
2483                    name: "coder".to_string(),
2484                    pid: 7,
2485                },
2486                StoppedMember {
2487                    name: "reviewer".to_string(),
2488                    pid: 8,
2489                },
2490            ],
2491            interrupted_by: None,
2492        };
2493        assert_eq!(
2494            failure.to_string(),
2495            "the formation could not be launched\n  'broken' (broken@0.1.0): its process exited \
2496             with status 1 before reporting; its last stderr lines were:\n      \
2497             error[E-RUN-008]: missing\n  stopped: coder (pid 7), reviewer (pid 8)"
2498        );
2499    }
2500
2501    #[test]
2502    fn ready_peer_debug_prints_no_door_token() {
2503        let peer = ReadyPeer {
2504            name: "coder".to_string(),
2505            capsule: "coder".to_string(),
2506            version: "1.2.0".to_string(),
2507            pid: 1,
2508            session_id: "ses_x".to_string(),
2509            url: "http://localhost:1".to_string(),
2510            workdir: PathBuf::from("/home/u/.murmur/formations/frm_x/coder"),
2511            door_token: Some(DoorToken::new("mdt1.secretpayload.mac".to_string())),
2512        };
2513        let debug = format!("{peer:?}");
2514        assert!(!debug.contains("secretpayload"), "{debug}");
2515        assert!(debug.contains("<redacted>"), "{debug}");
2516    }
2517
2518    /// A peer's directory is `<home>/formations/<frm_id>/<member>`, every level owner-only.
2519    #[test]
2520    fn a_member_directory_is_made_owner_only_under_its_formation() {
2521        use std::os::unix::fs::PermissionsExt;
2522
2523        let home = tempfile::tempdir().unwrap();
2524        let formations = home.path().join(FORMATIONS_DIR);
2525        let id = FormationId::mint();
2526        let workdir = make_member_workdir(&formations, &id, "worker").unwrap();
2527        assert_eq!(workdir, formations.join(id.as_str()).join("worker"));
2528        for dir in [&formations, &formations.join(id.as_str()), &workdir] {
2529            let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
2530            assert_eq!(mode, 0o700, "{}", dir.display());
2531        }
2532    }
2533
2534    /// Nothing already at a member's path becomes its preopen: a directory, a file or a symlink
2535    /// there refuses the member, naming the path.
2536    #[test]
2537    fn an_existing_member_path_is_refused_and_named() {
2538        let home = tempfile::tempdir().unwrap();
2539        let formations = home.path().join(FORMATIONS_DIR);
2540        let id = FormationId::mint();
2541        make_member_workdir(&formations, &id, "worker").unwrap();
2542        let again = make_member_workdir(&formations, &id, "worker").unwrap_err();
2543        let path = formations.join(id.as_str()).join("worker");
2544        assert!(again.contains(&path.display().to_string()), "{again}");
2545        std::os::unix::fs::symlink(home.path(), formations.join(id.as_str()).join("linked"))
2546            .unwrap();
2547        assert!(make_member_workdir(&formations, &id, "linked").is_err());
2548    }
2549
2550    /// A peer whose directory cannot be made is not started, and the launch is refused naming it.
2551    #[test]
2552    fn a_peer_whose_directory_cannot_be_made_refuses_the_launch() {
2553        let _guard = binary_guard();
2554        let dir = tempfile::tempdir().unwrap();
2555        let options = options(Duration::from_secs(5));
2556        let formations = options.formations_dir.clone().unwrap();
2557        let taken = formations.join(options.formation_id.as_str()).join("coder");
2558        std::fs::create_dir_all(&taken).unwrap();
2559        std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2560        let result = launch_plan(plan_of("coder"), options);
2561        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2562        let Err(failure) = result else {
2563            panic!("the launch was expected to be refused");
2564        };
2565        assert!(failure.stopped.is_empty(), "{failure:?}");
2566        match only_reason(&failure) {
2567            MemberFailureReason::NotStarted { error } => {
2568                assert!(error.contains(&taken.display().to_string()), "{error}")
2569            }
2570            other => panic!("expected NotStarted, got {other:?}"),
2571        }
2572    }
2573
2574    /// A member's directory is `formations/<frm_id>/<member>`, and the roots `mur trace show
2575    /// frm_<id>` searches are one `.murmur` per member directory that has one, sorted, and none for
2576    /// a formation with no directory.
2577    #[test]
2578    fn member_roots_are_the_member_directories_that_hold_sessions() {
2579        assert!(formation_member_roots(&FormationId::mint()).is_empty());
2580
2581        let id = FormationId::mint();
2582        let home = tempfile::tempdir().unwrap();
2583        let formations = home.path().join(FORMATIONS_DIR);
2584        for member in ["worker", "reviewer", "idle"] {
2585            assert_eq!(
2586                make_member_workdir(&formations, &id, member).unwrap(),
2587                formations.join(id.as_str()).join(member)
2588            );
2589        }
2590        for member in ["worker", "reviewer"] {
2591            std::fs::create_dir(formations.join(id.as_str()).join(member).join(".murmur")).unwrap();
2592        }
2593        assert_eq!(
2594            formation_member_roots_in(&formations, &id),
2595            [
2596                formations.join(id.as_str()).join("reviewer/.murmur"),
2597                formations.join(id.as_str()).join("worker/.murmur"),
2598            ]
2599        );
2600    }
2601
2602    /// The marker is three keys, written and read back unchanged.
2603    #[test]
2604    fn formation_marker_round_trips_as_three_keys() {
2605        let marker = FormationMarker {
2606            project_dir: PathBuf::from("/srv/project"),
2607            launcher_pid: 4242,
2608            launcher_start: "12345".to_string(),
2609        };
2610        let value = serde_json::to_value(&marker).unwrap();
2611        assert_eq!(
2612            value,
2613            serde_json::json!({
2614                "project_dir": "/srv/project",
2615                "launcher_pid": 4242,
2616                "launcher_start": "12345",
2617            })
2618        );
2619        let dir = tempfile::tempdir().unwrap();
2620        std::fs::write(dir.path().join(FORMATION_MARKER_FILE), value.to_string()).unwrap();
2621        assert_eq!(FormationMarker::read(dir.path()), Some(marker));
2622        std::fs::write(dir.path().join(FORMATION_MARKER_FILE), "{").unwrap();
2623        assert_eq!(FormationMarker::read(dir.path()), None);
2624        assert_eq!(FormationMarker::read(&dir.path().join("missing")), None);
2625    }
2626
2627    /// A launch with peers writes its marker at `0600` into its formation's directory before the
2628    /// first peer's directory is made, naming the canonical project directory and this process as
2629    /// its launcher — and writes it even when the launch is then refused.
2630    #[test]
2631    fn formation_marker_is_written_owner_only_before_the_first_peer_directory() {
2632        use std::os::unix::fs::PermissionsExt;
2633
2634        let _guard = binary_guard();
2635        let dir = tempfile::tempdir().unwrap();
2636        let project = tempfile::tempdir().unwrap();
2637        let mut options = options(Duration::from_secs(1));
2638        options.project_dir = project.path().join(".");
2639        let formation_dir = options
2640            .formations_dir
2641            .clone()
2642            .unwrap()
2643            .join(options.formation_id.as_str());
2644        std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2645        let result = launch_plan(plan_of("coder"), options);
2646        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2647        assert!(result.is_err(), "the stand-in never reports itself");
2648
2649        let path = formation_dir.join(FORMATION_MARKER_FILE);
2650        let metadata = std::fs::metadata(&path).unwrap();
2651        assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
2652        let marker = FormationMarker::read(&formation_dir).unwrap();
2653        let pid = std::process::id();
2654        assert_eq!(
2655            marker,
2656            FormationMarker {
2657                project_dir: std::fs::canonicalize(project.path()).unwrap(),
2658                launcher_pid: pid,
2659                launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
2660            }
2661        );
2662        let peer_dir = std::fs::metadata(formation_dir.join("coder")).unwrap();
2663        assert!(
2664            metadata.modified().unwrap() <= peer_dir.modified().unwrap(),
2665            "the marker was written after the peer's directory was made"
2666        );
2667    }
2668
2669    /// A plan with no peers makes no formation directory, so there is nowhere for a marker.
2670    #[test]
2671    fn formation_marker_is_never_written_without_peers() {
2672        let _guard = binary_guard();
2673        let dir = tempfile::tempdir().unwrap();
2674        let options = options(Duration::from_secs(5));
2675        let formations = options.formations_dir.clone().unwrap();
2676        std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2677        let plan = FormationPlan {
2678            peers: Vec::new(),
2679            entry: member("planner"),
2680            callees: Vec::new(),
2681        };
2682        let result = launch_plan(plan, options);
2683        std::env::remove_var(child_launch::MUR_BINARY_ENV);
2684        let formation = result.expect("a formation with no peers has nothing to wait for");
2685        drop(formation);
2686        assert!(
2687            std::fs::read_dir(&formations).map_or(true, |mut entries| entries.next().is_none()),
2688            "{}",
2689            formations.display()
2690        );
2691    }
2692
2693    fn marker_of(pid: u32, start: &str) -> FormationMarker {
2694        FormationMarker {
2695            project_dir: PathBuf::from("/srv/project"),
2696            launcher_pid: pid,
2697            launcher_start: start.to_string(),
2698        }
2699    }
2700
2701    /// A pid this test started and has already reaped: no process holds it.
2702    fn reaped_pid() -> u32 {
2703        let mut child = std::process::Command::new("true").spawn().unwrap();
2704        let pid = child.id();
2705        child.wait().unwrap();
2706        pid
2707    }
2708
2709    fn running_record(pid: u32, formation: Option<&FormationId>) -> crate::running::RunningRecord {
2710        crate::running::RunningRecord {
2711            session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
2712            url: "127.0.0.1:1".to_string(),
2713            pid,
2714            process_start: crate::running::process_start_token(pid).unwrap_or_default(),
2715            capsule_name: "worker".to_string(),
2716            capsule_version: "0.1.0".to_string(),
2717            workdir: PathBuf::from("/tmp/worker"),
2718            outlives_launcher: false,
2719            started_at: "2026-01-01T00:00:00Z".to_string(),
2720            door_token: None,
2721            credentials: Default::default(),
2722            formation_id: formation.cloned(),
2723            formation_lifeline: false,
2724            formation_launcher: None,
2725            spawned_by: None,
2726        }
2727    }
2728
2729    /// A formation is live while its launcher is — this process with its own start token — and
2730    /// not once its launcher's pid is held by nothing.
2731    #[test]
2732    fn prune_formations_liveness_reads_the_launcher_from_the_marker() {
2733        let id = FormationId::mint();
2734        let own = std::process::id();
2735        let token = crate::running::process_start_token(own).unwrap();
2736        assert!(formation_is_live(&id, &marker_of(own, &token), Some(&[])));
2737        assert!(!formation_is_live(
2738            &id,
2739            &marker_of(reaped_pid(), &token),
2740            Some(&[])
2741        ));
2742    }
2743
2744    /// A formation whose launcher is gone is live while a running record names it and that
2745    /// record's process is alive; a record naming another formation, or a gone process, is not
2746    /// enough.
2747    #[test]
2748    fn prune_formations_liveness_reads_running_records_that_name_the_formation() {
2749        let id = FormationId::mint();
2750        let dead = marker_of(reaped_pid(), "1");
2751        let mut sleeper = std::process::Command::new("sleep")
2752            .arg("30")
2753            .spawn()
2754            .unwrap();
2755        let live = running_record(sleeper.id(), Some(&id));
2756        let other = running_record(sleeper.id(), Some(&FormationId::mint()));
2757        let unnamed = running_record(sleeper.id(), None);
2758        let gone = running_record(reaped_pid(), Some(&id));
2759
2760        assert!(formation_is_live(&id, &dead, Some(&[live])));
2761        assert!(!formation_is_live(
2762            &id,
2763            &dead,
2764            Some(&[other, unnamed, gone])
2765        ));
2766        let _ = sleeper.kill();
2767        let _ = sleeper.wait();
2768    }
2769
2770    /// A running directory that cannot be listed says nothing about what is running, so every
2771    /// formation is live and nothing is removed.
2772    #[test]
2773    fn prune_formations_liveness_treats_an_unlistable_running_directory_as_all_live() {
2774        let id = FormationId::mint();
2775        assert!(formation_is_live(&id, &marker_of(reaped_pid(), "1"), None));
2776
2777        let home = tempfile::tempdir().unwrap();
2778        let project = tempfile::tempdir().unwrap();
2779        let formations = home.path().join(FORMATIONS_DIR);
2780        let old = FormationId::parse(&format!("frm_{:012x}{:020x}", 1_000_000u64, 1)).unwrap();
2781        let old_dir = formations.join(old.as_str());
2782        std::fs::create_dir_all(old_dir.join("worker")).unwrap();
2783        let marker = FormationMarker {
2784            project_dir: std::fs::canonicalize(project.path()).unwrap(),
2785            ..marker_of(reaped_pid(), "1")
2786        };
2787        std::fs::write(
2788            old_dir.join(FORMATION_MARKER_FILE),
2789            serde_json::to_string(&marker).unwrap(),
2790        )
2791        .unwrap();
2792        let policy = murmur_artifact::TraceRetainConfig {
2793            max_sessions: Some(1),
2794            max_age_secs: Some(1),
2795        };
2796
2797        // A regular file where the running directory should be cannot be listed.
2798        let running = home.path().join("running");
2799        std::fs::write(&running, "").unwrap();
2800        let current = FormationId::mint();
2801        assert!(prune_ended_formations_in(
2802            &formations,
2803            &running,
2804            &current,
2805            project.path(),
2806            &policy
2807        )
2808        .is_empty());
2809        assert!(old_dir.is_dir());
2810
2811        // A running directory that does not exist is an empty machine, and the ended one goes.
2812        std::fs::remove_file(&running).unwrap();
2813        let pruned =
2814            prune_ended_formations_in(&formations, &running, &current, project.path(), &policy);
2815        assert_eq!(pruned.len(), 1, "{pruned:?}");
2816        assert_eq!(pruned[0].formation_id, old);
2817        assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
2818        assert!(!old_dir.exists());
2819    }
2820}