1	//! Shell wake-watcher hosting (M5-D4b — CONTEXT §Shell sleep/wake): the
     2	//! offline half of the online/offline mutual exclusivity. While a shell
     3	//! instance is **offline** and its manifest declares a `wake_command`, the
     4	//! daemon runs that template as a long-running **wake-watcher** child whose
     5	//! sole job is to fire a wake; while the shell binary runs, no watcher does
     6	//! ([`reconcile_once`] flips between them — spt-core owns the flip).
     7	//!
     8	//! **Exit-opcode supervision:** the watcher exiting with [`WAKE_OPCODE`] ⇒
     9	//! the wake resolution ([`resolve_wake`]) brings the shell online. Any other
    10	//! exit is a crash ⇒ respawn with exponential backoff ([`backoff_ms`]) and
    11	//! **give-up** after [`WakeParams::give_up_after`] consecutive crashes (a
    12	//! durable [`WAKER_GAVE_UP_FILE`] marker the next shell activity — any
    13	//! (re)launch — clears; crash-bug safety, the watcher can never crash-loop
    14	//! forever). One watcher per offline instance: the in-process [`WakeSet`]
    15	//! entry plus the perch's [`WAKER_PID_FILE`] (which also lets a freshly
    16	//! booted daemon kill a dead daemon's orphaned watcher before adopting).
    17	//!
    18	//! **Scheduling (KH 7.4):** every watcher gets its own supervision thread —
    19	//! a hung waker stalls nothing; the reconcile loop itself is one bounded
    20	//! sweep per tick off the daemon's control surfaces. Watcher children are
    21	//! *supervised* (waited) children, never detached-immortal — the KH 5.6
    22	//! pipe-inherit shape (an unwaitable child holding a capture pipe) does not
    23	//! arise; stdio is null regardless.
    24	// [impl->REQ-SHELL-2]
    25	
    26	use std::collections::HashMap;
    27	use std::path::Path;
    28	use std::process::{Command, Stdio};
    29	use std::sync::atomic::{AtomicBool, Ordering};
    30	use std::sync::{Arc, Mutex};
    31	use std::thread::JoinHandle;
    32	use std::time::Duration;
    33	
    34	use spt_runtime::manifest::{AdapterKind, Manifest, Shell};
    35	use spt_runtime::registry::AdapterRecord;
    36	use spt_store::perch::ParentHint;
    37	use spt_store::shellinfo::{self, SHELL_STATUS_OFFLINE, SHELL_STATUS_ONLINE};
    38	
    39	/// The wake opcode (documented in `docs/MANIFEST.md` §Shell adapters): a
    40	/// wake-watcher exiting with **86** fires the wake resolution; every other
    41	/// exit is a crash.
    42	pub const WAKE_OPCODE: i32 = 86;
    43	
    44	/// The running watcher's OS pid, parked on the shell perch — the
    45	/// one-watcher-per-instance lock and the cross-daemon-restart kill handle.
    46	pub const WAKER_PID_FILE: &str = "waker.pid";
    47	
    48	/// The give-up latch: present ⇒ the watcher crash-looped past its budget and
    49	/// the reconciler must NOT restart it. Cleared by the next shell activity
    50	/// (any (re)launch — [`crate::shellhost::launch_shell`]).
    51	pub const WAKER_GAVE_UP_FILE: &str = "waker.gaveup";
    52	
    53	/// Watcher supervision knobs (injectable for tests; defaults are production).
    54	#[derive(Debug, Clone, Copy)]
    55	pub struct WakeParams {
    56	    /// First-crash respawn delay; doubles per consecutive crash.
    57	    pub backoff_base_ms: u64,
    58	    /// Backoff ceiling.
    59	    pub backoff_cap_ms: u64,
    60	    /// Consecutive crash-exits before the give-up latch drops.
    61	    pub give_up_after: u32,
    62	}
    63	
    64	impl Default for WakeParams {
    65	    fn default() -> Self {
    66	        Self {
    67	            backoff_base_ms: 1_000,
    68	            backoff_cap_ms: 60_000,
    69	            give_up_after: 6,
    70	        }
    71	    }
    72	}
    73	
    74	/// Exponential backoff for `failures` consecutive crash-exits:
    75	/// `base × 2^(failures-1)`, saturating at the cap (the
    76	/// `spt-net::pairing::ratelimit` recipe). Zero failures ⇒ no wait.
    77	// [impl->REQ-SHELL-2]
    78	pub fn backoff_ms(params: &WakeParams, failures: u32) -> u64 {
    79	    if failures == 0 {
    80	        return 0;
    81	    }
    82	    params
    83	        .backoff_base_ms
    84	        .saturating_mul(1u64 << (failures - 1).min(20))
    85	        .min(params.backoff_cap_ms)
    86	}
    87	
    88	/// The live watcher registry: one supervision thread per offline instance,
    89	/// keyed `owner/shell_id`. Shared between the daemon's reconcile loop and
    90	/// anything that must stop a watcher.
    91	#[derive(Default)]
    92	pub struct WakeSet {
    93	    inner: Mutex<HashMap<(String, String), WatcherHandle>>,
    94	}
    95	
    96	struct WatcherHandle {
    97	    stop: Arc<AtomicBool>,
    98	    thread: JoinHandle<()>,
    99	}
   100	
   101	impl WakeSet {
   102	    pub fn new() -> Self {
   103	        Self::default()
   104	    }
   105	
   106	    /// How many watchers are live (finished threads pruned first).
   107	    pub fn len(&self) -> usize {
   108	        let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner());
   109	        map.retain(|_, h| !h.thread.is_finished());
   110	        map.len()
   111	    }
   112	
   113	    pub fn is_empty(&self) -> bool {
   114	        self.len() == 0
   115	    }
   116	
   117	    fn contains(&self, owner: &str, shell_id: &str) -> bool {
   118	        let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner());
   119	        match map.get(&(owner.to_string(), shell_id.to_string())) {
   120	            Some(h) if !h.thread.is_finished() => true,
   121	            Some(_) => {
   122	                map.remove(&(owner.to_string(), shell_id.to_string()));
   123	                false
   124	            }
   125	            None => false,
   126	        }
   127	    }
   128	
   129	    fn insert(&self, owner: &str, shell_id: &str, handle: WatcherHandle) {
   130	        let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner());
   131	        map.insert((owner.to_string(), shell_id.to_string()), handle);
   132	    }
   133	
   134	    /// Signal one watcher to stop and kill its child (the perch pid file is
   135	    /// the kill handle). The thread sees the kill as a non-opcode exit with
   136	    /// its stop flag up and returns without counting a failure.
   137	    fn stop_watcher(&self, owlery: &Path, owner: &str, shell_id: &str) {
   138	        let handle = {
   139	            let mut map = self.inner.lock().unwrap_or_else(|p| p.into_inner());
   140	            map.remove(&(owner.to_string(), shell_id.to_string()))
   141	        };
   142	        if let Some(h) = handle {
   143	            h.stop.store(true, Ordering::SeqCst);
   144	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, shell_id);
   145	            kill_waker_at(&perch);
   146	            let _ = h.thread.join();
   147	        }
   148	    }
   149	}
   150	
   151	/// Park the waker's pid **and its birth stamp** as one record — `<pid>\n<birth>`
   152	/// — in the single write that already existed.
   153	///
   154	/// ONE file and ONE write, because the pair must never be torn: a stamp parked
   155	/// in a second file leaves a window where the pid is on disk and the stamp is
   156	/// not, and [`spt_store::liveness::relay_liveness`] reads a live pid with no
   157	/// stamp as `Held`. That window is exactly the unauthenticated kill this
   158	/// requirement removes, reintroduced as a race.
   159	///
   160	/// A backend that exposes no start time writes the pid alone — the same
   161	/// one-line shape every pre-fix record already has, which reads `Held` on
   162	/// existence and keeps today's behaviour until the next launch.
   163	// [impl->REQ-SHELL-KILL-AUTHENTICATED]
   164	pub fn record_waker_launch(shell_perch: &Path, pid: u32) {
   165	    let body = match spt_store::proc::process_started_at(pid) {
   166	        Some(birth) => format!("{pid}\n{birth}"),
   167	        None => pid.to_string(),
   168	    };
   169	    let _ = std::fs::write(shell_perch.join(WAKER_PID_FILE), body);
   170	}
   171	
   172	/// Read that record back — **the** parse, so the two-line format has exactly one
   173	/// reader shape (the `record_shell_launch`/`read_shell_launch` pattern).
   174	///
   175	/// `None` means "no usable pid on record": absent, empty, or unparseable. Every
   176	/// caller must treat that as FAIL-TOWARD-ALIVE — refuse the kill and leave the
   177	/// record standing. A whole-content `trim().parse()` (what every reader did
   178	/// before this) returns `None` on the two-line form, so a reader left un-migrated
   179	/// would kill nothing, retire the record anyway, and report success while
   180	/// orphaning a live waker.
   181	// [impl->REQ-SHELL-KILL-AUTHENTICATED]
   182	pub fn read_waker_launch(shell_perch: &Path) -> Option<(u32, Option<u64>)> {
   183	    let raw = std::fs::read_to_string(shell_perch.join(WAKER_PID_FILE)).ok()?;
   184	    let mut lines = raw.lines();
   185	    let pid = lines.next()?.trim().parse::<u32>().ok()?;
   186	    let birth = lines.next().and_then(|l| l.trim().parse::<u64>().ok());
   187	    Some((pid, birth))
   188	}
   189	
   190	/// Kill the recorded waker pid and retire the pid file — the
   191	/// mutual-exclusivity enforcement point and the orphan cleanup
   192	/// ([`crate::shellhost::launch_shell`] calls this before the binary rises;
   193	/// the reconciler calls it before adopting an instance a dead daemon left).
   194	///
   195	/// The kill is now **authenticated**: it fires only when the pid+birth pair
   196	/// says the process at that number is the waker we spawned. Gating on bare
   197	/// `is_process_alive` (what this did before) hands a recycled pid straight to
   198	/// `taskkill /T` (KNOWN-HAZARDS 7.58).
   199	///
   200	/// Same Linux caveat as [`crate::shellhost::kill_shell_at`]: the 10ms jiffy
   201	/// resolution of `/proc/<pid>/stat` field 22 means the pair NARROWS the
   202	/// mis-fire window to a pid recycled inside the recorded start's own tick,
   203	/// rather than eliminating it. Windows `FILETIME` is unaffected.
   204	// [impl->REQ-SHELL-2]
   205	// [impl->REQ-SHELL-KILL-AUTHENTICATED]
   206	pub fn kill_waker_at(shell_perch: &Path) {
   207	    let pid_file = shell_perch.join(WAKER_PID_FILE);
   208	    let Some((pid, birth)) = read_waker_launch(shell_perch) else {
   209	        // Absent is the ordinary pre-launch case and says nothing. A record that
   210	        // EXISTS and will not parse is a different animal: refuse, say so, and
   211	        // LEAVE it — retiring it here would orphan a possibly-live waker while
   212	        // reporting success.
   213	        if pid_file.exists() {
   214	            spt_proto::emit_line_err!(
   215	                "WAKER_KILL_REFUSED_UNREADABLE:{}: a waker record exists but no pid could be \
   216	                 parsed from it — refusing the kill and keeping the record",
   217	                shell_perch.display()
   218	            );
   219	        }
   220	        return;
   221	    };
   222	    if pid == 0 {
   223	        // A backend that exposed no pid: never a process, never a kill target.
   224	        let _ = std::fs::remove_file(&pid_file);
   225	        return;
   226	    }
   227	    // The SAME policy the shell chokepoint applies — one destructive decision,
   228	    // made in one place.
   229	    match crate::shellhost::kill_verdict(spt_store::liveness::relay_liveness(Some(pid), birth)) {
   230	        crate::shellhost::KillVerdict::Fire => {
   231	            crate::shellhost::kill_shell_pid(pid);
   232	            let _ = std::fs::remove_file(&pid_file);
   233	        }
   234	        crate::shellhost::KillVerdict::RetireOnly => {
   235	            spt_proto::emit_line_err!(
   236	                "WAKER_KILL_REFUSED_GONE:{}: recorded waker pid={pid} is not the process we \
   237	                 spawned (absent, or the pid was recycled) — retiring the record and killing \
   238	                 nothing",
   239	                shell_perch.display()
   240	            );
   241	            let _ = std::fs::remove_file(&pid_file);
   242	        }
   243	        crate::shellhost::KillVerdict::RefuseAndKeep => {
   244	            spt_proto::emit_line_err!(
   245	                "WAKER_KILL_REFUSED_UNPROVEN:{}: recorded waker pid={pid} could not be \
   246	                 identified — refusing the kill and keeping the record; the waker may still be \
   247	                 running",
   248	                shell_perch.display()
   249	            );
   250	        }
   251	    }
   252	}
   253	
   254	/// Clear the give-up latch — any (re)launch is "next shell activity".
   255	pub fn clear_gave_up(shell_perch: &Path) {
   256	    let _ = std::fs::remove_file(shell_perch.join(WAKER_GAVE_UP_FILE));
   257	}
   258	
   259	/// One watcher's supervision loop (runs on its own thread): spawn the
   260	/// tokenized `wake_command`, park the pid, wait; `exit(WAKE_OPCODE)` ⇒ run
   261	/// `resolve` once and finish; any other exit ⇒ respawn with exponential
   262	/// backoff until the give-up budget, then drop the durable latch. A raised
   263	/// `stop` flag ends the loop without counting the (killed) child as a crash.
   264	// [impl->REQ-SHELL-2]
   265	pub fn watcher_run(
   266	    owlery: &Path,
   267	    owner: &str,
   268	    shell_id: &str,
   269	    tokens: &[String],
   270	    params: &WakeParams,
   271	    stop: &AtomicBool,
   272	    resolve: impl FnOnce() -> Result<String, String>,
   273	) {
   274	    let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, shell_id);
   275	    let Some((program, args)) = tokens.split_first() else {
   276	        spt_proto::emit_line_err!("WAKER_EMPTY:{owner}/{shell_id}: empty wake_command");
   277	        return;
   278	    };
   279	    let mut failures = 0u32;
   280	    while !stop.load(Ordering::SeqCst) {
   281	        let mut cmd = Command::new(program);
   282	        cmd.args(args)
   283	            .stdin(Stdio::null())
   284	            .stdout(Stdio::null())
   285	            .stderr(Stdio::null());
   286	        // The daemon (a console-less DETACHED process) hosting a
   287	        // console-subsystem waker would otherwise pop a visible Terminal
   288	        // window per spawn on Windows 11 (the KH 5.6-adjacent D3e lesson —
   289	        // mock-shell's spt_cmd does the same).
   290	        #[cfg(windows)]
   291	        {
   292	            use std::os::windows::process::CommandExt;
   293	            cmd.creation_flags(0x0800_0000); // CREATE_NO_WINDOW
   294	        }
   295	        let child = cmd.spawn();
   296	        let mut child = match child {
   297	            Ok(c) => c,
   298	            Err(e) => {
   299	                // An unspawnable waker is a permanent condition — latch now.
   300	                spt_proto::emit_line_err!("WAKER_SPAWN_FAIL:{owner}/{shell_id}: {e} (giving up)");
   301	                let _ = std::fs::write(perch.join(WAKER_GAVE_UP_FILE), b"");
   302	                return;
   303	            }
   304	        };
   305	        // The birth stamp rides WITH the pid, in one write — the kill path
   306	        // authenticates the pair, and a stamp parked separately would leave a
   307	        // window that reads Held. [impl->REQ-SHELL-KILL-AUTHENTICATED]
   308	        record_waker_launch(&perch, child.id());
   309	        let status = child.wait();
   310	        let _ = std::fs::remove_file(perch.join(WAKER_PID_FILE));
   311	        if stop.load(Ordering::SeqCst) {
   312	            return; // stopped from outside — the kill is not a crash
   313	        }
   314	        match status.ok().and_then(|s| s.code()) {
   315	            Some(WAKE_OPCODE) => {
   316	                match resolve() {
   317	                    Ok(what) => spt_proto::emit_line_err!("SHELL_WAKE:{owner}/{shell_id}: {what}"),
   318	                    Err(e) => spt_proto::emit_line_err!("SHELL_WAKE_FAIL:{owner}/{shell_id}: {e}"),
   319	                }
   320	                return;
   321	            }
   322	            _ => {
   323	                failures += 1;
   324	                if failures >= params.give_up_after {
   325	                    spt_proto::emit_line_err!(
   326	                        "WAKER_GAVE_UP:{owner}/{shell_id}: {failures} consecutive crash-exits \
   327	                         (cleared by the next relink/launch)"
   328	                    );
   329	                    let _ = std::fs::write(perch.join(WAKER_GAVE_UP_FILE), b"");
   330	                    return;
   331	                }
   332	                // Backoff in small slices so a stop request lands promptly.
   333	                let mut left = backoff_ms(params, failures);
   334	                while left > 0 && !stop.load(Ordering::SeqCst) {
   335	                    let step = left.min(50);
   336	                    std::thread::sleep(Duration::from_millis(step));
   337	                    left -= step;
   338	                }
   339	            }
   340	        }
   341	    }
   342	}
   343	
   344	/// The **state-keyed wake resolution** (M5-D4c — CONTEXT §Shell sleep/wake):
   345	/// key on the owner's local rest record, then bring the shell online —
   346	///
   347	/// - owner **dormant** (resting warm, still running) ⇒ touch nothing on the
   348	///   endpoint, just relaunch the shell binary;
   349	/// - owner **suspended** ⇒ revive the owner first
   350	///   ([`crate::resting::daemon_rest_event`] `Wake` — its cascade relaunches
   351	///   the owner's *persistent* shells, so this fn launches only if the
   352	///   cascade did not already), then the shell;
   353	/// - owner **active** (or recordless — a pre-resting/interim perch) ⇒ just
   354	///   relaunch the shell;
   355	/// - **no local instance of the owner** ⇒ refuse, naming the deferral: the
   356	///   `shell_wake_spawn_anywhere` fresh-spawn branch rides
   357	///   instantiate-anywhere (the D1c grant shape is its seam), and the
   358	///   *active-elsewhere cross-node attach* arm upgrades with presence/MRA
   359	///   (D6) + cross-node link (D8c) — D4 resolves against the local node.
   360	///
   361	/// The shell's bind onlines the perch (the D3b contract) — this fn never
   362	/// flips status.
   363	// [impl->REQ-SHELL-2]
   364	pub fn resolve_wake(
   365	    owlery: &Path,
   366	    owner: &str,
   367	    shell_id: &str,
   368	    adapter_name: &str,
   369	    shell: &Shell,
   370	) -> Result<String, String> {
   371	    // Shells hang off flat-Self perches only (the D3a layout decision).
   372	    let owner_perch = owlery.join(owner);
   373	    if !owner_perch.join("info.json").exists() {
   374	        // No local instance — consult presence (M5-D6b, REQ-PRES-1): a
   375	        // routable instance of the owner on ANOTHER node gets the wake
   376	        // FORWARDED there through the D5b remote rest op (the target node's
   377	        // own cascade relaunches ITS persistent shells). This local shell
   378	        // stays offline — the cross-node shell *link* is D8c; say so.
   379	        // [impl->REQ-PRES-1]
   380	        if let Some(node) = remote_owner_node(owner) {
   381	            return match forward_wake(owner, &node) {
   382	                Ok(outcome) => Ok(format!(
   383	                    "WAKE_FORWARDED:{owner}@{node}: {outcome}; this shell stays offline \
   384	                     here (cross-node shell link lands at D8c)"
   385	                )),
   386	                Err(e) => Err(format!("WAKE_FORWARD_FAIL:{owner}@{node}: {e}")),
   387	            };
   388	        }
   389	        return Err(format!(
   390	            "WAKE_NO_REACHABLE_INSTANCE:{owner}: no instance of the owner on any \
   391	             reachable node — a fresh-spawn-to-wake rides the deferred \
   392	             instantiate-anywhere capability (the shell_wake_spawn_anywhere grant \
   393	             shape is its seam)"
   394	        ));
   395	    }
   396	    let state = crate::resting::read_rest(&owner_perch).map(|r| r.state);
   397	    let mut did = String::new();
   398	    if state == Some(crate::resting::RestState::Suspended) {
   399	        match crate::resting::daemon_rest_event(owner, crate::resting::RestEvent::Wake, None) {
   400	            Ok(_) => did.push_str("revived owner; "),
   401	            Err(e) => return Err(format!("revive owner: {e}")),
   402	        }
   403	    } else if state == Some(crate::resting::RestState::Dormant) {
   404	        did.push_str("owner dormant (left in place); ");
   405	    }
   406	    // Never double-launch a live binary: the revive's wake cascade may have
   407	    // relaunched this (persistent) instance already — and two reconcilers
   408	    // (a fresh daemon adopting + a stale watcher firing) may race a wake.
   409	    // A parked pid that probes alive means somebody won; stand down.
   410	    let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, shell_id);
   411	    let live_pid = std::fs::read_to_string(perch.join(crate::shellhost::SHELL_PID_FILE))
   412	        .ok()
   413	        .and_then(|s| s.trim().parse::<u32>().ok())
   414	        .filter(|&p| p != 0 && spt_store::proc::is_process_alive(p));
   415	    if let Some(p) = live_pid {
   416	        return Ok(format!("{did}already relaunched pid={p} (online at bind)"));
   417	    }
   418	    // [impl->REQ-INSTALL-11] the wake-triggered relaunch resolves against the
   419	    // adapter's install dir, like every other launch of the same instance.
   420	    let install_dir =
   421	        crate::shellhost::shell_install_dir(&spt_store::perch::adapters_dir(), adapter_name);
   422	    let pid = crate::shellhost::launch_shell(
   423	        owlery,
   424	        owner,
   425	        shell_id,
   426	        adapter_name,
   427	        install_dir.as_deref(),
   428	        shell,
   429	    )?;
   430	    Ok(format!("{did}relaunched pid={pid} (online at bind)"))
   431	}
   432	
   433	/// The node (hex) holding a routable instance of `owner` per the gossiped
   434	/// registry snapshots, preferring the most-recently-active row when several
   435	/// nodes hold one (the presence datum as the tiebreak; absent data ranks
   436	/// last). `None` = the owner exists nowhere reachable.
   437	// [impl->REQ-PRES-1]
   438	fn remote_owner_node(owner: &str) -> Option<String> {
   439	    let regs = crate::presence::load_registry_snapshots(&crate::presence::registry_snapshot_dir());
   440	    let local = crate::presence::local_node_hex();
   441	    let mut best: Option<(u64, String)> = None;
   442	    for reg in regs.values() {
   443	        for row in reg.instances(owner) {
   444	            if row.node == local || !row.status.routable() {
   445	                continue;
   446	            }
   447	            let ts = row.last_active_ms.unwrap_or(0);
   448	            let better = match &best {
   449	                Some((cur, _)) => ts > *cur,
   450	                None => true,
   451	            };
   452	            if better {
   453	                best = Some((ts, row.node.clone()));
   454	            }
   455	        }
   456	    }
   457	    best.map(|(_, node)| node)
   458	}
   459	
   460	/// Forward a wake to `node` through the D5b remote rest op (the remote-drive
   461	/// trust class — the target's access gate decides). One dial, one request,
   462	/// one reply; a refusal (no reply) surfaces as an error.
   463	// [impl->REQ-PRES-1]
   464	fn forward_wake(owner: &str, node: &str) -> Result<String, String> {
   465	    use spt_net::net::endpoint::addr_for_node_hex;
   466	    let addr = addr_for_node_hex(node)
   467	        .and_then(|a| serde_json::to_value(a).ok())
   468	        .ok_or_else(|| format!("no dialable address for {node}"))?;
   469	    let mut brain =
   470	        crate::brain::Brain::cold_start(&crate::endpoint::broker_socket_name(), now_ms())
   471	            .map_err(|e| format!("broker connect: {e}"))?;
   472	    // A-4b (REQ-OPID-TRACING-RETRY): the wake-forward is a tracing-only rest op —
   473	    // if a broker restart dropped the conn/stream after the op journaled, re-mint a
   474	    // FRESH `wake` op from the SAME source and run ONCE more. The dial rides INSIDE
   475	    // `run` because that restart drops the CONN too, not just the stream — a retry
   476	    // against the dead conn would surface a different error, not self-heal. A 2nd
   477	    // no-longer-held collapses to the helper's F-1 public string.
   478	    // [impl->REQ-OPID-TRACING-RETRY]
   479	    let outcome = crate::effect::with_tracing_retry(
   480	        // The wake-forward rest op's seq source is a fresh `now_ms()` — its OWN
   481	        // minting source, distinct from shellchan's spool-row counter (doyle ruling:
   482	        // the tag names the seq source, not the subsystem), so it stamps `wake`.
   483	        || Ok(crate::effect::MintedOp::new(crate::effect::Minter::Wake, now_ms())),
   484	        |op| {
   485	            let conn = brain.net_dial(addr.clone(), None)?;
   486	            // Keep the tracing string's seq consistent with the (possibly re-minted) op.
   487	            let op_id = format!("{}:wakefwd:{}", crate::presence::local_node_hex(), op.seq);
   488	            crate::resthost::request_rest(
   489	                &mut brain,
   490	                conn.conn_id,
   491	                owner,
   492	                spt_net::net::rest::REST_EVENT_WAKE,
   493	                &op_id,
   494	                op,
   495	            )
   496	        },
   497	    )
   498	    .map_err(|e| format!("rest op: {e}"))?;
   499	    match outcome {
   500	        crate::resthost::RestRequestOutcome::Edge(d) => Ok(format!("woke ({d})")),
   501	        crate::resthost::RestRequestOutcome::NoEdge => Ok("already awake".to_string()),
   502	        crate::resthost::RestRequestOutcome::Failed(e) => Err(e),
   503	        crate::resthost::RestRequestOutcome::NoReply => {
   504	            Err("refused or dropped (no reply)".to_string())
   505	        }
   506	    }
   507	}
   508	
   509	/// Resolve one adapter **option**'s `[shell]` section through the merged view
   510	/// (composite addressing): `adapter_option` is the stored `<adapter>[:profile]`
   511	/// string, split → parent lookup in `registered` → profile overlay (shipped or
   512	/// local). Returns an **owned** [`Shell`] — a profile produces a fresh merged
   513	/// manifest, not a borrow into the parent slice. A bare name resolves to the
   514	/// parent unmodified; a deregistered parent or a failed overlay yields `None`.
   515	fn shell_section_of(
   516	    registered: &[(AdapterRecord, Manifest)],
   517	    adapters_dir: &Path,
   518	    adapter_option: &str,
   519	) -> Option<Shell> {
   520	    spt_runtime::registry::resolve_option_in(registered, adapters_dir, adapter_option)
   521	        .ok()
   522	        .filter(|m| m.adapter.kind == AdapterKind::Shell)
   523	        .and_then(|m| m.shell)
   524	}
   525	
   526	fn now_ms() -> u64 {
   527	    crate::brain::now_ms()
   528	}
   529	
   530	/// Fill a `wake_command` template.
   531	///
   532	/// **Its catalog and the spawn template's share a vocabulary; they are not the
   533	/// same set, and never claiming they are is the point of saying so here.** The
   534	/// keys mean the same thing in both (`id`, `adapter_name`, `link_token`,
   535	/// `adapter_dir`), so an author learns one manifest language — but `perch_dir`
   536	/// is **spawn-only by design** (REQ-SHELL-PERCH-DIR: it resolves a *live* link's
   537	/// perch-relative transfer paths, and a waker has no live link). A comment
   538	/// asserting set-EQUALITY over two sets that are separately extended is the
   539	/// drift this project single-sources against; it had already been false since
   540	/// `perch_dir` landed.
   541	///
   542	/// The waker's token is minted **unparked**: it substitutes (template
   543	/// compatibility) without ever verifying against the perch, because an offline
   544	/// link has no live credential by design — the close retired it, and a waker
   545	/// wakes by *exit code*, not by driving the link.
   546	///
   547	/// `install_dir` resolves the program token and fills `{adapter_dir}`, exactly as
   548	/// on the spawn side ([`crate::shellhost::fill_spawn_command`]) — a released
   549	/// adapter whose spawn resolves and whose wake does not would go permanently
   550	/// unwakeable the moment it went offline.
   551	fn fill_wake_command(
   552	    shell_id: &str,
   553	    adapter_name: &str,
   554	    install_dir: Option<&Path>,
   555	    wake_command: &str,
   556	) -> Result<Vec<String>, String> {
   557	    let mut keys = std::collections::BTreeMap::from([
   558	        ("id".to_string(), shell_id.to_string()),
   559	        ("adapter_name".to_string(), adapter_name.to_string()),
   560	        (
   561	            "link_token".to_string(),
   562	            crate::shellhost::mint_link_token(),
   563	        ),
   564	    ]);
   565	    // [impl->REQ-INSTALL-11] opt-in and N-1-safe: a template that never names the
   566	    // key fills byte-identically.
   567	    if let Some(dir) = install_dir {
   568	        keys.insert("adapter_dir".to_string(), dir.display().to_string());
   569	    }
   570	    // [impl->REQ-HAZARD-TEMPLATE-ARGV-FILL] tokenize-template-then-fill-each: a
   571	    // multi-word/quote/semicolon {key} value is exactly one argv element.
   572	    let mut tokens =
   573	        spt_runtime::runtime::fill_template_tokens(wake_command, &keys).map_err(|e| e.to_string())?;
   574	    let Some(program) = tokens.first_mut() else {
   575	        return Err("empty wake_command".into());
   576	    };
   577	    // [impl->REQ-INSTALL-11] one resolution primitive, no parallel path.
   578	    if let Some(dir) = install_dir {
   579	        *program = spt_runtime::runtime::resolve_program_in_dir(program, dir);
   580	    }
   581	    Ok(tokens)
   582	}
   583	
   584	/// One reconcile sweep — the invariant holder for the online/offline mutual
   585	/// exclusivity (CONTEXT: "spt-core flips between them"):
   586	///
   587	/// - **offline + `wake_command` + no give-up latch + no live watcher** ⇒
   588	///   kill any orphaned waker pid (a dead daemon's leftover), then start a
   589	///   supervision thread.
   590	/// - **not offline (onlined / torn down / deregistered) + live watcher** ⇒
   591	///   stop it (flag + kill).
   592	///
   593	/// Runs at daemon boot and every tick — lifecycle flips happen in CLI-process
   594	/// library code, so the *reconciler* holds the invariant, not the flipping
   595	/// caller; a flip is at most one tick stale.
   596	// [impl->REQ-SHELL-2]
   597	/// Heal every local shell record whose `online` is a lie — leg (a) of
   598	/// releases#78.
   599	///
   600	/// `effective_status` already computes this flip on EVERY read and never writes
   601	/// it down, so the record and the display disagree by design: the display side
   602	/// derives and looks clean, while the wake cascade reads the RECORDED field and
   603	/// skips its relaunch because that field still says online. A machine death
   604	/// breaks no link, so `close_shell` never runs and nothing else ever corrects it.
   605	///
   606	/// THE WRITE IS GUARDED ON AN ACTUAL CHANGE. This runs every reconcile cycle
   607	/// (5s), and a heal that rewrote the record each time would be a continuous
   608	/// stream of identical writes over a rarely-changing condition — noise on disk,
   609	/// and a record whose mtime stops meaning "something happened".
   610	///
   611	/// It DERIVES deliberately, unlike the watcher-eligibility read below it: this
   612	/// is not an eligibility decision that could relaunch a binary, it is a
   613	/// correction of a field that is already wrong. Nothing is started here.
   614	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   615	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   616	fn heal_stale_online_records(owlery: &Path) {
   617	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
   618	        for (shell_id, info) in shellinfo::list_shells(owlery, &owner) {
   619	            if info.status != SHELL_STATUS_ONLINE {
   620	                continue;
   621	            }
   622	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
   623	            let truth = shellinfo::effective_status(&perch, &info);
   624	            if truth == info.status {
   625	                continue; // the record already tells the truth — write nothing
   626	            }
   627	            let mut healed = info.clone();
   628	            healed.status = truth.to_string();
   629	            if shellinfo::write_shell_info(&perch, &healed).is_ok() {
   630	                spt_proto::emit_line_err!("SHELL_RECORD_HEALED:{owner}/{shell_id}: online -> {truth}");
   631	            }
   632	        }
   633	    }
   634	}
   635	
   636	/// Slack absorbed when asking whether a launch predates the boot instant.
   637	///
   638	/// The boot instant is DERIVED, not read from a ledger — on Windows it is
   639	/// now-minus-uptime, which drifts by however long the two calls take and by
   640	/// whatever the clock did since — so a process launched in the first moments of
   641	/// a boot can compute as very slightly older than the boot itself. Without slack
   642	/// that reads as "a corpse from the previous boot" and the sweep would relaunch
   643	/// something that had just started. Ten seconds is far wider than any plausible
   644	/// derivation error and far narrower than the gap this discriminant exists to
   645	/// detect (a previous boot is minutes to days back).
   646	pub const BOOT_RESTORE_SLACK_MS: u64 = 10_000;
   647	
   648	/// Whether a recorded launch predates the current boot — leg (b)'s discriminant
   649	/// (releases#78).
   650	///
   651	/// This is what preserves the force-kill ruling BY CONSTRUCTION rather than by a
   652	/// second rule: a shell force-killed during steady-state operation was launched
   653	/// AFTER the current boot, so it can never satisfy this predicate and is never
   654	/// spontaneously relaunched — the mid-deploy exe-overwrite case that ruling
   655	/// protects is untouched. Only a fresh boot restores.
   656	///
   657	/// Pure over its inputs so the boundary is unit-testable without a reboot.
   658	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   659	pub fn launch_predates_boot(launched_ms: u64, boot_ms: u64, slack_ms: u64) -> bool {
   660	    launched_ms.saturating_add(slack_ms) < boot_ms
   661	}
   662	
   663	/// Is this instance watcher-eligible? The class-(c) site of
   664	/// REQ-HAZARD-SHELL-STALE-ONLINE, ruled onto the corpse-boot discriminant
   665	/// (doyle, releases#78 comment 5156810267).
   666	///
   667	/// It reads the RECORDED status — never the derived one — and then asks the SAME
   668	/// `launch_predates_boot` question leg (b) asks. Three arms, and each one is the
   669	/// answer to a case the other two get wrong:
   670	///
   671	/// - **force-killed THIS boot** ⇒ record healed to offline (honest), corpse
   672	///   launched after boot ⇒ NOT eligible. This is what the class-(c) freeze
   673	///   protects: on Windows an operator kills a shell precisely to free its exe
   674	///   for overwrite (shared install dir ⇒ a routine deploy step), and a watcher
   675	///   armed here would re-lock the file under them on the next spooled frame.
   676	///   Recovery stays demand-driven — `relink`, or a `shell cmd` that wakes.
   677	/// - **restart casualty** (corpse launched BEFORE this boot) ⇒ eligible, the
   678	///   operator-greenlit scope of #78 leg (b). Two routes, one answer: the boot
   679	///   sweep restores it once, and the watcher covers it on demand thereafter.
   680	/// - **cleanly closed** (no corpse — `close_shell` retires the pid file) ⇒
   681	///   eligible exactly as before this change.
   682	///
   683	/// WHY THE CORPSE IS DEFINED ON THE PARKED STAMPS and not on a live pid re-probe:
   684	/// `shell_pid_provably_dead` runs the PAIR test, so a RECYCLED pid (alive, but
   685	/// its native start stamp mismatches the one leg (a) parked) reads as a corpse —
   686	/// because our process IS dead. Without that, a race walks straight back into the
   687	/// forbidden shape: the heal fires on the true corpse, the OS recycles its pid
   688	/// before the next tick, this read sees "alive ⇒ no corpse", and a force-killed
   689	/// instance is adopted after all.
   690	///
   691	/// NO STAMP, or no boot oracle on this platform ⇒ NOT eligible when a corpse is
   692	/// present: the discriminant cannot be evaluated, and the safe direction is to
   693	/// leave a shell down rather than arm a relaunch of a binary an operator may have
   694	/// killed on purpose. Same stance, same reason, as the boot sweep's.
   695	// [impl->REQ-HAZARD-SHELL-STALE-ONLINE]
   696	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   697	fn watcher_eligible(perch: &Path, info: &shellinfo::ShellInfo, boot_ms: Option<u64>) -> bool {
   698	    if info.status != SHELL_STATUS_OFFLINE {
   699	        return false;
   700	    }
   701	    if !shellinfo::shell_pid_provably_dead(perch) {
   702	        return true; // no corpse: cleanly closed, or never launched
   703	    }
   704	    let (Some(boot_ms), Some(launch)) = (boot_ms, shellinfo::read_shell_launch(perch)) else {
   705	        return false;
   706	    };
   707	    launch_predates_boot(launch.launched_ms, boot_ms, BOOT_RESTORE_SLACK_MS)
   708	}
   709	
   710	/// Which trigger asked for a restore.
   711	///
   712	/// The two triggers run the SAME per-owner body and differ only in what woke
   713	/// them, so each NAMES ITSELF in its event line rather than sharing one: the
   714	/// field diagnosis of releases#228 was a COUNT of these events in a rotated
   715	/// daemon log, and a shared name would have left that reading unable to say
   716	/// which trigger had fired -- the one thing it needed to know.
   717	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
   718	pub enum RestoreTrigger {
   719	    /// The once-per-daemon-generation boot sweep (releases#78 leg (b)).
   720	    Boot,
   721	    /// An owner endpoint that came online since the last reconcile pass
   722	    /// (releases#228).
   723	    OwnerOnline,
   724	}
   725	
   726	impl RestoreTrigger {
   727	    fn restored_event(self) -> &'static str {
   728	        match self {
   729	            RestoreTrigger::Boot => "SHELL_BOOT_RESTORED",
   730	            RestoreTrigger::OwnerOnline => "SHELL_OWNER_ONLINE_RESTORED",
   731	        }
   732	    }
   733	
   734	    fn failed_event(self) -> &'static str {
   735	        match self {
   736	            RestoreTrigger::Boot => "SHELL_BOOT_RESTORE_FAIL",
   737	            RestoreTrigger::OwnerOnline => "SHELL_OWNER_ONLINE_RESTORE_FAIL",
   738	        }
   739	    }
   740	}
   741	
   742	/// ONE owner's slice of the persistent-shell restore decision -- the body BOTH
   743	/// triggers run.
   744	///
   745	/// It is shared rather than restated because the refusal arms ARE the ruling:
   746	/// doyle's condition on releases#228 was that the owner-online trigger carry the
   747	/// boot sweep's arms VERBATIM, and two copies of a conjunction drift the first
   748	/// time one side is edited. Every gate is a conjunction and each one is
   749	/// load-bearing:
   750	///
   751	/// - the adapter section says `persistent` — the contract promise is scoped to
   752	///   exactly those, and a non-persistent instance was never owed a restore;
   753	/// - the OWNER endpoint is online — an offline owner is owed nothing, which is
   754	///   the difference between healing a record and resurrecting a shell nobody is
   755	///   there to drive;
   756	/// - the instance is down IN FACT (the derived read, which the birth stamp from
   757	///   leg (a) now makes trustworthy across a pid reuse);
   758	/// - and its recorded launch PREDATES THE BOOT INSTANT.
   759	///
   760	/// An instance with NO launch stamp is deliberately NOT restored: the sweep
   761	/// cannot prove it predates boot, and the safe direction here is to leave a
   762	/// shell down rather than relaunch a binary an operator may have killed on
   763	/// purpose. Such an instance gains a stamp the first time it is launched by a
   764	/// build carrying leg (a), and is covered from then on.
   765	///
   766	/// `set` is the daemon's live watcher set, and it is `None` at boot ON PURPOSE:
   767	/// the sweep runs before the reconcile loop's first tick, so no watcher can yet
   768	/// exist for the instance it is about to launch. The owner-online trigger runs
   769	/// MID-generation, where one can -- the reconcile start side arms watchers with
   770	/// no owner-online conjunct of its own -- so it passes the set and the watcher is
   771	/// stopped before the binary rises, which is the same online/offline mutual
   772	/// exclusivity `REQ-SHELL-2` holds everywhere else.
   773	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   774	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   775	// [impl->REQ-SHELL-OWNER-ONLINE-RESTORE]
   776	#[allow(clippy::too_many_arguments)]
   777	fn restore_persistent_shells_of_owner(
   778	    owlery: &Path,
   779	    registered: &[(AdapterRecord, Manifest)],
   780	    adapters_dir: &Path,
   781	    owner: &str,
   782	    boot_ms: u64,
   783	    trigger: RestoreTrigger,
   784	    set: Option<&Arc<WakeSet>>,
   785	    restored: &mut Vec<String>,
   786	) {
   787	    let owner_perch = spt_store::perch::resolve_perch_path_in(owlery, owner, ParentHint::Infer);
   788	    if !spt_store::liveness::is_perch_alive(&owner_perch) {
   789	        return; // an offline owner is owed no persistent shell
   790	    }
   791	    for (shell_id, info) in shellinfo::list_shells(owlery, owner) {
   792	        let Some(shell) =
   793	            shell_section_of(registered, adapters_dir, &info.adapter_name).filter(|s| s.persistent)
   794	        else {
   795	            continue;
   796	        };
   797	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, &shell_id);
   798	        if shellinfo::is_shell_online(&perch, &info) {
   799	            continue; // already up in fact
   800	        }
   801	        let Some(launch) = shellinfo::read_shell_launch(&perch) else {
   802	            continue; // no stamp ⇒ cannot prove it predates boot ⇒ leave it down
   803	        };
   804	        if !launch_predates_boot(launch.launched_ms, boot_ms, BOOT_RESTORE_SLACK_MS) {
   805	            continue; // launched THIS boot: a force-kill, not a restart casualty
   806	        }
   807	        // Mutual exclusivity, the online side: the watcher dies before the binary
   808	        // rises. `launch_shell` already kills the waker CHILD, but the supervising
   809	        // thread would read that as a crash-exit and respawn it under the shell we
   810	        // are starting; only the set can retire the thread. [impl->REQ-SHELL-2]
   811	        if let Some(set) = set {
   812	            set.stop_watcher(owlery, owner, &shell_id);
   813	        }
   814	        let install_dir = crate::shellhost::shell_install_dir(adapters_dir, &info.adapter_name);
   815	        match crate::shellhost::launch_shell(
   816	            owlery,
   817	            owner,
   818	            &shell_id,
   819	            &info.adapter_name,
   820	            install_dir.as_deref(),
   821	            &shell,
   822	        ) {
   823	            Ok(_) => {
   824	                eprintln!("{}:{owner}/{shell_id}", trigger.restored_event());
   825	                restored.push(format!("{owner}/{shell_id}"));
   826	            }
   827	            Err(e) => eprintln!("{}:{owner}/{shell_id}: {e}", trigger.failed_event()),
   828	        }
   829	    }
   830	}
   831	
   832	/// The once-per-daemon-generation boot sweep: bring back the `persistent` shells
   833	/// a node restart stranded (leg (b) of releases#78).
   834	///
   835	/// Runs ONCE per daemon generation. It is not the ONLY restore trigger -- an owner
   836	/// endpoint that comes online after this sweep is covered by
   837	/// [`restore_persistent_shells_on_owner_online`] (releases#228), running the same
   838	/// [`restore_persistent_shells_of_owner`] body -- but it is the one that needs no
   839	/// trigger at all, since a node restart is what strands the shells to begin with.
   840	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   841	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   842	pub fn restore_persistent_shells_at_boot(
   843	    owlery: &Path,
   844	    registered: &[(AdapterRecord, Manifest)],
   845	    adapters_dir: &Path,
   846	) -> Vec<String> {
   847	    let Some(boot_ms) = spt_store::proc::boot_instant_ms() else {
   848	        // No boot oracle on this platform: the discriminant cannot be evaluated,
   849	        // so nothing is restored. Silence beats guessing at a relaunch.
   850	        return Vec::new();
   851	    };
   852	    let mut restored = Vec::new();
   853	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
   854	        restore_persistent_shells_of_owner(
   855	            owlery,
   856	            registered,
   857	            adapters_dir,
   858	            &owner,
   859	            boot_ms,
   860	            RestoreTrigger::Boot,
   861	            None,
   862	            &mut restored,
   863	        );
   864	    }
   865	    restored
   866	}
   867	
   868	/// Per-owner online state carried ACROSS reconcile passes, so the restore below
   869	/// can be EDGE-triggered on an owner's offline->online transition.
   870	///
   871	/// WHY AN EDGE AND NOT A LEVEL (doyle's condition, releases#228): a successful
   872	/// restore self-limits, because the launch restamps `launched_ms` to now and the
   873	/// corpse-boot discriminant refuses that instance from then on. A FAILING launch
   874	/// does not -- under a level trigger ("the owner is online") it would be retried
   875	/// every [`RECONCILE_INTERVAL_MS`] for as long as the owner stayed up. The edge
   876	/// bounds it to one attempt per owner-online event.
   877	#[derive(Debug, Default)]
   878	pub struct OwnerOnlineEdge {
   879	    /// Owner id -> online as of the last pass. An owner that disappears from the
   880	    /// perch listing drops out, so a later reappearance reads as a transition.
   881	    seen: HashMap<String, bool>,
   882	}
   883	
   884	impl OwnerOnlineEdge {
   885	    pub fn new() -> Self {
   886	        Self::default()
   887	    }
   888	
   889	    /// Observe every owner and return those that went offline->online since the
   890	    /// last observation. An owner seen for the FIRST time counts as a transition
   891	    /// when it is online: it was not online at the last pass, because it was not
   892	    /// there at all.
   893	    fn transitions(&mut self, owlery: &Path) -> Vec<String> {
   894	        let mut fired = Vec::new();
   895	        let mut now = HashMap::new();
   896	        for owner in spt_store::perch::list_self_perch_ids(owlery) {
   897	            let perch = spt_store::perch::resolve_perch_path_in(owlery, &owner, ParentHint::Infer);
   898	            let online = spt_store::liveness::is_perch_alive(&perch);
   899	            if online && !self.seen.get(&owner).copied().unwrap_or(false) {
   900	                fired.push(owner.clone());
   901	            }
   902	            now.insert(owner, online);
   903	        }
   904	        self.seen = now;
   905	        fired
   906	    }
   907	
   908	    /// Observe without firing -- the initial state the first tick compares against.
   909	    ///
   910	    /// Called BEFORE the boot sweep, and the ORDER is the safe direction rather
   911	    /// than an accident. Seeded AFTER the sweep, an owner that comes online in the
   912	    /// window between the two reads as already-online and never fires an edge --
   913	    /// precisely the miss releases#228 is about. Seeded before, the worst case is a
   914	    /// second attempt at an instance the sweep already restored, which the
   915	    /// down-in-fact arm refuses.
   916	    pub fn seed(&mut self, owlery: &Path) {
   917	        let _ = self.transitions(owlery);
   918	    }
   919	}
   920	
   921	/// Restore the `persistent` shells of every owner that came online since the last
   922	/// pass -- the steady-state half of the contract's promise (releases#228).
   923	///
   924	/// Daemons boot before endpoints do, so the boot sweep correctly skips an owner
   925	/// that is not up yet and, being once-per-generation, never revisits it; bringup
   926	/// from offline emits no rest edge either, so the ADR-0048 cascade does not cover
   927	/// it. This is the trigger that does. Re-running the sweep's conjuncts is safe by
   928	/// construction and not by convention: `launch_predates_boot` is MONOTONE within a
   929	/// machine boot (every launch path restamps `launched_ms`, a kill never restamps,
   930	/// `boot_ms` is the machine boot instant), so a later evaluation can only refuse
   931	/// MORE. A shell an operator force-killed during this boot is refused here exactly
   932	/// as it is refused at boot -- forever, and by the same predicate.
   933	// [impl->REQ-SHELL-OWNER-ONLINE-RESTORE]
   934	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   935	pub fn restore_persistent_shells_on_owner_online(
   936	    owlery: &Path,
   937	    registered: &[(AdapterRecord, Manifest)],
   938	    adapters_dir: &Path,
   939	    edge: &mut OwnerOnlineEdge,
   940	    set: &Arc<WakeSet>,
   941	) -> Vec<String> {
   942	    let owners = edge.transitions(owlery);
   943	    if owners.is_empty() {
   944	        return Vec::new();
   945	    }
   946	    let Some(boot_ms) = spt_store::proc::boot_instant_ms() else {
   947	        // Same stance as the sweep's: no boot oracle => the discriminant cannot be
   948	        // evaluated => restore nothing.
   949	        return Vec::new();
   950	    };
   951	    let mut restored = Vec::new();
   952	    for owner in owners {
   953	        restore_persistent_shells_of_owner(
   954	            owlery,
   955	            registered,
   956	            adapters_dir,
   957	            &owner,
   958	            boot_ms,
   959	            RestoreTrigger::OwnerOnline,
   960	            Some(set),
   961	            &mut restored,
   962	        );
   963	    }
   964	    restored
   965	}
   966	
   967	pub fn reconcile_once(
   968	    owlery: &Path,
   969	    registered: &[(AdapterRecord, Manifest)],
   970	    adapters_dir: &Path,
   971	    set: &Arc<WakeSet>,
   972	    params: &WakeParams,
   973	) {
   974	    // Leg (a) first, every cycle: the record must stop lying before anything reads
   975	    // it below. Ordering matters -- the watcher-eligibility read further down asks
   976	    // the RECORDED field, so a heal that ran after it would leave this cycle acting
   977	    // on the stale value it just corrected.
   978	    heal_stale_online_records(owlery);
   979	
   980	    // Stop side first: watchers whose instance is gone or no longer eligible.
   981	    let live: Vec<(String, String)> = {
   982	        let mut map = set.inner.lock().unwrap_or_else(|p| p.into_inner());
   983	        map.retain(|_, h| !h.thread.is_finished());
   984	        map.keys().cloned().collect()
   985	    };
   986	    for (owner, shell_id) in live {
   987	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
   988	        let still_eligible = shellinfo::read_shell_info(&perch)
   989	            .map(|i| i.status == SHELL_STATUS_OFFLINE)
   990	            .unwrap_or(false);
   991	        if !still_eligible {
   992	            set.stop_watcher(owlery, &owner, &shell_id);
   993	        }
   994	    }
   995	
   996	    // Start side: every ELIGIBLE instance with a wake_command and headroom.
   997	    //
   998	    // Eligibility is `watcher_eligible` — recorded status (never derived) AND the
   999	    // corpse-boot discriminant. Before leg (a) the recorded field was the whole
  1000	    // guard: a force-killed instance kept a stale `online` record, which is what
  1001	    // held it out of this set. That record now tells the truth, so the protection
  1002	    // moved to the discriminant that states it directly (doyle, releases#78
  1003	    // comment 5156810267) — the same `launch_predates_boot` leg (b) uses, so the
  1004	    // freeze's forbidden behaviour stays forbidden by construction rather than by
  1005	    // a second rule.
  1006	    let boot_ms = spt_store::proc::boot_instant_ms();
  1007	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
  1008	        for (shell_id, info) in shellinfo::list_shells(owlery, &owner) {
  1009	            if set.contains(&owner, &shell_id) {
  1010	                continue;
  1011	            }
  1012	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
  1013	            if !watcher_eligible(&perch, &info, boot_ms) {
  1014	                continue;
  1015	            }
  1016	            let Some(shell) = shell_section_of(registered, adapters_dir, &info.adapter_name) else {
  1017	                continue; // deregistered adapter: nothing to run
  1018	            };
  1019	            let Some(wake_command) = shell.wake_command.as_deref() else {
  1020	                continue;
  1021	            };
  1022	            if perch.join(WAKER_GAVE_UP_FILE).exists() {
  1023	                continue; // crash-latched until the next shell activity
  1024	            }
  1025	            // Adopt cleanly: a dead daemon's watcher must not double-run.
  1026	            kill_waker_at(&perch);
  1027	            // [impl->REQ-INSTALL-11] threading only: this loop already holds the
  1028	            // registered set the record's `source_dir` lives in.
  1029	            let install_dir = crate::shellhost::shell_install_dir(adapters_dir, &info.adapter_name);
  1030	            let tokens = match fill_wake_command(
  1031	                &shell_id,
  1032	                &info.adapter_name,
  1033	                install_dir.as_deref(),
  1034	                wake_command,
  1035	            ) {
  1036	                Ok(t) => t,
  1037	                Err(e) => {
  1038	                    spt_proto::emit_line_err!("WAKER_TEMPLATE:{owner}/{shell_id}: {e}");
  1039	                    continue;
  1040	                }
  1041	            };
  1042	            let stop = Arc::new(AtomicBool::new(false));
  1043	            let handle = {
  1044	                let stop = Arc::clone(&stop);
  1045	                let owlery = owlery.to_path_buf();
  1046	                let owner = owner.clone();
  1047	                let shell_id = shell_id.clone();
  1048	                let adapter_name = info.adapter_name.clone();
  1049	                let params = *params;
  1050	                std::thread::spawn(move || {
  1051	                    watcher_run(&owlery, &owner, &shell_id, &tokens, &params, &stop, || {
  1052	                        resolve_wake(&owlery, &owner, &shell_id, &adapter_name, &shell)
  1053	                    });
  1054	                })
  1055	            };
  1056	            set.insert(
  1057	                &owner,
  1058	                &shell_id,
  1059	                WatcherHandle {
  1060	                    stop,
  1061	                    thread: handle,
  1062	                },
  1063	            );
  1064	        }
  1065	    }
  1066	}
  1067	
  1068	/// The reconcile cadence (ms): lifecycle flips land in CLI processes, so the
  1069	/// daemon's loop is the invariant holder — a flip is at most one tick stale.
  1070	pub const RECONCILE_INTERVAL_MS: u64 = 5_000;
  1071	
  1072	/// Spawn the daemon's wake host: one thread sweeping [`reconcile_once`] at
  1073	/// boot and every [`RECONCILE_INTERVAL_MS`] until `stop`. The registered set
  1074	/// is re-read each sweep (adapter add/remove lands between ticks).
  1075	// [impl->REQ-SHELL-2]
  1076	pub fn spawn_wake_host(stop: Arc<AtomicBool>) -> JoinHandle<()> {
  1077	    std::thread::spawn(move || {
  1078	        let set = Arc::new(WakeSet::new());
  1079	        let params = WakeParams::default();
  1080	        let mut owner_edge = OwnerOnlineEdge::new();
  1081	        // ONCE per daemon generation, before the reconcile loop: bring back the
  1082	        // persistent shells a node restart stranded (releases#78 leg b). Outside
  1083	        // the loop because a node restart is a boot event -- but NOT because
  1084	        // once-ness protects the operator-force-kill ruling: it does not, and a
  1085	        // comment here used to say it did. `launch_predates_boot` is what protects
  1086	        // it, on every evaluation (`reconcile_once`'s start-side comment records
  1087	        // that ruling; doyle re-measured it on releases#228). The steady-state
  1088	        // half -- an owner that comes online AFTER this sweep -- is covered in the
  1089	        // loop below, which is exactly the re-evaluation the false comment forbade.
  1090	        {
  1091	            let owlery = spt_store::perch::owlery_dir();
  1092	            let adapters_dir = spt_store::perch::adapters_dir();
  1093	            let registered = spt_runtime::registry::registered(&adapters_dir);
  1094	            // Seeded BEFORE the sweep: an owner that comes online between the two
  1095	            // must read as a transition on the first tick, not as already-online.
  1096	            owner_edge.seed(&owlery);
  1097	            restore_persistent_shells_at_boot(&owlery, &registered, &adapters_dir);
  1098	        }
  1099	        while !stop.load(Ordering::SeqCst) {
  1100	            let owlery = spt_store::perch::owlery_dir();
  1101	            let adapters_dir = spt_store::perch::adapters_dir();
  1102	            let registered = spt_runtime::registry::registered(&adapters_dir);
  1103	            // Restore BEFORE reconciling, the same order the boot pass runs in: a
  1104	            // restored instance is not watcher material, and reconcile's start side
  1105	            // would otherwise arm a watcher for the shell we are about to launch.
  1106	            restore_persistent_shells_on_owner_online(
  1107	                &owlery,
  1108	                &registered,
  1109	                &adapters_dir,
  1110	                &mut owner_edge,
  1111	                &set,
  1112	            );
  1113	            reconcile_once(&owlery, &registered, &adapters_dir, &set, &params);
  1114	            // Sleep in slices so a stop lands promptly.
  1115	            let mut left = RECONCILE_INTERVAL_MS;
  1116	            while left > 0 && !stop.load(Ordering::SeqCst) {
  1117	                let step = left.min(100);
  1118	                std::thread::sleep(Duration::from_millis(step));
  1119	                left -= step;
  1120	            }
  1121	        }
  1122	    })
  1123	}
  1124	
  1125	#[cfg(test)]
  1126	mod tests {
  1127	    use super::*;
  1128	    use spt_store::shellinfo::{spawn_record, SHELL_STATUS_ONLINE};
  1129	
  1130	    /// BACKSTOP for the converge-waits below — a ceiling on an observable, not
  1131	    /// a budget any run is expected to consume (the healthy path exits in
  1132	    /// milliseconds, and the give-up latch exits the product-fault path at
  1133	    /// once).
  1134	    ///
  1135	    /// ANCHORED, not chosen: it is `slow-timeout.period` from
  1136	    /// `.config/nextest.toml`, i.e. exactly the point at which the HARNESS
  1137	    /// ITSELF starts calling this test slow. Past that, nextest is already
  1138	    /// complaining and a longer wait cannot be the right answer — so the number
  1139	    /// has a SOURCE rather than being "some multiple of the 2s that broke",
  1140	    /// which would be a threshold keyed on a measurement instead of on the
  1141	    /// thing. If that config value changes, this should follow it.
  1142	    const WAIT_BACKSTOP: Duration = Duration::from_secs(60);
  1143	
  1144	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE] the record heal writes the truth
  1145	    // down, and writes NOTHING when the record already tells it.
  1146	    //
  1147	    // THE GUARD IS ASSERTED BY COUNTING WRITES, not by checking the final state,
  1148	    // and that distinction is the point: this runs every 5s reconcile cycle, so an
  1149	    // implementation that rewrote the record unconditionally would satisfy any
  1150	    // state-only assertion while producing a continuous stream of identical writes
  1151	    // over a condition that changes maybe twice a day — and would destroy the one
  1152	    // thing an on-disk record's mtime is good for, which is saying when something
  1153	    // actually happened. mtime is the observable that tells the two apart.
  1154	    //
  1155	    // The healthy instance beside the stale one is not decoration either: without
  1156	    // it, an implementation that healed EVERY record to offline would pass.
  1157	    #[test]
  1158	    fn the_heal_writes_the_truth_once_and_then_leaves_the_record_alone() {
  1159	        crate::test_home::with_home(|_| {
  1160	            let owlery = spt_store::perch::owlery_dir();
  1161	            let perch_path = spt_store::perch::resolve_perch_path("ling", ParentHint::Infer);
  1162	            std::fs::create_dir_all(&perch_path).unwrap();
  1163	            spt_store::info::write_info(
  1164	                &perch_path,
  1165	                &spt_store::info::InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1166	            )
  1167	            .unwrap();
  1168	
  1169	            // A STRANDED instance: record online, pid provably a corpse.
  1170	            let stale = spawn_record(&owlery, "ling", "alchemy", None).unwrap();
  1171	            let stale_perch =
  1172	                spt_store::perch::resolve_shell_perch_path_in(&owlery, "ling", &stale);
  1173	            let mut info = shellinfo::read_shell_info(&stale_perch).unwrap();
  1174	            info.status = SHELL_STATUS_ONLINE.to_string();
  1175	            shellinfo::write_shell_info(&stale_perch, &info).unwrap();
  1176	            std::fs::write(stale_perch.join(shellinfo::SHELL_PID_FILE), "1").unwrap();
  1177	            shellinfo::record_shell_launch(&stale_perch, 1, 1_000);
  1178	
  1179	            // A HEALTHY instance: record online over OUR OWN live pid, stamped by
  1180	            // its own launch. The heal must not touch it.
  1181	            let live = spawn_record(&owlery, "ling", "pacer", None).unwrap();
  1182	            let live_perch = spt_store::perch::resolve_shell_perch_path_in(&owlery, "ling", &live);
  1183	            let mut linfo = shellinfo::read_shell_info(&live_perch).unwrap();
  1184	            linfo.status = SHELL_STATUS_ONLINE.to_string();
  1185	            shellinfo::write_shell_info(&live_perch, &linfo).unwrap();
  1186	            std::fs::write(
  1187	                live_perch.join(shellinfo::SHELL_PID_FILE),
  1188	                std::process::id().to_string(),
  1189	            )
  1190	            .unwrap();
  1191	            shellinfo::record_shell_launch(&live_perch, std::process::id(), 1_000);
  1192	
  1193	            let mtime = |p: &std::path::Path| {
  1194	                std::fs::metadata(spt_store::perch::info_file_at(p))
  1195	                    .and_then(|m| m.modified())
  1196	                    .unwrap()
  1197	            };
  1198	            let live_before = mtime(&live_perch);
  1199	
  1200	            heal_stale_online_records(&owlery);
  1201	
  1202	            assert_eq!(
  1203	                shellinfo::read_shell_info(&stale_perch).unwrap().status,
  1204	                SHELL_STATUS_OFFLINE,
  1205	                "the stranded record is healed to the truth the display already showed"
  1206	            );
  1207	            assert_eq!(
  1208	                shellinfo::read_shell_info(&live_perch).unwrap().status,
  1209	                SHELL_STATUS_ONLINE,
  1210	                "and a genuinely live instance is left online"
  1211	            );
  1212	            assert_eq!(
  1213	                mtime(&live_perch),
  1214	                live_before,
  1215	                "the healthy record was not REWRITTEN — an unconditional heal would \
  1216	                 pass every assertion above and fail this one"
  1217	            );
  1218	
  1219	            // Second cycle over the now-correct records: nothing may be written.
  1220	            let stale_after_heal = mtime(&stale_perch);
  1221	            let live_after_heal = mtime(&live_perch);
  1222	            heal_stale_online_records(&owlery);
  1223	            assert_eq!(
  1224	                (mtime(&stale_perch), mtime(&live_perch)),
  1225	                (stale_after_heal, live_after_heal),
  1226	                "a second cycle over records that already tell the truth writes nothing"
  1227	            );
  1228	        });
  1229	    }
  1230	
  1231	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE] the boot discriminant's boundary,
  1232	    // asserted from BOTH sides of it plus the slack band.
  1233	    //
  1234	    // The pair is what makes it evidence: a predicate answering "yes" to everything
  1235	    // would restore force-killed shells (breaking the class-c ruling this
  1236	    // discriminant exists to preserve), and one answering "no" to everything would
  1237	    // restore nothing and still pass any single-sided row. The slack row is the
  1238	    // third case and it is not decoration -- the boot instant is DERIVED (Windows:
  1239	    // now minus uptime), so a process launched in the first moments after boot can
  1240	    // compute as marginally older than boot itself, and without slack the sweep
  1241	    // would relaunch something that had only just started.
  1242	    #[test]
  1243	    fn only_a_launch_from_before_this_boot_is_a_restart_casualty() {
  1244	        const BOOT: u64 = 1_000_000;
  1245	
  1246	        assert!(
  1247	            launch_predates_boot(BOOT - 60_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1248	            "launched a minute before boot ⇒ a restart casualty, restore it"
  1249	        );
  1250	        assert!(
  1251	            !launch_predates_boot(BOOT + 30_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1252	            "launched AFTER boot ⇒ a force-kill during steady state, never a \
  1253	             spontaneous relaunch (the class-c ruling, preserved by construction)"
  1254	        );
  1255	        assert!(
  1256	            !launch_predates_boot(BOOT - 1_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1257	            "inside the slack band ⇒ treated as this boot, since the boot instant \
  1258	             is derived and drifts by a little"
  1259	        );
  1260	        assert!(
  1261	            !launch_predates_boot(BOOT, BOOT, BOOT_RESTORE_SLACK_MS),
  1262	            "exactly at the boot instant is not before it"
  1263	        );
  1264	    }
  1265	
  1266	    // [unit->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS] THE RESTART SHAPE, built
  1267	    // WITHOUT a suspend edge — the row the existing cascade test cannot express.
  1268	    //
  1269	    // `rest_edges_cascade_shells_with_divergence` suspends first, and the suspend
  1270	    // WRITES the offline record, so by the time its wake edge fires the instance is
  1271	    // already in the state the to_active arm requires: that fixture can never
  1272	    // observe this class. This one constructs the restart shape directly — a record
  1273	    // left ONLINE over a corpse, a launch stamp from before boot, an online owner —
  1274	    // and calls no rest path anywhere in its setup.
  1275	    //
  1276	    // Three instances, because the sweep is a CONJUNCTION and a single-instance
  1277	    // fixture cannot tell a correct sweep from one that restores everything: the
  1278	    // stranded one must come back, the one force-killed THIS boot must not, and the
  1279	    // one whose owner is offline must not.
  1280	    #[test]
  1281	    fn the_boot_sweep_restores_only_the_restart_casualty_of_an_online_owner() {
  1282	        let tmp = tempfile::tempdir().unwrap();
  1283	        let owlery = tmp.path();
  1284	        let boot = spt_store::proc::boot_instant_ms().expect("a boot instant on this platform");
  1285	
  1286	        // An ONLINE owner (seed_owner writes the `{}` record is_perch_alive reads as
  1287	        // alive by interim parity) and an explicitly OFFLINE one.
  1288	        seed_owner(owlery, "ling");
  1289	        let gone = owlery.join("gone");
  1290	        std::fs::create_dir_all(&gone).unwrap();
  1291	        std::fs::write(gone.join("info.json"), "{\"status\":\"offline\"}").unwrap();
  1292	
  1293	        #[cfg(windows)]
  1294	        let noop = "cmd /c exit 0";
  1295	        #[cfg(unix)]
  1296	        let noop = "true";
  1297	        let adapters_dir = tmp.path().join("adapters");
  1298	        let src = tmp.path().join("srcs").join("mock-boot");
  1299	        std::fs::create_dir_all(&src).unwrap();
  1300	        std::fs::write(
  1301	            src.join("manifest.toml"),
  1302	            format!(
  1303	                "[adapter]\nname = \"mock-boot\"\nkind = \"shell\"\nversion = \"1\"\n\
  1304	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1305	                 persistent = true\n"
  1306	            ),
  1307	        )
  1308	        .unwrap();
  1309	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1310	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1311	
  1312	        // Every instance is recorded ONLINE over a dead pid — the restart shape —
  1313	        // and they differ ONLY in the gate under test.
  1314	        let mk = |owner: &str, launched_ms: u64| -> String {
  1315	            let id = spawn_record(owlery, owner, "mock-boot", None).unwrap();
  1316	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, &id);
  1317	            let mut i = shellinfo::read_shell_info(&perch).unwrap();
  1318	            i.status = SHELL_STATUS_ONLINE.to_string();
  1319	            shellinfo::write_shell_info(&perch, &i).unwrap();
  1320	            std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), "2000000000").unwrap();
  1321	            let launch = shellinfo::ShellLaunch {
  1322	                pid_started_at: Some(1),
  1323	                launched_ms,
  1324	            };
  1325	            std::fs::write(
  1326	                perch.join(shellinfo::SHELL_LAUNCH_FILE),
  1327	                serde_json::to_string(&launch).unwrap(),
  1328	            )
  1329	            .unwrap();
  1330	            id
  1331	        };
  1332	        let stranded = mk("ling", boot.saturating_sub(600_000)); // 10 min before boot
  1333	        let force_killed = mk("ling", boot + 120_000); // 2 min AFTER boot
  1334	        let orphan_owner = mk("gone", boot.saturating_sub(600_000));
  1335	
  1336	        let restored = restore_persistent_shells_at_boot(owlery, &registered, &adapters_dir);
  1337	
  1338	        assert!(
  1339	            restored.contains(&format!("ling/{stranded}")),
  1340	            "the restart casualty of an ONLINE owner is restored — the case the \
  1341	             cascade test cannot reach: {restored:?}"
  1342	        );
  1343	        // Owner-QUALIFIED comparisons throughout: shell ids are minted per owner, so
  1344	        // "mock-boot-0" exists under BOTH owners here and a bare substring match
  1345	        // collides across them (it did, on this row's first run).
  1346	        assert!(
  1347	            !restored.contains(&format!("ling/{force_killed}")),
  1348	            "an instance launched THIS boot is a force-kill, not a restart casualty, \
  1349	             and must never be spontaneously relaunched: {restored:?}"
  1350	        );
  1351	        assert!(
  1352	            !restored.contains(&format!("gone/{orphan_owner}")),
  1353	            "an OFFLINE owner is owed no persistent shell: {restored:?}"
  1354	        );
  1355	    }
  1356	
  1357	    // [unit->REQ-SHELL-OWNER-ONLINE-RESTORE]
  1358	    // [unit->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
  1359	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1360	    // THE SEAM releases#78 LEFT: the sweep runs while the owner endpoint is still
  1361	    // down (daemons boot before endpoints do), spends its one generation restoring
  1362	    // NOTHING, and the owner comes online afterwards with no rest edge anywhere in
  1363	    // the setup -- so neither the sweep nor the ADR-0048 cascade can cover it.
  1364	    //
  1365	    // Three arms in ONE fixture, and each is the answer to a case the others get
  1366	    // wrong (the BAROMETER W2 warning this milestone is held to: a single-instance
  1367	    // fixture cannot tell a correct trigger from one that restores everything):
  1368	    //
  1369	    // - the restart casualty of the now-online owner is RESTORED -- the row the
  1370	    //   ticket is about;
  1371	    // - a sibling force-killed THIS boot is NOT -- the refusal arm carried verbatim
  1372	    //   from the sweep, which an implementation that simply relaunches everything
  1373	    //   for an online owner would fail;
  1374	    // - a SECOND pass with the owner still online restores a freshly stranded
  1375	    //   instance NOT AT ALL -- the trigger is an EDGE, not the level "the owner is
  1376	    //   online". That is the arm no success case can express: a successful restore
  1377	    //   restamps itself out of eligibility, so only an instance that never had its
  1378	    //   edge can tell a level trigger from an edge one.
  1379	    #[test]
  1380	    fn an_owner_coming_online_after_a_spent_sweep_restores_only_its_restart_casualty() {
  1381	        let tmp = tempfile::tempdir().unwrap();
  1382	        let owlery = tmp.path();
  1383	        let boot = spt_store::proc::boot_instant_ms().expect("a boot instant on this platform");
  1384	
  1385	        // The owner is OFFLINE at sweep time -- the ordering that strands the shell.
  1386	        let owner_dir = owlery.join("ling");
  1387	        std::fs::create_dir_all(&owner_dir).unwrap();
  1388	        let offline_owner = "{\"status\":\"offline\"}";
  1389	        std::fs::write(owner_dir.join("info.json"), offline_owner).unwrap();
  1390	
  1391	        #[cfg(windows)]
  1392	        let noop = "cmd /c exit 0";
  1393	        #[cfg(unix)]
  1394	        let noop = "true";
  1395	        let adapters_dir = tmp.path().join("adapters");
  1396	        let src = tmp.path().join("srcs").join("mock-owner-online");
  1397	        std::fs::create_dir_all(&src).unwrap();
  1398	        std::fs::write(
  1399	            src.join("manifest.toml"),
  1400	            format!(
  1401	                "[adapter]\nname = \"mock-owner-online\"\nkind = \"shell\"\nversion = \"1\"\n\
  1402	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1403	                 persistent = true\n"
  1404	            ),
  1405	        )
  1406	        .unwrap();
  1407	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1408	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1409	
  1410	        // The restart shape, built with NO suspend/rest path anywhere: recorded
  1411	        // ONLINE over a dead pid, with a parked launch stamp.
  1412	        let mk = |launched_ms: u64| -> String {
  1413	            let id = spawn_record(owlery, "ling", "mock-owner-online", None).unwrap();
  1414	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "ling", &id);
  1415	            let mut i = shellinfo::read_shell_info(&perch).unwrap();
  1416	            i.status = SHELL_STATUS_ONLINE.to_string();
  1417	            shellinfo::write_shell_info(&perch, &i).unwrap();
  1418	            std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), "2000000000").unwrap();
  1419	            let launch = shellinfo::ShellLaunch {
  1420	                pid_started_at: Some(1),
  1421	                launched_ms,
  1422	            };
  1423	            std::fs::write(
  1424	                perch.join(shellinfo::SHELL_LAUNCH_FILE),
  1425	                serde_json::to_string(&launch).unwrap(),
  1426	            )
  1427	            .unwrap();
  1428	            id
  1429	        };
  1430	        let stranded = mk(boot.saturating_sub(600_000)); // 10 min BEFORE boot
  1431	        let force_killed = mk(boot + 120_000); // 2 min AFTER boot
  1432	
  1433	        // The daemon's own order: seed the edge, then spend the sweep.
  1434	        let mut edge = OwnerOnlineEdge::new();
  1435	        edge.seed(owlery);
  1436	        let swept = restore_persistent_shells_at_boot(owlery, &registered, &adapters_dir);
  1437	        assert!(
  1438	            swept.is_empty(),
  1439	            "the sweep is SPENT and restored nothing: its owner-online conjunct \
  1440	             refused, correctly, and it never runs again this generation: {swept:?}"
  1441	        );
  1442	
  1443	        // The owner comes online. No rest edge is emitted by bringup — this write is
  1444	        // the endpoint record changing, nothing more.
  1445	        std::fs::write(owner_dir.join("info.json"), "{}").unwrap();
  1446	
  1447	        let set = Arc::new(WakeSet::new());
  1448	        let restored =
  1449	            restore_persistent_shells_on_owner_online(owlery, &registered, &adapters_dir, &mut edge, &set);
  1450	
  1451	        assert!(
  1452	            restored.contains(&format!("ling/{stranded}")),
  1453	            "the owner came online after the sweep, so THIS is the trigger that owes \
  1454	             the restore — the seam releases#228 reported: {restored:?}"
  1455	        );
  1456	        assert!(
  1457	            !restored.contains(&format!("ling/{force_killed}")),
  1458	            "an instance launched THIS boot is a force-kill, refused by the same \
  1459	             discriminant the sweep uses — re-evaluating can only refuse MORE: \
  1460	             {restored:?}"
  1461	        );
  1462	
  1463	        // ...and the trigger is an EDGE: the owner stays online, a NEW casualty is
  1464	        // stranded, and the next pass does nothing. A level trigger would take it.
  1465	        let late = mk(boot.saturating_sub(600_000));
  1466	        let again =
  1467	            restore_persistent_shells_on_owner_online(owlery, &registered, &adapters_dir, &mut edge, &set);
  1468	        assert!(
  1469	            again.is_empty(),
  1470	            "no NEW owner-online transition ⇒ no restore attempt at all; this is what \
  1471	             bounds a failing launch to one attempt per owner-online event instead of \
  1472	             one every five seconds forever (ling/{late} stayed down): {again:?}"
  1473	        );
  1474	    }
  1475	
  1476	    fn fast_params() -> WakeParams {
  1477	        WakeParams {
  1478	            backoff_base_ms: 10,
  1479	            backoff_cap_ms: 40,
  1480	            give_up_after: 3,
  1481	        }
  1482	    }
  1483	
  1484	    // [unit->REQ-INSTALL-11] the wake site, matching the spawn site: a bare
  1485	    // program token binds to the shipped binary and {adapter_dir} fills. Wiring
  1486	    // spawn alone would leave a released adapter launchable but PERMANENTLY
  1487	    // UNWAKEABLE the moment it went offline — the worse half of the gap, because
  1488	    // the wake path is the one that runs with nobody watching.
  1489	    #[test]
  1490	    fn a_wake_template_resolves_its_program_and_adapter_dir_against_the_install_dir() {
  1491	        let tmp = tempfile::tempdir().unwrap();
  1492	        let install = tmp.path().join("adapter dir"); // spaces: the argv-fill shape
  1493	        std::fs::create_dir_all(&install).unwrap();
  1494	        let shipped = if cfg!(windows) { "waker.exe" } else { "waker" };
  1495	        std::fs::write(install.join(shipped), b"").unwrap();
  1496	
  1497	        let tokens =
  1498	            fill_wake_command("sh-1", "mock-shell", Some(&install), "waker --root {adapter_dir}")
  1499	                .expect("adapter_dir is a wake substitution key");
  1500	        assert_eq!(
  1501	            tokens[0],
  1502	            install.join(shipped).display().to_string(),
  1503	            "the bare token must bind to the SHIPPED binary, not fall through to PATH"
  1504	        );
  1505	        assert_eq!(
  1506	            tokens[2],
  1507	            install.display().to_string(),
  1508	            "the install dir (spaces included) is exactly one argv element"
  1509	        );
  1510	    }
  1511	
  1512	    // [unit->REQ-INSTALL-11] the catalogs share a VOCABULARY, they are not one
  1513	    // set — the relationship the corrected `fill_wake_command` docstring states,
  1514	    // pinned so a future edit cannot quietly restore the set-equality claim.
  1515	    // `adapter_dir` is in BOTH; `perch_dir` is spawn-only by design (a waker has
  1516	    // no live link, so there is no perch-relative transfer path to resolve).
  1517	    #[test]
  1518	    fn the_wake_catalog_shares_the_spawn_vocabulary_but_is_not_the_same_set() {
  1519	        let install = std::path::PathBuf::from("/adapters/mock");
  1520	        assert!(
  1521	            fill_wake_command("sh-1", "mock-shell", Some(&install), "w {adapter_dir}").is_ok(),
  1522	            "adapter_dir is shared by both catalogs"
  1523	        );
  1524	        assert!(
  1525	            fill_wake_command("sh-1", "mock-shell", Some(&install), "w {perch_dir}").is_err(),
  1526	            "perch_dir is SPAWN-ONLY — a waker has no live link to resolve paths against"
  1527	        );
  1528	    }
  1529	
  1530	    fn seed_owner(owlery: &Path, owner: &str) {
  1531	        let p = owlery.join(owner);
  1532	        std::fs::create_dir_all(&p).unwrap();
  1533	        std::fs::write(p.join("info.json"), "{}").unwrap();
  1534	    }
  1535	
  1536	    /// Tokens for a waker that exits with `code` immediately.
  1537	    fn exit_with(code: i32) -> Vec<String> {
  1538	        #[cfg(windows)]
  1539	        return vec!["cmd".into(), "/c".into(), format!("exit {code}")];
  1540	        #[cfg(unix)]
  1541	        return vec!["sh".into(), "-c".into(), format!("exit {code}")];
  1542	    }
  1543	
  1544	    /// Tokens for a waker that appends one line to `tally` then exits 1.
  1545	    fn crash_tallying(tally: &Path) -> Vec<String> {
  1546	        #[cfg(windows)]
  1547	        return vec![
  1548	            "powershell".into(),
  1549	            "-NoProfile".into(),
  1550	            "-Command".into(),
  1551	            format!("Add-Content -Path '{}' -Value x; exit 1", tally.display()),
  1552	        ];
  1553	        #[cfg(unix)]
  1554	        return vec![
  1555	            "sh".into(),
  1556	            "-c".into(),
  1557	            format!("echo x >> '{}'; exit 1", tally.display()),
  1558	        ];
  1559	    }
  1560	
  1561	    // [unit->REQ-SHELL-2] the backoff curve: base × 2^(n-1), saturating at
  1562	    // the cap; zero failures wait nothing.
  1563	    #[test]
  1564	    fn backoff_curve_doubles_to_the_cap() {
  1565	        let p = WakeParams {
  1566	            backoff_base_ms: 100,
  1567	            backoff_cap_ms: 1_500,
  1568	            give_up_after: 6,
  1569	        };
  1570	        assert_eq!(backoff_ms(&p, 0), 0);
  1571	        assert_eq!(backoff_ms(&p, 1), 100);
  1572	        assert_eq!(backoff_ms(&p, 2), 200);
  1573	        assert_eq!(backoff_ms(&p, 4), 800);
  1574	        assert_eq!(backoff_ms(&p, 5), 1_500, "saturates at the cap");
  1575	        assert_eq!(
  1576	            backoff_ms(&p, 60),
  1577	            1_500,
  1578	            "huge failure counts can't overflow"
  1579	        );
  1580	    }
  1581	
  1582	    // [unit->REQ-SHELL-2] exit-opcode supervision: exit(86) fires the wake
  1583	    // resolution exactly once and the loop ends; the pid file is retired.
  1584	    #[test]
  1585	    fn watcher_opcode_exit_fires_resolution_once() {
  1586	        let tmp = tempfile::tempdir().unwrap();
  1587	        let owlery = tmp.path();
  1588	        seed_owner(owlery, "doyle");
  1589	        let id = spawn_record(owlery, "doyle", "mock-shell", None).unwrap();
  1590	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  1591	
  1592	        let fired = std::sync::atomic::AtomicU32::new(0);
  1593	        let stop = AtomicBool::new(false);
  1594	        watcher_run(
  1595	            owlery,
  1596	            "doyle",
  1597	            &id,
  1598	            &exit_with(WAKE_OPCODE),
  1599	            &fast_params(),
  1600	            &stop,
  1601	            || {
  1602	                fired.fetch_add(1, Ordering::SeqCst);
  1603	                Ok("test-resolved".into())
  1604	            },
  1605	        );
  1606	        assert_eq!(
  1607	            fired.load(Ordering::SeqCst),
  1608	            1,
  1609	            "resolution fired exactly once"
  1610	        );
  1611	        assert!(!perch.join(WAKER_PID_FILE).exists(), "pid file retired");
  1612	        assert!(
  1613	            !perch.join(WAKER_GAVE_UP_FILE).exists(),
  1614	            "an opcode exit is not a crash"
  1615	        );
  1616	    }
  1617	
  1618	    // [unit->REQ-SHELL-2] crash-exit supervision: a crashing waker respawns
  1619	    // (with backoff) exactly give_up_after times, then the durable give-up
  1620	    // latch drops and the resolution never fires; clear_gave_up re-arms.
  1621	    #[test]
  1622	    fn watcher_crash_exits_respawn_then_give_up() {
  1623	        let tmp = tempfile::tempdir().unwrap();
  1624	        let owlery = tmp.path();
  1625	        seed_owner(owlery, "doyle");
  1626	        let id = spawn_record(owlery, "doyle", "mock-shell", None).unwrap();
  1627	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  1628	        let tally = perch.join("waker-tally.txt");
  1629	
  1630	        let stop = AtomicBool::new(false);
  1631	        watcher_run(
  1632	            owlery,
  1633	            "doyle",
  1634	            &id,
  1635	            &crash_tallying(&tally),
  1636	            &fast_params(),
  1637	            &stop,
  1638	            || -> Result<String, String> { panic!("a crash-exit must never resolve a wake") },
  1639	        );
  1640	        let runs = std::fs::read_to_string(&tally).unwrap().lines().count();
  1641	        assert_eq!(runs, 3, "respawned exactly to the give-up budget");
  1642	        assert!(
  1643	            perch.join(WAKER_GAVE_UP_FILE).exists(),
  1644	            "the durable latch dropped"
  1645	        );
  1646	
  1647	        clear_gave_up(&perch);
  1648	        assert!(
  1649	            !perch.join(WAKER_GAVE_UP_FILE).exists(),
  1650	            "shell activity re-arms"
  1651	        );
  1652	    }
  1653	
  1654	    /// A `[shell]` section parsed from TOML (the manifest is the only
  1655	    /// constructor — its serde defaults are part of the contract).
  1656	    fn shell_section(spawn: &str, extra: &str) -> Shell {
  1657	        let toml_src = format!(
  1658	            "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1659	             min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{spawn}'\n{extra}\n"
  1660	        );
  1661	        spt_runtime::Manifest::from_toml_str(&toml_src)
  1662	            .unwrap()
  1663	            .shell
  1664	            .unwrap()
  1665	    }
  1666	
  1667	    #[cfg(windows)]
  1668	    const NOOP: &str = "cmd /c exit 0";
  1669	    #[cfg(unix)]
  1670	    const NOOP: &str = "true";
  1671	
  1672	    // [unit->REQ-SHELL-2] wake resolution, the no-reachable branch: no local
  1673	    // instance of the owner ⇒ refuse naming the instantiate-anywhere
  1674	    // deferral — never a silent no-op, never a fresh spawn.
  1675	    #[test]
  1676	    fn resolve_wake_refuses_without_a_reachable_owner() {
  1677	        let tmp = tempfile::tempdir().unwrap();
  1678	        let owlery = tmp.path();
  1679	        // The shell perch exists; the OWNER's perch does not (home node gone).
  1680	        let id = spawn_record(owlery, "ghost", "mock-wake", None).unwrap();
  1681	        let err =
  1682	            resolve_wake(owlery, "ghost", &id, "mock-wake", &shell_section(NOOP, "")).unwrap_err();
  1683	        assert!(err.contains("WAKE_NO_REACHABLE_INSTANCE"), "{err}");
  1684	        assert!(
  1685	            err.contains("instantiate-anywhere"),
  1686	            "the deferral is named: {err}"
  1687	        );
  1688	    }
  1689	
  1690	    // [unit->REQ-SHELL-2] wake resolution, the dormant branch: the owner is
  1691	    // resting warm — touch nothing on the endpoint, just relaunch the shell.
  1692	    #[test]
  1693	    fn resolve_wake_leaves_a_dormant_owner_and_relaunches() {
  1694	        use spt_store::info::{write_info, InfoJson};
  1695	        let tmp = tempfile::tempdir().unwrap();
  1696	        let owlery = tmp.path();
  1697	        let owner_perch = owlery.join("ling");
  1698	        std::fs::create_dir_all(&owner_perch).unwrap();
  1699	        write_info(
  1700	            &owner_perch,
  1701	            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1702	        )
  1703	        .unwrap();
  1704	        // Dormant is WARM (bound, live, resting in place): pin the owner online
  1705	        // so the liveness-aware derivation reads Dormant, not the host-dependent
  1706	        // pid-4242 cold-probe (which would collapse to Suspended, mis-routing the
  1707	        // wake to the suspended branch). REQ-EFFECTIVE-INSTANCE-STATE.
  1708	        spt_store::info::set_status(&owner_perch, spt_store::liveness::STATUS_ONLINE).unwrap();
  1709	        crate::resting::write_rest(&owner_perch, crate::resting::RestState::Dormant, 1_000)
  1710	            .unwrap();
  1711	
  1712	        let id = spawn_record(owlery, "ling", "mock-wake", None).unwrap();
  1713	        let msg = resolve_wake(owlery, "ling", &id, "mock-wake", &shell_section(NOOP, ""))
  1714	            .expect("resolves");
  1715	        assert!(msg.contains("owner dormant"), "{msg}");
  1716	        assert!(msg.contains("relaunched"), "{msg}");
  1717	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "ling", &id);
  1718	        assert!(
  1719	            perch.join(crate::shellhost::SHELL_PID_FILE).exists(),
  1720	            "binary relaunched"
  1721	        );
  1722	        assert_eq!(
  1723	            crate::resting::read_rest(&owner_perch).unwrap().state,
  1724	            crate::resting::RestState::Dormant,
  1725	            "a dormant owner is left in place"
  1726	        );
  1727	    }
  1728	
  1729	    // [unit->REQ-SHELL-2] wake resolution, the suspended branch: revive the
  1730	    // owner FIRST (the real daemon_rest_event — its wake cascade relaunches
  1731	    // persistent shells, and the resolution then must NOT double-launch a
  1732	    // live binary); a non-persistent shell is launched by the resolution
  1733	    // itself after the revive.
  1734	    #[test]
  1735	    fn resolve_wake_revives_a_suspended_owner_without_double_launch() {
  1736	        crate::test_home::with_home(|home| {
  1737	            use spt_store::info::{write_info, InfoJson};
  1738	            use spt_store::perch;
  1739	            let owlery = perch::owlery_dir();
  1740	            let owner_perch =
  1741	                perch::resolve_perch_path("ling", spt_store::perch::ParentHint::Infer);
  1742	            std::fs::create_dir_all(&owner_perch).unwrap();
  1743	            write_info(
  1744	                &owner_perch,
  1745	                &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1746	            )
  1747	            .unwrap();
  1748	            // Pin the owner online (warm) so the effective state is
  1749	            // intent-refined by the written rest record: the initial Suspended
  1750	            // record ⇒ from=Suspended (the first Wake is a real revive edge),
  1751	            // and after the revive writes Active the later re-suspend is a real
  1752	            // Active→Suspended edge. A cold (offline) pin would collapse EVERY
  1753	            // intent to Suspended, no-edging the re-suspend. Never the
  1754	            // host-dependent pid-4242 probe. REQ-EFFECTIVE-INSTANCE-STATE.
  1755	            spt_store::info::set_status(&owner_perch, spt_store::liveness::STATUS_ONLINE).unwrap();
  1756	            crate::resting::write_rest(&owner_perch, crate::resting::RestState::Suspended, 1_000)
  1757	                .unwrap();
  1758	
  1759	            // A registered PERSISTENT sleeper adapter — the revive's wake
  1760	            // cascade relaunches it and the binary stays alive.
  1761	            #[cfg(windows)]
  1762	            let sleeper = "ping -n 30 127.0.0.1";
  1763	            #[cfg(unix)]
  1764	            let sleeper = "sleep 30";
  1765	            let adapters = perch::adapters_dir();
  1766	            for (name, spawn, extra) in [
  1767	                ("mock-wake", sleeper, "persistent = true"),
  1768	                ("mock-noper", NOOP, ""), // the cascade must skip this one
  1769	            ] {
  1770	                let src = home.join("srcs").join(name);
  1771	                std::fs::create_dir_all(&src).unwrap();
  1772	                std::fs::write(
  1773	                    src.join("manifest.toml"),
  1774	                    format!(
  1775	                        "[adapter]\nname = \"{name}\"\nkind = \"shell\"\nversion = \"1\"\n\
  1776	                         min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{spawn}'\n{extra}\n"
  1777	                    ),
  1778	                )
  1779	                .unwrap();
  1780	                spt_runtime::registry::register(&adapters, &src, 1).unwrap();
  1781	            }
  1782	            let shell = shell_section(sleeper, "persistent = true");
  1783	
  1784	            let id = spawn_record(&owlery, "ling", "mock-wake", None).unwrap();
  1785	            let msg = resolve_wake(&owlery, "ling", &id, "mock-wake", &shell).expect("resolves");
  1786	            assert!(msg.contains("revived owner"), "{msg}");
  1787	            assert!(
  1788	                msg.contains("already relaunched"),
  1789	                "a live cascade launch is never doubled: {msg}"
  1790	            );
  1791	            assert_eq!(
  1792	                crate::resting::read_rest(&owner_perch).unwrap().state,
  1793	                crate::resting::RestState::Active,
  1794	                "suspended owner revived"
  1795	            );
  1796	
  1797	            // Re-suspend and resolve a NON-persistent shell of the same
  1798	            // owner: the revive's cascade skips it (its registered adapter
  1799	            // is not persistent), so the resolution launches it itself.
  1800	            crate::resting::daemon_rest_event("ling", crate::resting::RestEvent::Suspend, None)
  1801	                .expect("ok")
  1802	                .expect("edge");
  1803	            let shell_np = shell_section(NOOP, "");
  1804	            let id2 = spawn_record(&owlery, "ling", "mock-noper", None).unwrap();
  1805	            let msg =
  1806	                resolve_wake(&owlery, "ling", &id2, "mock-noper", &shell_np).expect("resolves");
  1807	            assert!(msg.contains("revived owner"), "{msg}");
  1808	            assert!(msg.contains("relaunched pid="), "{msg}");
  1809	        });
  1810	    }
  1811	
  1812	    // [unit->REQ-HAZARD-SHELL-STALE-ONLINE] a force-killed instance is NOT adopted
  1813	    // and its binary is NOT relaunched behind the operator's back — the class-(c)
  1814	    // protection, which BAROMETER W2 moved off the stale record and onto the
  1815	    // corpse-boot discriminant (this row's assertions are unchanged from the day it
  1816	    // caught that conflict: it went red when leg (a) healed the record out from
  1817	    // under the old guard, and it is green again on the replacement). The corpse
  1818	    // here carries NO launch stamp, so the discriminant cannot prove it a restart
  1819	    // casualty and the safe direction holds it out. Deliberate, operator-ratified
  1820	    // (flynn 2026-07-25): every reason to deliberately stop a shell is a reason not
  1821	    // to want it back a tick later — mid-deploy the relaunch would run the OLD
  1822	    // binary out of the file being replaced (worse than a failed install), and it
  1823	    // turns quarantining a misbehaving shell into a restart loop. Recovery is
  1824	    // demand-driven instead: `relink` (unblocked by the derived gate) or a `shell
  1825	    // cmd` that wakes. The offline sibling proves the rig would adopt if eligible.
  1826	    #[test]
  1827	    fn reconcile_never_adopts_a_stale_online_instance() {
  1828	        const DEAD_PID: u32 = 2_000_000_000;
  1829	        let tmp = tempfile::tempdir().unwrap();
  1830	        let owlery = tmp.path();
  1831	        seed_owner(owlery, "doyle");
  1832	
  1833	        #[cfg(windows)]
  1834	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1835	        #[cfg(unix)]
  1836	        let (noop, sleeper) = ("true", "sleep 30");
  1837	        let adapters_dir = tmp.path().join("adapters");
  1838	        let src = tmp.path().join("srcs").join("mock-wake");
  1839	        std::fs::create_dir_all(&src).unwrap();
  1840	        std::fs::write(
  1841	            src.join("manifest.toml"),
  1842	            format!(
  1843	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1844	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1845	                 persistent = true\nwake_command = '{sleeper}'\n"
  1846	            ),
  1847	        )
  1848	        .unwrap();
  1849	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1850	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1851	
  1852	        // The stale-online instance: bound, then its binary was force-killed.
  1853	        let dead = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1854	        let dead_perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &dead);
  1855	        let mut info = shellinfo::read_shell_info(&dead_perch).unwrap();
  1856	        info.status = SHELL_STATUS_ONLINE.to_string();
  1857	        shellinfo::write_shell_info(&dead_perch, &info).unwrap();
  1858	        std::fs::write(
  1859	            dead_perch.join(shellinfo::SHELL_PID_FILE),
  1860	            DEAD_PID.to_string(),
  1861	        )
  1862	        .unwrap();
  1863	
  1864	        let set = Arc::new(WakeSet::new());
  1865	        let params = fast_params();
  1866	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1867	        assert_eq!(
  1868	            set.len(),
  1869	            0,
  1870	            "a force-killed instance must NOT be adopted — no spontaneous relaunch"
  1871	        );
  1872	        assert!(
  1873	            !dead_perch.join(WAKER_PID_FILE).exists(),
  1874	            "no waker child was started for it either"
  1875	        );
  1876	
  1877	        // Control: a genuinely OFFLINE sibling is still adopted, so the absence
  1878	        // above is the derivation policy, not a broken rig.
  1879	        let off = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1880	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1881	        assert_eq!(set.len(), 1, "the offline sibling {off} still gets a watcher");
  1882	        set.stop_watcher(owlery, "doyle", &off);
  1883	    }
  1884	
  1885	    // [unit->REQ-HAZARD-SHELL-STALE-ONLINE] [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1886	    // BOTH HALVES IN ONE ROW: for a same-boot corpse the heal FIRED (the record now
  1887	    // reads offline — the honest value leg (a) owes it) AND the watcher count is
  1888	    // still 0. Asserted together on purpose. Split across two rows either half
  1889	    // passes alone on a broken build: without the heal the record still says online
  1890	    // and a status-only read refuses adoption for the WRONG reason (the old
  1891	    // accidental protection, which is exactly what this milestone removed), while a
  1892	    // watcher-0 assertion on its own is satisfied by an implementation that adopts
  1893	    // nothing at all.
  1894	    //
  1895	    // The pre-boot sibling is the contrast that keeps the discriminant from being
  1896	    // vacuous: an implementation treating EVERY corpse as ineligible passes the
  1897	    // same-boot half and silently strands every restart casualty's watcher. Both
  1898	    // instances are the identical shape — record online over a corpse with a parked
  1899	    // launch stamp — and differ ONLY in which side of the boot instant it falls on.
  1900	    #[test]
  1901	    fn the_heal_fires_yet_a_same_boot_corpse_is_never_adopted() {
  1902	        const DEAD_PID: u32 = 2_000_000_000;
  1903	        let tmp = tempfile::tempdir().unwrap();
  1904	        let owlery = tmp.path();
  1905	        seed_owner(owlery, "doyle");
  1906	        let boot = spt_store::proc::boot_instant_ms().expect("a boot instant on this platform");
  1907	
  1908	        #[cfg(windows)]
  1909	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1910	        #[cfg(unix)]
  1911	        let (noop, sleeper) = ("true", "sleep 30");
  1912	        let adapters_dir = tmp.path().join("adapters");
  1913	        let src = tmp.path().join("srcs").join("mock-wake");
  1914	        std::fs::create_dir_all(&src).unwrap();
  1915	        std::fs::write(
  1916	            src.join("manifest.toml"),
  1917	            format!(
  1918	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1919	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1920	                 persistent = true\nwake_command = '{sleeper}'\n"
  1921	            ),
  1922	        )
  1923	        .unwrap();
  1924	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1925	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1926	
  1927	        let perch_of =
  1928	            |id: &str| spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", id);
  1929	        let mk = |launched_ms: u64| -> String {
  1930	            let id = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1931	            let perch = perch_of(&id);
  1932	            let mut i = shellinfo::read_shell_info(&perch).unwrap();
  1933	            i.status = SHELL_STATUS_ONLINE.to_string();
  1934	            shellinfo::write_shell_info(&perch, &i).unwrap();
  1935	            std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), DEAD_PID.to_string()).unwrap();
  1936	            let launch = shellinfo::ShellLaunch {
  1937	                pid_started_at: Some(1),
  1938	                launched_ms,
  1939	            };
  1940	            std::fs::write(
  1941	                perch.join(shellinfo::SHELL_LAUNCH_FILE),
  1942	                serde_json::to_string(&launch).unwrap(),
  1943	            )
  1944	            .unwrap();
  1945	            id
  1946	        };
  1947	        let force_killed = mk(boot + 120_000); // killed 2 min INTO this boot
  1948	        let casualty = mk(boot.saturating_sub(600_000)); // launched 10 min BEFORE it
  1949	
  1950	        let set = Arc::new(WakeSet::new());
  1951	        let params = fast_params();
  1952	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1953	
  1954	        assert_eq!(
  1955	            shellinfo::read_shell_info(&perch_of(&force_killed))
  1956	                .unwrap()
  1957	                .status,
  1958	            SHELL_STATUS_OFFLINE,
  1959	            "the heal FIRED — the record stopped lying about a corpse"
  1960	        );
  1961	        assert!(
  1962	            !set.contains("doyle", &force_killed),
  1963	            "...and the honest record still did not buy it a watcher: killed THIS \
  1964	             boot is the mid-deploy case the class-(c) freeze protects"
  1965	        );
  1966	        assert!(
  1967	            !perch_of(&force_killed).join(WAKER_PID_FILE).exists(),
  1968	            "no waker child was started for it either"
  1969	        );
  1970	        assert!(
  1971	            set.contains("doyle", &casualty),
  1972	            "a corpse launched BEFORE this boot is a restart casualty — eligible, \
  1973	             per #78 leg (b)'s greenlit scope"
  1974	        );
  1975	        assert_eq!(set.len(), 1, "and it is the ONLY instance adopted");
  1976	        set.stop_watcher(owlery, "doyle", &casualty);
  1977	    }
  1978	
  1979	    // [unit->REQ-SHELL-2] the reconciler holds the mutual exclusivity: an
  1980	    // offline instance with a wake_command gets exactly ONE watcher (a second
  1981	    // sweep never doubles it); a give-up latch suppresses adoption; onlining
  1982	    // the instance stops the watcher on the next sweep.
  1983	    #[test]
  1984	    fn reconcile_flips_watchers_with_instance_state() {
  1985	        let tmp = tempfile::tempdir().unwrap();
  1986	        let owlery = tmp.path();
  1987	        seed_owner(owlery, "doyle");
  1988	
  1989	        // A registered shell adapter whose waker sleeps (stays alive between
  1990	        // sweeps) — the long-running watcher case.
  1991	        #[cfg(windows)]
  1992	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1993	        #[cfg(unix)]
  1994	        let (noop, sleeper) = ("true", "sleep 30");
  1995	        let adapters_dir = tmp.path().join("adapters");
  1996	        let src = tmp.path().join("srcs").join("mock-wake");
  1997	        std::fs::create_dir_all(&src).unwrap();
  1998	        std::fs::write(
  1999	            src.join("manifest.toml"),
  2000	            format!(
  2001	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  2002	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  2003	                 wake_command = '{sleeper}'\n"
  2004	            ),
  2005	        )
  2006	        .unwrap();
  2007	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  2008	        let registered = spt_runtime::registry::registered(&adapters_dir);
  2009	
  2010	        let id = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  2011	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  2012	
  2013	        let set = Arc::new(WakeSet::new());
  2014	        let params = fast_params();
  2015	
  2016	        // Offline (the spawn_record default) ⇒ one watcher; never two.
  2017	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2018	        assert_eq!(set.len(), 1, "offline + wake_command ⇒ a watcher");
  2019	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2020	        assert_eq!(set.len(), 1, "a second sweep never doubles it");
  2021	        // The child parked its pid (it is the long sleeper, still up).
  2022	        //
  2023	        // READ-AND-PARSE, never `exists()`. THIS is the defect that went red on
  2024	        // kitsubito, and it is a CREATE-BEFORE-WRITE race, not a slow spawn:
  2025	        // `std::fs::write` CREATES the pid file and THEN writes into it, so a
  2026	        // reader can observe a zero-byte file. The old gate did exactly that —
  2027	        // `exists()` answered true at ~15ms, `read_to_string` returned "", and
  2028	        // `.parse()` panicked `ParseIntError { kind: Empty }`. The whole test
  2029	        // failed in 15 MILLISECONDS: its 2s ceiling was never approached, so
  2030	        // this was never a timing budget problem. Load only WIDENS the
  2031	        // create→write window and makes the race observable — a contributor,
  2032	        // never the mechanism.
  2033	        //
  2034	        // Parsing IS the readiness test: an empty file means NOT YET PARKED,
  2035	        // which is why this loop cannot be written as exists-then-read.
  2036	        //
  2037	        // The backstop below is defence against a DIFFERENT, unobserved failure
  2038	        // — a watcher whose child never spawns at all. `reconcile_once` returns
  2039	        // once the watcher THREAD is registered and the thread spawns the child
  2040	        // itself, so there is nothing to join and the wait is open-ended in
  2041	        // principle. Two terminal outcomes, so a failure says WHICH happened:
  2042	        // the pid parks, or the watcher LATCHES give-up (a product fault, and a
  2043	        // permanent one — exit at once rather than burn the backstop).
  2044	        let gave_up = perch.join(WAKER_GAVE_UP_FILE);
  2045	        let deadline = std::time::Instant::now() + WAIT_BACKSTOP;
  2046	        let mut parked: Option<u32> = None;
  2047	        let mut spawn_gave_up = false;
  2048	        while std::time::Instant::now() < deadline {
  2049	            // Read-and-PARSE, not `exists()`: the file appears before the write
  2050	            // lands, so an exists-gate can hand the next line an empty string.
  2051	            // Through the SHARED parse — a whole-content `trim().parse()` here
  2052	            // would read the two-line record as un-parked forever.
  2053	            if let Some((pid, _birth)) = read_waker_launch(&perch) {
  2054	                parked = Some(pid);
  2055	                break;
  2056	            }
  2057	            if gave_up.exists() {
  2058	                spawn_gave_up = true;
  2059	                break;
  2060	            }
  2061	            std::thread::sleep(Duration::from_millis(20));
  2062	        }
  2063	        assert!(
  2064	            !spawn_gave_up,
  2065	            "the waker child could NOT BE SPAWNED (WAKER_GAVE_UP latched) — a \
  2066	             product fault, not a slow box"
  2067	        );
  2068	        let waker_pid = parked.expect(
  2069	            "the waker never parked a pid and never latched give-up: the watcher \
  2070	             thread had not yet spawned its child. If this fires, the box is \
  2071	             saturated beyond a 60s spawn — suspect the pool, not this seam",
  2072	        );
  2073	        assert!(
  2074	            spt_store::proc::is_process_alive(waker_pid),
  2075	            "the waker runs while offline"
  2076	        );
  2077	
  2078	        // Online the instance ⇒ the next sweep stops the watcher + kills the
  2079	        // child (mutual exclusivity, the binary side wins).
  2080	        let mut info = shellinfo::read_shell_info(&perch).unwrap();
  2081	        info.status = SHELL_STATUS_ONLINE.to_string();
  2082	        shellinfo::write_shell_info(&perch, &info).unwrap();
  2083	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2084	        assert_eq!(set.len(), 0, "onlined instance keeps no watcher");
  2085	        // Same shape as the park wait, one assertion down: dropping the watcher
  2086	        // is synchronous (`set.len()` above proves it), but the CHILD's death is
  2087	        // the OS's business and lands whenever it lands. Converge on it rather
  2088	        // than asserting it on the next instruction.
  2089	        let deadline = std::time::Instant::now() + WAIT_BACKSTOP;
  2090	        let mut child_dead = false;
  2091	        while std::time::Instant::now() < deadline {
  2092	            if !spt_store::proc::is_process_alive(waker_pid) {
  2093	                child_dead = true;
  2094	                break;
  2095	            }
  2096	            std::thread::sleep(Duration::from_millis(20));
  2097	        }
  2098	        assert!(
  2099	            child_dead,
  2100	            "the waker child was killed when its instance came online (mutual \
  2101	             exclusivity — the binary side wins)"
  2102	        );
  2103	
  2104	        // Back offline but crash-latched ⇒ no adoption.
  2105	        let mut info = shellinfo::read_shell_info(&perch).unwrap();
  2106	        info.status = SHELL_STATUS_OFFLINE.to_string();
  2107	        shellinfo::write_shell_info(&perch, &info).unwrap();
  2108	        std::fs::write(perch.join(WAKER_GAVE_UP_FILE), b"").unwrap();
  2109	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2110	        assert_eq!(set.len(), 0, "the give-up latch suppresses adoption");
  2111	    }
  2112	
  2113	    // [unit->REQ-MANIFEST-2] the wake seam resolves the **merged view**: a
  2114	    // shipped profile that adds `wake_command` makes a `<adapter>:<profile>`
  2115	    // instance wakeable, while the bare parent (no wake_command) stays inert —
  2116	    // proof the daemon re-resolves the stored composite, not the parent.
  2117	    #[test]
  2118	    fn reconcile_resolves_profile_overlay() {
  2119	        let tmp = tempfile::tempdir().unwrap();
  2120	        let owlery = tmp.path();
  2121	        seed_owner(owlery, "doyle");
  2122	
  2123	        #[cfg(windows)]
  2124	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  2125	        #[cfg(unix)]
  2126	        let (noop, sleeper) = ("true", "sleep 30");
  2127	        let adapters_dir = tmp.path().join("adapters");
  2128	        let src = tmp.path().join("srcs").join("mock-wakeprof");
  2129	        std::fs::create_dir_all(&src).unwrap();
  2130	        std::fs::write(
  2131	            src.join("manifest.toml"),
  2132	            format!(
  2133	                "[adapter]\nname = \"mock-wakeprof\"\nkind = \"shell\"\nversion = \"1\"\n\
  2134	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\n\
  2135	                 [profiles.waker]\n[profiles.waker.shell]\nwake_command = '{sleeper}'\n"
  2136	            ),
  2137	        )
  2138	        .unwrap();
  2139	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  2140	        let registered = spt_runtime::registry::registered(&adapters_dir);
  2141	
  2142	        // A bare-parent instance: no wake_command in the parent ⇒ never woken.
  2143	        spawn_record(owlery, "doyle", "mock-wakeprof", None).unwrap();
  2144	        // A profiled instance: mint a clean id off the parent, then store the
  2145	        // composite as its adapter_name (the id stays colon-free; the carrier
  2146	        // holds the option — the #8 identity split).
  2147	        let pid = spawn_record(owlery, "doyle", "mock-wakeprof", None).unwrap();
  2148	        let pperch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &pid);
  2149	        let mut info = shellinfo::read_shell_info(&pperch).unwrap();
  2150	        info.adapter_name = "mock-wakeprof:waker".to_string();
  2151	        shellinfo::write_shell_info(&pperch, &info).unwrap();
  2152	
  2153	        let set = Arc::new(WakeSet::new());
  2154	        let params = fast_params();
  2155	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2156	        assert_eq!(
  2157	            set.len(),
  2158	            1,
  2159	            "only the profiled instance (overlay adds wake_command) gets a watcher"
  2160	        );
  2161	        assert!(set.contains("doyle", &pid), "the profiled instance is the one watched");
  2162	
  2163	        set.stop_watcher(owlery, "doyle", &pid);
  2164	    }
  2165	
  2166	    // ──────────────────────────────────────────────────────────────────────
  2167	    // Waker kill authentication (REQ-SHELL-KILL-AUTHENTICATED)
  2168	    // ──────────────────────────────────────────────────────────────────────
  2169	
  2170	    fn waker_stand_in() -> std::process::Child {
  2171	        #[cfg(windows)]
  2172	        {
  2173	            use std::os::windows::process::CommandExt;
  2174	            Command::new("cmd")
  2175	                .args(["/c", "ping -n 60 127.0.0.1"])
  2176	                // stdin too — see the note on `live_stand_in`.
  2177	                .stdin(Stdio::null())
  2178	                .stdout(Stdio::null())
  2179	                .stderr(Stdio::null())
  2180	                .creation_flags(0x0800_0000) // CREATE_NO_WINDOW
  2181	                .spawn()
  2182	                .expect("spawned the stand-in waker")
  2183	        }
  2184	        #[cfg(unix)]
  2185	        {
  2186	            Command::new("sleep")
  2187	                .arg("60")
  2188	                .stdin(Stdio::null())
  2189	                .stdout(Stdio::null())
  2190	                .stderr(Stdio::null())
  2191	                .spawn()
  2192	                .expect("spawned the stand-in waker")
  2193	        }
  2194	    }
  2195	
  2196	    fn waker_died_within(child: &mut std::process::Child, ms: u64) -> bool {
  2197	        let deadline = std::time::Instant::now() + Duration::from_millis(ms);
  2198	        loop {
  2199	            if matches!(child.try_wait(), Ok(Some(_))) {
  2200	                return true;
  2201	            }
  2202	            if std::time::Instant::now() >= deadline {
  2203	                return false;
  2204	            }
  2205	            std::thread::sleep(Duration::from_millis(25));
  2206	        }
  2207	    }
  2208	
  2209	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] The pair round-trips through the ONE
  2210	    // shared parse, and a legacy one-line record still reads — the format change
  2211	    // must not orphan the records already on disk.
  2212	    #[test]
  2213	    fn waker_record_round_trips_and_reads_the_legacy_shape() {
  2214	        let d = tempfile::tempdir().unwrap();
  2215	        let perch = d.path();
  2216	
  2217	        let me = std::process::id();
  2218	        record_waker_launch(perch, me);
  2219	        let (pid, birth) = read_waker_launch(perch).expect("the record parses");
  2220	        assert_eq!(pid, me);
  2221	        assert_eq!(
  2222	            birth,
  2223	            spt_store::proc::process_started_at(me),
  2224	            "the stamp parked beside the pid is the pid's own birth"
  2225	        );
  2226	
  2227	        // The pre-fix shape: one line, no stamp.
  2228	        std::fs::write(perch.join(WAKER_PID_FILE), me.to_string()).unwrap();
  2229	        assert_eq!(
  2230	            read_waker_launch(perch),
  2231	            Some((me, None)),
  2232	            "a legacy one-line record still yields its pid, with no stamp to compare"
  2233	        );
  2234	    }
  2235	
  2236	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] A recycled waker pid is spared, and
  2237	    // its record retired. Without the WRITE-side stamp this row is unreachable:
  2238	    // a bare pid reads Held and the kill fires.
  2239	    #[test]
  2240	    fn kill_waker_at_spares_a_recycled_pid() {
  2241	        let d = tempfile::tempdir().unwrap();
  2242	        let perch = d.path();
  2243	        let mut stranger = waker_stand_in();
  2244	        // The recycled shape: a LIVE pid beside a stamp belonging to something
  2245	        // else — written by hand, since `record_waker_launch` would pair honestly.
  2246	        std::fs::write(
  2247	            perch.join(WAKER_PID_FILE),
  2248	            format!("{}\n123", stranger.id()),
  2249	        )
  2250	        .unwrap();
  2251	
  2252	        kill_waker_at(perch);
  2253	
  2254	        let died = waker_died_within(&mut stranger, 750);
  2255	        crate::shellhost::kill_shell_pid(stranger.id());
  2256	        let _ = stranger.wait();
  2257	        assert!(!died, "a waker record whose stamp mismatches must not kill");
  2258	        assert!(
  2259	            !perch.join(WAKER_PID_FILE).exists(),
  2260	            "a provably-not-ours waker record is retired"
  2261	        );
  2262	    }
  2263	
  2264	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] Non-vacuity for the waker half: a
  2265	    // genuine pair still dies. Asserted beside the row above, because a gate
  2266	    // that spares everything passes that one on its own.
  2267	    #[test]
  2268	    fn kill_waker_at_still_kills_a_matching_pair() {
  2269	        let d = tempfile::tempdir().unwrap();
  2270	        let perch = d.path();
  2271	        let mut ours = waker_stand_in();
  2272	        record_waker_launch(perch, ours.id());
  2273	
  2274	        kill_waker_at(perch);
  2275	
  2276	        let died = waker_died_within(&mut ours, 5_000);
  2277	        crate::shellhost::kill_shell_pid(ours.id());
  2278	        let _ = ours.wait();
  2279	        assert!(died, "the authenticated waker kill must still kill our own");
  2280	    }
  2281	
  2282	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] BOTH halves of the parse-failure arm,
  2283	    // asserted together: refuse the kill AND KEEP the record. Retiring it while
  2284	    // killing nothing is the shape that orphans a live waker and reports
  2285	    // success — strictly worse than the defect being fixed, and a kill-count
  2286	    // assertion alone would pass it.
  2287	    #[test]
  2288	    fn kill_waker_at_refuses_and_keeps_an_unparseable_record() {
  2289	        let d = tempfile::tempdir().unwrap();
  2290	        let perch = d.path();
  2291	        let mut stranger = waker_stand_in();
  2292	        std::fs::write(perch.join(WAKER_PID_FILE), "not-a-pid\n").unwrap();
  2293	
  2294	        kill_waker_at(perch);
  2295	
  2296	        let died = waker_died_within(&mut stranger, 500);
  2297	        crate::shellhost::kill_shell_pid(stranger.id());
  2298	        let _ = stranger.wait();
  2299	        assert!(!died, "an unreadable record kills nothing");
  2300	        assert!(
  2301	            perch.join(WAKER_PID_FILE).exists(),
  2302	            "an unreadable record is KEPT — retiring it would orphan a live waker while \
  2303	             reporting success"
  2304	        );
  2305	    }
  2306	}