fix(daemon): stop one pane's panic from costing every pane on the machine

The daemon holds every shell on the box behind mutexes — the pty master, the
child handle, the writer, the pane state. Seventy-four of those locks were
taken with `.lock().unwrap()`, so a panic in any one critical section
poisoned the mutex and every later taker died on it too. One bug in one
pane's thread, and the daemon can no longer serve any of them.

Poisoning buys nothing here. Whatever inconsistency the panic left is there
either way; the flag only decides whether the next thread also dies. A
garbled write to one pane is recoverable, and losing every session on the
machine is not.

Most of the code already agreed — about seventy places carried on, spelled
out by hand — so this is mostly a drift fix. `Locked::locked` gives the
policy one name, the daemon uses it throughout, and a test walks
`daemon/` and fails with the file and line if a panicking lock comes back.
Somewhere that genuinely cannot tolerate the inconsistency can still take
the poison with `lock()`; the point is that it be a decision rather than
the default.

crates/tty7-core/src/core/threads.rs

@@ -7,3 +7,118 @@ pub fn promote_to_user_interactive() {
         libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, 0);
     }
 }
+
+/// Takes a lock, and goes on taking it after another thread has panicked
+/// holding it.
+///
+/// `Mutex::lock().unwrap()` is the wrong default for anything the daemon owns.
+/// Poisoning protects nothing: whatever inconsistency a panic left behind is
+/// there either way, and the flag only decides whether the *next* thread to
+/// ask also dies. In a process that holds every shell on the machine, that
+/// turns one bug in one pane into a daemon that can no longer serve any of
+/// them — the pty master, the child handle, the writer and the pane state all
+/// sit behind mutexes, and a panic in any critical section takes the lot.
+///
+/// So the policy is to carry on. A garbled write to one pane is recoverable;
+/// losing every session on the box is not. This is what most of the code
+/// already did by hand, spelled `unwrap_or_else(|e| e.into_inner())` in about
+/// seventy places, against about eighty that panicked instead — the drift is
+/// what this exists to stop.
+///
+/// Where the inconsistency genuinely cannot be tolerated, take the poison
+/// explicitly with `lock()` and decide there; the point is that it be a
+/// decision rather than a default.
+pub trait Locked<T> {
+    /// The guard, whether or not the lock is poisoned.
+    fn locked(&self) -> std::sync::MutexGuard<'_, T>;
+}
+
+impl<T> Locked<T> for std::sync::Mutex<T> {
+    fn locked(&self) -> std::sync::MutexGuard<'_, T> {
+        self.lock().unwrap_or_else(|e| e.into_inner())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Locked as _;
+
+    /// A mutex another thread panicked under is still usable, and still holds
+    /// what that thread had written before it went.
+    #[test]
+    fn a_poisoned_lock_is_still_a_lock() {
+        let shared = std::sync::Arc::new(std::sync::Mutex::new(vec![1, 2, 3]));
+        let poisoner = std::sync::Arc::clone(&shared);
+        let died = std::thread::spawn(move || {
+            let mut held = poisoner.lock().expect("the first lock is clean");
+            held.push(4);
+            panic!("the thread goes down holding it");
+        })
+        .join();
+        assert!(died.is_err(), "the thread did panic");
+        assert!(shared.lock().is_err(), "and the mutex is poisoned");
+
+        assert_eq!(
+            *shared.locked(),
+            vec![1, 2, 3, 4],
+            "the lock still opens, and the write that got through is still there"
+        );
+        shared.locked().push(5);
+        assert_eq!(*shared.locked(), vec![1, 2, 3, 4, 5], "and it stays usable");
+    }
+
+    /// Nothing the daemon owns takes a lock that can panic on poison.
+    ///
+    /// The policy above is only worth stating if it holds, and it drifted
+    /// once already: about seventy places carried on and about eighty died,
+    /// with nothing to say which was intended. A guard is cheaper than
+    /// re-deciding it per review.
+    ///
+    /// Scoped to `daemon/` deliberately. That is the process holding every
+    /// shell on the machine, where the cascade costs the most; a tool that
+    /// panics takes only itself with it.
+    #[test]
+    fn the_daemon_takes_no_lock_that_dies_of_poison() {
+        fn walk(dir: &std::path::Path, found: &mut Vec<String>) {
+            let Ok(entries) = std::fs::read_dir(dir) else {
+                return;
+            };
+            for entry in entries.flatten() {
+                let path = entry.path();
+                if path.is_dir() {
+                    walk(&path, found);
+                    continue;
+                }
+                if path.extension().is_none_or(|e| e != "rs")
+                    || path.file_name().is_some_and(|n| n == "tests.rs")
+                {
+                    continue;
+                }
+                let Ok(text) = std::fs::read_to_string(&path) else {
+                    continue;
+                };
+                let mut in_tests = false;
+                for (n, line) in text.lines().enumerate() {
+                    in_tests |= line.contains("#[cfg(test)]");
+                    if !in_tests && line.contains(".lock().unwrap()") {
+                        found.push(format!("{}:{}", path.display(), n + 1));
+                    }
+                }
+            }
+        }
+
+        let daemon = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("src")
+            .join("daemon");
+        assert!(daemon.is_dir(), "the daemon sources moved: {daemon:?}");
+        let mut found = Vec::new();
+        walk(&daemon, &mut found);
+        assert!(
+            found.is_empty(),
+            "these take a lock that panics once another thread has poisoned it, \
+             which is how one pane's bug becomes every pane's; use `Locked::locked` \
+             or take the poison deliberately:\n{}",
+            found.join("\n")
+        );
+    }
+}

crates/tty7-core/src/daemon/pane.rs

@@ -17,6 +17,7 @@ use crate::daemon::protocol::{
     ShellSpec, WinSize,
 };
 use crate::daemon::shell_integration;
+use crate::core::threads::Locked as _;
 
 #[cfg(windows)]
 fn default_prog() -> CommandBuilder {
@@ -644,14 +645,14 @@ impl OutputGate {
     pub fn sub(&self, n: usize) {
         let prev = self.queued.fetch_sub(n as i64, Ordering::Relaxed);
         if prev >= Self::HIGH_WATER && prev - (n as i64) < Self::HIGH_WATER {
-            let _park = self.park.lock().unwrap();
+            let _park = self.park.locked();
             self.drained.notify_all();
         }
     }
 
     fn reset(&self) {
         self.queued.store(0, Ordering::Relaxed);
-        let _park = self.park.lock().unwrap();
+        let _park = self.park.locked();
         self.drained.notify_all();
     }
 
@@ -668,7 +669,7 @@ impl OutputGate {
             return;
         }
         let deadline = std::time::Instant::now() + Self::MAX_WAIT;
-        let mut park = self.park.lock().unwrap();
+        let mut park = self.park.locked();
         while self.queued.load(Ordering::Relaxed) >= Self::HIGH_WATER {
             let left = deadline.saturating_duration_since(std::time::Instant::now());
             if left.is_zero() {
@@ -973,7 +974,7 @@ impl DeathReporter {
     }
 
     fn probe_exit_code(&self, probe: impl FnMut() -> Option<i32> + Send + 'static) {
-        *self.exit_code.lock().unwrap() = Some(Box::new(probe));
+        *self.exit_code.locked() = Some(Box::new(probe));
     }
 
     #[cfg(windows)]
@@ -991,7 +992,7 @@ impl DeathReporter {
             .unwrap()
             .as_mut()
             .and_then(|probe| probe());
-        let mut st = state.lock().unwrap();
+        let mut st = state.locked();
         st.alive = false;
         st.exit_code = code;
         let pane = st.id;
@@ -1007,7 +1008,7 @@ impl DeathReporter {
         if subscribed {
             return;
         }
-        if let Some(on_dead) = self.on_dead.lock().unwrap().take() {
+        if let Some(on_dead) = self.on_dead.locked().take() {
             on_dead();
         }
     }
@@ -1599,7 +1600,7 @@ impl DaemonPane {
             },
             death,
         );
-        *pane.reader.lock().unwrap() = Some(reader);
+        *pane.reader.locked() = Some(reader);
 
         pane
     }
@@ -1625,7 +1626,7 @@ impl DaemonPane {
             .ok()?
             .as_ref()
             .and_then(|master| master.as_raw_fd())?;
-        let st = self.state.lock().unwrap();
+        let st = self.state.locked();
         Some(Carried {
             id: self.id,
             owner: self.owner.clone(),
@@ -1769,7 +1770,7 @@ impl DaemonPane {
         let broker = {
             let state = state.clone();
             crate::daemon::ssh::PromptBroker::new(Box::new(move |msg: DaemonMsg| {
-                match &state.lock().unwrap().subscriber {
+                match &state.locked().subscriber {
                     Some(sub) => sub.send(msg).is_ok(),
                     None => false,
                 }
@@ -1807,7 +1808,7 @@ impl DaemonPane {
             },
             death,
         );
-        *pane.reader.lock().unwrap() = Some(reader);
+        *pane.reader.locked() = Some(reader);
 
         crate::daemon::ssh::SshManager::global().spawn_native_session(
             id,
@@ -1830,7 +1831,7 @@ impl DaemonPane {
 
     pub(crate) fn ssh_connection(&self) -> Option<Arc<crate::daemon::ssh::SshConnection>> {
         match &self.backend {
-            PaneBackend::NativeSsh(b) => b.connection.lock().unwrap().upgrade(),
+            PaneBackend::NativeSsh(b) => b.connection.locked().upgrade(),
             PaneBackend::Pty(_) => None,
         }
     }
@@ -1870,7 +1871,7 @@ impl DaemonPane {
                 // local can `ssh` out mid-session, so the flag is refreshed from
                 // the same remote-context poll below. Seed it from the pane's
                 // current context so the first probe answers correctly.
-                let starts_local = state.lock().unwrap().remote.is_none();
+                let starts_local = state.locked().remote.is_none();
                 let mut graphics = GraphicsSniffer::new_local(starts_local);
                 let mut buf = [0u8; 65536];
 
@@ -1993,7 +1994,7 @@ impl DaemonPane {
                             }
                             let remote = if poll_now {
                                 let managed = {
-                                    let st = state.lock().unwrap();
+                                    let st = state.locked();
                                     st.remote
                                         .as_ref()
                                         .is_some_and(|remote| remote.kind != RemoteKind::Ssh)
@@ -2014,7 +2015,7 @@ impl DaemonPane {
                                 || remote.is_some()
                                 || agent.is_some()
                                 || probed_cwd.is_some();
-                            let mut st = state.lock().unwrap();
+                            let mut st = state.locked();
                             let facts_before = may_change_facts.then(|| observed_facts(&st));
                             st.ring.append(bytes);
                             fan_out_output(&mut st, bytes, frames, &gate);
@@ -2070,14 +2071,14 @@ impl DaemonPane {
     }
 
     pub fn attach(&self, subscriber: Sender<DaemonMsg>) -> u64 {
-        let mut st = self.state.lock().unwrap();
+        let mut st = self.state.locked();
         let epoch = attach_subscriber(&mut st, subscriber);
         self.gate.reset();
         epoch
     }
 
     pub fn detach(&self, epoch: u64) -> bool {
-        let mut st = self.state.lock().unwrap();
+        let mut st = self.state.locked();
         if st.subscriber_epoch == epoch {
             st.subscriber = None;
             self.gate.reset();
@@ -2086,21 +2087,21 @@ impl DaemonPane {
     }
 
     pub fn observe(&self, observer: Sender<DaemonMsg>, gate: Arc<OutputGate>) -> u64 {
-        let mut st = self.state.lock().unwrap();
+        let mut st = self.state.locked();
         observe_subscriber(&mut st, observer, gate)
     }
 
     pub(crate) fn unobserve(&self, observer_id: u64) {
-        let mut st = self.state.lock().unwrap();
+        let mut st = self.state.locked();
         st.observers.retain(|obs| obs.id != observer_id);
     }
 
     pub fn controls(&self, epoch: u64) -> bool {
-        self.state.lock().unwrap().subscriber_epoch == epoch
+        self.state.locked().subscriber_epoch == epoch
     }
 
     pub fn agent_state(&self) -> Option<crate::daemon::control::PaneAgentState> {
-        agent_state_snapshot(&self.state.lock().unwrap())
+        agent_state_snapshot(&self.state.locked())
     }
 
     pub fn gate(&self) -> Arc<OutputGate> {
@@ -2135,7 +2136,7 @@ impl DaemonPane {
     }
 
     pub fn resize(&self, size: WinSize) {
-        resize_state(&mut self.state.lock().unwrap(), size);
+        resize_state(&mut self.state.locked(), size);
         match &self.backend {
             PaneBackend::Pty(p) => {
                 if let Ok(master) = p.master.lock()
@@ -2149,12 +2150,12 @@ impl DaemonPane {
     }
 
     pub fn alive(&self) -> bool {
-        self.state.lock().unwrap().alive
+        self.state.locked().alive
     }
 
     pub fn info(&self) -> PaneInfo {
         let (cwd, osc_title, alive) = {
-            let st = self.state.lock().unwrap();
+            let st = self.state.locked();
             (st.cwd.clone(), st.osc_title.clone(), st.alive)
         };
         PaneInfo {
@@ -2168,7 +2169,7 @@ impl DaemonPane {
     }
 
     pub(crate) fn remote_context(&self) -> Option<RemoteContext> {
-        let cached = self.state.lock().unwrap().remote.clone();
+        let cached = self.state.locked().remote.clone();
         cached.or_else(|| self.foreground_remote_context())
     }
 
@@ -2179,7 +2180,7 @@ impl DaemonPane {
     /// snapshot is fine — the snapshot returns its own mark, and that pair is
     /// what gets recorded.
     pub(crate) fn scrollback_mark(&self) -> u64 {
-        self.state.lock().unwrap().ring.appended
+        self.state.locked().ring.appended
     }
 
     /// The pane's screen, capped for storage, with the mark that says how much
@@ -2191,7 +2192,7 @@ impl DaemonPane {
     /// that stopped producing at that moment would keep the stale copy for
     /// good, because its mark would never move again.
     pub(crate) fn scrollback_snapshot(&self) -> (Vec<crate::daemon::scrollback::Segment>, u64) {
-        let st = self.state.lock().unwrap();
+        let st = self.state.locked();
         let mut segments = st.ring.snapshot();
         crate::daemon::scrollback::trim_to(&mut segments, crate::daemon::scrollback::SNAPSHOT_CAP);
         (segments, st.ring.appended)
@@ -2345,7 +2346,7 @@ impl Drop for DaemonPane {
         {
             let _ = child.wait();
         }
-        if let Some(handle) = self.reader.lock().unwrap().take() {
+        if let Some(handle) = self.reader.locked().take() {
             join_bounded(handle, Duration::from_secs(2));
         }
         if let PaneBackend::Pty(p) = &mut self.backend

crates/tty7-core/src/daemon/server.rs

@@ -8,6 +8,7 @@ use crate::daemon::pane::DaemonPane;
 use crate::daemon::protocol::{ClientMsg, DaemonMsg, DaemonVersion, RemoteKind};
 use crate::daemon::ssh::SshConnection;
 use crate::daemon::transport::{self, Stream};
+use crate::core::threads::Locked as _;
 
 struct Registry {
     panes: Mutex<HashMap<u64, Arc<DaemonPane>>>,
@@ -57,20 +58,20 @@ impl Registry {
     }
 
     fn insert(&self, pane: Arc<DaemonPane>) {
-        self.panes.lock().unwrap().insert(pane.id, pane);
+        self.panes.locked().insert(pane.id, pane);
     }
 
     fn get(&self, id: u64) -> Option<Arc<DaemonPane>> {
-        self.panes.lock().unwrap().get(&id).cloned()
+        self.panes.locked().get(&id).cloned()
     }
 
     fn remove(&self, id: u64) -> Option<Arc<DaemonPane>> {
-        self.panes.lock().unwrap().remove(&id)
+        self.panes.locked().remove(&id)
     }
 
     fn drain_and_kill(&self) {
         let panes: Vec<Arc<DaemonPane>> = {
-            let mut guard = self.panes.lock().unwrap();
+            let mut guard = self.panes.locked();
             guard.drain().map(|(_, p)| p).collect()
         };
         for pane in panes {
@@ -79,7 +80,7 @@ impl Registry {
     }
 
     fn all(&self) -> Vec<Arc<DaemonPane>> {
-        self.panes.lock().unwrap().values().cloned().collect()
+        self.panes.locked().values().cloned().collect()
     }
 
     fn list(&self) -> Vec<crate::daemon::protocol::PaneInfo> {
@@ -94,7 +95,7 @@ impl Registry {
 
 impl crate::host::server::PaneDirectory for Registry {
     fn pane_count(&self) -> u64 {
-        self.panes.lock().unwrap().len() as u64
+        self.panes.locked().len() as u64
     }
 
     fn panes(&self) -> Vec<crate::daemon::protocol::PaneInfo> {
@@ -102,7 +103,7 @@ impl crate::host::server::PaneDirectory for Registry {
     }
 
     fn agent_states(&self) -> Vec<crate::daemon::control::PaneAgentState> {
-        let panes: Vec<Arc<DaemonPane>> = self.panes.lock().unwrap().values().cloned().collect();
+        let panes: Vec<Arc<DaemonPane>> = self.panes.locked().values().cloned().collect();
         let mut states: Vec<_> = panes.iter().filter_map(|p| p.agent_state()).collect();
         states.sort_by_key(|s| s.pane_id);
         states

crates/tty7-core/src/daemon/ssh/broker.rs

@@ -6,6 +6,7 @@ use std::time::Duration;
 use tokio::sync::oneshot;
 
 use crate::daemon::protocol::{AuthPromptKind, AuthResponse, DaemonMsg, SshPhase};
+use crate::core::threads::Locked as _;
 
 const PROMPT_TIMEOUT: Duration = Duration::from_secs(120);
 const DELIVERY_WINDOW: Duration = Duration::from_secs(15);
@@ -27,27 +28,27 @@ impl PromptBroker {
     }
 
     pub fn has_pending(&self) -> bool {
-        !self.pending.lock().unwrap().is_empty()
+        !self.pending.locked().is_empty()
     }
 
     pub async fn prompt(&self, kind: AuthPromptKind) -> AuthResponse {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (tx, rx) = oneshot::channel();
-        self.pending.lock().unwrap().insert(id, tx);
+        self.pending.locked().insert(id, tx);
 
         let frame = DaemonMsg::AuthPrompt {
             request_id: id,
             prompt: kind,
         };
         if !self.deliver_with_retry(frame).await {
-            self.pending.lock().unwrap().remove(&id);
+            self.pending.locked().remove(&id);
             return AuthResponse::Cancelled;
         }
 
         match tokio::time::timeout(PROMPT_TIMEOUT, rx).await {
             Ok(Ok(resp)) => resp,
             _ => {
-                self.pending.lock().unwrap().remove(&id);
+                self.pending.locked().remove(&id);
                 AuthResponse::Cancelled
             }
         }
@@ -78,7 +79,7 @@ impl PromptBroker {
     }
 
     pub fn deliver(&self, request_id: u64, response: AuthResponse) {
-        if let Some(tx) = self.pending.lock().unwrap().remove(&request_id) {
+        if let Some(tx) = self.pending.locked().remove(&request_id) {
             let _ = tx.send(response);
         }
     }

crates/tty7-core/src/daemon/ssh/forward.rs

@@ -15,6 +15,7 @@ use crate::daemon::protocol::{
 
 use super::session::SshConnection;
 use super::{ConnectionKey, SshManager};
+use crate::core::threads::Locked as _;
 
 async fn accept_retrying(listener: &TcpListener) -> Option<(TcpStream, std::net::SocketAddr)> {
     let mut failures = 0u32;
@@ -166,7 +167,7 @@ impl RemoteForwardTable {
     }
 
     pub(super) fn rekey(&self, bind_host: &str, from_port: u16, to_port: u16) {
-        let mut map = self.inner.lock().unwrap();
+        let mut map = self.inner.locked();
         if let Some(target) = map.remove(&(bind_host.to_string(), from_port)) {
             map.insert((bind_host.to_string(), to_port), target);
         }
@@ -177,7 +178,7 @@ impl RemoteForwardTable {
         connected_address: &str,
         connected_port: u16,
     ) -> Option<(String, u16)> {
-        let map = self.inner.lock().unwrap();
+        let map = self.inner.locked();
         if let Some(t) = map.get(&(connected_address.to_string(), connected_port)) {
             return Some(t.clone());
         }
@@ -276,7 +277,7 @@ impl ForwardEntry {
             target_host: self.target_host.clone(),
             target_port: self.target_port,
             description: self.description.clone(),
-            status: self.status.lock().unwrap().clone(),
+            status: self.status.locked().clone(),
         }
     }
 }
@@ -381,7 +382,7 @@ impl SshForwardRegistry {
     }
 
     fn list_owned(&self, owner: &ForwardOwner, view_pane: u64) -> Vec<ManagedForward> {
-        let owners = self.owners.lock().unwrap();
+        let owners = self.owners.locked();
         let mut list: Vec<_> = owners
             .get(owner)
             .into_iter()
@@ -399,7 +400,7 @@ impl SshForwardRegistry {
         forward_id: u64,
     ) -> Vec<ManagedForward> {
         let removed = {
-            let mut owners = self.owners.lock().unwrap();
+            let mut owners = self.owners.locked();
             owners.get_mut(owner).and_then(|entries| {
                 let pos = entries.iter().position(|e| e.id == forward_id)?;
                 Some(entries.remove(pos))
@@ -412,7 +413,7 @@ impl SshForwardRegistry {
     }
 
     async fn teardown_owned(&self, owner: &ForwardOwner) {
-        let entries = self.owners.lock().unwrap().remove(owner);
+        let entries = self.owners.locked().remove(owner);
         for entry in entries.into_iter().flatten() {
             Self::cancel_entry(entry).await;
         }
@@ -479,7 +480,7 @@ impl SshForwardRegistry {
                     }
                 });
             };
-            *task_status.lock().unwrap() = loop_exit_status(exit);
+            *task_status.locked() = loop_exit_status(exit);
         });
         (bound, status, ForwardCancel::Task(handle))
     }
@@ -535,7 +536,7 @@ impl SshForwardRegistry {
                     }
                 });
             };
-            *task_status.lock().unwrap() = loop_exit_status(exit);
+            *task_status.locked() = loop_exit_status(exit);
         });
         (bound, status, ForwardCancel::Task(handle))
     }
@@ -625,7 +626,7 @@ impl SshForwardRegistry {
         };
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (bind_port, status, cancel) = self.start_local(&conn, &rule).await;
-        if let ForwardStatus::Error(e) = &*status.lock().unwrap() {
+        if let ForwardStatus::Error(e) = &*status.locked() {
             return Err(io::Error::other(e.clone()));
         }
         let entry = ForwardEntry {
@@ -657,7 +658,7 @@ impl SshForwardRegistry {
         remote_host: &str,
         remote_port: u16,
     ) -> Option<u16> {
-        let owners = self.owners.lock().unwrap();
+        let owners = self.owners.locked();
         owners
             .get(owner)?
             .iter()
@@ -668,7 +669,7 @@ impl SshForwardRegistry {
                     && e.target_port == remote_port
                     // Now that a dead loop says so, this stops handing out the
                     // port of a forward that no longer serves anything.
-                    && matches!(*e.status.lock().unwrap(), ForwardStatus::Listening)
+                    && matches!(*e.status.locked(), ForwardStatus::Listening)
             })
             .map(|e| e.bind_port)
     }
@@ -677,7 +678,7 @@ impl SshForwardRegistry {
 impl SshManager {
     pub(crate) fn existing_connection(&self, spec: &NativeSshSpec) -> Option<Arc<SshConnection>> {
         let key = ConnectionKey::from_spec(spec);
-        let slot = self.conns.lock().unwrap().get(&key).cloned()?;
+        let slot = self.conns.locked().get(&key).cloned()?;
         let guard = slot.try_lock().ok()?;
         let conn = guard.upgrade()?;
         conn.is_alive().then_some(conn)

crates/tty7-core/src/daemon/ssh/mod.rs

@@ -34,6 +34,7 @@ use crate::daemon::shell_integration::remote;
 use forward::RemoteForwardTable;
 use handler::ClientHandler;
 use session::drive_channel;
+use crate::core::threads::Locked as _;
 
 const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
 
@@ -233,7 +234,7 @@ impl SshManager {
             .await
             .map_err(|e| format!("{e}"))?;
 
-        *conn_slot.lock().unwrap() = Arc::downgrade(&conn);
+        *conn_slot.locked() = Arc::downgrade(&conn);
 
         broker.status(SshPhase::Connected);
 
@@ -252,7 +253,7 @@ impl SshManager {
                     .await
                     .map_err(|e| format!("{e}"))?;
                 conn = fresh;
-                *conn_slot.lock().unwrap() = Arc::downgrade(&conn);
+                *conn_slot.locked() = Arc::downgrade(&conn);
                 conn.open_session_channel()
                     .await
                     .map_err(|e| format!("open shell channel failed: {e}"))?
@@ -411,11 +412,11 @@ impl SshManager {
     }
 
     fn evict_connection(&self, key: &ConnectionKey) {
-        self.conns.lock().unwrap().remove(key);
+        self.conns.locked().remove(key);
     }
 
     pub fn routes(&self) -> Vec<crate::daemon::control::RouteInfo> {
-        let conns = self.conns.lock().unwrap();
+        let conns = self.conns.locked();
         let mut routes: Vec<_> = conns
             .iter()
             .map(|(key, slot)| {
@@ -442,7 +443,7 @@ impl SshManager {
 
     async fn remote_bootstrap(&self, conn: &Arc<SshConnection>) -> Option<String> {
         let key = conn.key().clone();
-        let cached = { self.probes.lock().unwrap().get(&key).cloned() };
+        let cached = { self.probes.locked().get(&key).cloned() };
         let probed = match cached {
             Some(hit) => hit,
             // Only an answer is remembered. This map is on the process-wide
@@ -459,7 +460,7 @@ impl SshManager {
                         }
                         None => log::debug!("ssh {key:?}: no remote shell integration"),
                     }
-                    self.probes.lock().unwrap().insert(key, answer.clone());
+                    self.probes.locked().insert(key, answer.clone());
                     answer
                 }
                 None => {
@@ -502,7 +503,7 @@ impl SshManager {
             let mut guard = match reuse {
                 true => {
                     let slot: ConnSlot = {
-                        let mut map = self.conns.lock().unwrap();
+                        let mut map = self.conns.locked();
                         map.entry(key.clone())
                             .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(Weak::new())))
                             .clone()

crates/tty7-core/src/daemon/ssh/sftp.rs

@@ -15,6 +15,7 @@ use crate::daemon::protocol::{
 };
 
 use super::{ConnectionKey, SshConnection, SshManager};
+use crate::core::threads::Locked as _;
 
 const CHUNK: usize = 256 * 1024;
 
@@ -196,34 +197,34 @@ impl Job {
     }
 
     fn set_total(&self, total: u64) {
-        self.progress.lock().unwrap().set_total(total);
+        self.progress.locked().set_total(total);
     }
 
     fn set_current(&self, path: impl Into<String>) {
-        self.progress.lock().unwrap().set_current(path);
+        self.progress.locked().set_current(path);
     }
 
     fn add_bytes(&self, n: u64) {
-        self.progress.lock().unwrap().add_bytes(n);
+        self.progress.locked().add_bytes(n);
     }
 
     fn finish(&self) {
-        self.progress.lock().unwrap().finish();
-        *self.done_at.lock().unwrap() = Some(Instant::now());
+        self.progress.locked().finish();
+        *self.done_at.locked() = Some(Instant::now());
     }
 
     fn fail(&self, reason: impl Into<String>) {
-        self.progress.lock().unwrap().fail(reason);
-        *self.done_at.lock().unwrap() = Some(Instant::now());
+        self.progress.locked().fail(reason);
+        *self.done_at.locked() = Some(Instant::now());
     }
 
     fn mark_cancelled(&self) {
-        self.progress.lock().unwrap().cancel();
-        *self.done_at.lock().unwrap() = Some(Instant::now());
+        self.progress.locked().cancel();
+        *self.done_at.locked() = Some(Instant::now());
     }
 
     fn snapshot(&self) -> SftpJobProgress {
-        let p = self.progress.lock().unwrap();
+        let p = self.progress.locked();
         SftpJobProgress {
             job_id: self.id,
             pane_id: self.pane_id,
@@ -240,7 +241,7 @@ impl Job {
 
     fn is_expired(&self) -> bool {
         matches!(
-            *self.done_at.lock().unwrap(),
+            *self.done_at.locked(),
             Some(t) if t.elapsed() > JOB_RETENTION
         )
     }
@@ -337,7 +338,7 @@ impl SftpManager {
             progress: Mutex::new(JobProgress::new()),
             done_at: Mutex::new(None),
         });
-        self.jobs.lock().unwrap().insert(id, job.clone());
+        self.jobs.locked().insert(id, job.clone());
 
         SshManager::global().handle().spawn(async move {
             run_transfer(sftp, spec, job).await;
@@ -347,7 +348,7 @@ impl SftpManager {
 
     pub fn cancel(&self, job_id: u64) -> Vec<SftpJobProgress> {
         let pane = {
-            let jobs = self.jobs.lock().unwrap();
+            let jobs = self.jobs.locked();
             if let Some(job) = jobs.get(&job_id) {
                 job.cancel.store(true, Ordering::SeqCst);
                 Some(job.pane_id)
@@ -362,7 +363,7 @@ impl SftpManager {
     }
 
     pub(crate) fn list_jobs(&self, pane_id: u64) -> Vec<SftpJobProgress> {
-        let mut jobs = self.jobs.lock().unwrap();
+        let mut jobs = self.jobs.locked();
         jobs.retain(|_, job| !job.is_expired());
         let mut out: Vec<SftpJobProgress> = jobs
             .values()
@@ -392,7 +393,7 @@ impl SftpManager {
 
     async fn session_for(&self, conn: &Arc<SshConnection>) -> Result<Arc<SftpSession>, String> {
         let slot = {
-            let mut map = self.sessions.lock().unwrap();
+            let mut map = self.sessions.locked();
             map.entry(conn.key().clone())
                 .or_insert_with(|| {
                     Arc::new(SessionSlot {
@@ -417,7 +418,7 @@ impl SftpManager {
     }
 
     fn invalidate(&self, key: &ConnectionKey) {
-        self.sessions.lock().unwrap().remove(key);
+        self.sessions.locked().remove(key);
     }
 }