warning: in the working copy of 'crates/spt-daemon/src/broker.rs', LF will be replaced by CRLF the next time Git touches it warning: in the working copy of 'crates/spt-daemon/src/effect.rs', LF will be replaced by CRLF the next time Git touches it warning: in the working copy of 'crates/spt-daemon/src/pairhost.rs', LF will be replaced by CRLF the next time Git touches it warning: in the working copy of 'crates/spt-daemon/src/servicehost.rs', LF will be replaced by CRLF the next time Git touches it diff --git a/crates/spt-daemon/src/broker.rs b/crates/spt-daemon/src/broker.rs index 1b1acc4..c71b5af 100644 --- a/crates/spt-daemon/src/broker.rs +++ b/crates/spt-daemon/src/broker.rs @@ -34,7 +34,9 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::io; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; -use std::sync::mpsc::{channel, sync_channel, Receiver, RecvTimeoutError, Sender, SyncSender, TrySendError}; +use std::sync::mpsc::{ + channel, sync_channel, Receiver, RecvTimeoutError, Sender, SyncSender, TrySendError, +}; use std::sync::{Arc, Mutex, OnceLock}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; @@ -52,26 +54,31 @@ use crate::effect::{EffectJournal, EffectKey, EffectKind, Minter, Outcome}; use crate::frame::{Envelope, Role}; use crate::msg::{ applied_envelope, decode_bytes, displaced_envelope, endpoint_injected_envelope, - evicted_envelope, net_dialed_envelope, output_envelope, - size_envelope, subscribed_envelope, sync_output_envelope, AdapterApplyReq, BrainRestarted, EndpointInputReq, - ErrorEvent, ExitEvent, InputReq, KillReq, NetDialReq, NetPresenceSubscribeReq, NetSent, TeardownReq, KIND_TEARDOWN, - BrokerImageReply, CoordinatorImageAnnounce, CoordinatorImageAnnounceReply, CoordinatorImageReply, - StallEvictsReply, NetStatusReply, NetStreamOpenReq, NetStreamOpened, NetStreamSendReq, NetStreamSubscribeReq, - NetStreamsReply, NetStreamOpenerReply, NetStreamOpenerReq, NetStreamRetireReq, NetStreamRetired, NetStreamUnsubscribeReq, NetStreamUnsubscribed, MetMember, PairCodeSubmit, PairJoinReply, PairJoinReq, PairMeetReq, ResizeReq, SessionInfo, SessionsReply, SpawnReq, - SpawnConflict, Spawned, SubscribeOutcome, SubscribeReq, UnsubscribeReq, KIND_BRAIN_RESTART, KIND_BRAIN_RESTARTED, KIND_ENDPOINT_INPUT, KIND_ERROR, KIND_EXIT, - KIND_INPUT, KIND_KILL, KIND_NET_DIAL, KIND_NET_DIAL_LOOPBACK, KIND_NET_DIAL_SUBMIT, - KIND_NET_DIAL_SUBMITTED, KIND_NET_PRESENCE_SUBSCRIBE, - KIND_NET_SENT, - KIND_ADAPTER_APPLY, KIND_APPLIED, - KIND_BROKER_IMAGE, KIND_BROKER_IMAGE_REPLY, KIND_STALL_EVICTS, KIND_STALL_EVICTS_REPLY, - KIND_COORDINATOR_IMAGE, KIND_COORDINATOR_IMAGE_ANNOUNCE, KIND_COORDINATOR_IMAGE_ANNOUNCE_REPLY, - KIND_COORDINATOR_IMAGE_REPLY, - KIND_NET_STATUS, KIND_NET_STATUS_REPLY, KIND_NET_STREAMS, KIND_NET_STREAMS_REPLY, - KIND_NET_STREAM_OPEN, KIND_NET_STREAM_OPENED, KIND_NET_STREAM_OPENER, KIND_NET_STREAM_OPENER_REPLY, KIND_NET_STREAM_RETIRE, KIND_NET_STREAM_RETIRED, KIND_NET_STREAM_SEND, KIND_NET_STREAM_SUBSCRIBE, - KIND_NET_STREAM_UNSUBSCRIBE, KIND_NET_STREAM_UNSUBSCRIBED, - KIND_MET_MEMBER, KIND_PAIR_CODE_SUBMIT, KIND_PAIR_JOIN, KIND_PAIR_JOINED, KIND_PAIR_MEET, KIND_RESIZE, KIND_SESSIONS, KIND_SESSIONS_REPLY, KIND_SPAWN, - BringUpReq, BroughtUpReply, BRING_UP_ADMITTED, BRING_UP_ADMIT_PREFIX, BRING_UP_ALREADY_LIVE, BRING_UP_REFUSED, KIND_BRING_UP, KIND_BROUGHT_UP, - KIND_SPAWNED, KIND_SPAWN_CONFLICT, KIND_SPAWN_FRESH, KIND_SUBSCRIBE, KIND_UNSUBSCRIBE, + evicted_envelope, net_dialed_envelope, output_envelope, size_envelope, subscribed_envelope, + sync_output_envelope, AdapterApplyReq, BrainRestarted, BringUpReq, BrokerImageReply, + BroughtUpReply, CoordinatorImageAnnounce, CoordinatorImageAnnounceReply, CoordinatorImageReply, + EndpointInputReq, ErrorEvent, ExitEvent, InputReq, KillReq, MetMember, NetDialReq, + NetPresenceSubscribeReq, NetSent, NetStatusReply, NetStreamOpenReq, NetStreamOpened, + NetStreamOpenerReply, NetStreamOpenerReq, NetStreamRetireReq, NetStreamRetired, + NetStreamSendReq, NetStreamSubscribeReq, NetStreamUnsubscribeReq, NetStreamUnsubscribed, + NetStreamsReply, PairCodeSubmit, PairJoinReply, PairJoinReq, PairMeetReq, ResizeReq, + SessionInfo, SessionsReply, SpawnConflict, SpawnReq, Spawned, StallEvictsReply, + SubscribeOutcome, SubscribeReq, TeardownReq, UnsubscribeReq, BRING_UP_ADMITTED, + BRING_UP_ADMIT_PREFIX, BRING_UP_ALREADY_LIVE, BRING_UP_REFUSED, KIND_ADAPTER_APPLY, + KIND_APPLIED, KIND_BRAIN_RESTART, KIND_BRAIN_RESTARTED, KIND_BRING_UP, KIND_BROKER_IMAGE, + KIND_BROKER_IMAGE_REPLY, KIND_BROUGHT_UP, KIND_COORDINATOR_IMAGE, + KIND_COORDINATOR_IMAGE_ANNOUNCE, KIND_COORDINATOR_IMAGE_ANNOUNCE_REPLY, + KIND_COORDINATOR_IMAGE_REPLY, KIND_ENDPOINT_INPUT, KIND_ERROR, KIND_EXIT, KIND_INPUT, + KIND_KILL, KIND_MET_MEMBER, KIND_NET_DIAL, KIND_NET_DIAL_LOOPBACK, KIND_NET_DIAL_SUBMIT, + KIND_NET_DIAL_SUBMITTED, KIND_NET_PRESENCE_SUBSCRIBE, KIND_NET_SENT, KIND_NET_STATUS, + KIND_NET_STATUS_REPLY, KIND_NET_STREAMS, KIND_NET_STREAMS_REPLY, KIND_NET_STREAM_OPEN, + KIND_NET_STREAM_OPENED, KIND_NET_STREAM_OPENER, KIND_NET_STREAM_OPENER_REPLY, + KIND_NET_STREAM_RETIRE, KIND_NET_STREAM_RETIRED, KIND_NET_STREAM_SEND, + KIND_NET_STREAM_SUBSCRIBE, KIND_NET_STREAM_UNSUBSCRIBE, KIND_NET_STREAM_UNSUBSCRIBED, + KIND_PAIR_CODE_SUBMIT, KIND_PAIR_JOIN, KIND_PAIR_JOINED, KIND_PAIR_MEET, KIND_RESIZE, + KIND_SESSIONS, KIND_SESSIONS_REPLY, KIND_SPAWN, KIND_SPAWNED, KIND_SPAWN_CONFLICT, + KIND_SPAWN_FRESH, KIND_STALL_EVICTS, KIND_STALL_EVICTS_REPLY, KIND_SUBSCRIBE, KIND_TEARDOWN, + KIND_UNSUBSCRIBE, }; use crate::nethost::{NetHost, NET_EFFECT_SESSION}; use crate::translation::{key_to_bytes, InjectFloor, KeyCmd, ToBinary, TranslationChild}; @@ -308,8 +315,8 @@ fn zombie_verdict( past_grace: bool, ) -> bool { match wrapper_alive { - None => false, // no probeable pid — never guess - Some(false) => true, // dead root, surviving record — always a zombie + None => false, // no probeable pid — never guess + Some(false) => true, // dead root, surviving record — always a zombie Some(true) => adapter_labeled && past_grace && !has_live_descendants, } } @@ -362,7 +369,12 @@ pub fn session_is_zombie(pid: Option, adapter_labeled: bool, spawned_ms_ago .any(|d| spt_store::proc::is_process_alive(*d)) }); let past_grace = Duration::from_millis(spawned_ms_ago) >= spawn_client_grace(); - zombie_verdict(wrapper_alive, adapter_labeled, has_live_descendants, past_grace) + zombie_verdict( + wrapper_alive, + adapter_labeled, + has_live_descendants, + past_grace, + ) } /// The RC-origin input-fence verdict (ADR-0044 decision 3, @@ -493,7 +505,11 @@ const INJECT_MISS_STRIKE_BUDGET: u32 = 3; /// invariant, not its exact length. fn inject_miss_strike_budget() -> u32 { match std::env::var("SPT_INJECT_MISS_STRIKE_BUDGET") { - Ok(n) => n.parse::().ok().filter(|b| *b >= 1).unwrap_or(INJECT_MISS_STRIKE_BUDGET), + Ok(n) => n + .parse::() + .ok() + .filter(|b| *b >= 1) + .unwrap_or(INJECT_MISS_STRIKE_BUDGET), Err(_) => INJECT_MISS_STRIKE_BUDGET, } } @@ -593,7 +609,11 @@ const INJECT_TEXT_CHUNK: usize = 256; fn inject_text_chunk() -> usize { match std::env::var("SPT_INJECT_TEXT_CHUNK") { - Ok(n) => n.parse::().ok().filter(|c| *c > 0).unwrap_or(INJECT_TEXT_CHUNK), + Ok(n) => n + .parse::() + .ok() + .filter(|c| *c > 0) + .unwrap_or(INJECT_TEXT_CHUNK), Err(_) => INJECT_TEXT_CHUNK, } } @@ -1465,9 +1485,7 @@ impl OutputLog { // attach, which is precisely what "bring the engine room up" means. // The TOTP gate is the other half of the same seat rule and lives at // the same door, for the same reason the locks do. - crate::attach::EngineRoomLock::Pass => { - return self.bringup_refusal(code, conn, ticket) - } + crate::attach::EngineRoomLock::Pass => return self.bringup_refusal(code, conn, ticket), crate::attach::EngineRoomLock::NoViewport => { "the engine room has no viewport — it is never watched, at any origin" } @@ -1510,12 +1528,7 @@ impl OutputLog { /// a workaround. // [impl->REQ-ER-BRINGUP-TOTP-GATE] // [impl->REQ-ER-BRINGUP-ATTEMPT-BOUND] - fn bringup_refusal( - &self, - code: Option<&str>, - conn: u64, - ticket: AdmitTicket, - ) -> SeatGate { + fn bringup_refusal(&self, code: Option<&str>, conn: u64, ticket: AdmitTicket) -> SeatGate { use spt_store::engineroom as er; match ticket { // A bring-up this broker admitted seats its controller without @@ -2055,7 +2068,13 @@ impl OutputLog { } } - fn become_controller(&mut self, sub: SharedSend, by: Option, from_seq: u64, attach_gen: u64) { + fn become_controller( + &mut self, + sub: SharedSend, + by: Option, + from_seq: u64, + attach_gen: u64, + ) { // Drop the prior controller sink first (its writer's live loop ends when // tx drops), and bump the generation so (a) an in-flight deadline-evict // for the old controller can't unseat this one and (b) a prior writer @@ -2073,7 +2092,11 @@ impl OutputLog { sub.describe(&format!( "controller session={} endpoint={} by={}", self.session_id, - if self.endpoint.is_empty() { "-" } else { &self.endpoint }, + if self.endpoint.is_empty() { + "-" + } else { + &self.endpoint + }, by.as_deref().unwrap_or("local") )); sub.lifecycle_event( @@ -2149,7 +2172,7 @@ impl OutputLog { _writer: writer, attach_gen, revoked_by, - }); + }); self.stamp_driven_by(); // The seat changed hands (or was filled): the incoming controller does // not inherit the outgoing one's authority. An equal-generation re-serve @@ -2333,7 +2356,11 @@ impl OutputLog { sub.describe(&format!( "viewer session={} endpoint={} vid={vid}", self.session_id, - if self.endpoint.is_empty() { "-" } else { &self.endpoint } + if self.endpoint.is_empty() { + "-" + } else { + &self.endpoint + } )); sub.lifecycle_event( "viewer-attach", @@ -2370,8 +2397,9 @@ impl OutputLog { let evicted = Arc::new(AtomicBool::new(false)); let writer_evicted = Arc::clone(&evicted); let session_id = self.session_id; - let writer = - thread::spawn(move || viewer_writer(writer_send, session_id, initial, rx, writer_evicted)); + let writer = thread::spawn(move || { + viewer_writer(writer_send, session_id, initial, rx, writer_evicted) + }); self.viewers.insert( vid, ViewerSink { @@ -2466,7 +2494,11 @@ impl OutputLog { "SUBSCRIBE_DECISION: session={} endpoint={} by={} conn={} intent={} \ old_by={} old_gen={} req_gen={} decision={}", self.session_id, - if self.endpoint.is_empty() { "-" } else { &self.endpoint }, + if self.endpoint.is_empty() { + "-" + } else { + &self.endpoint + }, by_lbl, conn, intent_lbl, @@ -2993,7 +3025,6 @@ impl OutputLog { } } - /// Stamp the perch's `driven_by` to the current controller's identity (the /// broker is the single writer — resolves the clear-race). The remote-drive /// detection fact (REQ-REACH-1) moved here from `serve_attach` so a displaced @@ -3141,7 +3172,10 @@ fn viewer_writer( fn drop(&mut self) { self.send.lifecycle_event( "writer-exit", - &format!("role=viewer session={} reason={}", self.session_id, self.reason), + &format!( + "role=viewer session={} reason={}", + self.session_id, self.reason + ), ); } } @@ -3404,7 +3438,10 @@ fn controller_writer( note_controller_write_retired(sid, send.id(), &e); send.lifecycle_event( "writer-exit", - &format!("role=controller session={sid} reason=write-failed kind={:?}", e.kind()), + &format!( + "role=controller session={sid} reason=write-failed kind={:?}", + e.kind() + ), ); return; } @@ -3443,7 +3480,10 @@ fn controller_writer( // [impl->REQ-CONN-POISON-ATTRIBUTION] send.lifecycle_event( "writer-exit", - &format!("role=controller session={sid} reason=write-failed kind={:?}", e.kind()), + &format!( + "role=controller session={sid} reason=write-failed kind={:?}", + e.kind() + ), ); return; } @@ -3738,7 +3778,6 @@ fn recover_log(m: &Mutex) -> std::sync::MutexGuard<'_, OutputLog> { } } - /// WHICH of the home subnet's two seeds admitted a bring-up code. /// /// A bool was enough until releases#102: an ADMIN-TOTP bring-up empowers the @@ -3807,10 +3846,7 @@ pub(crate) fn classify_cred(member_verified: bool, admin_verified: bool) -> Brin /// ceremony, and nothing here reaches the joiner-facing wire. // [impl->REQ-ER-BRINGUP-TOTP-GATE] // [impl->REQ-ENGINEROOM-ADMIN-BRINGUP-EMPOWERS] -fn bringup_code_verifies( - room: &spt_store::engineroom::EngineRoom, - presented: &str, -) -> BringUpCred { +fn bringup_code_verifies(room: &spt_store::engineroom::EngineRoom, presented: &str) -> BringUpCred { use spt_net::net::pairing::totp::code_matches_window; let subnets = spt_store::subnet::SubnetStore::load(); let Some(rec) = subnets.find(&room.home_subnet) else { @@ -5306,8 +5342,7 @@ impl Broker { // AND the off-lock converge_perch_stamps below see the // cleared state and the stale info.json stamp clears. let _ = log.reap_dead_controller(); - let stamp_gen = - stamp_slot(&endpoint).gen.load(Ordering::Acquire); + let stamp_gen = stamp_slot(&endpoint).gen.load(Ordering::Acquire); SessSnap { id, endpoint, @@ -5427,7 +5462,9 @@ impl Broker { for id in &my_cb_streams { let _ = host.send_stream(*id, &[], true); let _ = host.retire_stream_terminal(*id); - eprintln!("STREAM_CONNBOUND_RETIRE:{id}: opener conn exited — FIN + terminal retire"); + eprintln!( + "STREAM_CONNBOUND_RETIRE:{id}: opener conn exited — FIN + terminal retire" + ); } // And presence: the liveness log + its ring persist (D4c). if my_presence_sub { @@ -5457,7 +5494,11 @@ impl Broker { /// [`KIND_SPAWN_CONFLICT`] — NEVER `Spawned(existing)`. `Ok(None)` = the /// conflict was sent (no session to auto-subscribe). // [impl->REQ-SPAWN-FRESH-TRUTHFUL] - fn dispatch_spawn_fresh(&self, env: Envelope, send: &SharedSend) -> Result, String> { + fn dispatch_spawn_fresh( + &self, + env: Envelope, + send: &SharedSend, + ) -> Result, String> { let req: SpawnReq = serde_json::from_value(env.payload).map_err(|e| format!("bad spawn payload: {e}"))?; self.dispatch_spawn_policy(req, send, true) @@ -5564,7 +5605,9 @@ impl Broker { let adapters_dir = spt_store::perch::adapters_dir(); let (record, manifest) = match spt_runtime::registry::resolve_option(&adapters_dir, &room.adapter) { - Ok((r, m)) if m.adapter.kind == spt_runtime::manifest::AdapterKind::Harness => (r, m), + Ok((r, m)) if m.adapter.kind == spt_runtime::manifest::AdapterKind::Harness => { + (r, m) + } _ => { reply( BRING_UP_REFUSED, @@ -5719,12 +5762,7 @@ impl Broker { /// than to sleep through it, and a sleeping test proves the TTL only for the /// value it slept for. // [impl->REQ-ER-BRINGUP-SPAWNS-SESSION] - fn mint_bringup_admit_at( - &self, - session: u64, - verdict: TicketVerdict, - now: Instant, - ) -> String { + fn mint_bringup_admit_at(&self, session: u64, verdict: TicketVerdict, now: Instant) -> String { let seed = spt_proto::identity::Identity::generate().seed(); let ticket = format!( "{BRING_UP_ADMIT_PREFIX}{}", @@ -5817,16 +5855,17 @@ impl Broker { let gate = { let sessions = recover(&self.sessions); let mut inflight = recover(&self.wake_inflight); - let live = sessions.iter().find(|(_, h)| h.endpoint == req.endpoint).map( - |(sid, h)| { + let live = sessions + .iter() + .find(|(_, h)| h.endpoint == req.endpoint) + .map(|(sid, h)| { ( *sid, h.session.process_id(), !h.adapter.is_empty(), h.spawned_at.elapsed().as_millis() as u64, ) - }, - ); + }); match wake_gate_decision( live.is_some(), inflight.contains(&req.endpoint), @@ -5835,7 +5874,12 @@ impl Broker { WakeGate::AlreadyLive => { let (sid, spid, adapter_labeled, spawned_ms_ago) = live.expect("live is Some on AlreadyLive"); - Gate::AlreadyLive { sid, spid, adapter_labeled, spawned_ms_ago } + Gate::AlreadyLive { + sid, + spid, + adapter_labeled, + spawned_ms_ago, + } } WakeGate::Racing => Gate::Racing, WakeGate::Claim => { @@ -5849,7 +5893,12 @@ impl Broker { }; match gate { Gate::Claimed(guard) => break Some(guard), - Gate::AlreadyLive { sid, spid, adapter_labeled, spawned_ms_ago } => { + Gate::AlreadyLive { + sid, + spid, + adapter_labeled, + spawned_ms_ago, + } => { // ONE liveness authority (ADR-0041 decision 6, // REQ-ENDPOINT-CYCLE-HONEST): before refusing/deduping by // citing the claimed session, PROBE its client tree — off @@ -5922,8 +5971,11 @@ impl Broker { ); let frame = Envelope::new( KIND_SPAWNED, - serde_json::to_value(Spawned { session_id: sid, pid: spid }) - .expect("Spawned serializes"), + serde_json::to_value(Spawned { + session_id: sid, + pid: spid, + }) + .expect("Spawned serializes"), ); send_frame(send, &frame); return Ok(Some(sid)); @@ -6122,18 +6174,15 @@ impl Broker { // C-1: the shared bounded-respawn give-up counter (starts at 0; the worker // resets it on a healthy commit, the dispatch respawn path increments it). let translation_respawns = Arc::new(AtomicU32::new(0)); - let translation = req - .translation_binary - .as_deref() - .and_then(|argv| { - build_translation( - argv, - &req.endpoint, - &input, - Arc::clone(&translation_respawns), - &log, - ) - }); + let translation = req.translation_binary.as_deref().and_then(|argv| { + build_translation( + argv, + &req.endpoint, + &input, + Arc::clone(&translation_respawns), + &log, + ) + }); recover(&self.sessions).insert( id, @@ -6275,7 +6324,11 @@ impl Broker { let h = sessions .get(&req.session_id) .ok_or_else(|| format!("no such session {}", req.session_id))?; - (Arc::clone(&h.input), h.translation.clone(), Arc::clone(&h.log)) + ( + Arc::clone(&h.input), + h.translation.clone(), + Arc::clone(&h.log), + ) }; // RC-ORIGIN INPUT FENCE (ADR-0044 decision 3, the required defense): // an rc-tagged input must come from the ACTIVE controller lease's @@ -6416,7 +6469,9 @@ impl Broker { // recovered session shows clean; re-stamped if the new binary faults too). let perch = resolve_perch_path(endpoint, ParentHint::Infer); let _ = spt_store::info::set_translation_fault(&perch, None); - eprintln!("TRANSLATION_RESPAWN:{endpoint}: rebuilt faulted binary (attempt {n}/{budget})"); + eprintln!( + "TRANSLATION_RESPAWN:{endpoint}: rebuilt faulted binary (attempt {n}/{budget})" + ); } // Swap it in under the lock (the session may have exited mid-build). let mut map = recover(&self.sessions); @@ -6511,7 +6566,11 @@ impl Broker { "ENDPOINT_INJECT:{} ({} bytes → translation binary{})", req.endpoint, bytes.len(), - if req.native && !idle { ", native mid-active" } else { "" } + if req.native && !idle { + ", native mid-active" + } else { + "" + } ); send_frame( send, @@ -6536,20 +6595,29 @@ impl Broker { "ENDPOINT_INJECT:{}: endpoint ACTIVE -> spool (deferred hint), not injected", req.endpoint ); - send_frame(send, &endpoint_injected_envelope(&req.endpoint, false, true)); + send_frame( + send, + &endpoint_injected_envelope(&req.endpoint, false, true), + ); } else { eprintln!( "ENDPOINT_INJECT:{}: no working translation binary (absent/faulted/worker-gone) -> SPOOLED (idle window), not injected", req.endpoint ); - send_frame(send, &endpoint_injected_envelope(&req.endpoint, false, false)); + send_frame( + send, + &endpoint_injected_envelope(&req.endpoint, false, false), + ); } Ok(()) } // No hosted session for this endpoint — tell the caller to spool // NON-deferred (idle-eligible; a non-hosted target has no active window). None => { - send_frame(send, &endpoint_injected_envelope(&req.endpoint, false, false)); + send_frame( + send, + &endpoint_injected_envelope(&req.endpoint, false, false), + ); Ok(()) } } @@ -6678,8 +6746,10 @@ impl Broker { let supervised = crate::brainproc::supervised_generation(); let accepted = match serde_json::from_value::(env.payload) { Ok(a) if coordinator_announce_accepted(a.generation, supervised) => { - *self.coordinator_image.lock().expect("coordinator image lock") = - Some((a.generation, a.version)); + *self + .coordinator_image + .lock() + .expect("coordinator image lock") = Some((a.generation, a.version)); true } _ => false, @@ -6746,8 +6816,7 @@ impl Broker { }; let frame = Envelope::new( KIND_BRAIN_RESTARTED, - serde_json::to_value(BrainRestarted { honored }) - .expect("BrainRestarted serializes"), + serde_json::to_value(BrainRestarted { honored }).expect("BrainRestarted serializes"), ); send_frame(send, &frame); } @@ -6848,7 +6917,10 @@ impl Broker { .to_hex(); host.submit_dial(addr, remote_id_hex); // Immediate bare ack — the dial spawned; its outcome is a presence event. - send_frame(send, &Envelope::new(KIND_NET_DIAL_SUBMITTED, serde_json::Value::Null)); + send_frame( + send, + &Envelope::new(KIND_NET_DIAL_SUBMITTED, serde_json::Value::Null), + ); Ok(()) } @@ -7338,8 +7410,7 @@ impl Broker { sessions .iter() .find(|(id, h)| { - req.session_id == Some(**id) - || want_endpoint.is_some_and(|e| e == h.endpoint) + req.session_id == Some(**id) || want_endpoint.is_some_and(|e| e == h.endpoint) }) .map(|(_, h)| (h.session.process_id(), Arc::clone(&h.session))) }; @@ -7431,12 +7502,10 @@ impl Broker { &rec.name, ) .ok() - .and_then(|m| m.service) - else { + .and_then(|m| m.service) else { continue; }; - let outcome = - crate::servicehost::quiesce_for_update(set, &rec.name, &service); + let outcome = crate::servicehost::quiesce_for_update(set, &rec.name, &service); eprintln!("SERVICE_QUIESCE:{}: {outcome:?}", rec.name); if !outcome.clear_to_swap() { // Release every hold we took, including this one: an @@ -7575,15 +7644,30 @@ mod tests { #[test] fn stamp_divergence_gates_writes() { // Converged already → no writes. - assert_eq!(stamp_divergence(None, false, 0, None, false, 0), (false, false)); + assert_eq!( + stamp_divergence(None, false, 0, None, false, 0), + (false, false) + ); // The stamp-before-bind loss: perch says controlled=false, session IS driven. - assert_eq!(stamp_divergence(None, false, 0, None, true, 0), (true, false)); + assert_eq!( + stamp_divergence(None, false, 0, None, true, 0), + (true, false) + ); // A remote controller's driven_by appears → control write. - assert_eq!(stamp_divergence(None, true, 0, Some("n"), true, 0), (true, false)); + assert_eq!( + stamp_divergence(None, true, 0, Some("n"), true, 0), + (true, false) + ); // Viewer count changed only → viewer write only. - assert_eq!(stamp_divergence(None, true, 0, None, true, 2), (false, true)); + assert_eq!( + stamp_divergence(None, true, 0, None, true, 2), + (false, true) + ); // Both diverge. - assert_eq!(stamp_divergence(Some("a"), false, 1, None, true, 3), (true, true)); + assert_eq!( + stamp_divergence(Some("a"), false, 1, None, true, 3), + (true, true) + ); } // [unit->REQ-UPDATE-RUNNING-IMAGE-SURFACE] the coordinator-image @@ -7948,7 +8032,10 @@ mod tests { ); let ev: crate::msg::ViewerEvictedEvent = serde_json::from_value(env.payload).expect("marker payload"); - assert_eq!(ev.session_id, 7, "the marker names the evicted viewer's session"); + assert_eq!( + ev.session_id, 7, + "the marker names the evicted viewer's session" + ); } // ── NORMAL close: flag false → no marker; the client read hits EOF. ── @@ -8012,7 +8099,11 @@ mod tests { ); // First append fits the depth-1 queue (no overflow, not evicted yet). - assert_eq!(log.append(b"chunk-0"), None, "no controller; first chunk fits"); + assert_eq!( + log.append(b"chunk-0"), + None, + "no controller; first chunk fits" + ); assert!( !observed.load(Ordering::Acquire), "a viewer keeping within its queue is NOT flagged" @@ -8021,7 +8112,11 @@ mod tests { // Second append OVERFLOWS the (still-undrained) depth-1 queue → eviction: // the flag is SET (so the writer skips-to-live) and the sink is removed. - assert_eq!(log.append(b"chunk-1"), None, "no controller; eviction returns no ctrl job"); + assert_eq!( + log.append(b"chunk-1"), + None, + "no controller; eviction returns no ctrl job" + ); assert!( observed.load(Ordering::Acquire), "an overflow eviction must SET the sink's `evicted` flag BEFORE dropping \ @@ -8071,7 +8166,14 @@ mod tests { fn exit_enqueues_behind_queued_output_per_sink() { let (send, mut client, _recv) = controller_socket_pair(); let mut log = OutputLog::new(9, DEFAULT_LOG_CHUNKS, String::new(), (24, 80)); - let out = log.resolve_subscribe(Arc::clone(&send), 0, AttachIntent::Control, Some("op".into()), 200, None); + let out = log.resolve_subscribe( + Arc::clone(&send), + 0, + AttachIntent::Control, + Some("op".into()), + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); // Queue output THEN the Exit through the same fanout the exit waiter uses. @@ -8080,7 +8182,11 @@ mod tests { } let frame = Envelope::new( crate::msg::KIND_EXIT, - serde_json::to_value(ExitEvent { session_id: 9, code: Some(0) }).unwrap(), + serde_json::to_value(ExitEvent { + session_id: 9, + code: Some(0), + }) + .unwrap(), ); let fanout = log.exit_fanout(); let (tx, sink) = fanout.controller.expect("controller queue"); @@ -8117,17 +8223,34 @@ mod tests { let (taker, _cc, _rc2) = controller_socket_pair(); let mut log = OutputLog::new(1, DEFAULT_LOG_CHUNKS, String::new(), (24, 80)); - let out = log.resolve_subscribe(Arc::clone(&live), 0, AttachIntent::Control, Some("op".into()), 200, None); + let out = log.resolve_subscribe( + Arc::clone(&live), + 0, + AttachIntent::Control, + Some("op".into()), + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); // NEWER gen + plain Control → loud SUPERSESSION (the live replacement // viewport — the T6 shape): the slot moves; outcome is Controller. // (The revoked marker is CONSUMED by the exiting writer — emission is // proven by `revoked_incumbent_writer_emits_the_terminal_displaced`.) - let out = log.resolve_subscribe(Arc::clone(&ctrl2), 0, AttachIntent::Control, Some("op".into()), 300, None); + let out = log.resolve_subscribe( + Arc::clone(&ctrl2), + 0, + AttachIntent::Control, + Some("op".into()), + 300, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); let c = log.controller.as_ref().expect("replacement holds the slot"); - assert!(Arc::ptr_eq(&c.send, &ctrl2), "the newer viewport superseded"); + assert!( + Arc::ptr_eq(&c.send, &ctrl2), + "the newer viewport superseded" + ); assert_eq!(c.attach_gen, 300); assert!(!log.is_controller(&live), "the fence moved with the slot"); @@ -8140,15 +8263,28 @@ mod tests { "older-gen {intent:?} must refuse busy, got {out:?}" ); } - let c = log.controller.as_ref().expect("incumbent survives the replays"); + let c = log + .controller + .as_ref() + .expect("incumbent survives the replays"); assert_eq!(c.attach_gen, 300); // NEWER gen + explicit Take → loud supersession, TookControl outcome. - let out = log.resolve_subscribe(Arc::clone(&taker), 0, AttachIntent::Take, Some("op".into()), 400, None); + let out = log.resolve_subscribe( + Arc::clone(&taker), + 0, + AttachIntent::Take, + Some("op".into()), + 400, + None, + ); assert!(matches!(out, SubscribeOutcome::TookControl), "got {out:?}"); let c = log.controller.as_ref().expect("taker holds the slot"); assert!(Arc::ptr_eq(&c.send, &taker)); - assert_eq!(c.attach_gen, 400, "the slot carries the taker's lease generation"); + assert_eq!( + c.attach_gen, 400, + "the slot carries the taker's lease generation" + ); // [unit->REQ-INPUT-CONTROLLER-FENCE] assert!(!log.is_controller(&ctrl2)); assert!(log.is_controller(&taker)); @@ -8169,16 +8305,35 @@ mod tests { // A real controller with a REAL writer thread (empty ring → empty // initial batch; the writer parks on its live queue). - let out = log.resolve_subscribe(Arc::clone(&a_send), 0, AttachIntent::Control, Some("op".into()), 200, None); + let out = log.resolve_subscribe( + Arc::clone(&a_send), + 0, + AttachIntent::Control, + Some("op".into()), + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); // Distinct-lease Take: the old sink drops (tx closes) and its writer // must write the terminal Displaced to A's conn on exit. - let out = log.resolve_subscribe(Arc::clone(&taker), 0, AttachIntent::Take, Some("op".into()), 300, None); + let out = log.resolve_subscribe( + Arc::clone(&taker), + 0, + AttachIntent::Take, + Some("op".into()), + 300, + None, + ); assert!(matches!(out, SubscribeOutcome::TookControl), "got {out:?}"); let env = read_frame(&mut a_client).expect("A's conn carries the terminal frame"); - assert_eq!(env.kind, crate::msg::KIND_DISPLACED, "terminal Displaced, got {}", env.kind); + assert_eq!( + env.kind, + crate::msg::KIND_DISPLACED, + "terminal Displaced, got {}", + env.kind + ); let ev: crate::msg::DisplacedEvent = serde_json::from_value(env.payload).expect("displaced payload"); assert_eq!(ev.session_id, 7); @@ -8233,16 +8388,15 @@ mod tests { // A VIEWER with a generous channel we CAN drain — to prove it stays fed. let (vsend, _vclient, _vrecv) = controller_socket_pair(); let (vtx, vrx) = sync_channel::(VIEWER_CHANNEL_DEPTH); - log.viewers - .insert( - 0, - ViewerSink { - tx: vtx, - send: vsend, - evicted: Arc::new(AtomicBool::new(false)), - _writer: thread::spawn(|| {}), - }, - ); + log.viewers.insert( + 0, + ViewerSink { + tx: vtx, + send: vsend, + evicted: Arc::new(AtomicBool::new(false)), + _writer: thread::spawn(|| {}), + }, + ); // Append 50 chunks. The controller channel (depth 2) fills after 2; every // further append DROPS (returns None — within the deadline, never evict) and @@ -8287,9 +8441,17 @@ mod tests { fn contiguous_advance_freezes_on_a_gap() { let dt = AtomicU64::new(0); contiguous_advance(&dt, 0); - assert_eq!(dt.load(Ordering::Acquire), 1, "seq 0 (== cursor) advances to 1"); + assert_eq!( + dt.load(Ordering::Acquire), + 1, + "seq 0 (== cursor) advances to 1" + ); contiguous_advance(&dt, 1); - assert_eq!(dt.load(Ordering::Acquire), 2, "contiguous seq 1 advances to 2"); + assert_eq!( + dt.load(Ordering::Acquire), + 2, + "contiguous seq 1 advances to 2" + ); // GAP: cursor is 2 but seq 5 arrives (3,4 dropped while Full). FREEZE at 2 — // a high-watermark jump to 6 would skip 3,4 on resume = a B2 violation. contiguous_advance(&dt, 5); @@ -8300,10 +8462,18 @@ mod tests { ); // Re-delivering the frozen seq (2, via ring replay) resumes contiguous advance. contiguous_advance(&dt, 2); - assert_eq!(dt.load(Ordering::Acquire), 3, "re-delivering the frozen seq resumes"); + assert_eq!( + dt.load(Ordering::Acquire), + 3, + "re-delivering the frozen seq resumes" + ); // A rewind re-send (seq < cursor) is a no-op. contiguous_advance(&dt, 0); - assert_eq!(dt.load(Ordering::Acquire), 3, "a rewind re-send cannot lower the cursor"); + assert_eq!( + dt.load(Ordering::Acquire), + 3, + "a rewind re-send cannot lower the cursor" + ); } /// W1 — `advance_delivered` moves the shared cursor monotonically (D4-1) via @@ -8381,15 +8551,18 @@ mod tests { // interleaved with single keystrokes. let inputs: Vec> = vec![ b"first".to_vec(), - b"\x03".to_vec(), // Ctrl-C + b"\x03".to_vec(), // Ctrl-C b"PASTE-BLOCK-AAAA".to_vec(), b"z".to_vec(), - b"\r".to_vec(), // Enter + b"\r".to_vec(), // Enter b"PASTE-BLOCK-BBBB".to_vec(), b"last".to_vec(), ]; for rec in &inputs { - assert!(w.enqueue(rec.clone()), "depth 256: every enqueue is accepted"); + assert!( + w.enqueue(rec.clone()), + "depth 256: every enqueue is accepted" + ); } // Close the FIFO so the drain loop terminates, then drain through the SOLE // writer exactly as `input_writer` does. @@ -8426,7 +8599,10 @@ mod tests { // The first DEPTH enqueues fit (accepted, no backpressure yet). for i in 0..DEPTH { - assert!(w.enqueue(vec![i as u8]), "enqueue {i} fits within the bound"); + assert!( + w.enqueue(vec![i as u8]), + "enqueue {i} fits within the bound" + ); } assert!( !w.backpressure.load(Ordering::Acquire), @@ -8446,7 +8622,10 @@ mod tests { ); // A second overflow while still saturated stays dropped + backpressured (the // stamp is rising-edge-only, but the STATE remains true — no flap to false). - assert!(!w.enqueue(b"OVERFLOW-2".to_vec()), "still dropping while full"); + assert!( + !w.enqueue(b"OVERFLOW-2".to_vec()), + "still dropping while full" + ); assert!( w.backpressure.load(Ordering::Acquire), "backpressure stays asserted while the queue remains saturated" @@ -8484,7 +8663,10 @@ mod tests { panic!("poison the inject floor"); }) .join(); - assert!(floor.is_poisoned(), "precondition: the floor mutex is poisoned"); + assert!( + floor.is_poisoned(), + "precondition: the floor mutex is poisoned" + ); // The fix: the recovered guard is fully usable — open() takes, is_held reads. lock_floor(&floor).open(); assert!( @@ -8511,12 +8693,23 @@ mod tests { panic!("poison the sessions map mid-attach"); }) .join(); - assert!(sessions.is_poisoned(), "precondition: the sessions mutex is poisoned"); + assert!( + sessions.is_poisoned(), + "precondition: the sessions mutex is poisoned" + ); // The fix: the recovered guard is fully usable — the prior row survives and a // NEW attach can still insert/look up (no permanent wedge). - assert_eq!(recover(&sessions).get(&7).copied(), Some(70), "prior state survives recovery"); + assert_eq!( + recover(&sessions).get(&7).copied(), + Some(70), + "prior state survives recovery" + ); recover(&sessions).insert(9, 90); - assert_eq!(recover(&sessions).get(&9).copied(), Some(90), "the next attach still opens"); + assert_eq!( + recover(&sessions).get(&9).copied(), + Some(90), + "the next attach still opens" + ); } /// The physical screen a FRESH client terminal shows after applying the log's @@ -8563,16 +8756,29 @@ mod tests { log.commit_resize(4, 10); let screen = repaint_screen(&log); - assert_eq!(log.grid.geometry(), (4, 10), "the grid lands on the new geometry"); + assert_eq!( + log.grid.geometry(), + (4, 10), + "the grid lands on the new geometry" + ); assert_eq!(log.size, (4, 10), "the stored letterbox size follows"); - assert_eq!(log.geometry_epoch, 1, "a committed resize opens a new epoch"); + assert_eq!( + log.geometry_epoch, 1, + "a committed resize opens a new epoch" + ); assert_eq!( screen[0], "ABCDEFGHIJ", "the old-geometry row is TRUNCATED by the resize, never re-wrapped" ); assert_eq!(screen[1], "", "…so nothing wrapped onto row 2"); - assert_eq!(screen[2], "0123456789", "the new-geometry write wraps at 10"); - assert_eq!(screen[3], "ABCDE", "…and its tail is on row 4, not truncated away"); + assert_eq!( + screen[2], "0123456789", + "the new-geometry write wraps at 10" + ); + assert_eq!( + screen[3], "ABCDE", + "…and its tail is on row 4, not truncated away" + ); } // [unit->REQ-RC-RESIZE-GEOMETRY-EPOCH] The GROW direction, where the rejected @@ -8616,11 +8822,25 @@ mod tests { log.abort_resize(); // the surface refused the resize let screen = repaint_screen(&log); - assert_eq!(log.grid.geometry(), (3, 20), "the grid stays at the old geometry"); - assert_eq!(log.size, (3, 20), "the stored size is not advanced by a refusal"); + assert_eq!( + log.grid.geometry(), + (3, 20), + "the grid stays at the old geometry" + ); + assert_eq!( + log.size, + (3, 20), + "the stored size is not advanced by a refusal" + ); assert_eq!(log.geometry_epoch, 0, "a refused resize opens NO epoch"); - assert_eq!(screen[0], "ABCDEFGHIJKLMNO", "held output replayed at the old width"); - assert_eq!(screen[1], "XY", "post-issue bytes belong to the old geometry too"); + assert_eq!( + screen[0], "ABCDEFGHIJKLMNO", + "held output replayed at the old width" + ); + assert_eq!( + screen[1], "XY", + "post-issue bytes belong to the old geometry too" + ); } // [unit->REQ-RC-RESIZE-GEOMETRY-EPOCH] The barrier is single-occupancy: a @@ -8638,7 +8858,10 @@ mod tests { ); log.mark_resize_issued(); log.commit_resize(3, 10); - assert!(log.begin_resize(3, 14).is_ok(), "the barrier reopens after the commit"); + assert!( + log.begin_resize(3, 14).is_ok(), + "the barrier reopens after the commit" + ); } /// Poll the shared `delivered_through` cursor until it reaches `want` (the @@ -8733,7 +8956,10 @@ mod tests { baselines the watermark jump instead of reject-gapping it" ); let bytes = decode_bytes(&ev.data_b64).unwrap(); - assert!(bytes.starts_with(b"\x1b[?1049"), "synthesized repaint, not raw"); + assert!( + bytes.starts_with(b"\x1b[?1049"), + "synthesized repaint, not raw" + ); // LEG 3: one successful sync write advances the cursor-of-record past the // WHOLE suppressed range as-if-written. @@ -8828,11 +9054,17 @@ mod tests { let mut log = floor_rig(); let (send2, mut client2, _recv2) = controller_socket_pair(); let resume_from = log.delivered_through.load(Ordering::Acquire); - assert_eq!(resume_from, 1, "precondition: the detached cursor is below the floor"); + assert_eq!( + resume_from, 1, + "precondition: the detached cursor is below the floor" + ); log.become_controller(Arc::clone(&send2), None, resume_from, 0); let f = read_frame(&mut client2).expect("the resume initial frame"); let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); - assert_eq!(ev.seq, 2, "the repaint rides the watermark pseudo-seq (next_seq - 1)"); + assert_eq!( + ev.seq, 2, + "the repaint rides the watermark pseudo-seq (next_seq - 1)" + ); assert!(ev.sync, "the below-floor repaint batch rides the wire flag"); let bytes = decode_bytes(&ev.data_b64).unwrap(); assert!( @@ -8846,7 +9078,10 @@ mod tests { assert!(log.append(b"live-after-resume").is_none()); // seq 3 let f = read_frame(&mut client2).expect("the live frame after the repaint"); let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); - assert_eq!(ev.seq, 3, "live frames stream raw + in-order after the repaint"); + assert_eq!( + ev.seq, 3, + "live frames stream raw + in-order after the repaint" + ); log.clear_controller(); drop(log); } @@ -8864,7 +9099,10 @@ mod tests { let f = read_frame(&mut client2).expect("the at-floor resume frame"); let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); assert_eq!(ev.seq, 3); - assert!(!ev.sync, "a raw ring slice is UNFLAGGED — strict B2 semantics untouched"); + assert!( + !ev.sync, + "a raw ring slice is UNFLAGGED — strict B2 semantics untouched" + ); assert_eq!( decode_bytes(&ev.data_b64).unwrap(), b"after-commit-raw", @@ -8888,9 +9126,14 @@ mod tests { assert_eq!(f.kind, crate::msg::KIND_SIZE); let f = read_frame(&mut view_client).expect("viewer initial frame"); let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); - assert_eq!(ev.seq, 2, "below-floor viewer gets the repaint at the pseudo-seq"); + assert_eq!( + ev.seq, 2, + "below-floor viewer gets the repaint at the pseudo-seq" + ); assert!( - decode_bytes(&ev.data_b64).unwrap().starts_with(b"\x1b[?1049"), + decode_bytes(&ev.data_b64) + .unwrap() + .starts_with(b"\x1b[?1049"), "the synthesized repaint, not raw suppressed ring bytes" ); @@ -8936,14 +9179,29 @@ mod tests { let ev: crate::msg::OutputEvent = serde_json::from_value(f.payload).unwrap(); let bytes = decode_bytes(&ev.data_b64).unwrap(); let s = String::from_utf8_lossy(&bytes); - assert!(s.contains("\x1b]2;mid-window-title\x07"), "title change surfaces: {s:?}"); - assert!(s.contains("\x1b[?25l"), "cursor-visibility toggle surfaces: {s:?}"); + assert!( + s.contains("\x1b]2;mid-window-title\x07"), + "title change surfaces: {s:?}" + ); + assert!( + s.contains("\x1b[?25l"), + "cursor-visibility toggle surfaces: {s:?}" + ); // A CROSS-geometry commit resets the scroll region (grid resize // semantics, matching the client terminal's own reset on the letterbox // move) — the region CONTRACT still surfaces, as the explicit reset. - assert!(s.contains("\x1b[r"), "the region contract surfaces explicitly: {s:?}"); - assert!(s.contains("\x1b[?1049h"), "alt-screen switch surfaces: {s:?}"); - assert!(s.contains("ALT"), "window content surfaces at the target geometry: {s:?}"); + assert!( + s.contains("\x1b[r"), + "the region contract surfaces explicitly: {s:?}" + ); + assert!( + s.contains("\x1b[?1049h"), + "alt-screen switch surfaces: {s:?}" + ); + assert!( + s.contains("ALT"), + "window content surfaces at the target geometry: {s:?}" + ); drop(log); // A SAME-geometry transition (a SIGWINCH re-assert: begin/commit at the @@ -8955,7 +9213,8 @@ mod tests { log.add_viewer(Arc::clone(&view_send), 0); let _ = read_frame(&mut view_client).expect("initial size"); let _ = read_frame(&mut view_client).expect("initial repaint"); - log.begin_resize(4, 20).expect("same-geometry barrier closes"); + log.begin_resize(4, 20) + .expect("same-geometry barrier closes"); assert!(log.append(b"\x1b[1;3r").is_none()); log.mark_resize_issued(); log.commit_resize(4, 20); @@ -9005,8 +9264,14 @@ mod tests { "the deferred toggles flush INSIDE the sync frame, after the repaint, \ in emission order: {s:?}" ); - assert!(!s.contains("\x1b]8;"), "the hyperlink pair drops balanced: {s:?}"); - assert!(s.contains("LINK"), "the link TEXT still renders via the repaint: {s:?}"); + assert!( + !s.contains("\x1b]8;"), + "the hyperlink pair drops balanced: {s:?}" + ); + assert!( + s.contains("LINK"), + "the link TEXT still renders via the repaint: {s:?}" + ); drop(log); } @@ -9088,7 +9353,11 @@ mod tests { log.mark_resize_issued(); log.abort_resize(); // the surface refused the resize - assert_eq!(log.grid.geometry(), (3, 20), "abort: grid stays at the old geometry"); + assert_eq!( + log.grid.geometry(), + (3, 20), + "abort: grid stays at the old geometry" + ); assert_eq!(log.size, (3, 20), "abort: stored size untouched"); // Viewer: the FIRST post-abort frame is the sync frame — no `size` frame @@ -9153,7 +9422,10 @@ mod tests { OutputLog::new(1, DEFAULT_LOG_CHUNKS, "reaped".to_string(), (24, 80)).stamp_reaped(); let after = spt_store::info::read_info(&perch).unwrap(); - assert!(!after.controlled, "control stamps never outlive their session"); + assert!( + !after.controlled, + "control stamps never outlive their session" + ); assert_eq!(after.driven_by, None); assert_eq!( after.controllable, @@ -9184,9 +9456,18 @@ mod tests { torn.append(b"x"); // next_seq now 1, ring=[(0,x)] torn.ring.push_back((999, b"garbage".to_vec())); // back(999) >= next_seq(1) → torn let next_before = torn.next_seq; - assert!(torn.clamp_or_reset(), "a torn ring (last seq >= next_seq) is reset"); - assert!(torn.ring.is_empty(), "the torn ring is emptied — no garbage served"); - assert_eq!(torn.next_seq, next_before, "next_seq is preserved (cursors never rewind)"); + assert!( + torn.clamp_or_reset(), + "a torn ring (last seq >= next_seq) is reset" + ); + assert!( + torn.ring.is_empty(), + "the torn ring is emptied — no garbage served" + ); + assert_eq!( + torn.next_seq, next_before, + "next_seq is preserved (cursors never rewind)" + ); // Torn: over-cap ring (an interrupted prune). let mut over = OutputLog::new(3, 2, String::new(), (24, 80)); @@ -9216,8 +9497,14 @@ mod tests { .join(); assert!(log.is_poisoned(), "precondition: the log mutex is poisoned"); let g = recover_log(&log); - assert!(g.ring.is_empty(), "recover_log clamps the torn ring to empty on recovery"); - assert_eq!(g.next_seq, 1, "next_seq preserved through the poison-recover clamp"); + assert!( + g.ring.is_empty(), + "recover_log clamps the torn ring to empty on recovery" + ); + assert_eq!( + g.next_seq, 1, + "next_seq preserved through the poison-recover clamp" + ); } // [unit->REQ-TRANSLATE-COMMIT-MISS-TOLERANCE] C-1 respool-once / dead-letter: a @@ -9234,10 +9521,16 @@ mod tests { assert_eq!(note_miss_respool(&mut set, "env-A"), MissRespool::Respool); assert!(set.contains("env-A")); // Second miss of the SAME envelope → dead-letter, NOT respooled again. - assert_eq!(note_miss_respool(&mut set, "env-A"), MissRespool::DeadLetter); + assert_eq!( + note_miss_respool(&mut set, "env-A"), + MissRespool::DeadLetter + ); // A DIFFERENT envelope respools once on its own. assert_eq!(note_miss_respool(&mut set, "env-B"), MissRespool::Respool); - assert_eq!(note_miss_respool(&mut set, "env-B"), MissRespool::DeadLetter); + assert_eq!( + note_miss_respool(&mut set, "env-B"), + MissRespool::DeadLetter + ); // A committed envelope is forgotten → a later miss respools it afresh. set.remove("env-A"); assert_eq!(note_miss_respool(&mut set, "env-A"), MissRespool::Respool); @@ -9268,7 +9561,10 @@ mod tests { let payload = b"XLATE_OK line one\nline two\nline three"; // Full echo present → verified. let full_echo = b"prompt> XLATE_OK line one\nline two\nline three\n"; - assert!(echo_verified(payload, full_echo), "a fully-echoed head verifies"); + assert!( + echo_verified(payload, full_echo), + "a fully-echoed head verifies" + ); // Head swallowed: only a suffix echoed (the field bug — mid-word start). The // leading prefix is ABSENT → verify MISS. let tail_only = b"three\r\n"; // the ~322B suffix class, head gone @@ -9277,11 +9573,23 @@ mod tests { "a swallowed head (prefix absent from echo) fails verify — the head-loss tell" ); // A keys-only sequence types no echoable text → vacuously verified. - assert!(echo_verified(b"", b""), "empty payload is vacuously verified"); - assert!(echo_verified(b"", b"noise"), "empty payload verifies regardless of echo"); + assert!( + echo_verified(b"", b""), + "empty payload is vacuously verified" + ); + assert!( + echo_verified(b"", b"noise"), + "empty payload verifies regardless of echo" + ); // A short payload (below the prefix window) matches on its whole self. - assert!(echo_verified(b"hi", b"...hi..."), "a short payload matches whole"); - assert!(!echo_verified(b"hi", b"...ho..."), "a short payload absent → miss"); + assert!( + echo_verified(b"hi", b"...hi..."), + "a short payload matches whole" + ); + assert!( + !echo_verified(b"hi", b"...ho..."), + "a short payload absent → miss" + ); } // [unit->REQ-INJECT-MULTILINE-INTEGRITY] Layer 1 re-arm: the settle-gate must re-run @@ -9296,14 +9604,26 @@ mod tests { fn should_settle_rearms_on_observable_pty() { // Echoing/interactive PTY (probe observed → not unobservable): re-settle EVERY // delivery's first byte — this is the class the `/clear` head-swallow bites. - assert!(should_settle(1, false), "observable PTY re-settles before each delivery"); + assert!( + should_settle(1, false), + "observable PTY re-settles before each delivery" + ); // Non-echoing ConPTY (probe unobservable, latched): skip the steady-state settle — // no reader-reattach race, and settling would burn the full deadline every time. - assert!(!should_settle(1, true), "unobservable-probe PTY skips the steady-state settle"); + assert!( + !should_settle(1, true), + "unobservable-probe PTY skips the steady-state settle" + ); // A RE-DRIVE always settles regardless of the latch (only reached on a swallowed // head → readiness must be re-confirmed before retyping). - assert!(should_settle(2, true), "a re-drive settles even when the probe is unobservable"); - assert!(should_settle(2, false), "a re-drive settles on an observable PTY too"); + assert!( + should_settle(2, true), + "a re-drive settles even when the probe is unobservable" + ); + assert!( + should_settle(2, false), + "a re-drive settles on an observable PTY too" + ); } // [unit->REQ-INJECT-MULTILINE-INTEGRITY] the subslice search the head match rides: @@ -9314,9 +9634,18 @@ mod tests { assert!(contains_subslice(b"abcdef", b"cde")); assert!(contains_subslice(b"abcdef", b"abc")); assert!(contains_subslice(b"abcdef", b"def")); - assert!(!contains_subslice(b"abcdef", b"ce"), "non-contiguous is not a subslice"); - assert!(contains_subslice(b"abc", b""), "empty needle is vacuously present"); - assert!(!contains_subslice(b"ab", b"abc"), "needle longer than haystack is absent"); + assert!( + !contains_subslice(b"abcdef", b"ce"), + "non-contiguous is not a subslice" + ); + assert!( + contains_subslice(b"abc", b""), + "empty needle is vacuously present" + ); + assert!( + !contains_subslice(b"ab", b"abc"), + "needle longer than haystack is absent" + ); } // [unit->REQ-INJECT-MULTILINE-INTEGRITY] the output-log tap the settle-gate + @@ -9354,17 +9683,31 @@ mod tests { // Small: whole, single emit. let mut parts: Vec> = Vec::new(); chunk_text(b"small", 256, |p| parts.push(p.to_vec())); - assert_eq!(parts, vec![b"small".to_vec()], "a small payload is one whole write"); + assert_eq!( + parts, + vec![b"small".to_vec()], + "a small payload is one whole write" + ); // Large: split into ceil(len/chunk) ordered parts, reassembling exactly. let payload: Vec = (0..1000u16).map(|i| (i % 251) as u8).collect(); let chunk = 256; let mut got: Vec> = Vec::new(); chunk_text(&payload, chunk, |p| got.push(p.to_vec())); - assert_eq!(got.len(), 1000_usize.div_ceil(chunk), "ceil(len/chunk) parts"); - assert!(got.iter().take(got.len() - 1).all(|p| p.len() == chunk), "all but last are full"); + assert_eq!( + got.len(), + 1000_usize.div_ceil(chunk), + "ceil(len/chunk) parts" + ); + assert!( + got.iter().take(got.len() - 1).all(|p| p.len() == chunk), + "all but last are full" + ); let reassembled: Vec = got.concat(); - assert_eq!(reassembled, payload, "in-order reassembly is byte-identical — no head/tail loss"); + assert_eq!( + reassembled, payload, + "in-order reassembly is byte-identical — no head/tail loss" + ); } // [unit->REQ-HAZARD-INJECT-WORKER-POISON] B6 leg (ii): a PANIC inside the inject @@ -9663,8 +10006,14 @@ mod tests { let mut log = OutputLog::new(1, DEFAULT_LOG_CHUNKS, String::new(), (24, 80)); // Establish: conn A holds the lease at generation 500. - let (out, decision) = - log.resolve_subscribe_inner(Arc::clone(&a), 0, AttachIntent::Control, Some("op".into()), 500, None); + let (out, decision) = log.resolve_subscribe_inner( + Arc::clone(&a), + 0, + AttachIntent::Control, + Some("op".into()), + 500, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); assert_eq!(decision, "controller"); let epoch_after_establish = log.controller_epoch.load(Ordering::Acquire); @@ -9673,12 +10022,30 @@ mod tests { // replayed. The seat is untouched: same sink, same generation, and the // controller EPOCH does not move (a bump is what stops the live writer // mid-batch, so an unchanged epoch IS "the writer was never disturbed"). - let (out, decision) = - log.resolve_subscribe_inner(Arc::clone(&a), 0, AttachIntent::Control, Some("op".into()), 500, None); - assert!(matches!(out, SubscribeOutcome::Controller), "the wire answer stays Controller (N-1 tolerant), got {out:?}"); - assert_eq!(decision, "idempotent", "the breadcrumb distinguishes reuse from replacement"); - let seat = log.controller.as_ref().expect("the seat survives its own replay"); - assert!(Arc::ptr_eq(&seat.send, &a), "the SAME sink is preserved — not a fresh one over a dropped writer"); + let (out, decision) = log.resolve_subscribe_inner( + Arc::clone(&a), + 0, + AttachIntent::Control, + Some("op".into()), + 500, + None, + ); + assert!( + matches!(out, SubscribeOutcome::Controller), + "the wire answer stays Controller (N-1 tolerant), got {out:?}" + ); + assert_eq!( + decision, "idempotent", + "the breadcrumb distinguishes reuse from replacement" + ); + let seat = log + .controller + .as_ref() + .expect("the seat survives its own replay"); + assert!( + Arc::ptr_eq(&seat.send, &a), + "the SAME sink is preserved — not a fresh one over a dropped writer" + ); assert_eq!(seat.attach_gen, 500); assert_eq!( log.controller_epoch.load(Ordering::Acquire), @@ -9688,25 +10055,49 @@ mod tests { // CELL 2 — a DIFFERENT conn at the same generation is the fix-6 successor: // today's silent swap must NOT regress into idempotence. - let (out, decision) = - log.resolve_subscribe_inner(Arc::clone(&b), 0, AttachIntent::Control, Some("op".into()), 500, None); + let (out, decision) = log.resolve_subscribe_inner( + Arc::clone(&b), + 0, + AttachIntent::Control, + Some("op".into()), + 500, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); - assert_eq!(decision, "controller", "a different carrier is a re-serve, not a replay"); + assert_eq!( + decision, "controller", + "a different carrier is a re-serve, not a replay" + ); assert!( Arc::ptr_eq(&log.controller.as_ref().unwrap().send, &b), "the successor conn takes the seat (ADR-0038 fix 6)" ); // CELL 3 — strictly newer generation still supersedes loudly... - let (out, decision) = - log.resolve_subscribe_inner(Arc::clone(&a), 0, AttachIntent::Take, Some("op".into()), 900, None); + let (out, decision) = log.resolve_subscribe_inner( + Arc::clone(&a), + 0, + AttachIntent::Take, + Some("op".into()), + 900, + None, + ); assert!(matches!(out, SubscribeOutcome::TookControl), "got {out:?}"); assert_eq!(decision, "took"); // CELL 4 — ...and strictly older is still refused busy. - let (out, decision) = - log.resolve_subscribe_inner(Arc::clone(&b), 0, AttachIntent::Control, Some("op".into()), 500, None); - assert!(matches!(out, SubscribeOutcome::BusyControlled { .. }), "got {out:?}"); + let (out, decision) = log.resolve_subscribe_inner( + Arc::clone(&b), + 0, + AttachIntent::Control, + Some("op".into()), + 500, + None, + ); + assert!( + matches!(out, SubscribeOutcome::BusyControlled { .. }), + "got {out:?}" + ); assert_eq!(decision, "busy"); log.clear_controller(); @@ -9733,7 +10124,9 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 700, None); + 700, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); assert_eq!(decision, "controller"); @@ -9744,7 +10137,9 @@ mod tests { 12, AttachIntent::Control, Some("op".into()), - 700, None); + 700, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); assert_eq!( decision, "controller", @@ -9755,7 +10150,11 @@ mod tests { 12, "and the seat records the floor it was re-established from" ); - assert_eq!(log.controller.as_ref().unwrap().attach_gen, 700, "the generation is preserved (fix 6)"); + assert_eq!( + log.controller.as_ref().unwrap().attach_gen, + 700, + "the generation is preserved (fix 6)" + ); log.clear_controller(); } @@ -9781,7 +10180,9 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 200, None); + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); // A replayed stale Request (same identity, OLDER generation) is @@ -9791,13 +10192,18 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 100, None); + 100, + None, + ); let SubscribeOutcome::BusyControlled { by } = out else { panic!("a stale same-identity generation must refuse busy, got {out:?}"); }; assert_eq!(by, "op"); let c = log.controller.as_ref().expect("incumbent survives"); - assert!(Arc::ptr_eq(&c.send, &live), "the newer controller keeps the slot"); + assert!( + Arc::ptr_eq(&c.send, &live), + "the newer controller keeps the slot" + ); assert_eq!(c.attach_gen, 200); // EQUAL generation = the same Request reconstructed (dispatcher @@ -9807,7 +10213,9 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 200, None); + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); assert!(Arc::ptr_eq(&log.controller.as_ref().unwrap().send, &stale)); @@ -9818,8 +10226,13 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 0, None); - assert!(matches!(out, SubscribeOutcome::Controller), "legacy gen 0 re-takes, got {out:?}"); + 0, + None, + ); + assert!( + matches!(out, SubscribeOutcome::Controller), + "legacy gen 0 re-takes, got {out:?}" + ); log.clear_controller(); } @@ -9849,7 +10262,9 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 200, None); + 200, + None, + ); assert!(matches!(out, SubscribeOutcome::Controller), "got {out:?}"); // Mid-serve gap resume: the SAME worker re-subscribes from its floor @@ -9859,8 +10274,13 @@ mod tests { 3, AttachIntent::Control, Some("op".into()), - 200, None); - assert!(matches!(out, SubscribeOutcome::Controller), "resume re-takes, got {out:?}"); + 200, + None, + ); + assert!( + matches!(out, SubscribeOutcome::Controller), + "resume re-takes, got {out:?}" + ); assert_eq!( log.controller.as_ref().unwrap().attach_gen, 200, @@ -9873,7 +10293,9 @@ mod tests { 0, AttachIntent::Control, Some("op".into()), - 100, None); + 100, + None, + ); assert!( matches!(out, SubscribeOutcome::BusyControlled { .. }), "a stale replay must stay refused across the resume, got {out:?}" @@ -9916,12 +10338,18 @@ mod tests { // The matching generation releases normally. log.detach_if_gen(&send, Some(200)); - assert!(log.controller.is_none(), "the owning generation's release clears"); + assert!( + log.controller.is_none(), + "the owning generation's release clears" + ); // None (N-1 / conn cleanup) keeps ptr-identity behavior. log.become_controller(Arc::clone(&send), Some("op".into()), 0, 300); log.detach_if_gen(&send, None); - assert!(log.controller.is_none(), "gen-less detach keeps the legacy ptr clear"); + assert!( + log.controller.is_none(), + "gen-less detach keeps the legacy ptr clear" + ); } /// #6 CORE (REQ-BROKER-SCREEN-GRID, ADR-0031), the integrated broker seam: @@ -9957,7 +10385,11 @@ mod tests { // The initial batch is ONE output frame carrying the synthesized repaint. let frame = read_frame(&mut client).expect("a repaint frame on attach"); - assert_eq!(frame.kind, crate::msg::KIND_OUTPUT, "the initial batch is an output frame"); + assert_eq!( + frame.kind, + crate::msg::KIND_OUTPUT, + "the initial batch is an output frame" + ); let ev: crate::msg::OutputEvent = serde_json::from_value(frame.payload).expect("output payload"); let repaint = decode_bytes(&ev.data_b64).expect("repaint b64 decodes"); @@ -10033,13 +10465,19 @@ mod tests { let mut auth = avt::Vt::new(cols as usize, rows as usize); auth.feed_str(std::str::from_utf8(frame1).unwrap()); auth.feed_str(std::str::from_utf8(frame2).unwrap()); - let auth: Vec = auth.view().map(|l| l.text().trim_end().to_string()).collect(); + let auth: Vec = auth + .view() + .map(|l| l.text().trim_end().to_string()) + .collect(); // Candidate: what the attaching client's terminal actually shows. let mut seen = avt::Vt::new(cols as usize, rows as usize); seen.feed_str(std::str::from_utf8(&repaint).unwrap()); seen.feed_str(std::str::from_utf8(&diff).unwrap()); - let seen: Vec = seen.view().map(|l| l.text().trim_end().to_string()).collect(); + let seen: Vec = seen + .view() + .map(|l| l.text().trim_end().to_string()) + .collect(); assert_eq!( seen, auth, @@ -10076,7 +10514,10 @@ mod tests { } thread::sleep(Duration::from_millis(1)); } - assert!(writer.is_finished(), "precondition: the writer thread has exited"); + assert!( + writer.is_finished(), + "precondition: the writer thread has exited" + ); log.controller = Some(ControllerSink { attach_gen: 0, establish_from_seq: 0, @@ -10093,14 +10534,20 @@ mod tests { Some("remote-node"), "precondition: a stale remote controller is present" ); - assert!(log.reap_dead_controller(), "a dead-writer controller is reaped"); + assert!( + log.reap_dead_controller(), + "a dead-writer controller is reaped" + ); assert!(!log.has_controller(), "the controller slot is cleared"); assert_eq!( log.controller_by(), None, "controller_by is now honest (None) → converge_perch_stamps clears the stamp" ); - assert!(!log.reap_dead_controller(), "idempotent: nothing left to reap"); + assert!( + !log.reap_dead_controller(), + "idempotent: nothing left to reap" + ); } // [unit->REQ-DRIVEN-BY-OWN-NODE-NORMALIZE] own-node latch is TRUTHFUL (doyle @@ -10134,7 +10581,10 @@ mod tests { Some(own_hex.as_str()), "an own-node controller latches driven_by to its own hex (CONTEXT:386)" ); - assert!(after.controlled, "controlled stays true (any-controller truth)"); + assert!( + after.controlled, + "controlled stays true (any-controller truth)" + ); log.clear_controller(); // A remote hex latches identically. @@ -10316,8 +10766,14 @@ mod tests { let before = STALL_EVICT_COUNT.load(Ordering::Relaxed); let (nsend, _nc, _nr) = controller_socket_pair(); - let outcome = - log.resolve_subscribe(nsend, 0, AttachIntent::Control, Some("newcomer".to_string()), 0, None); + let outcome = log.resolve_subscribe( + nsend, + 0, + AttachIntent::Control, + Some("newcomer".to_string()), + 0, + None, + ); assert_eq!( outcome, @@ -10346,7 +10802,11 @@ mod tests { "a blocked-past-deadline writer is reaped though it has not exited" ); assert!(!log.has_controller(), "the controller slot is cleared"); - assert_eq!(log.controller_by(), None, "driven_by truth is now honest (None)"); + assert_eq!( + log.controller_by(), + None, + "driven_by truth is now honest (None)" + ); } /// W3a endpoint selection (REQ-ADAPTER-LIVE-UPDATE, ADR-0025): from the @@ -10370,19 +10830,39 @@ mod tests { // sort+dedup, not just incidental ordering. let rows = vec![ // ep-b matches (sorts AFTER ep-a despite appearing first). - ("ep-b".to_string(), "claude-spt".to_string(), Some(dir_b.clone())), + ( + "ep-b".to_string(), + "claude-spt".to_string(), + Some(dir_b.clone()), + ), // ep-a, two sessions, SAME endpoint+dir → must dedup to one entry. - ("ep-a".to_string(), "claude-spt".to_string(), Some(dir_a.clone())), - ("ep-a".to_string(), "claude-spt".to_string(), Some(dir_a.clone())), + ( + "ep-a".to_string(), + "claude-spt".to_string(), + Some(dir_a.clone()), + ), + ( + "ep-a".to_string(), + "claude-spt".to_string(), + Some(dir_a.clone()), + ), // Non-matching adapter → excluded entirely. - ("ep-c".to_string(), "codex-spt".to_string(), Some(PathBuf::from("/install/ep-c"))), + ( + "ep-c".to_string(), + "codex-spt".to_string(), + Some(PathBuf::from("/install/ep-c")), + ), // Matching adapter but NO install_dir → excluded (nothing to swap). ("ep-d".to_string(), "claude-spt".to_string(), None), // PROFILE-COMPOSITE row (F015B): a `--adapter claude-spt:ccs` endpoint // stores the composite `claude-spt:ccs`, but the update carries the // PARENT record name `claude-spt` — it MUST match on the parent (an // exact-match skew is the silent-no-op bug). - ("ep-e".to_string(), "claude-spt:ccs".to_string(), Some(dir_e.clone())), + ( + "ep-e".to_string(), + "claude-spt:ccs".to_string(), + Some(dir_e.clone()), + ), ]; let got = select_endpoints_running_adapter(rows, "claude-spt"); @@ -10623,8 +11103,15 @@ mod tests { let _ = spt_store::empower::drop_all(sid); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let (out, note) = - log.resolve_subscribe_gated(sub, 0, AttachIntent::Control, None, 100, Some(member), AdmitTicket::None); + let (out, note) = log.resolve_subscribe_gated( + sub, + 0, + AttachIntent::Control, + None, + 100, + Some(member), + AdmitTicket::None, + ); assert!( matches!(out, SubscribeOutcome::Controller), "the member code still takes the controls, got {out:?}" @@ -10651,8 +11138,15 @@ mod tests { let _ = spt_store::empower::drop_all(sid); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let (out, note) = - log.resolve_subscribe_gated(sub, 0, AttachIntent::Control, None, 100, Some(admin.clone()), AdmitTicket::None); + let (out, note) = log.resolve_subscribe_gated( + sub, + 0, + AttachIntent::Control, + None, + 100, + Some(admin.clone()), + AdmitTicket::None, + ); assert!( matches!(out, SubscribeOutcome::Controller), "the admin code takes the controls, got {out:?}" @@ -10700,10 +11194,20 @@ mod tests { let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let (out, note) = - log.resolve_subscribe_gated(sub, 0, AttachIntent::Take, None, 100, Some(admin), AdmitTicket::None); + let (out, note) = log.resolve_subscribe_gated( + sub, + 0, + AttachIntent::Take, + None, + 100, + Some(admin), + AdmitTicket::None, + ); assert!( - matches!(out, SubscribeOutcome::Controller | SubscribeOutcome::TookControl), + matches!( + out, + SubscribeOutcome::Controller | SubscribeOutcome::TookControl + ), "the take is seated, got {out:?}" ); assert!( @@ -10881,8 +11385,15 @@ mod tests { let redeemed = broker.redeem_bringup_admit(Some(&live), 1); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let (out, note) = - log.resolve_subscribe_gated(sub, 0, AttachIntent::Control, None, 100, None, redeemed); + let (out, note) = log.resolve_subscribe_gated( + sub, + 0, + AttachIntent::Control, + None, + 100, + None, + redeemed, + ); assert!( matches!(out, SubscribeOutcome::Controller), "the seat resolves, got {out:?}" @@ -10974,7 +11485,14 @@ mod tests { clear_bringup_ledger(); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let out = log.resolve_subscribe(sub, 0, AttachIntent::Control, None, 100, Some(member.clone())); + let out = log.resolve_subscribe( + sub, + 0, + AttachIntent::Control, + None, + 100, + Some(member.clone()), + ); assert!( matches!(out, SubscribeOutcome::Controller), "a member code brings the engine room up, got {out:?}" @@ -10988,8 +11506,7 @@ mod tests { assert_eq!(briefed.len(), 1, "exactly one briefing, got {briefed:?}"); assert_eq!(briefed[0].from, spt_store::briefing::BRIEFING_AUTHOR); assert!( - briefed[0].body.contains("Posture right now") - && briefed[0].body.contains("SCOPE"), + briefed[0].body.contains("Posture right now") && briefed[0].body.contains("SCOPE"), "carrying the posture and the ruleset table: {}", briefed[0].body ); @@ -11011,8 +11528,13 @@ mod tests { clear_bringup_ledger(); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let wrong = if member == "000000" { "111111" } else { "000000" }; - let out = log.resolve_subscribe(sub, 0, AttachIntent::Control, None, 100, Some(wrong.into())); + let wrong = if member == "000000" { + "111111" + } else { + "000000" + }; + let out = + log.resolve_subscribe(sub, 0, AttachIntent::Control, None, 100, Some(wrong.into())); assert!( matches!(out, SubscribeOutcome::BusyControlled { .. }), "a wrong code takes nothing, got {out:?}" @@ -11025,7 +11547,14 @@ mod tests { ); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let out = log.resolve_subscribe(sub, 0, AttachIntent::Control, None, 100, Some(member.clone())); + let out = log.resolve_subscribe( + sub, + 0, + AttachIntent::Control, + None, + 100, + Some(member.clone()), + ); assert!( matches!(out, SubscribeOutcome::BusyControlled { .. }), "the right code inside the backoff is refused like any other, got {out:?}" @@ -11085,8 +11614,11 @@ mod tests { std::process::id(), SEQ.fetch_add(1, Ordering::Relaxed) ); - Broker::bind_in(&name, spt_store::perch::spt_home().join("admit-effects.log")) - .expect("bind admit test broker") + Broker::bind_in( + &name, + spt_store::perch::spt_home().join("admit-effects.log"), + ) + .expect("bind admit test broker") } /// Read the `subscribed` verdict off a connection, skipping the output @@ -11219,7 +11751,11 @@ mod tests { // longer redeem, so a broker that has admitted bring-ups for months // holds only what is still live. let expired_by_now = broker.mint_bringup_admit_at(13, mechanics_verdict(), t0); - let live = broker.mint_bringup_admit_at(13, mechanics_verdict(), t0 + BRINGUP_ADMIT_TTL + Duration::from_secs(1)); + let live = broker.mint_bringup_admit_at( + 13, + mechanics_verdict(), + t0 + BRINGUP_ADMIT_TTL + Duration::from_secs(1), + ); let held = recover(&broker.bringup_admits); assert!( held.contains_key(&live) && !held.contains_key(&expired_by_now), @@ -11319,7 +11855,10 @@ mod tests { // The four distinct ways a ticket-shaped string fails to redeem. let spent = broker.mint_bringup_admit(1, mechanics_verdict()); - assert_eq!(broker.redeem_bringup_admit(Some(&spent), 1), AdmitTicket::Redeemed(mechanics_verdict())); + assert_eq!( + broker.redeem_bringup_admit(Some(&spent), 1), + AdmitTicket::Redeemed(mechanics_verdict()) + ); let wrong_session = broker.mint_bringup_admit(1, mechanics_verdict()); let expired = broker.mint_bringup_admit_at(1, mechanics_verdict(), t0); let causes: [(&str, AdmitTicket); 4] = [ @@ -11327,7 +11866,10 @@ mod tests { "never minted", broker.redeem_bringup_admit(Some("admit:0123456789abcdef"), 1), ), - ("already spent", broker.redeem_bringup_admit(Some(&spent), 1)), + ( + "already spent", + broker.redeem_bringup_admit(Some(&spent), 1), + ), ( "wrong session", broker.redeem_bringup_admit(Some(&wrong_session), 42), @@ -11368,7 +11910,10 @@ mod tests { ); } assert_eq!( - sentences.iter().collect::>().len(), + sentences + .iter() + .collect::>() + .len(), 1, "one sentence for all four causes — the refusal must not say which, \ or it reports whether a bring-up is in flight: {sentences:?}" @@ -11577,8 +12122,11 @@ mod tests { crate::test_home::with_home(|_| { let (member, admin) = provision_engine_room(); let name = format!("spt-daemon-er-grant-{}.sock", std::process::id()); - let broker = Broker::bind_in(&name, spt_store::perch::spt_home().join("grant-effects.log")) - .expect("bind serving broker"); + let broker = Broker::bind_in( + &name, + spt_store::perch::spt_home().join("grant-effects.log"), + ) + .expect("bind serving broker"); let serving = Arc::clone(&broker); std::thread::spawn(move || { let _ = serving.serve(); @@ -11598,7 +12146,9 @@ mod tests { // (a) MEMBER code: comes up, grants nothing, claims nothing. clear_bringup_ledger(); let (asked_on, mut client, _r) = controller_socket_pair(); - broker.dispatch_bring_up(bring_up(&member), &asked_on).expect("the verb answers"); + broker + .dispatch_bring_up(bring_up(&member), &asked_on) + .expect("the verb answers"); let reply = brought_up_reply(&mut client); assert_eq!(reply.outcome, BRING_UP_ADMITTED, "{}", reply.detail); assert!( @@ -11625,7 +12175,9 @@ mod tests { // holding the controls would be a sentence that can turn out false. clear_bringup_ledger(); let (asked_on, mut client, _r) = controller_socket_pair(); - broker.dispatch_bring_up(bring_up(&admin), &asked_on).expect("the verb answers"); + broker + .dispatch_bring_up(bring_up(&admin), &asked_on) + .expect("the verb answers"); let reply = brought_up_reply(&mut client); assert_eq!(reply.outcome, BRING_UP_ADMITTED, "{}", reply.detail); assert!( @@ -11718,8 +12270,9 @@ mod tests { clear_bringup_ledger(); let ledger_file = spt_store::perch::engine_room_gate_file(); let name = format!("spt-daemon-er-e2e-{}.sock", std::process::id()); - let broker = Broker::bind_in(&name, spt_store::perch::spt_home().join("e2e-effects.log")) - .expect("bind serving broker"); + let broker = + Broker::bind_in(&name, spt_store::perch::spt_home().join("e2e-effects.log")) + .expect("bind serving broker"); let serving = Arc::clone(&broker); // The spawn dials this broker as a CLIENT (harnesshost → // Brain::cold_start), so the accept loop has to be up for the @@ -11846,7 +12399,10 @@ mod tests { .expect("the verb answers"); let again = brought_up_reply(&mut asked_client); assert_eq!(again.outcome, BRING_UP_ALREADY_LIVE, "{}", again.detail); - assert_eq!(again.session_id, sid, "naming the session already hosting it"); + assert_eq!( + again.session_id, sid, + "naming the session already hosting it" + ); assert!( again.admit.is_none(), "no ticket: this caller brought nothing up, and the seat gate still \ @@ -11887,7 +12443,8 @@ mod tests { for by in [None, Some("PEERHEX".to_string())] { let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let out = log.resolve_subscribe(sub, 0, AttachIntent::Viewer, by.clone(), 100, None); + let out = + log.resolve_subscribe(sub, 0, AttachIntent::Viewer, by.clone(), 100, None); assert!( matches!(out, SubscribeOutcome::BusyControlled { .. }), "the engine room is never watched (by={by:?}), got {out:?}" @@ -11916,8 +12473,7 @@ mod tests { clear_bringup_ledger(); let mut log = engine_room_log(); let (sub, _c, _r) = controller_socket_pair(); - let out = - log.resolve_subscribe(sub, 0, intent, None, 100, Some(member.clone())); + let out = log.resolve_subscribe(sub, 0, intent, None, 100, Some(member.clone())); assert!( matches!(out, SubscribeOutcome::Controller), "local {intent:?} takes the controls, got {out:?}" @@ -11947,13 +12503,8 @@ mod tests { let id = spt_store::engineroom::ENGINE_ROOM_ID; let perch = resolve_perch_path(id, ParentHint::Infer); std::fs::create_dir_all(&perch).unwrap(); - let rec = spt_store::info::InfoJson::new( - id, - "t", - std::process::id(), - "sid-1", - "live_agent", - ); + let rec = + spt_store::info::InfoJson::new(id, "t", std::process::id(), "sid-1", "live_agent"); spt_store::info::write_info(&perch, &rec).unwrap(); let empower = |subnet: &str| { let mut held = spt_store::empower::Empowerments::default(); diff --git a/crates/spt-daemon/src/effect.rs b/crates/spt-daemon/src/effect.rs index 6b59493..6ffb766 100644 --- a/crates/spt-daemon/src/effect.rs +++ b/crates/spt-daemon/src/effect.rs @@ -465,35 +465,23 @@ impl EffectJournal { /// Whether `key`'s effect has already been applied + recorded. pub fn is_applied(&self, key: EffectKey) -> bool { - self.lock_recover() - .applied - .contains(&key) + self.lock_recover().applied.contains(&key) } /// A snapshot of the applied-set (for introspection / tests). Unordered. pub fn applied_keys(&self) -> Vec { - self.lock_recover() - .applied - .iter() - .copied() - .collect() + self.lock_recover().applied.iter().copied().collect() } /// Count of distinct effects applied. pub fn applied_count(&self) -> usize { - self.lock_recover() - .applied - .len() + self.lock_recover().applied.len() } /// Keys with an unfinished `PENDING` (broker crashed mid-effect). Empty on the /// brain-crash path; surfaced for the future broker-restart recovery. pub fn pending_keys(&self) -> Vec { - self.lock_recover() - .pending - .iter() - .copied() - .collect() + self.lock_recover().pending.iter().copied().collect() } } @@ -895,11 +883,13 @@ mod tests { let net_key = k(1, 2); assert_eq!( - j.apply_once(pty_key, EffectKind::PtyWrite, || Ok(())).unwrap(), + j.apply_once(pty_key, EffectKind::PtyWrite, || Ok(())) + .unwrap(), Outcome::Applied ); assert_eq!( - j.apply_once(net_key, EffectKind::NetSend, || Ok(())).unwrap(), + j.apply_once(net_key, EffectKind::NetSend, || Ok(())) + .unwrap(), Outcome::Applied ); @@ -930,19 +920,27 @@ mod tests { // PtyWrite is not. Keys are written as "sid op" tokens on the journal line. let text = std::fs::read_to_string(j.path()).expect("read journal file"); // New line shape carries the minter tag: "PENDING 1 cli 2 net-send". - let net_token = format!("{} {} {}", net_key.class, net_key.minter.as_tag(), net_key.op); - let pty_token = format!("{} {} {}", pty_key.class, pty_key.minter.as_tag(), pty_key.op); + let net_token = format!( + "{} {} {}", + net_key.class, + net_key.minter.as_tag(), + net_key.op + ); + let pty_token = format!( + "{} {} {}", + pty_key.class, + pty_key.minter.as_tag(), + pty_key.op + ); assert!( text.lines().any(|l| { - (l.starts_with("PENDING ") || l.starts_with("DONE ")) - && l.contains(&net_token) + (l.starts_with("PENDING ") || l.starts_with("DONE ")) && l.contains(&net_token) }), "the durable NetSend effect must be journaled to disk; file was: {text:?}" ); assert!( !text.lines().any(|l| { - (l.starts_with("PENDING ") || l.starts_with("DONE ")) - && l.contains(&pty_token) + (l.starts_with("PENDING ") || l.starts_with("DONE ")) && l.contains(&pty_token) }), "an ephemeral PtyWrite must NOT pay the durable journal write (no per-\ keystroke fsync) — REQ-HAZARD-EFFECT-JOURNAL-PTY-WEDGE. File was: {text:?}" @@ -1068,7 +1066,10 @@ mod tests { minter: Minter::Rc, op: 3, }; - assert!(j_new.is_applied(rc_key), "new line recovers to the tagged key"); + assert!( + j_new.is_applied(rc_key), + "new line recovers to the tagged key" + ); assert!( !j_new.is_applied(legacy_key), "a new rc line is not a legacy key" @@ -1083,7 +1084,10 @@ mod tests { .unwrap(); let j_mixed = EffectJournal::open(&mixed_path).unwrap(); assert!(j_mixed.is_applied(legacy_key), "mixed: legacy key present"); - assert!(j_mixed.is_applied(new_producer_key), "mixed: cli key present"); + assert!( + j_mixed.is_applied(new_producer_key), + "mixed: cli key present" + ); assert_eq!( j_mixed.applied_count(), 2, @@ -1139,7 +1143,8 @@ mod tests { op: OP, }; assert_eq!( - j.apply_once(replay, EffectKind::NetDial, || Ok(())).unwrap(), + j.apply_once(replay, EffectKind::NetDial, || Ok(())) + .unwrap(), Outcome::Deduped, "a same-minter same-op replay is still deduped" ); @@ -1169,17 +1174,22 @@ mod tests { minter: Minter::Rc, op: colliding_int, }; - assert_ne!(shell_key, rc_key, "same session+int, different minter = distinct key"); + assert_ne!( + shell_key, rc_key, + "same session+int, different minter = distinct key" + ); // Spool delivery lands first (Spool is durable — a real journal write). assert_eq!( - j.apply_once(shell_key, EffectKind::Spool, || Ok(())).unwrap(), + j.apply_once(shell_key, EffectKind::Spool, || Ok(())) + .unwrap(), Outcome::Applied, "the shell spool delivery applies" ); // The rc operator's identically-numbered op must ALSO apply, not dedupe. assert_eq!( - j.apply_once(rc_key, EffectKind::PtyWrite, || Ok(())).unwrap(), + j.apply_once(rc_key, EffectKind::PtyWrite, || Ok(())) + .unwrap(), Outcome::Applied, "the rc operator op must NOT be swallowed by the shell op's key (the bug)" ); @@ -1224,7 +1234,11 @@ mod tests { assert_eq!(out, "healed"); let ops = minted.borrow(); - assert_eq!(ops.len(), 2, "run invoked exactly twice (initial + ONE retry)"); + assert_eq!( + ops.len(), + 2, + "run invoked exactly twice (initial + ONE retry)" + ); assert_eq!(ops[0].minter, Minter::Rc); assert_eq!( ops[1].minter, diff --git a/crates/spt-daemon/src/pairhost.rs b/crates/spt-daemon/src/pairhost.rs index 748a14e..5c537dc 100644 --- a/crates/spt-daemon/src/pairhost.rs +++ b/crates/spt-daemon/src/pairhost.rs @@ -1042,8 +1042,14 @@ mod tests { let v = meet_failure_detail("IPv4-only", "meet probe timed out", 4, 75, 75, clock, true); assert!(v.contains("4 rendezvous attempt"), "attempt count: {v}"); assert!(v.contains("75s"), "elapsed/deadline: {v}"); - assert!(v.contains("bound families: IPv4-only"), "families (ties W1): {v}"); - assert!(v.contains("last error: meet probe timed out"), "last error kept: {v}"); + assert!( + v.contains("bound families: IPv4-only"), + "families (ties W1): {v}" + ); + assert!( + v.contains("last error: meet probe timed out"), + "last error kept: {v}" + ); // REQ-JOIN-VERBOSE-CLOCK: the joiner's step, signed offset, and // correction state all carried in the verbose block. assert!(v.contains("joiner clock:"), "clock line present: {v}"); @@ -1133,8 +1139,15 @@ mod tests { ) .await; assert_eq!(got, Ok(7u32), "the post-refresh final sweep lands"); - assert_eq!(calls.get(), 2, "one exhausting probe + one post-refresh retry"); - assert!(refreshed.get(), "the refresh hook fired between the two probes"); + assert_eq!( + calls.get(), + 2, + "one exhausting probe + one post-refresh retry" + ); + assert!( + refreshed.get(), + "the refresh hook fired between the two probes" + ); } // [unit->REQ-HAZARD-CEREMONY-CLOCK-STEP] and if the final post-refresh sweep @@ -1158,6 +1171,10 @@ mod tests { ) .await; assert_eq!(got, Err("dead subnet"), "exhaustion error preserved"); - assert_eq!(calls.get(), 2, "exhausting probe + one final retry, then stop"); + assert_eq!( + calls.get(), + 2, + "exhausting probe + one final retry, then stop" + ); } } diff --git a/crates/spt-daemon/src/servicehost.rs b/crates/spt-daemon/src/servicehost.rs index 3dc3d25..566032f 100644 --- a/crates/spt-daemon/src/servicehost.rs +++ b/crates/spt-daemon/src/servicehost.rs @@ -472,10 +472,7 @@ pub fn fill_service_command( ) -> Result, String> { let keys = std::collections::BTreeMap::from([ ("adapter_name".to_string(), adapter_name.to_string()), - ( - "adapter_dir".to_string(), - install_dir.display().to_string(), - ), + ("adapter_dir".to_string(), install_dir.display().to_string()), ]); // [impl->REQ-HAZARD-TEMPLATE-ARGV-FILL] tokenize-template-then-fill-each: a // multi-word/quote/semicolon {key} value is exactly one argv element. @@ -605,7 +602,10 @@ impl OrphanSweep { /// previous instance is PROVEN gone (or provably never was): an unresolved /// sweep blocks the spawn rather than risking two live instances. pub fn clear_to_spawn(&self) -> bool { - matches!(self, Self::NoRecord | Self::AlreadyDead | Self::Killed | Self::NotOurs(_)) + matches!( + self, + Self::NoRecord | Self::AlreadyDead | Self::Killed | Self::NotOurs(_) + ) } } @@ -766,24 +766,23 @@ pub fn supervisor_run( // Each run gets a clean sheet: a fault must be explained by THIS run's // output, never by a previous one's still sitting in the file. reclaim_capture(&capture); - let mut child = - match crate::daemon::detached_no_inherit_env( - program, - args, - &env, - SERVICE_ENV_SCRUB, - Some(&capture), - ) { - Ok(c) => c, - Err(e) => { - let e = format!("spawn {program}: {e}"); - eprintln!("SERVICE_STARTUP_FAULT:{option}: {e}"); - return Some(StandDown { - latch: Latch::StartupFault, - detail: Some(e), - }); - } - }; + let mut child = match crate::daemon::detached_no_inherit_env( + program, + args, + &env, + SERVICE_ENV_SCRUB, + Some(&capture), + ) { + Ok(c) => c, + Err(e) => { + let e = format!("spawn {program}: {e}"); + eprintln!("SERVICE_STARTUP_FAULT:{option}: {e}"); + return Some(StandDown { + latch: Latch::StartupFault, + detail: Some(e), + }); + } + }; // Park the kill handle BEFORE the wait: a daemon that dies mid-run must // leave its successor something path-verifiable to reap. let image = spt_store::proc::exe_path(child.pid()); @@ -1027,9 +1026,7 @@ impl ServiceSet { /// The latch currently suppressing this option, [`Latch::None`] if none. // [impl->REQ-RESIDENT-SERVICE] pub fn latch(&self, option: &str) -> Latch { - self.stand_down(option) - .map(|s| s.latch) - .unwrap_or_default() + self.stand_down(option).map(|s| s.latch).unwrap_or_default() } /// The whole stand-down record — the latch AND the evidence for it. This is @@ -1073,8 +1070,7 @@ impl ServiceSet { pub fn is_held(&self, option: &str) -> bool { let key = spt_store::perch::encode_adapter_option(option); let map = self.holds.lock().unwrap_or_else(|p| p.into_inner()); - map.get(&key) - .is_some_and(|f| f.load(Ordering::SeqCst)) + map.get(&key).is_some_and(|f| f.load(Ordering::SeqCst)) } /// Engage the hold. Step 1 of [`quiesce_order`], and it must land BEFORE the @@ -1289,7 +1285,8 @@ pub fn reconcile_once( // Resolve through the OPTION seam even for a bare name, so this path is // option-general by construction rather than adapter-only with an // option-shaped signature bolted on later. - let Ok(manifest) = spt_runtime::registry::resolve_option_in(registered, adapters_dir, &option) + let Ok(manifest) = + spt_runtime::registry::resolve_option_in(registered, adapters_dir, &option) else { continue; // unresolvable manifest: not a service question }; @@ -1318,9 +1315,7 @@ pub fn reconcile_once( // and the operator asking "why will my service not start" would get the // fault's name and nothing else. let mut detail = match decision.outcome { - ServiceOutcome::StartupFault | ServiceOutcome::Latched => { - stood.and_then(|s| s.detail) - } + ServiceOutcome::StartupFault | ServiceOutcome::Latched => stood.and_then(|s| s.detail), _ => None, }; let outcome = match decision.outcome { @@ -2289,7 +2284,8 @@ mod tests { }; std::fs::write(dir.join(shipped), b"").unwrap(); - let tokens = fill_service_command("cc", dir, &svc("svcbin --serve {adapter_name}")).unwrap(); + let tokens = + fill_service_command("cc", dir, &svc("svcbin --serve {adapter_name}")).unwrap(); assert_eq!( tokens[0], dir.join(shipped).display().to_string(), @@ -2329,7 +2325,10 @@ mod tests { let home = Path::new("/spt-home"); let a_dir = spt_store::perch::resolve_service_dir_in(home, "cc:dev"); let b_dir = spt_store::perch::resolve_service_dir_in(home, "cc_dev"); - assert_ne!(a_dir, b_dir, "the collision-adversarial pair must stay apart"); + assert_ne!( + a_dir, b_dir, + "the collision-adversarial pair must stay apart" + ); let env = service_env_at(home, "cc:dev", &a_dir); assert_eq!( @@ -2477,17 +2476,23 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let tokens: Vec = long_running().split(' ').map(String::from).collect(); let (program, args) = tokens.split_first().unwrap(); - let child = - crate::daemon::detached_no_inherit_env(program, args, &[], &[], None).expect("spawn orphan"); + let child = crate::daemon::detached_no_inherit_env(program, args, &[], &[], None) + .expect("spawn orphan"); let pid = child.pid(); // Park exactly what a supervisor parks, then FORGET the handle — this is // a dead daemon's orphan, which nobody holds a handle to. let image = spt_store::proc::exe_path(pid); - assert!(image.is_some(), "the image oracle must answer for our own child"); + assert!( + image.is_some(), + "the image oracle must answer for our own child" + ); park_identity(tmp.path(), pid, image.as_deref()); drop(child); - assert_eq!(kill_orphan_service_at(tmp.path(), "cc"), OrphanSweep::Killed); + assert_eq!( + kill_orphan_service_at(tmp.path(), "cc"), + OrphanSweep::Killed + ); assert!( !spt_store::proc::is_process_alive(pid), "Killed is only reported when the post-kill read says so" @@ -2501,7 +2506,10 @@ mod tests { #[test] fn empty_and_dead_orphan_records_read_apart() { let tmp = tempfile::tempdir().unwrap(); - assert_eq!(kill_orphan_service_at(tmp.path(), "cc"), OrphanSweep::NoRecord); + assert_eq!( + kill_orphan_service_at(tmp.path(), "cc"), + OrphanSweep::NoRecord + ); park_identity(tmp.path(), 0, None); assert_eq!( kill_orphan_service_at(tmp.path(), "cc"), @@ -2669,9 +2677,8 @@ mod tests { }) }; assert!( - wait_until(|| read_parked_identity(&dir).is_some_and(|(pid, _)| { - pid != 0 && spt_store::proc::is_process_alive(pid) - })), + wait_until(|| read_parked_identity(&dir) + .is_some_and(|(pid, _)| { pid != 0 && spt_store::proc::is_process_alive(pid) })), "the supervised child never came up" ); let pid = read_parked_identity(&dir).unwrap().0; @@ -2890,11 +2897,25 @@ mod tests { let set = ServiceSet::new(); let params = ServiceParams::default(); - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); - assert_eq!(out.len(), 1, "one candidate per registered adapter: {out:?}"); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); + assert_eq!( + out.len(), + 1, + "one candidate per registered adapter: {out:?}" + ); assert_eq!(out[0].option, "a", "the RAW option is what is reported"); assert_eq!(out[0].outcome, ServiceOutcome::Started); - assert_eq!(out[0].detail, None, "a plain Started invents no reassurance"); + assert_eq!( + out[0].detail, None, + "a plain Started invents no reassurance" + ); assert!(set.contains("a")); let dir = spt_store::perch::resolve_service_dir("a"); @@ -2904,8 +2925,14 @@ mod tests { ); let first = read_parked_identity(&dir).expect("parked").0; - let again = - reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); + let again = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); assert_eq!(again[0].outcome, ServiceOutcome::AlreadyRunning); assert_eq!(set.len(), 1, "one supervisor per option"); assert_eq!( @@ -2935,14 +2962,28 @@ mod tests { let set = ServiceSet::new(); let params = ServiceParams::default(); - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); assert_eq!(out[0].outcome, ServiceOutcome::BindDeferred); assert!( set.is_empty(), "a deferred service is reported, never supervised" ); - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Bind, None, ¶ms); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Bind, + None, + ¶ms, + ); assert_eq!(out[0].outcome, ServiceOutcome::Started); assert!(set.contains("a")); @@ -2972,7 +3013,14 @@ mod tests { let set = ServiceSet::new(); let params = fast_latch_params(); - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); assert_eq!(out[0].outcome, ServiceOutcome::Started); assert!( @@ -2987,7 +3035,14 @@ mod tests { ); // A NON-clearing opportunity: report the suppression, raise nothing. - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Bind, None, ¶ms); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Bind, + None, + ¶ms, + ); assert_eq!(out[0].outcome, ServiceOutcome::StartupFault); assert_eq!( out[0].detail, None, @@ -3253,7 +3308,14 @@ mod tests { assert!(!set.contains("a")); // THE ASSERTION: a clearing opportunity does not start a held option. - let out = reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); + let out = reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); assert_eq!( out.iter().map(|o| o.outcome).collect::>(), [ServiceOutcome::Held], @@ -3739,7 +3801,14 @@ mod tests { )]; let set = ServiceSet::new(); let params = fast_latch_params(); - reconcile_once(&adapters, ®istered, &set, Opportunity::Boot, None, ¶ms); + reconcile_once( + &adapters, + ®istered, + &set, + Opportunity::Boot, + None, + ¶ms, + ); assert!( wait_until(|| set.latch("a") == Latch::StartupFault), "the fixture never latched" @@ -3800,8 +3869,18 @@ mod tests { crate::test_home::with_home(|home| { let (adapters, install) = sweep_dirs(home); let registered = vec![ - reg("a", &install, true, Some((long_running(), ServiceStart::Boot))), - reg("b", &install, false, Some((long_running(), ServiceStart::Bind))), + reg( + "a", + &install, + true, + Some((long_running(), ServiceStart::Boot)), + ), + reg( + "b", + &install, + false, + Some((long_running(), ServiceStart::Bind)), + ), reg("c", &install, true, None), ]; let set = ServiceSet::new(); [raw output: artifact://441]