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 live, birth-matching launch wins even before its bind handshake.
   410	    let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, shell_id);
   411	    if let Some(p) = crate::shellhost::live_launch_winner(&perch) {
   412	        return Ok(format!("{did}already relaunched pid={p} (online at bind)"));
   413	    }
   414	    // [impl->REQ-INSTALL-11] the wake-triggered relaunch resolves against the
   415	    // adapter's install dir, like every other launch of the same instance.
   416	    let install_dir =
   417	        crate::shellhost::shell_install_dir(&spt_store::perch::adapters_dir(), adapter_name);
   418	    let pid = crate::shellhost::launch_shell(
   419	        owlery,
   420	        owner,
   421	        shell_id,
   422	        adapter_name,
   423	        install_dir.as_deref(),
   424	        shell,
   425	    )?;
   426	    Ok(format!("{did}relaunched pid={pid} (online at bind)"))
   427	}
   428	
   429	/// The node (hex) holding a routable instance of `owner` per the gossiped
   430	/// registry snapshots, preferring the most-recently-active row when several
   431	/// nodes hold one (the presence datum as the tiebreak; absent data ranks
   432	/// last). `None` = the owner exists nowhere reachable.
   433	// [impl->REQ-PRES-1]
   434	fn remote_owner_node(owner: &str) -> Option<String> {
   435	    let regs = crate::presence::load_registry_snapshots(&crate::presence::registry_snapshot_dir());
   436	    let local = crate::presence::local_node_hex();
   437	    let mut best: Option<(u64, String)> = None;
   438	    for reg in regs.values() {
   439	        for row in reg.instances(owner) {
   440	            if row.node == local || !row.status.routable() {
   441	                continue;
   442	            }
   443	            let ts = row.last_active_ms.unwrap_or(0);
   444	            let better = match &best {
   445	                Some((cur, _)) => ts > *cur,
   446	                None => true,
   447	            };
   448	            if better {
   449	                best = Some((ts, row.node.clone()));
   450	            }
   451	        }
   452	    }
   453	    best.map(|(_, node)| node)
   454	}
   455	
   456	/// Forward a wake to `node` through the D5b remote rest op (the remote-drive
   457	/// trust class — the target's access gate decides). One dial, one request,
   458	/// one reply; a refusal (no reply) surfaces as an error.
   459	// [impl->REQ-PRES-1]
   460	fn forward_wake(owner: &str, node: &str) -> Result<String, String> {
   461	    use spt_net::net::endpoint::addr_for_node_hex;
   462	    let addr = addr_for_node_hex(node)
   463	        .and_then(|a| serde_json::to_value(a).ok())
   464	        .ok_or_else(|| format!("no dialable address for {node}"))?;
   465	    let mut brain =
   466	        crate::brain::Brain::cold_start(&crate::endpoint::broker_socket_name(), now_ms())
   467	            .map_err(|e| format!("broker connect: {e}"))?;
   468	    // A-4b (REQ-OPID-TRACING-RETRY): the wake-forward is a tracing-only rest op —
   469	    // if a broker restart dropped the conn/stream after the op journaled, re-mint a
   470	    // FRESH `wake` op from the SAME source and run ONCE more. The dial rides INSIDE
   471	    // `run` because that restart drops the CONN too, not just the stream — a retry
   472	    // against the dead conn would surface a different error, not self-heal. A 2nd
   473	    // no-longer-held collapses to the helper's F-1 public string.
   474	    // [impl->REQ-OPID-TRACING-RETRY]
   475	    let outcome = crate::effect::with_tracing_retry(
   476	        // The wake-forward rest op's seq source is a fresh `now_ms()` — its OWN
   477	        // minting source, distinct from shellchan's spool-row counter (doyle ruling:
   478	        // the tag names the seq source, not the subsystem), so it stamps `wake`.
   479	        || Ok(crate::effect::MintedOp::new(crate::effect::Minter::Wake, now_ms())),
   480	        |op| {
   481	            let conn = brain.net_dial(addr.clone(), None)?;
   482	            // Keep the tracing string's seq consistent with the (possibly re-minted) op.
   483	            let op_id = format!("{}:wakefwd:{}", crate::presence::local_node_hex(), op.seq);
   484	            crate::resthost::request_rest(
   485	                &mut brain,
   486	                conn.conn_id,
   487	                owner,
   488	                spt_net::net::rest::REST_EVENT_WAKE,
   489	                &op_id,
   490	                op,
   491	            )
   492	        },
   493	    )
   494	    .map_err(|e| format!("rest op: {e}"))?;
   495	    match outcome {
   496	        crate::resthost::RestRequestOutcome::Edge(d) => Ok(format!("woke ({d})")),
   497	        crate::resthost::RestRequestOutcome::NoEdge => Ok("already awake".to_string()),
   498	        crate::resthost::RestRequestOutcome::Failed(e) => Err(e),
   499	        crate::resthost::RestRequestOutcome::NoReply => {
   500	            Err("refused or dropped (no reply)".to_string())
   501	        }
   502	    }
   503	}
   504	
   505	/// Resolve one adapter **option**'s `[shell]` section through the merged view
   506	/// (composite addressing): `adapter_option` is the stored `<adapter>[:profile]`
   507	/// string, split → parent lookup in `registered` → profile overlay (shipped or
   508	/// local). Returns an **owned** [`Shell`] — a profile produces a fresh merged
   509	/// manifest, not a borrow into the parent slice. A bare name resolves to the
   510	/// parent unmodified; a deregistered parent or a failed overlay yields `None`.
   511	fn shell_section_of(
   512	    registered: &[(AdapterRecord, Manifest)],
   513	    adapters_dir: &Path,
   514	    adapter_option: &str,
   515	) -> Option<Shell> {
   516	    spt_runtime::registry::resolve_option_in(registered, adapters_dir, adapter_option)
   517	        .ok()
   518	        .filter(|m| m.adapter.kind == AdapterKind::Shell)
   519	        .and_then(|m| m.shell)
   520	}
   521	
   522	fn now_ms() -> u64 {
   523	    crate::brain::now_ms()
   524	}
   525	
   526	/// Fill a `wake_command` template.
   527	///
   528	/// **Its catalog and the spawn template's share a vocabulary; they are not the
   529	/// same set, and never claiming they are is the point of saying so here.** The
   530	/// keys mean the same thing in both (`id`, `adapter_name`, `link_token`,
   531	/// `adapter_dir`), so an author learns one manifest language — but `perch_dir`
   532	/// is **spawn-only by design** (REQ-SHELL-PERCH-DIR: it resolves a *live* link's
   533	/// perch-relative transfer paths, and a waker has no live link). A comment
   534	/// asserting set-EQUALITY over two sets that are separately extended is the
   535	/// drift this project single-sources against; it had already been false since
   536	/// `perch_dir` landed.
   537	///
   538	/// The waker's token is minted **unparked**: it substitutes (template
   539	/// compatibility) without ever verifying against the perch, because an offline
   540	/// link has no live credential by design — the close retired it, and a waker
   541	/// wakes by *exit code*, not by driving the link.
   542	///
   543	/// `install_dir` resolves the program token and fills `{adapter_dir}`, exactly as
   544	/// on the spawn side ([`crate::shellhost::fill_spawn_command`]) — a released
   545	/// adapter whose spawn resolves and whose wake does not would go permanently
   546	/// unwakeable the moment it went offline.
   547	fn fill_wake_command(
   548	    shell_id: &str,
   549	    adapter_name: &str,
   550	    install_dir: Option<&Path>,
   551	    wake_command: &str,
   552	) -> Result<Vec<String>, String> {
   553	    let mut keys = std::collections::BTreeMap::from([
   554	        ("id".to_string(), shell_id.to_string()),
   555	        ("adapter_name".to_string(), adapter_name.to_string()),
   556	        (
   557	            "link_token".to_string(),
   558	            crate::shellhost::mint_link_token(),
   559	        ),
   560	    ]);
   561	    // [impl->REQ-INSTALL-11] opt-in and N-1-safe: a template that never names the
   562	    // key fills byte-identically.
   563	    if let Some(dir) = install_dir {
   564	        keys.insert("adapter_dir".to_string(), dir.display().to_string());
   565	    }
   566	    // [impl->REQ-HAZARD-TEMPLATE-ARGV-FILL] tokenize-template-then-fill-each: a
   567	    // multi-word/quote/semicolon {key} value is exactly one argv element.
   568	    let mut tokens =
   569	        spt_runtime::runtime::fill_template_tokens(wake_command, &keys).map_err(|e| e.to_string())?;
   570	    let Some(program) = tokens.first_mut() else {
   571	        return Err("empty wake_command".into());
   572	    };
   573	    // [impl->REQ-INSTALL-11] one resolution primitive, no parallel path.
   574	    if let Some(dir) = install_dir {
   575	        *program = spt_runtime::runtime::resolve_program_in_dir(program, dir);
   576	    }
   577	    Ok(tokens)
   578	}
   579	
   580	/// One reconcile sweep — the invariant holder for the online/offline mutual
   581	/// exclusivity (CONTEXT: "spt-core flips between them"):
   582	///
   583	/// - **offline + `wake_command` + no give-up latch + no live watcher** ⇒
   584	///   kill any orphaned waker pid (a dead daemon's leftover), then start a
   585	///   supervision thread.
   586	/// - **not offline (onlined / torn down / deregistered) + live watcher** ⇒
   587	///   stop it (flag + kill).
   588	///
   589	/// Runs at daemon boot and every tick — lifecycle flips happen in CLI-process
   590	/// library code, so the *reconciler* holds the invariant, not the flipping
   591	/// caller; a flip is at most one tick stale.
   592	// [impl->REQ-SHELL-2]
   593	/// Heal every local shell record whose `online` is a lie — leg (a) of
   594	/// releases#78.
   595	///
   596	/// `effective_status` already computes this flip on EVERY read and never writes
   597	/// it down, so the record and the display disagree by design: the display side
   598	/// derives and looks clean, while the wake cascade reads the RECORDED field and
   599	/// skips its relaunch because that field still says online. A machine death
   600	/// breaks no link, so `close_shell` never runs and nothing else ever corrects it.
   601	///
   602	/// THE WRITE IS GUARDED ON AN ACTUAL CHANGE. This runs every reconcile cycle
   603	/// (5s), and a heal that rewrote the record each time would be a continuous
   604	/// stream of identical writes over a rarely-changing condition — noise on disk,
   605	/// and a record whose mtime stops meaning "something happened".
   606	///
   607	/// It DERIVES deliberately, unlike the watcher-eligibility read below it: this
   608	/// is not an eligibility decision that could relaunch a binary, it is a
   609	/// correction of a field that is already wrong. Nothing is started here.
   610	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   611	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   612	fn heal_stale_online_records(owlery: &Path, events: &mut dyn std::io::Write) {
   613	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
   614	        for (shell_id, info) in shellinfo::list_shells(owlery, &owner) {
   615	            if info.status != SHELL_STATUS_ONLINE {
   616	                continue;
   617	            }
   618	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
   619	            let truth = shellinfo::effective_status(&perch, &info);
   620	            if truth == info.status {
   621	                continue; // the record already tells the truth — write nothing
   622	            }
   623	            let mut healed = info.clone();
   624	            healed.status = truth.to_string();
   625	            if shellinfo::write_shell_info(&perch, &healed).is_ok() {
   626	                // [impl->REQ-SHELL-HEAL-ENUMERATION]
   627	                let _ = spt_proto::emit_line!(events, "SHELL_RECORD_HEALED:{owner}/{shell_id}: online -> {truth}");
   628	            }
   629	        }
   630	    }
   631	}
   632	
   633	/// Slack absorbed when asking whether a launch predates the boot instant.
   634	///
   635	/// The boot instant is DERIVED, not read from a ledger — on Windows it is
   636	/// now-minus-uptime, which drifts by however long the two calls take and by
   637	/// whatever the clock did since — so a process launched in the first moments of
   638	/// a boot can compute as very slightly older than the boot itself. Without slack
   639	/// that reads as "a corpse from the previous boot" and a nonpersistent watcher
   640	/// could be armed after a same-boot death. Ten seconds is far wider than plausible
   641	/// derivation error and far narrower than the gap this discriminant exists to
   642	/// detect (a previous boot is minutes to days back).
   643	pub const BOOT_RESTORE_SLACK_MS: u64 = 10_000;
   644	
   645	/// Whether a nonpersistent shell's recorded launch predates the current boot.
   646	/// This preserves its same-boot force-kill freeze; persistent shells do not use
   647	/// launch age as eligibility (releases#287).
   648	// [impl->REQ-HAZARD-SHELL-STALE-ONLINE]
   649	pub fn launch_predates_boot(launched_ms: u64, boot_ms: u64, slack_ms: u64) -> bool {
   650	    launched_ms.saturating_add(slack_ms) < boot_ms
   651	}
   652	
   653	/// Watchers require an offline instance with no live launch awaiting bind.
   654	/// Persistent shells bypass the corpse-age freeze (releases#287); nonpersistent
   655	/// corpses still need a dated pre-boot launch. Missing dates fail closed only
   656	/// for that nonpersistent freeze, never for persistent restoration.
   657	// [impl->REQ-HAZARD-SHELL-STALE-ONLINE]
   658	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   659	fn watcher_eligible(
   660	    perch: &Path,
   661	    info: &shellinfo::ShellInfo,
   662	    persistent: bool,
   663	    boot_ms: Option<u64>,
   664	) -> bool {
   665	    if info.status != SHELL_STATUS_OFFLINE {
   666	        return false;
   667	    }
   668	    if crate::shellhost::live_launch_winner(perch).is_some() {
   669	        return false; // binding: the shell binary and watcher must not overlap
   670	    }
   671	    if persistent || !shellinfo::shell_pid_provably_dead(perch) {
   672	        return true; // persistence overrides the freeze; clean closes need no age
   673	    }
   674	    let (Some(boot_ms), Some(launch)) = (boot_ms, shellinfo::read_shell_launch(perch)) else {
   675	        return false;
   676	    };
   677	    launch_predates_boot(launch.launched_ms, boot_ms, BOOT_RESTORE_SLACK_MS)
   678	}
   679	
   680	/// Which trigger asked for a restore.
   681	///
   682	/// The two triggers run the SAME per-owner body and differ only in what woke
   683	/// them, so each NAMES ITSELF in its event line rather than sharing one: the
   684	/// field diagnosis of releases#228 was a COUNT of these events in a rotated
   685	/// daemon log, and a shared name would have left that reading unable to say
   686	/// which trigger had fired -- the one thing it needed to know.
   687	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
   688	pub enum RestoreTrigger {
   689	    /// The once-per-daemon-generation boot sweep (releases#78 leg (b)).
   690	    Boot,
   691	    /// An owner endpoint that came online since the last reconcile pass
   692	    /// (releases#228).
   693	    OwnerOnline,
   694	}
   695	
   696	impl RestoreTrigger {
   697	    fn restored_event(self) -> &'static str {
   698	        match self {
   699	            RestoreTrigger::Boot => "SHELL_BOOT_RESTORED",
   700	            RestoreTrigger::OwnerOnline => "SHELL_OWNER_ONLINE_RESTORED",
   701	        }
   702	    }
   703	
   704	    fn failed_event(self) -> &'static str {
   705	        match self {
   706	            RestoreTrigger::Boot => "SHELL_BOOT_RESTORE_FAIL",
   707	            RestoreTrigger::OwnerOnline => "SHELL_OWNER_ONLINE_RESTORE_FAIL",
   708	        }
   709	    }
   710	}
   711	
   712	/// ONE owner's slice of the persistent-shell restore decision -- the body BOTH
   713	/// triggers run.
   714	///
   715	/// Both triggers restore every persistent instance whose owner is online and
   716	/// whose binary is down in fact. Birth-safe liveness excludes a live launch
   717	/// awaiting bind as well as an already-online shell. Machine boot and launch
   718	/// age are irrelevant, including missing stamps (releases#287 operator ruling).
   719	///
   720	/// `set` is the daemon's live watcher set, and it is `None` at boot ON PURPOSE:
   721	/// the sweep runs before the reconcile loop's first tick, so no watcher can yet
   722	/// exist for the instance it is about to launch. The owner-online trigger runs
   723	/// MID-generation, where one can -- the reconcile start side arms watchers with
   724	/// no owner-online conjunct of its own -- so it passes the set and the watcher is
   725	/// stopped before the binary rises, which is the same online/offline mutual
   726	/// exclusivity `REQ-SHELL-2` holds everywhere else.
   727	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   728	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   729	// [impl->REQ-SHELL-OWNER-ONLINE-RESTORE]
   730	fn restore_persistent_shells_of_owner(
   731	    owlery: &Path,
   732	    registered: &[(AdapterRecord, Manifest)],
   733	    adapters_dir: &Path,
   734	    owner: &str,
   735	    trigger: RestoreTrigger,
   736	    set: Option<&Arc<WakeSet>>,
   737	    restored: &mut Vec<String>,
   738	) {
   739	    let owner_perch = spt_store::perch::resolve_perch_path_in(owlery, owner, ParentHint::Infer);
   740	    if !spt_store::liveness::is_perch_alive(&owner_perch) {
   741	        return; // an offline owner is owed no persistent shell
   742	    }
   743	    for (shell_id, info) in shellinfo::list_shells(owlery, owner) {
   744	        let Some(shell) =
   745	            shell_section_of(registered, adapters_dir, &info.adapter_name).filter(|s| s.persistent)
   746	        else {
   747	            continue;
   748	        };
   749	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, &shell_id);
   750	        if shellinfo::is_shell_online(&perch, &info)
   751	            || crate::shellhost::live_launch_winner(&perch).is_some()
   752	        {
   753	            continue; // already online or binding in fact
   754	        }
   755	        // Mutual exclusivity, the online side: the watcher dies before the binary
   756	        // rises. `launch_shell` already kills the waker CHILD, but the supervising
   757	        // thread would read that as a crash-exit and respawn it under the shell we
   758	        // are starting; only the set can retire the thread. [impl->REQ-SHELL-2]
   759	        if let Some(set) = set {
   760	            set.stop_watcher(owlery, owner, &shell_id);
   761	        }
   762	        let install_dir = crate::shellhost::shell_install_dir(adapters_dir, &info.adapter_name);
   763	        match crate::shellhost::launch_shell(
   764	            owlery,
   765	            owner,
   766	            &shell_id,
   767	            &info.adapter_name,
   768	            install_dir.as_deref(),
   769	            &shell,
   770	        ) {
   771	            Ok(_) => {
   772	                eprintln!("{}:{owner}/{shell_id}", trigger.restored_event());
   773	                restored.push(format!("{owner}/{shell_id}"));
   774	            }
   775	            Err(e) => eprintln!("{}:{owner}/{shell_id}: {e}", trigger.failed_event()),
   776	        }
   777	    }
   778	}
   779	
   780	/// The once-per-daemon-generation boot sweep: bring back the `persistent` shells
   781	/// a node restart stranded (leg (b) of releases#78).
   782	///
   783	/// Runs ONCE per daemon generation. It is not the ONLY restore trigger -- an owner
   784	/// endpoint that comes online after this sweep is covered by
   785	/// [`restore_persistent_shells_on_owner_online`] (releases#228), running the same
   786	/// [`restore_persistent_shells_of_owner`] body -- but it is the one that needs no
   787	/// trigger at all, since a node restart is what strands the shells to begin with.
   788	// [impl->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
   789	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   790	pub fn restore_persistent_shells_at_boot(
   791	    owlery: &Path,
   792	    registered: &[(AdapterRecord, Manifest)],
   793	    adapters_dir: &Path,
   794	) -> Vec<String> {
   795	    let mut restored = Vec::new();
   796	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
   797	        restore_persistent_shells_of_owner(
   798	            owlery,
   799	            registered,
   800	            adapters_dir,
   801	            &owner,
   802	            RestoreTrigger::Boot,
   803	            None,
   804	            &mut restored,
   805	        );
   806	    }
   807	    restored
   808	}
   809	
   810	/// Per-owner online state carried ACROSS reconcile passes, so the restore below
   811	/// can be EDGE-triggered on an owner's offline->online transition.
   812	///
   813	/// An edge bounds a failing launch to one attempt per owner-online event,
   814	/// rather than retrying every reconcile tick while the owner remains online.
   815	/// A successful launch is protected by birth-safe liveness even before bind.
   816	#[derive(Debug, Default)]
   817	pub struct OwnerOnlineEdge {
   818	    /// Owner id -> online as of the last pass. An owner that disappears from the
   819	    /// perch listing drops out, so a later reappearance reads as a transition.
   820	    seen: HashMap<String, bool>,
   821	}
   822	
   823	impl OwnerOnlineEdge {
   824	    pub fn new() -> Self {
   825	        Self::default()
   826	    }
   827	
   828	    /// Observe every owner and return those that went offline->online since the
   829	    /// last observation. An owner seen for the FIRST time counts as a transition
   830	    /// when it is online: it was not online at the last pass, because it was not
   831	    /// there at all.
   832	    fn transitions(&mut self, owlery: &Path) -> Vec<String> {
   833	        let mut fired = Vec::new();
   834	        let mut now = HashMap::new();
   835	        for owner in spt_store::perch::list_self_perch_ids(owlery) {
   836	            let perch = spt_store::perch::resolve_perch_path_in(owlery, &owner, ParentHint::Infer);
   837	            let online = spt_store::liveness::is_perch_alive(&perch);
   838	            if online && !self.seen.get(&owner).copied().unwrap_or(false) {
   839	                fired.push(owner.clone());
   840	            }
   841	            now.insert(owner, online);
   842	        }
   843	        self.seen = now;
   844	        fired
   845	    }
   846	
   847	    /// Observe without firing -- the initial state the first tick compares against.
   848	    ///
   849	    /// Called BEFORE the boot sweep, and the ORDER is the safe direction rather
   850	    /// than an accident. Seeded AFTER the sweep, an owner that comes online in the
   851	    /// window between the two reads as already-online and never fires an edge --
   852	    /// precisely the miss releases#228 is about. Seeded before, the worst case is a
   853	    /// second attempt at an instance the sweep already restored, which the
   854	    /// down-in-fact arm refuses.
   855	    pub fn seed(&mut self, owlery: &Path) {
   856	        let _ = self.transitions(owlery);
   857	    }
   858	}
   859	
   860	/// Restore the `persistent` shells of every owner that came online since the last
   861	/// pass -- the steady-state half of the contract's promise (releases#228).
   862	///
   863	/// Daemons boot before endpoints do, so the boot sweep correctly skips an owner
   864	/// that is not up yet and, being once-per-generation, never revisits it; bringup
   865	/// from offline emits no rest edge either, so the ADR-0048 cascade does not cover
   866	/// it. Both triggers use the same down-in-fact decision, without a machine-boot
   867	/// or launch-stamp eligibility gate (releases#287).
   868	// [impl->REQ-SHELL-OWNER-ONLINE-RESTORE]
   869	// [impl->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
   870	pub fn restore_persistent_shells_on_owner_online(
   871	    owlery: &Path,
   872	    registered: &[(AdapterRecord, Manifest)],
   873	    adapters_dir: &Path,
   874	    edge: &mut OwnerOnlineEdge,
   875	    set: &Arc<WakeSet>,
   876	) -> Vec<String> {
   877	    let owners = edge.transitions(owlery);
   878	    if owners.is_empty() {
   879	        return Vec::new();
   880	    }
   881	    let mut restored = Vec::new();
   882	    for owner in owners {
   883	        restore_persistent_shells_of_owner(
   884	            owlery,
   885	            registered,
   886	            adapters_dir,
   887	            &owner,
   888	            RestoreTrigger::OwnerOnline,
   889	            Some(set),
   890	            &mut restored,
   891	        );
   892	    }
   893	    restored
   894	}
   895	
   896	pub fn reconcile_once(
   897	    owlery: &Path,
   898	    registered: &[(AdapterRecord, Manifest)],
   899	    adapters_dir: &Path,
   900	    set: &Arc<WakeSet>,
   901	    params: &WakeParams,
   902	) {
   903	    // Leg (a) first, every cycle: the record must stop lying before anything reads
   904	    // it below. Ordering matters -- the watcher-eligibility read further down asks
   905	    // the RECORDED field, so a heal that ran after it would leave this cycle acting
   906	    // on the stale value it just corrected.
   907	    heal_stale_online_records(owlery, &mut std::io::stderr().lock());
   908	
   909	    // Stop side first: watchers whose instance is gone or no longer eligible.
   910	    let live: Vec<(String, String)> = {
   911	        let mut map = set.inner.lock().unwrap_or_else(|p| p.into_inner());
   912	        map.retain(|_, h| !h.thread.is_finished());
   913	        map.keys().cloned().collect()
   914	    };
   915	    for (owner, shell_id) in live {
   916	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
   917	        let still_eligible = shellinfo::read_shell_info(&perch)
   918	            .map(|i| i.status == SHELL_STATUS_OFFLINE)
   919	            .unwrap_or(false);
   920	        if !still_eligible {
   921	            set.stop_watcher(owlery, &owner, &shell_id);
   922	        }
   923	    }
   924	
   925	    // Start side: every ELIGIBLE instance with a wake_command and headroom.
   926	    //
   927	    // Persistent instances bypass the force-kill freeze; only nonpersistent
   928	    // corpses need the pre-boot discriminant (releases#287).
   929	    let boot_ms = spt_store::proc::boot_instant_ms();
   930	    for owner in spt_store::perch::list_self_perch_ids(owlery) {
   931	        for (shell_id, info) in shellinfo::list_shells(owlery, &owner) {
   932	            if set.contains(&owner, &shell_id) {
   933	                continue;
   934	            }
   935	            let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, &owner, &shell_id);
   936	            let Some(shell) = shell_section_of(registered, adapters_dir, &info.adapter_name) else {
   937	                continue; // deregistered adapter: nothing to run
   938	            };
   939	            if !watcher_eligible(&perch, &info, shell.persistent, boot_ms) {
   940	                continue;
   941	            }
   942	            let Some(wake_command) = shell.wake_command.as_deref() else {
   943	                continue;
   944	            };
   945	            if perch.join(WAKER_GAVE_UP_FILE).exists() {
   946	                continue; // crash-latched until the next shell activity
   947	            }
   948	            // Adopt cleanly: a dead daemon's watcher must not double-run.
   949	            kill_waker_at(&perch);
   950	            // [impl->REQ-INSTALL-11] threading only: this loop already holds the
   951	            // registered set the record's `source_dir` lives in.
   952	            let install_dir = crate::shellhost::shell_install_dir(adapters_dir, &info.adapter_name);
   953	            let tokens = match fill_wake_command(
   954	                &shell_id,
   955	                &info.adapter_name,
   956	                install_dir.as_deref(),
   957	                wake_command,
   958	            ) {
   959	                Ok(t) => t,
   960	                Err(e) => {
   961	                    spt_proto::emit_line_err!("WAKER_TEMPLATE:{owner}/{shell_id}: {e}");
   962	                    continue;
   963	                }
   964	            };
   965	            let stop = Arc::new(AtomicBool::new(false));
   966	            let handle = {
   967	                let stop = Arc::clone(&stop);
   968	                let owlery = owlery.to_path_buf();
   969	                let owner = owner.clone();
   970	                let shell_id = shell_id.clone();
   971	                let adapter_name = info.adapter_name.clone();
   972	                let params = *params;
   973	                std::thread::spawn(move || {
   974	                    watcher_run(&owlery, &owner, &shell_id, &tokens, &params, &stop, || {
   975	                        resolve_wake(&owlery, &owner, &shell_id, &adapter_name, &shell)
   976	                    });
   977	                })
   978	            };
   979	            set.insert(
   980	                &owner,
   981	                &shell_id,
   982	                WatcherHandle {
   983	                    stop,
   984	                    thread: handle,
   985	                },
   986	            );
   987	        }
   988	    }
   989	}
   990	
   991	/// The reconcile cadence (ms): lifecycle flips land in CLI processes, so the
   992	/// daemon's loop is the invariant holder — a flip is at most one tick stale.
   993	pub const RECONCILE_INTERVAL_MS: u64 = 5_000;
   994	
   995	/// Spawn the daemon's wake host: one thread sweeping [`reconcile_once`] at
   996	/// boot and every [`RECONCILE_INTERVAL_MS`] until `stop`. The registered set
   997	/// is re-read each sweep (adapter add/remove lands between ticks).
   998	// [impl->REQ-SHELL-2]
   999	pub fn spawn_wake_host(stop: Arc<AtomicBool>) -> JoinHandle<()> {
  1000	    std::thread::spawn(move || {
  1001	        let set = Arc::new(WakeSet::new());
  1002	        let params = WakeParams::default();
  1003	        let mut owner_edge = OwnerOnlineEdge::new();
  1004	        // Once per daemon generation, restore down persistent shells of owners
  1005	        // already online. Owners arriving later are covered by the edge below.
  1006	        // Neither trigger gates persistence on machine boot or launch age.
  1007	        {
  1008	            let owlery = spt_store::perch::owlery_dir();
  1009	            let adapters_dir = spt_store::perch::adapters_dir();
  1010	            let registered = spt_runtime::registry::registered(&adapters_dir);
  1011	            // Seeded BEFORE the sweep: an owner that comes online between the two
  1012	            // must read as a transition on the first tick, not as already-online.
  1013	            owner_edge.seed(&owlery);
  1014	            restore_persistent_shells_at_boot(&owlery, &registered, &adapters_dir);
  1015	        }
  1016	        while !stop.load(Ordering::SeqCst) {
  1017	            let owlery = spt_store::perch::owlery_dir();
  1018	            let adapters_dir = spt_store::perch::adapters_dir();
  1019	            let registered = spt_runtime::registry::registered(&adapters_dir);
  1020	            // Restore BEFORE reconciling, the same order the boot pass runs in: a
  1021	            // restored instance is not watcher material, and reconcile's start side
  1022	            // would otherwise arm a watcher for the shell we are about to launch.
  1023	            restore_persistent_shells_on_owner_online(
  1024	                &owlery,
  1025	                &registered,
  1026	                &adapters_dir,
  1027	                &mut owner_edge,
  1028	                &set,
  1029	            );
  1030	            reconcile_once(&owlery, &registered, &adapters_dir, &set, &params);
  1031	            // Sleep in slices so a stop lands promptly.
  1032	            let mut left = RECONCILE_INTERVAL_MS;
  1033	            while left > 0 && !stop.load(Ordering::SeqCst) {
  1034	                let step = left.min(100);
  1035	                std::thread::sleep(Duration::from_millis(step));
  1036	                left -= step;
  1037	            }
  1038	        }
  1039	    })
  1040	}
  1041	
  1042	#[cfg(test)]
  1043	mod tests {
  1044	    use super::*;
  1045	    use spt_store::shellinfo::{spawn_record, SHELL_STATUS_ONLINE};
  1046	
  1047	    /// BACKSTOP for the converge-waits below — a ceiling on an observable, not
  1048	    /// a budget any run is expected to consume (the healthy path exits in
  1049	    /// milliseconds, and the give-up latch exits the product-fault path at
  1050	    /// once).
  1051	    ///
  1052	    /// ANCHORED, not chosen: it is `slow-timeout.period` from
  1053	    /// `.config/nextest.toml`, i.e. exactly the point at which the HARNESS
  1054	    /// ITSELF starts calling this test slow. Past that, nextest is already
  1055	    /// complaining and a longer wait cannot be the right answer — so the number
  1056	    /// has a SOURCE rather than being "some multiple of the 2s that broke",
  1057	    /// which would be a threshold keyed on a measurement instead of on the
  1058	    /// thing. If that config value changes, this should follow it.
  1059	    const WAIT_BACKSTOP: Duration = Duration::from_secs(60);
  1060	
  1061	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE] the record heal writes the truth
  1062	    // down, and writes NOTHING when the record already tells it.
  1063	    //
  1064	    // THE GUARD IS ASSERTED BY COUNTING WRITES, not by checking the final state,
  1065	    // and that distinction is the point: this runs every 5s reconcile cycle, so an
  1066	    // implementation that rewrote the record unconditionally would satisfy any
  1067	    // state-only assertion while producing a continuous stream of identical writes
  1068	    // over a condition that changes maybe twice a day — and would destroy the one
  1069	    // thing an on-disk record's mtime is good for, which is saying when something
  1070	    // actually happened. mtime is the observable that tells the two apart.
  1071	    //
  1072	    // The healthy instance beside the stale one is not decoration either: without
  1073	    // it, an implementation that healed EVERY record to offline would pass.
  1074	    // [unit->REQ-SHELL-HEAL-ENUMERATION]
  1075	    #[test]
  1076	    fn the_heal_writes_the_truth_once_and_then_leaves_the_record_alone() {
  1077	        crate::test_home::with_home(|_| {
  1078	            let owlery = spt_store::perch::owlery_dir();
  1079	            let perch_path = spt_store::perch::resolve_perch_path("ling", ParentHint::Infer);
  1080	            std::fs::create_dir_all(&perch_path).unwrap();
  1081	            spt_store::info::write_info(
  1082	                &perch_path,
  1083	                &spt_store::info::InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1084	            )
  1085	            .unwrap();
  1086	
  1087	            let stale = stranded_shell(&owlery, "ling", "alchemy", Some(1_000));
  1088	            let sibling = stranded_shell(&owlery, "ling", "alchemy", None);
  1089	            let pacer = stranded_shell(&owlery, "ling", "pacer", Some(2_000));
  1090	            seed_owner(&owlery, "perri");
  1091	            let other = stranded_shell(&owlery, "perri", "alchemy", None);
  1092	            let stale_perch =
  1093	                spt_store::perch::resolve_shell_perch_path_in(&owlery, "ling", &stale);
  1094	
  1095	            // A HEALTHY instance: record online over OUR OWN live pid, stamped by
  1096	            // its own launch. The heal must not touch it.
  1097	            let live = spawn_record(&owlery, "ling", "pacer", None).unwrap();
  1098	            let live_perch = spt_store::perch::resolve_shell_perch_path_in(&owlery, "ling", &live);
  1099	            let mut linfo = shellinfo::read_shell_info(&live_perch).unwrap();
  1100	            linfo.status = SHELL_STATUS_ONLINE.to_string();
  1101	            shellinfo::write_shell_info(&live_perch, &linfo).unwrap();
  1102	            std::fs::write(
  1103	                live_perch.join(shellinfo::SHELL_PID_FILE),
  1104	                std::process::id().to_string(),
  1105	            )
  1106	            .unwrap();
  1107	            shellinfo::record_shell_launch(&live_perch, std::process::id(), 1_000).unwrap();
  1108	
  1109	            let mtime = |p: &std::path::Path| {
  1110	                std::fs::metadata(spt_store::perch::info_file_at(p))
  1111	                    .and_then(|m| m.modified())
  1112	                    .unwrap()
  1113	            };
  1114	            let live_before = mtime(&live_perch);
  1115	
  1116	            let mut events = Vec::new();
  1117	            heal_stale_online_records(&owlery, &mut events);
  1118	            let output = String::from_utf8(events).unwrap();
  1119	            let mut reported: Vec<_> = output.lines().collect();
  1120	            reported.sort();
  1121	            let mut expected = vec![
  1122	                format!("SHELL_RECORD_HEALED:ling/{stale}: online -> offline"),
  1123	                format!("SHELL_RECORD_HEALED:ling/{sibling}: online -> offline"),
  1124	                format!("SHELL_RECORD_HEALED:ling/{pacer}: online -> offline"),
  1125	                format!("SHELL_RECORD_HEALED:perri/{other}: online -> offline"),
  1126	            ];
  1127	            expected.sort();
  1128	            assert_eq!(reported, expected, "every healed instance, no healthy representative");
  1129	
  1130	            assert_eq!(
  1131	                shellinfo::read_shell_info(&stale_perch).unwrap().status,
  1132	                SHELL_STATUS_OFFLINE,
  1133	                "the stranded record is healed to the truth the display already showed"
  1134	            );
  1135	            assert_eq!(
  1136	                shellinfo::read_shell_info(&live_perch).unwrap().status,
  1137	                SHELL_STATUS_ONLINE,
  1138	                "and a genuinely live instance is left online"
  1139	            );
  1140	            assert_eq!(
  1141	                mtime(&live_perch),
  1142	                live_before,
  1143	                "the healthy record was not REWRITTEN — an unconditional heal would \
  1144	                 pass every assertion above and fail this one"
  1145	            );
  1146	
  1147	            // Second cycle over the now-correct records: nothing may be written.
  1148	            let stale_after_heal = mtime(&stale_perch);
  1149	            let live_after_heal = mtime(&live_perch);
  1150	            let mut repeated = Vec::new();
  1151	            heal_stale_online_records(&owlery, &mut repeated);
  1152	            assert!(repeated.is_empty(), "already-healed records emit no duplicate events");
  1153	            assert_eq!(
  1154	                (mtime(&stale_perch), mtime(&live_perch)),
  1155	                (stale_after_heal, live_after_heal),
  1156	                "a second cycle over records that already tell the truth writes nothing"
  1157	            );
  1158	        });
  1159	    }
  1160	
  1161	    // [unit->REQ-HAZARD-SHELL-STALE-ONLINE]
  1162	    // Only the nonpersistent watcher freeze uses this boundary and slack band.
  1163	    #[test]
  1164	    fn only_a_launch_from_before_this_boot_is_a_restart_casualty() {
  1165	        const BOOT: u64 = 1_000_000;
  1166	
  1167	        assert!(
  1168	            launch_predates_boot(BOOT - 60_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1169	            "launched a minute before boot ⇒ a restart casualty, restore it"
  1170	        );
  1171	        assert!(
  1172	            !launch_predates_boot(BOOT + 30_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1173	            "launched AFTER boot ⇒ a force-kill during steady state, never a \
  1174	             spontaneous relaunch (the class-c ruling, preserved by construction)"
  1175	        );
  1176	        assert!(
  1177	            !launch_predates_boot(BOOT - 1_000, BOOT, BOOT_RESTORE_SLACK_MS),
  1178	            "inside the slack band ⇒ treated as this boot, since the boot instant \
  1179	             is derived and drifts by a little"
  1180	        );
  1181	        assert!(
  1182	            !launch_predates_boot(BOOT, BOOT, BOOT_RESTORE_SLACK_MS),
  1183	            "exactly at the boot instant is not before it"
  1184	        );
  1185	    }
  1186	
  1187	    fn register_restore_adapter(adapters_dir: &Path, source: &Path, name: &str, persistent: bool) {
  1188	        #[cfg(windows)]
  1189	        let noop = "cmd /c exit 0";
  1190	        #[cfg(unix)]
  1191	        let noop = "true";
  1192	        std::fs::create_dir_all(source).unwrap();
  1193	        std::fs::write(
  1194	            source.join("manifest.toml"),
  1195	            format!(
  1196	                "[adapter]\nname = \"{name}\"\nkind = \"shell\"\nversion = \"1\"\n\
  1197	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1198	                 persistent = {persistent}\n"
  1199	            ),
  1200	        )
  1201	        .unwrap();
  1202	        spt_runtime::registry::register(adapters_dir, source, 1).unwrap();
  1203	    }
  1204	
  1205	    /// Restart-shaped record, with NO suspend edge to pre-heal its status.
  1206	    fn stranded_shell(owlery: &Path, owner: &str, adapter: &str, launched_ms: Option<u64>) -> String {
  1207	        let id = spawn_record(owlery, owner, adapter, None).unwrap();
  1208	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, owner, &id);
  1209	        let mut info = shellinfo::read_shell_info(&perch).unwrap();
  1210	        info.status = SHELL_STATUS_ONLINE.to_string();
  1211	        shellinfo::write_shell_info(&perch, &info).unwrap();
  1212	        std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), "2000000000").unwrap();
  1213	        if let Some(launched_ms) = launched_ms {
  1214	            shellinfo::record_shell_launch(&perch, 2_000_000_000, launched_ms).unwrap();
  1215	        }
  1216	        id
  1217	    }
  1218	
  1219	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1220	    // [unit->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
  1221	    #[test]
  1222	    fn boot_restores_all_down_persistent_instances_without_launch_age_gates() {
  1223	        let tmp = tempfile::tempdir().unwrap();
  1224	        let owlery = tmp.path();
  1225	        let boot = spt_store::proc::boot_instant_ms().expect("boot instant");
  1226	        let adapters_dir = tmp.path().join("adapters");
  1227	        register_restore_adapter(&adapters_dir, &tmp.path().join("keeper"), "keeper", true);
  1228	        register_restore_adapter(&adapters_dir, &tmp.path().join("ordinary"), "ordinary", false);
  1229	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1230	        seed_owner(owlery, "ling");
  1231	        let gone = owlery.join("gone");
  1232	        std::fs::create_dir_all(&gone).unwrap();
  1233	        std::fs::write(gone.join("info.json"), "{\"status\":\"offline\"}").unwrap();
  1234	
  1235	        let before = stranded_shell(owlery, "ling", "keeper", Some(boot.saturating_sub(600_000)));
  1236	        let after = stranded_shell(owlery, "ling", "keeper", Some(boot + 120_000));
  1237	        let unstamped = stranded_shell(owlery, "ling", "keeper", None);
  1238	        stranded_shell(owlery, "gone", "keeper", None);
  1239	        stranded_shell(owlery, "ling", "ordinary", Some(boot + 120_000));
  1240	
  1241	        // A live binary awaiting bind still has an offline record. Removing the
  1242	        // boot-age gate must not turn the sweep/edge overlap into a second launch.
  1243	        let binding = spawn_record(owlery, "ling", "keeper", None).unwrap();
  1244	        let binding_perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "ling", &binding);
  1245	        std::fs::write(
  1246	            binding_perch.join(shellinfo::SHELL_PID_FILE),
  1247	            std::process::id().to_string(),
  1248	        )
  1249	        .unwrap();
  1250	        shellinfo::record_shell_launch(&binding_perch, std::process::id(), boot + 120_000).unwrap();
  1251	
  1252	        let mut restored = restore_persistent_shells_at_boot(owlery, &registered, &adapters_dir);
  1253	        restored.sort();
  1254	        let mut expected = vec![
  1255	            format!("ling/{before}"),
  1256	            format!("ling/{after}"),
  1257	            format!("ling/{unstamped}"),
  1258	        ];
  1259	        expected.sort();
  1260	        assert_eq!(restored, expected);
  1261	        assert_eq!(shellinfo::read_shell_pid(&binding_perch), Some(std::process::id()));
  1262	    }
  1263	
  1264	    // [unit->REQ-SHELL-OWNER-ONLINE-RESTORE]
  1265	    // [unit->REQ-HAZARD-RESTART-STRANDS-PERSISTENT-SHELLS]
  1266	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1267	    #[test]
  1268	    fn owner_online_restores_every_same_boot_or_unstamped_corpse_after_spent_sweep() {
  1269	        let tmp = tempfile::tempdir().unwrap();
  1270	        let owlery = tmp.path();
  1271	        let boot = spt_store::proc::boot_instant_ms().expect("boot instant");
  1272	        let owner_dir = owlery.join("ling");
  1273	        std::fs::create_dir_all(&owner_dir).unwrap();
  1274	        std::fs::write(owner_dir.join("info.json"), "{\"status\":\"offline\"}").unwrap();
  1275	        let adapters_dir = tmp.path().join("adapters");
  1276	        register_restore_adapter(&adapters_dir, &tmp.path().join("keeper"), "keeper", true);
  1277	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1278	        let first = stranded_shell(owlery, "ling", "keeper", Some(boot + 46_000));
  1279	        let second = stranded_shell(owlery, "ling", "keeper", Some(boot + 120_000));
  1280	        let unstamped = stranded_shell(owlery, "ling", "keeper", None);
  1281	        let mut edge = OwnerOnlineEdge::new();
  1282	        edge.seed(owlery);
  1283	        assert!(restore_persistent_shells_at_boot(owlery, &registered, &adapters_dir).is_empty());
  1284	
  1285	        let set = Arc::new(WakeSet::new());
  1286	        assert!(restore_persistent_shells_on_owner_online(
  1287	            owlery, &registered, &adapters_dir, &mut edge, &set,
  1288	        )
  1289	        .is_empty());
  1290	        std::fs::write(owner_dir.join("info.json"), "{}").unwrap();
  1291	        let mut restored =
  1292	            restore_persistent_shells_on_owner_online(owlery, &registered, &adapters_dir, &mut edge, &set);
  1293	        restored.sort();
  1294	        let mut expected = vec![
  1295	            format!("ling/{first}"),
  1296	            format!("ling/{second}"),
  1297	            format!("ling/{unstamped}"),
  1298	        ];
  1299	        expected.sort();
  1300	        assert_eq!(restored, expected, "enumerate every restored instance of this owner");
  1301	
  1302	        // Still an EDGE, not a permanent retry loop for failed launches.
  1303	        stranded_shell(owlery, "ling", "keeper", Some(boot + 180_000));
  1304	        assert!(restore_persistent_shells_on_owner_online(
  1305	            owlery, &registered, &adapters_dir, &mut edge, &set,
  1306	        )
  1307	        .is_empty());
  1308	    }
  1309	
  1310	    fn fast_params() -> WakeParams {
  1311	        WakeParams {
  1312	            backoff_base_ms: 10,
  1313	            backoff_cap_ms: 40,
  1314	            give_up_after: 3,
  1315	        }
  1316	    }
  1317	
  1318	    // [unit->REQ-INSTALL-11] the wake site, matching the spawn site: a bare
  1319	    // program token binds to the shipped binary and {adapter_dir} fills. Wiring
  1320	    // spawn alone would leave a released adapter launchable but PERMANENTLY
  1321	    // UNWAKEABLE the moment it went offline — the worse half of the gap, because
  1322	    // the wake path is the one that runs with nobody watching.
  1323	    #[test]
  1324	    fn a_wake_template_resolves_its_program_and_adapter_dir_against_the_install_dir() {
  1325	        let tmp = tempfile::tempdir().unwrap();
  1326	        let install = tmp.path().join("adapter dir"); // spaces: the argv-fill shape
  1327	        std::fs::create_dir_all(&install).unwrap();
  1328	        let shipped = if cfg!(windows) { "waker.exe" } else { "waker" };
  1329	        std::fs::write(install.join(shipped), b"").unwrap();
  1330	
  1331	        let tokens =
  1332	            fill_wake_command("sh-1", "mock-shell", Some(&install), "waker --root {adapter_dir}")
  1333	                .expect("adapter_dir is a wake substitution key");
  1334	        assert_eq!(
  1335	            tokens[0],
  1336	            install.join(shipped).display().to_string(),
  1337	            "the bare token must bind to the SHIPPED binary, not fall through to PATH"
  1338	        );
  1339	        assert_eq!(
  1340	            tokens[2],
  1341	            install.display().to_string(),
  1342	            "the install dir (spaces included) is exactly one argv element"
  1343	        );
  1344	    }
  1345	
  1346	    // [unit->REQ-INSTALL-11] the catalogs share a VOCABULARY, they are not one
  1347	    // set — the relationship the corrected `fill_wake_command` docstring states,
  1348	    // pinned so a future edit cannot quietly restore the set-equality claim.
  1349	    // `adapter_dir` is in BOTH; `perch_dir` is spawn-only by design (a waker has
  1350	    // no live link, so there is no perch-relative transfer path to resolve).
  1351	    #[test]
  1352	    fn the_wake_catalog_shares_the_spawn_vocabulary_but_is_not_the_same_set() {
  1353	        let install = std::path::PathBuf::from("/adapters/mock");
  1354	        assert!(
  1355	            fill_wake_command("sh-1", "mock-shell", Some(&install), "w {adapter_dir}").is_ok(),
  1356	            "adapter_dir is shared by both catalogs"
  1357	        );
  1358	        assert!(
  1359	            fill_wake_command("sh-1", "mock-shell", Some(&install), "w {perch_dir}").is_err(),
  1360	            "perch_dir is SPAWN-ONLY — a waker has no live link to resolve paths against"
  1361	        );
  1362	    }
  1363	
  1364	    fn seed_owner(owlery: &Path, owner: &str) {
  1365	        let p = owlery.join(owner);
  1366	        std::fs::create_dir_all(&p).unwrap();
  1367	        std::fs::write(p.join("info.json"), "{}").unwrap();
  1368	    }
  1369	
  1370	    /// Tokens for a waker that exits with `code` immediately.
  1371	    fn exit_with(code: i32) -> Vec<String> {
  1372	        #[cfg(windows)]
  1373	        return vec!["cmd".into(), "/c".into(), format!("exit {code}")];
  1374	        #[cfg(unix)]
  1375	        return vec!["sh".into(), "-c".into(), format!("exit {code}")];
  1376	    }
  1377	
  1378	    /// Tokens for a waker that appends one line to `tally` then exits 1.
  1379	    fn crash_tallying(tally: &Path) -> Vec<String> {
  1380	        #[cfg(windows)]
  1381	        return vec![
  1382	            "powershell".into(),
  1383	            "-NoProfile".into(),
  1384	            "-Command".into(),
  1385	            format!("Add-Content -Path '{}' -Value x; exit 1", tally.display()),
  1386	        ];
  1387	        #[cfg(unix)]
  1388	        return vec![
  1389	            "sh".into(),
  1390	            "-c".into(),
  1391	            format!("echo x >> '{}'; exit 1", tally.display()),
  1392	        ];
  1393	    }
  1394	
  1395	    // [unit->REQ-SHELL-2] the backoff curve: base × 2^(n-1), saturating at
  1396	    // the cap; zero failures wait nothing.
  1397	    #[test]
  1398	    fn backoff_curve_doubles_to_the_cap() {
  1399	        let p = WakeParams {
  1400	            backoff_base_ms: 100,
  1401	            backoff_cap_ms: 1_500,
  1402	            give_up_after: 6,
  1403	        };
  1404	        assert_eq!(backoff_ms(&p, 0), 0);
  1405	        assert_eq!(backoff_ms(&p, 1), 100);
  1406	        assert_eq!(backoff_ms(&p, 2), 200);
  1407	        assert_eq!(backoff_ms(&p, 4), 800);
  1408	        assert_eq!(backoff_ms(&p, 5), 1_500, "saturates at the cap");
  1409	        assert_eq!(
  1410	            backoff_ms(&p, 60),
  1411	            1_500,
  1412	            "huge failure counts can't overflow"
  1413	        );
  1414	    }
  1415	
  1416	    // [unit->REQ-SHELL-2] exit-opcode supervision: exit(86) fires the wake
  1417	    // resolution exactly once and the loop ends; the pid file is retired.
  1418	    #[test]
  1419	    fn watcher_opcode_exit_fires_resolution_once() {
  1420	        let tmp = tempfile::tempdir().unwrap();
  1421	        let owlery = tmp.path();
  1422	        seed_owner(owlery, "doyle");
  1423	        let id = spawn_record(owlery, "doyle", "mock-shell", None).unwrap();
  1424	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  1425	
  1426	        let fired = std::sync::atomic::AtomicU32::new(0);
  1427	        let stop = AtomicBool::new(false);
  1428	        watcher_run(
  1429	            owlery,
  1430	            "doyle",
  1431	            &id,
  1432	            &exit_with(WAKE_OPCODE),
  1433	            &fast_params(),
  1434	            &stop,
  1435	            || {
  1436	                fired.fetch_add(1, Ordering::SeqCst);
  1437	                Ok("test-resolved".into())
  1438	            },
  1439	        );
  1440	        assert_eq!(
  1441	            fired.load(Ordering::SeqCst),
  1442	            1,
  1443	            "resolution fired exactly once"
  1444	        );
  1445	        assert!(!perch.join(WAKER_PID_FILE).exists(), "pid file retired");
  1446	        assert!(
  1447	            !perch.join(WAKER_GAVE_UP_FILE).exists(),
  1448	            "an opcode exit is not a crash"
  1449	        );
  1450	    }
  1451	
  1452	    // [unit->REQ-SHELL-2] crash-exit supervision: a crashing waker respawns
  1453	    // (with backoff) exactly give_up_after times, then the durable give-up
  1454	    // latch drops and the resolution never fires; clear_gave_up re-arms.
  1455	    #[test]
  1456	    fn watcher_crash_exits_respawn_then_give_up() {
  1457	        let tmp = tempfile::tempdir().unwrap();
  1458	        let owlery = tmp.path();
  1459	        seed_owner(owlery, "doyle");
  1460	        let id = spawn_record(owlery, "doyle", "mock-shell", None).unwrap();
  1461	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  1462	        let tally = perch.join("waker-tally.txt");
  1463	
  1464	        let stop = AtomicBool::new(false);
  1465	        watcher_run(
  1466	            owlery,
  1467	            "doyle",
  1468	            &id,
  1469	            &crash_tallying(&tally),
  1470	            &fast_params(),
  1471	            &stop,
  1472	            || -> Result<String, String> { panic!("a crash-exit must never resolve a wake") },
  1473	        );
  1474	        let runs = std::fs::read_to_string(&tally).unwrap().lines().count();
  1475	        assert_eq!(runs, 3, "respawned exactly to the give-up budget");
  1476	        assert!(
  1477	            perch.join(WAKER_GAVE_UP_FILE).exists(),
  1478	            "the durable latch dropped"
  1479	        );
  1480	
  1481	        clear_gave_up(&perch);
  1482	        assert!(
  1483	            !perch.join(WAKER_GAVE_UP_FILE).exists(),
  1484	            "shell activity re-arms"
  1485	        );
  1486	    }
  1487	
  1488	    /// A `[shell]` section parsed from TOML (the manifest is the only
  1489	    /// constructor — its serde defaults are part of the contract).
  1490	    fn shell_section(spawn: &str, extra: &str) -> Shell {
  1491	        let toml_src = format!(
  1492	            "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1493	             min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{spawn}'\n{extra}\n"
  1494	        );
  1495	        spt_runtime::Manifest::from_toml_str(&toml_src)
  1496	            .unwrap()
  1497	            .shell
  1498	            .unwrap()
  1499	    }
  1500	
  1501	    #[cfg(windows)]
  1502	    const NOOP: &str = "cmd /c exit 0";
  1503	    #[cfg(unix)]
  1504	    const NOOP: &str = "true";
  1505	
  1506	    // [unit->REQ-SHELL-2] wake resolution, the no-reachable branch: no local
  1507	    // instance of the owner ⇒ refuse naming the instantiate-anywhere
  1508	    // deferral — never a silent no-op, never a fresh spawn.
  1509	    #[test]
  1510	    fn resolve_wake_refuses_without_a_reachable_owner() {
  1511	        let tmp = tempfile::tempdir().unwrap();
  1512	        let owlery = tmp.path();
  1513	        // The shell perch exists; the OWNER's perch does not (home node gone).
  1514	        let id = spawn_record(owlery, "ghost", "mock-wake", None).unwrap();
  1515	        let err =
  1516	            resolve_wake(owlery, "ghost", &id, "mock-wake", &shell_section(NOOP, "")).unwrap_err();
  1517	        assert!(err.contains("WAKE_NO_REACHABLE_INSTANCE"), "{err}");
  1518	        assert!(
  1519	            err.contains("instantiate-anywhere"),
  1520	            "the deferral is named: {err}"
  1521	        );
  1522	    }
  1523	
  1524	    // [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1525	    #[test]
  1526	    fn resolve_wake_does_not_adopt_a_recycled_pid_as_a_launch() {
  1527	        let tmp = tempfile::tempdir().unwrap();
  1528	        let owlery = tmp.path();
  1529	        seed_owner(owlery, "ling");
  1530	        let id = spawn_record(owlery, "ling", "mock-wake", None).unwrap();
  1531	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "ling", &id);
  1532	        let pid = std::process::id();
  1533	        std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), pid.to_string()).unwrap();
  1534	        shellinfo::record_shell_launch(&perch, pid, 1_000).unwrap();
  1535	        let mut launch = shellinfo::read_shell_launch(&perch).unwrap();
  1536	        launch.pid_started_at = Some(
  1537	            launch.pid_started_at.expect("native birth on supported platforms") ^ 1,
  1538	        );
  1539	        std::fs::write(
  1540	            perch.join(shellinfo::SHELL_LAUNCH_FILE),
  1541	            serde_json::to_string(&launch).unwrap(),
  1542	        )
  1543	        .unwrap();
  1544	        // A real spawn failure must surface, not a successful adoption of the
  1545	        // unrelated live process. No child needs cleanup on this error path.
  1546	        let missing = tmp.path().join("missing-shell-executable");
  1547	        let command = format!("\"{}\"", missing.display());
  1548	        let err = resolve_wake(
  1549	            owlery, "ling", &id, "mock-wake", &shell_section(&command, "persistent = true"),
  1550	        )
  1551	        .unwrap_err();
  1552	        assert!(err.contains("missing-shell-executable"), "{err}");
  1553	        assert!(spt_store::proc::is_process_alive(pid));
  1554	    }
  1555	
  1556	    // [unit->REQ-SHELL-2] wake resolution, the dormant branch: the owner is
  1557	    // resting warm — touch nothing on the endpoint, just relaunch the shell.
  1558	    #[test]
  1559	    fn resolve_wake_leaves_a_dormant_owner_and_relaunches() {
  1560	        use spt_store::info::{write_info, InfoJson};
  1561	        let tmp = tempfile::tempdir().unwrap();
  1562	        let owlery = tmp.path();
  1563	        let owner_perch = owlery.join("ling");
  1564	        std::fs::create_dir_all(&owner_perch).unwrap();
  1565	        write_info(
  1566	            &owner_perch,
  1567	            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1568	        )
  1569	        .unwrap();
  1570	        // Dormant is WARM (bound, live, resting in place): pin the owner online
  1571	        // so the liveness-aware derivation reads Dormant, not the host-dependent
  1572	        // pid-4242 cold-probe (which would collapse to Suspended, mis-routing the
  1573	        // wake to the suspended branch). REQ-EFFECTIVE-INSTANCE-STATE.
  1574	        spt_store::info::set_status(&owner_perch, spt_store::liveness::STATUS_ONLINE).unwrap();
  1575	        crate::resting::write_rest(&owner_perch, crate::resting::RestState::Dormant, 1_000)
  1576	            .unwrap();
  1577	
  1578	        let id = spawn_record(owlery, "ling", "mock-wake", None).unwrap();
  1579	        let msg = resolve_wake(owlery, "ling", &id, "mock-wake", &shell_section(NOOP, ""))
  1580	            .expect("resolves");
  1581	        assert!(msg.contains("owner dormant"), "{msg}");
  1582	        assert!(msg.contains("relaunched"), "{msg}");
  1583	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "ling", &id);
  1584	        assert!(
  1585	            perch.join(crate::shellhost::SHELL_PID_FILE).exists(),
  1586	            "binary relaunched"
  1587	        );
  1588	        assert_eq!(
  1589	            crate::resting::read_rest(&owner_perch).unwrap().state,
  1590	            crate::resting::RestState::Dormant,
  1591	            "a dormant owner is left in place"
  1592	        );
  1593	    }
  1594	
  1595	    // [unit->REQ-SHELL-2] wake resolution, the suspended branch: revive the
  1596	    // owner FIRST (the real daemon_rest_event — its wake cascade relaunches
  1597	    // persistent shells, and the resolution then must NOT double-launch a
  1598	    // live binary); a non-persistent shell is launched by the resolution
  1599	    // itself after the revive.
  1600	    #[test]
  1601	    fn resolve_wake_revives_a_suspended_owner_without_double_launch() {
  1602	        crate::test_home::with_home(|home| {
  1603	            use spt_store::info::{write_info, InfoJson};
  1604	            use spt_store::perch;
  1605	            let owlery = perch::owlery_dir();
  1606	            let owner_perch =
  1607	                perch::resolve_perch_path("ling", spt_store::perch::ParentHint::Infer);
  1608	            std::fs::create_dir_all(&owner_perch).unwrap();
  1609	            write_info(
  1610	                &owner_perch,
  1611	                &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
  1612	            )
  1613	            .unwrap();
  1614	            // Pin the owner online (warm) so the effective state is
  1615	            // intent-refined by the written rest record: the initial Suspended
  1616	            // record ⇒ from=Suspended (the first Wake is a real revive edge),
  1617	            // and after the revive writes Active the later re-suspend is a real
  1618	            // Active→Suspended edge. A cold (offline) pin would collapse EVERY
  1619	            // intent to Suspended, no-edging the re-suspend. Never the
  1620	            // host-dependent pid-4242 probe. REQ-EFFECTIVE-INSTANCE-STATE.
  1621	            spt_store::info::set_status(&owner_perch, spt_store::liveness::STATUS_ONLINE).unwrap();
  1622	            crate::resting::write_rest(&owner_perch, crate::resting::RestState::Suspended, 1_000)
  1623	                .unwrap();
  1624	
  1625	            // A registered PERSISTENT sleeper adapter — the revive's wake
  1626	            // cascade relaunches it and the binary stays alive.
  1627	            #[cfg(windows)]
  1628	            let sleeper = "ping -n 30 127.0.0.1";
  1629	            #[cfg(unix)]
  1630	            let sleeper = "sleep 30";
  1631	            let adapters = perch::adapters_dir();
  1632	            for (name, spawn, extra) in [
  1633	                ("mock-wake", sleeper, "persistent = true"),
  1634	                ("mock-noper", NOOP, ""), // the cascade must skip this one
  1635	            ] {
  1636	                let src = home.join("srcs").join(name);
  1637	                std::fs::create_dir_all(&src).unwrap();
  1638	                std::fs::write(
  1639	                    src.join("manifest.toml"),
  1640	                    format!(
  1641	                        "[adapter]\nname = \"{name}\"\nkind = \"shell\"\nversion = \"1\"\n\
  1642	                         min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{spawn}'\n{extra}\n"
  1643	                    ),
  1644	                )
  1645	                .unwrap();
  1646	                spt_runtime::registry::register(&adapters, &src, 1).unwrap();
  1647	            }
  1648	            let shell = shell_section(sleeper, "persistent = true");
  1649	
  1650	            let id = spawn_record(&owlery, "ling", "mock-wake", None).unwrap();
  1651	            let msg = resolve_wake(&owlery, "ling", &id, "mock-wake", &shell).expect("resolves");
  1652	            assert!(msg.contains("revived owner"), "{msg}");
  1653	            assert!(
  1654	                msg.contains("already relaunched"),
  1655	                "a live cascade launch is never doubled: {msg}"
  1656	            );
  1657	            assert_eq!(
  1658	                crate::resting::read_rest(&owner_perch).unwrap().state,
  1659	                crate::resting::RestState::Active,
  1660	                "suspended owner revived"
  1661	            );
  1662	
  1663	            // Re-suspend and resolve a NON-persistent shell of the same
  1664	            // owner: the revive's cascade skips it (its registered adapter
  1665	            // is not persistent), so the resolution launches it itself.
  1666	            crate::resting::daemon_rest_event("ling", crate::resting::RestEvent::Suspend, None)
  1667	                .expect("ok")
  1668	                .expect("edge");
  1669	            let shell_np = shell_section(NOOP, "");
  1670	            let id2 = spawn_record(&owlery, "ling", "mock-noper", None).unwrap();
  1671	            let msg =
  1672	                resolve_wake(&owlery, "ling", &id2, "mock-noper", &shell_np).expect("resolves");
  1673	            assert!(msg.contains("revived owner"), "{msg}");
  1674	            assert!(msg.contains("relaunched pid="), "{msg}");
  1675	        });
  1676	    }
  1677	
  1678	    // [unit->REQ-HAZARD-SHELL-STALE-ONLINE] a nonpersistent force-killed instance is NOT adopted
  1679	    // and its binary is NOT relaunched behind the operator's back — the class-(c)
  1680	    // protection, which BAROMETER W2 moved off the stale record and onto the
  1681	    // corpse-boot discriminant (this row's assertions are unchanged from the day it
  1682	    // caught that conflict: it went red when leg (a) healed the record out from
  1683	    // under the old guard, and it is green again on the replacement). The corpse
  1684	    // here carries NO launch stamp, so the discriminant cannot prove it a restart
  1685	    // casualty and the safe direction holds it out. Deliberate, operator-ratified
  1686	    // (flynn 2026-07-25): every reason to deliberately stop a shell is a reason not
  1687	    // to want it back a tick later — mid-deploy the relaunch would run the OLD
  1688	    // binary out of the file being replaced (worse than a failed install), and it
  1689	    // turns quarantining a misbehaving shell into a restart loop. Recovery is
  1690	    // demand-driven instead: `relink` (unblocked by the derived gate) or a `shell
  1691	    // cmd` that wakes. The offline sibling proves the rig would adopt if eligible.
  1692	    #[test]
  1693	    fn reconcile_never_adopts_a_stale_online_instance() {
  1694	        const DEAD_PID: u32 = 2_000_000_000;
  1695	        let tmp = tempfile::tempdir().unwrap();
  1696	        let owlery = tmp.path();
  1697	        seed_owner(owlery, "doyle");
  1698	
  1699	        #[cfg(windows)]
  1700	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1701	        #[cfg(unix)]
  1702	        let (noop, sleeper) = ("true", "sleep 30");
  1703	        let adapters_dir = tmp.path().join("adapters");
  1704	        let src = tmp.path().join("srcs").join("mock-wake");
  1705	        std::fs::create_dir_all(&src).unwrap();
  1706	        std::fs::write(
  1707	            src.join("manifest.toml"),
  1708	            format!(
  1709	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1710	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1711	                 persistent = false\nwake_command = '{sleeper}'\n"
  1712	            ),
  1713	        )
  1714	        .unwrap();
  1715	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1716	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1717	
  1718	        // The stale-online instance: bound, then its binary was force-killed.
  1719	        let dead = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1720	        let dead_perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &dead);
  1721	        let mut info = shellinfo::read_shell_info(&dead_perch).unwrap();
  1722	        info.status = SHELL_STATUS_ONLINE.to_string();
  1723	        shellinfo::write_shell_info(&dead_perch, &info).unwrap();
  1724	        std::fs::write(
  1725	            dead_perch.join(shellinfo::SHELL_PID_FILE),
  1726	            DEAD_PID.to_string(),
  1727	        )
  1728	        .unwrap();
  1729	
  1730	        let set = Arc::new(WakeSet::new());
  1731	        let params = fast_params();
  1732	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1733	        assert_eq!(
  1734	            set.len(),
  1735	            0,
  1736	            "a force-killed instance must NOT be adopted — no spontaneous relaunch"
  1737	        );
  1738	        assert!(
  1739	            !dead_perch.join(WAKER_PID_FILE).exists(),
  1740	            "no waker child was started for it either"
  1741	        );
  1742	
  1743	        // Control: a genuinely OFFLINE sibling is still adopted, so the absence
  1744	        // above is the derivation policy, not a broken rig.
  1745	        let off = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1746	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1747	        assert_eq!(set.len(), 1, "the offline sibling {off} still gets a watcher");
  1748	        set.stop_watcher(owlery, "doyle", &off);
  1749	    }
  1750	
  1751	    // [unit->REQ-HAZARD-SHELL-STALE-ONLINE] [unit->REQ-SHELL-PERSISTENT-BOOT-RESTORE]
  1752	    // BOTH HALVES IN ONE ROW: for a same-boot corpse the heal FIRED (the record now
  1753	    // reads offline — the honest value leg (a) owes it) AND the watcher count is
  1754	    // still 0. Asserted together on purpose. Split across two rows either half
  1755	    // passes alone on a broken build: without the heal the record still says online
  1756	    // and a status-only read refuses adoption for the WRONG reason (the old
  1757	    // accidental protection, which is exactly what this milestone removed), while a
  1758	    // watcher-0 assertion on its own is satisfied by an implementation that adopts
  1759	    // nothing at all.
  1760	    //
  1761	    // The pre-boot sibling is the contrast that keeps the discriminant from being
  1762	    // vacuous: an implementation treating EVERY corpse as ineligible passes the
  1763	    // same-boot half and silently strands every restart casualty's watcher. Both
  1764	    // instances are the identical shape — record online over a corpse with a parked
  1765	    // launch stamp — and differ ONLY in which side of the boot instant it falls on.
  1766	    #[test]
  1767	    fn nonpersistent_same_boot_freeze_does_not_apply_to_persistent_watchers() {
  1768	        const DEAD_PID: u32 = 2_000_000_000;
  1769	        let tmp = tempfile::tempdir().unwrap();
  1770	        let owlery = tmp.path();
  1771	        seed_owner(owlery, "doyle");
  1772	        let boot = spt_store::proc::boot_instant_ms().expect("a boot instant on this platform");
  1773	
  1774	        #[cfg(windows)]
  1775	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1776	        #[cfg(unix)]
  1777	        let (noop, sleeper) = ("true", "sleep 30");
  1778	        let adapters_dir = tmp.path().join("adapters");
  1779	        let src = tmp.path().join("srcs").join("mock-wake");
  1780	        std::fs::create_dir_all(&src).unwrap();
  1781	        std::fs::write(
  1782	            src.join("manifest.toml"),
  1783	            format!(
  1784	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1785	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1786	                 persistent = false\nwake_command = '{sleeper}'\n"
  1787	            ),
  1788	        )
  1789	        .unwrap();
  1790	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1791	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1792	
  1793	        let perch_of =
  1794	            |id: &str| spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", id);
  1795	        let mk = |launched_ms: u64| -> String {
  1796	            let id = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1797	            let perch = perch_of(&id);
  1798	            let mut i = shellinfo::read_shell_info(&perch).unwrap();
  1799	            i.status = SHELL_STATUS_ONLINE.to_string();
  1800	            shellinfo::write_shell_info(&perch, &i).unwrap();
  1801	            std::fs::write(perch.join(shellinfo::SHELL_PID_FILE), DEAD_PID.to_string()).unwrap();
  1802	            let launch = shellinfo::ShellLaunch {
  1803	                pid: None,
  1804	                pid_started_at: Some(1),
  1805	                launched_ms,
  1806	            };
  1807	            std::fs::write(
  1808	                perch.join(shellinfo::SHELL_LAUNCH_FILE),
  1809	                serde_json::to_string(&launch).unwrap(),
  1810	            )
  1811	            .unwrap();
  1812	            id
  1813	        };
  1814	        let force_killed = mk(boot + 120_000); // killed 2 min INTO this boot
  1815	        let casualty = mk(boot.saturating_sub(600_000)); // launched 10 min BEFORE it
  1816	
  1817	        let set = Arc::new(WakeSet::new());
  1818	        let params = fast_params();
  1819	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1820	
  1821	        assert_eq!(
  1822	            shellinfo::read_shell_info(&perch_of(&force_killed))
  1823	                .unwrap()
  1824	                .status,
  1825	            SHELL_STATUS_OFFLINE,
  1826	            "the heal FIRED — the record stopped lying about a corpse"
  1827	        );
  1828	        assert!(
  1829	            !set.contains("doyle", &force_killed),
  1830	            "...and the honest record still did not buy it a watcher: killed THIS \
  1831	             boot is the mid-deploy case the class-(c) freeze protects"
  1832	        );
  1833	        assert!(
  1834	            !perch_of(&force_killed).join(WAKER_PID_FILE).exists(),
  1835	            "no waker child was started for it either"
  1836	        );
  1837	        assert!(
  1838	            set.contains("doyle", &casualty),
  1839	            "a corpse launched BEFORE this boot is a restart casualty — eligible, \
  1840	             per #78 leg (b)'s greenlit scope"
  1841	        );
  1842	        assert_eq!(set.len(), 1, "and it is the ONLY instance adopted");
  1843	        set.stop_watcher(owlery, "doyle", &casualty);
  1844	
  1845	        // The SAME corpse is eligible when the manifest declares persistence;
  1846	        // no restart or new launch stamp is needed to release that freeze.
  1847	        let mut persistent_registered = registered.clone();
  1848	        for (_, manifest) in &mut persistent_registered {
  1849	            manifest.shell.as_mut().unwrap().persistent = true;
  1850	        }
  1851	        std::fs::remove_file(perch_of(&force_killed).join(shellinfo::SHELL_LAUNCH_FILE)).unwrap();
  1852	        let healed = shellinfo::read_shell_info(&perch_of(&force_killed)).unwrap();
  1853	        assert!(
  1854	            watcher_eligible(&perch_of(&force_killed), &healed, true, None),
  1855	            "persistent watcher eligibility needs neither launch stamp nor boot oracle"
  1856	        );
  1857	        reconcile_once(owlery, &persistent_registered, &adapters_dir, &set, &params);
  1858	        assert!(set.contains("doyle", &force_killed));
  1859	        set.stop_watcher(owlery, "doyle", &force_killed);
  1860	        set.stop_watcher(owlery, "doyle", &casualty);
  1861	    }
  1862	
  1863	    // [unit->REQ-SHELL-2] the reconciler holds the mutual exclusivity: an
  1864	    // offline instance with a wake_command gets exactly ONE watcher (a second
  1865	    // sweep never doubles it); a give-up latch suppresses adoption; onlining
  1866	    // the instance stops the watcher on the next sweep.
  1867	    #[test]
  1868	    fn reconcile_flips_watchers_with_instance_state() {
  1869	        let tmp = tempfile::tempdir().unwrap();
  1870	        let owlery = tmp.path();
  1871	        seed_owner(owlery, "doyle");
  1872	
  1873	        // A registered shell adapter whose waker sleeps (stays alive between
  1874	        // sweeps) — the long-running watcher case.
  1875	        #[cfg(windows)]
  1876	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  1877	        #[cfg(unix)]
  1878	        let (noop, sleeper) = ("true", "sleep 30");
  1879	        let adapters_dir = tmp.path().join("adapters");
  1880	        let src = tmp.path().join("srcs").join("mock-wake");
  1881	        std::fs::create_dir_all(&src).unwrap();
  1882	        std::fs::write(
  1883	            src.join("manifest.toml"),
  1884	            format!(
  1885	                "[adapter]\nname = \"mock-wake\"\nkind = \"shell\"\nversion = \"1\"\n\
  1886	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\
  1887	                 wake_command = '{sleeper}'\n"
  1888	            ),
  1889	        )
  1890	        .unwrap();
  1891	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  1892	        let registered = spt_runtime::registry::registered(&adapters_dir);
  1893	
  1894	        let id = spawn_record(owlery, "doyle", "mock-wake", None).unwrap();
  1895	        let perch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &id);
  1896	
  1897	        let set = Arc::new(WakeSet::new());
  1898	        let params = fast_params();
  1899	
  1900	        // Offline (the spawn_record default) ⇒ one watcher; never two.
  1901	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1902	        assert_eq!(set.len(), 1, "offline + wake_command ⇒ a watcher");
  1903	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1904	        assert_eq!(set.len(), 1, "a second sweep never doubles it");
  1905	        // The child parked its pid (it is the long sleeper, still up).
  1906	        //
  1907	        // READ-AND-PARSE, never `exists()`. THIS is the defect that went red on
  1908	        // kitsubito, and it is a CREATE-BEFORE-WRITE race, not a slow spawn:
  1909	        // `std::fs::write` CREATES the pid file and THEN writes into it, so a
  1910	        // reader can observe a zero-byte file. The old gate did exactly that —
  1911	        // `exists()` answered true at ~15ms, `read_to_string` returned "", and
  1912	        // `.parse()` panicked `ParseIntError { kind: Empty }`. The whole test
  1913	        // failed in 15 MILLISECONDS: its 2s ceiling was never approached, so
  1914	        // this was never a timing budget problem. Load only WIDENS the
  1915	        // create→write window and makes the race observable — a contributor,
  1916	        // never the mechanism.
  1917	        //
  1918	        // Parsing IS the readiness test: an empty file means NOT YET PARKED,
  1919	        // which is why this loop cannot be written as exists-then-read.
  1920	        //
  1921	        // The backstop below is defence against a DIFFERENT, unobserved failure
  1922	        // — a watcher whose child never spawns at all. `reconcile_once` returns
  1923	        // once the watcher THREAD is registered and the thread spawns the child
  1924	        // itself, so there is nothing to join and the wait is open-ended in
  1925	        // principle. Two terminal outcomes, so a failure says WHICH happened:
  1926	        // the pid parks, or the watcher LATCHES give-up (a product fault, and a
  1927	        // permanent one — exit at once rather than burn the backstop).
  1928	        let gave_up = perch.join(WAKER_GAVE_UP_FILE);
  1929	        let deadline = std::time::Instant::now() + WAIT_BACKSTOP;
  1930	        let mut parked: Option<u32> = None;
  1931	        let mut spawn_gave_up = false;
  1932	        while std::time::Instant::now() < deadline {
  1933	            // Read-and-PARSE, not `exists()`: the file appears before the write
  1934	            // lands, so an exists-gate can hand the next line an empty string.
  1935	            // Through the SHARED parse — a whole-content `trim().parse()` here
  1936	            // would read the two-line record as un-parked forever.
  1937	            if let Some((pid, _birth)) = read_waker_launch(&perch) {
  1938	                parked = Some(pid);
  1939	                break;
  1940	            }
  1941	            if gave_up.exists() {
  1942	                spawn_gave_up = true;
  1943	                break;
  1944	            }
  1945	            std::thread::sleep(Duration::from_millis(20));
  1946	        }
  1947	        assert!(
  1948	            !spawn_gave_up,
  1949	            "the waker child could NOT BE SPAWNED (WAKER_GAVE_UP latched) — a \
  1950	             product fault, not a slow box"
  1951	        );
  1952	        let waker_pid = parked.expect(
  1953	            "the waker never parked a pid and never latched give-up: the watcher \
  1954	             thread had not yet spawned its child. If this fires, the box is \
  1955	             saturated beyond a 60s spawn — suspect the pool, not this seam",
  1956	        );
  1957	        assert!(
  1958	            spt_store::proc::is_process_alive(waker_pid),
  1959	            "the waker runs while offline"
  1960	        );
  1961	
  1962	        // Online the instance ⇒ the next sweep stops the watcher + kills the
  1963	        // child (mutual exclusivity, the binary side wins).
  1964	        let mut info = shellinfo::read_shell_info(&perch).unwrap();
  1965	        info.status = SHELL_STATUS_ONLINE.to_string();
  1966	        shellinfo::write_shell_info(&perch, &info).unwrap();
  1967	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1968	        assert_eq!(set.len(), 0, "onlined instance keeps no watcher");
  1969	        // Same shape as the park wait, one assertion down: dropping the watcher
  1970	        // is synchronous (`set.len()` above proves it), but the CHILD's death is
  1971	        // the OS's business and lands whenever it lands. Converge on it rather
  1972	        // than asserting it on the next instruction.
  1973	        let deadline = std::time::Instant::now() + WAIT_BACKSTOP;
  1974	        let mut child_dead = false;
  1975	        while std::time::Instant::now() < deadline {
  1976	            if !spt_store::proc::is_process_alive(waker_pid) {
  1977	                child_dead = true;
  1978	                break;
  1979	            }
  1980	            std::thread::sleep(Duration::from_millis(20));
  1981	        }
  1982	        assert!(
  1983	            child_dead,
  1984	            "the waker child was killed when its instance came online (mutual \
  1985	             exclusivity — the binary side wins)"
  1986	        );
  1987	
  1988	        // Back offline but crash-latched ⇒ no adoption.
  1989	        let mut info = shellinfo::read_shell_info(&perch).unwrap();
  1990	        info.status = SHELL_STATUS_OFFLINE.to_string();
  1991	        shellinfo::write_shell_info(&perch, &info).unwrap();
  1992	        std::fs::write(perch.join(WAKER_GAVE_UP_FILE), b"").unwrap();
  1993	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  1994	        assert_eq!(set.len(), 0, "the give-up latch suppresses adoption");
  1995	    }
  1996	
  1997	    // [unit->REQ-MANIFEST-2] the wake seam resolves the **merged view**: a
  1998	    // shipped profile that adds `wake_command` makes a `<adapter>:<profile>`
  1999	    // instance wakeable, while the bare parent (no wake_command) stays inert —
  2000	    // proof the daemon re-resolves the stored composite, not the parent.
  2001	    #[test]
  2002	    fn reconcile_resolves_profile_overlay() {
  2003	        let tmp = tempfile::tempdir().unwrap();
  2004	        let owlery = tmp.path();
  2005	        seed_owner(owlery, "doyle");
  2006	
  2007	        #[cfg(windows)]
  2008	        let (noop, sleeper) = ("cmd /c exit 0", "ping -n 30 127.0.0.1");
  2009	        #[cfg(unix)]
  2010	        let (noop, sleeper) = ("true", "sleep 30");
  2011	        let adapters_dir = tmp.path().join("adapters");
  2012	        let src = tmp.path().join("srcs").join("mock-wakeprof");
  2013	        std::fs::create_dir_all(&src).unwrap();
  2014	        std::fs::write(
  2015	            src.join("manifest.toml"),
  2016	            format!(
  2017	                "[adapter]\nname = \"mock-wakeprof\"\nkind = \"shell\"\nversion = \"1\"\n\
  2018	                 min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n\n\
  2019	                 [profiles.waker]\n[profiles.waker.shell]\nwake_command = '{sleeper}'\n"
  2020	            ),
  2021	        )
  2022	        .unwrap();
  2023	        spt_runtime::registry::register(&adapters_dir, &src, 1).unwrap();
  2024	        let registered = spt_runtime::registry::registered(&adapters_dir);
  2025	
  2026	        // A bare-parent instance: no wake_command in the parent ⇒ never woken.
  2027	        spawn_record(owlery, "doyle", "mock-wakeprof", None).unwrap();
  2028	        // A profiled instance: mint a clean id off the parent, then store the
  2029	        // composite as its adapter_name (the id stays colon-free; the carrier
  2030	        // holds the option — the #8 identity split).
  2031	        let pid = spawn_record(owlery, "doyle", "mock-wakeprof", None).unwrap();
  2032	        let pperch = spt_store::perch::resolve_shell_perch_path_in(owlery, "doyle", &pid);
  2033	        let mut info = shellinfo::read_shell_info(&pperch).unwrap();
  2034	        info.adapter_name = "mock-wakeprof:waker".to_string();
  2035	        shellinfo::write_shell_info(&pperch, &info).unwrap();
  2036	
  2037	        let set = Arc::new(WakeSet::new());
  2038	        let params = fast_params();
  2039	        reconcile_once(owlery, &registered, &adapters_dir, &set, &params);
  2040	        assert_eq!(
  2041	            set.len(),
  2042	            1,
  2043	            "only the profiled instance (overlay adds wake_command) gets a watcher"
  2044	        );
  2045	        assert!(set.contains("doyle", &pid), "the profiled instance is the one watched");
  2046	
  2047	        set.stop_watcher(owlery, "doyle", &pid);
  2048	    }
  2049	
  2050	    // ──────────────────────────────────────────────────────────────────────
  2051	    // Waker kill authentication (REQ-SHELL-KILL-AUTHENTICATED)
  2052	    // ──────────────────────────────────────────────────────────────────────
  2053	
  2054	    fn waker_stand_in() -> std::process::Child {
  2055	        #[cfg(windows)]
  2056	        {
  2057	            use std::os::windows::process::CommandExt;
  2058	            Command::new("cmd")
  2059	                .args(["/c", "ping -n 60 127.0.0.1"])
  2060	                // stdin too — see the note on `live_stand_in`.
  2061	                .stdin(Stdio::null())
  2062	                .stdout(Stdio::null())
  2063	                .stderr(Stdio::null())
  2064	                .creation_flags(0x0800_0000) // CREATE_NO_WINDOW
  2065	                .spawn()
  2066	                .expect("spawned the stand-in waker")
  2067	        }
  2068	        #[cfg(unix)]
  2069	        {
  2070	            Command::new("sleep")
  2071	                .arg("60")
  2072	                .stdin(Stdio::null())
  2073	                .stdout(Stdio::null())
  2074	                .stderr(Stdio::null())
  2075	                .spawn()
  2076	                .expect("spawned the stand-in waker")
  2077	        }
  2078	    }
  2079	
  2080	    fn waker_died_within(child: &mut std::process::Child, ms: u64) -> bool {
  2081	        let deadline = std::time::Instant::now() + Duration::from_millis(ms);
  2082	        loop {
  2083	            if matches!(child.try_wait(), Ok(Some(_))) {
  2084	                return true;
  2085	            }
  2086	            if std::time::Instant::now() >= deadline {
  2087	                return false;
  2088	            }
  2089	            std::thread::sleep(Duration::from_millis(25));
  2090	        }
  2091	    }
  2092	
  2093	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] The pair round-trips through the ONE
  2094	    // shared parse, and a legacy one-line record still reads — the format change
  2095	    // must not orphan the records already on disk.
  2096	    #[test]
  2097	    fn waker_record_round_trips_and_reads_the_legacy_shape() {
  2098	        let d = tempfile::tempdir().unwrap();
  2099	        let perch = d.path();
  2100	
  2101	        let me = std::process::id();
  2102	        record_waker_launch(perch, me);
  2103	        let (pid, birth) = read_waker_launch(perch).expect("the record parses");
  2104	        assert_eq!(pid, me);
  2105	        assert_eq!(
  2106	            birth,
  2107	            spt_store::proc::process_started_at(me),
  2108	            "the stamp parked beside the pid is the pid's own birth"
  2109	        );
  2110	
  2111	        // The pre-fix shape: one line, no stamp.
  2112	        std::fs::write(perch.join(WAKER_PID_FILE), me.to_string()).unwrap();
  2113	        assert_eq!(
  2114	            read_waker_launch(perch),
  2115	            Some((me, None)),
  2116	            "a legacy one-line record still yields its pid, with no stamp to compare"
  2117	        );
  2118	    }
  2119	
  2120	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] A recycled waker pid is spared, and
  2121	    // its record retired. Without the WRITE-side stamp this row is unreachable:
  2122	    // a bare pid reads Held and the kill fires.
  2123	    #[test]
  2124	    fn kill_waker_at_spares_a_recycled_pid() {
  2125	        let d = tempfile::tempdir().unwrap();
  2126	        let perch = d.path();
  2127	        let mut stranger = waker_stand_in();
  2128	        // The recycled shape: a LIVE pid beside a stamp belonging to something
  2129	        // else — written by hand, since `record_waker_launch` would pair honestly.
  2130	        std::fs::write(
  2131	            perch.join(WAKER_PID_FILE),
  2132	            format!("{}\n123", stranger.id()),
  2133	        )
  2134	        .unwrap();
  2135	
  2136	        kill_waker_at(perch);
  2137	
  2138	        let died = waker_died_within(&mut stranger, 750);
  2139	        crate::shellhost::kill_shell_pid(stranger.id());
  2140	        let _ = stranger.wait();
  2141	        assert!(!died, "a waker record whose stamp mismatches must not kill");
  2142	        assert!(
  2143	            !perch.join(WAKER_PID_FILE).exists(),
  2144	            "a provably-not-ours waker record is retired"
  2145	        );
  2146	    }
  2147	
  2148	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] Non-vacuity for the waker half: a
  2149	    // genuine pair still dies. Asserted beside the row above, because a gate
  2150	    // that spares everything passes that one on its own.
  2151	    #[test]
  2152	    fn kill_waker_at_still_kills_a_matching_pair() {
  2153	        let d = tempfile::tempdir().unwrap();
  2154	        let perch = d.path();
  2155	        let mut ours = waker_stand_in();
  2156	        record_waker_launch(perch, ours.id());
  2157	
  2158	        kill_waker_at(perch);
  2159	
  2160	        let died = waker_died_within(&mut ours, 5_000);
  2161	        crate::shellhost::kill_shell_pid(ours.id());
  2162	        let _ = ours.wait();
  2163	        assert!(died, "the authenticated waker kill must still kill our own");
  2164	    }
  2165	
  2166	    // [unit->REQ-SHELL-KILL-AUTHENTICATED] BOTH halves of the parse-failure arm,
  2167	    // asserted together: refuse the kill AND KEEP the record. Retiring it while
  2168	    // killing nothing is the shape that orphans a live waker and reports
  2169	    // success — strictly worse than the defect being fixed, and a kill-count
  2170	    // assertion alone would pass it.
  2171	    #[test]
  2172	    fn kill_waker_at_refuses_and_keeps_an_unparseable_record() {
  2173	        let d = tempfile::tempdir().unwrap();
  2174	        let perch = d.path();
  2175	        let mut stranger = waker_stand_in();
  2176	        std::fs::write(perch.join(WAKER_PID_FILE), "not-a-pid\n").unwrap();
  2177	
  2178	        kill_waker_at(perch);
  2179	
  2180	        let died = waker_died_within(&mut stranger, 500);
  2181	        crate::shellhost::kill_shell_pid(stranger.id());
  2182	        let _ = stranger.wait();
  2183	        assert!(!died, "an unreadable record kills nothing");
  2184	        assert!(
  2185	            perch.join(WAKER_PID_FILE).exists(),
  2186	            "an unreadable record is KEPT — retiring it would orphan a live waker while \
  2187	             reporting success"
  2188	        );
  2189	    }
  2190	}