feat(daemon): size leases, and Take Back on the desktop

An observer can now run a pane at its own size (ClientMsg::Lease, feature size-lease). The pty and every observer go to that size. The controller keeps its grid, its resizes are remembered rather than applied, and the pane goes back to the last of them when the lease ends: the observer lets go, its connection closes, or the controller takes it back. A controller hears about leases only after asking (Watch), since an older client cannot decode DaemonMsg::Lease. The desktop asks on every attach, spawn and relink where the daemon advertises the feature, locally or through the host hello for remote workspaces, and shows the pane as in use on the phone with a Take Back button.
2026-11-06 08:15:35 +00:00 · 2026-09-27 12:50:40 +02:00
parent 7ad13295c5
commit 39ff2e6643
12 changed files with 520 additions and 12 deletions
@@ -3,8 +3,8 @@ use std::path::PathBuf;
 use std::time::Duration;

 use crate::daemon::protocol::{
-    ClientMsg, DaemonMsg, DaemonVersion, PaneInfo, PaneProcs, ShellSpec, WinSize, is_error_kind,
-    peek_frame_kind, take_frame,
+    ClientMsg, DaemonMsg, DaemonVersion, LeaseRequest, PaneInfo, PaneProcs, ShellSpec, WinSize,
+    is_error_kind, peek_frame_kind, take_frame,
 };
 use crate::daemon::router::{RouteAction, RouteChannel, RouteHeader, RouteTarget, negotiate};
 use crate::daemon::transport;
@@ -292,6 +292,11 @@ impl PaneSession {
        self.input.resize(size)
    }

+    /// See [`PaneInput::lease`].
+    pub fn lease(&mut self, request: LeaseRequest) -> io::Result<()> {
+        self.input.lease(request)
+    }
+
    pub fn detach(self) -> io::Result<()> {
        self.input.detach()
    }
@@ -332,6 +337,13 @@ impl PaneInput {
        ClientMsg::Resize(size).encode(&mut self.writer)
    }

+    /// Only to a daemon that advertised
+    /// [`FEATURE_SIZE_LEASE`](crate::daemon::protocol::FEATURE_SIZE_LEASE):
+    /// an older one cannot decode it and drops the connection.
+    pub fn lease(&mut self, request: LeaseRequest) -> io::Result<()> {
+        ClientMsg::Lease(request).encode(&mut self.writer)
+    }
+
    pub fn detach(mut self) -> io::Result<()> {
        ClientMsg::Detach.encode(&mut self.writer)
    }
@@ -17,8 +17,8 @@ use crate::core::kitty_graphics::{GraphicsSniffer, Segment, Sniffed};
 use crate::core::osc::OscTokenizer;
 use crate::core::term_modes::TerminalModes;
 use crate::daemon::protocol::{
-    AuthResponse, DaemonMsg, MAX_FRAME, NativeSshSpec, PaneInfo, RemoteContext, RemoteKind,
-    ShellSpec, WinSize,
+    AuthResponse, DaemonMsg, LeaseRequest, MAX_FRAME, NativeSshSpec, PaneInfo, RemoteContext,
+    RemoteKind, ShellSpec, WinSize,
 };
 use crate::daemon::shell_integration;

@@ -673,6 +673,19 @@ struct Observer {
    gate: Arc<OutputGate>,
 }

+/// An observer showing the pane at its own size — a phone, typically, while
+/// its user is away from the desk.
+///
+/// The pty runs at the observer's size and the observers see it; the
+/// controller keeps its own grid, is told who holds the pane, and is the only
+/// one that can end it early ([`LeaseRequest::TakeBack`]). What the controller
+/// asks for meanwhile is kept, not applied, and is what the pane goes back to.
+struct SizeLease {
+    observer: u64,
+    by: String,
+    desk: WinSize,
+}
+
 struct PaneState {
    id: u64,
    ring: ReplayRing,
@@ -689,6 +702,11 @@ struct PaneState {
    clipboard_write_from_spec: Option<bool>,
    observers: Vec<Observer>,
    observer_seq: u64,
+    lease: Option<SizeLease>,
+    /// The controller asked to hear about leases: it can say
+    /// [`DaemonMsg::Lease`], which a client from before them cannot decode.
+    /// Reset with every new controller.
+    lease_watch: bool,
    cwd: Option<PathBuf>,
    /// The last title the pane reported over OSC 0/2, for the machine tree to
    /// record. See [`crate::core::machine::PaneRecord::osc_title`].
@@ -764,6 +782,91 @@ fn resize_state(st: &mut PaneState, size: WinSize) {
        .retain(|obs| obs.tx.send(DaemonMsg::Size(size)).is_ok());
 }

+fn tell_observer(st: &mut PaneState, observer: u64, msg: DaemonMsg) {
+    if let Some(obs) = st.observers.iter().find(|obs| obs.id == observer) {
+        let _ = obs.tx.send(msg);
+    }
+}
+
+fn tell_controller_lease(st: &PaneState) {
+    if !st.lease_watch {
+        return;
+    }
+    if let Some(sub) = &st.subscriber {
+        let _ = sub.send(DaemonMsg::Lease(st.lease.as_ref().map(|l| l.by.clone())));
+    }
+}
+
+/// An observer takes the pane at `size`, or changes the size it holds it at.
+/// Returns the size to put the pty at, once the state lock is let go.
+///
+/// A second observer taking it over displaces the first, who is told, and the
+/// size the desk asked for carries over, so ending the lease still restores
+/// the desk rather than the first observer's size.
+fn lease_take(st: &mut PaneState, observer: u64, size: WinSize, by: String) -> Option<WinSize> {
+    if !st.observers.iter().any(|obs| obs.id == observer) {
+        return None;
+    }
+    let desk = match st.lease.take() {
+        Some(prev) => {
+            if prev.observer != observer {
+                tell_observer(st, prev.observer, DaemonMsg::Lease(None));
+            }
+            prev.desk
+        }
+        None => st.ring.size(),
+    };
+    st.lease = Some(SizeLease {
+        observer,
+        by: by.clone(),
+        desk,
+    });
+    tell_controller_lease(st);
+    // The observers see the pane at the leased size, sealed into the ring so
+    // a replay does too. The controller keeps its grid and gets no Size: a
+    // client that is told its pane changed size resizes back to fit its
+    // window, and the two would take turns.
+    st.ring.resize(size);
+    st.observers
+        .retain(|obs| obs.tx.send(DaemonMsg::Size(size)).is_ok());
+    tell_observer(st, observer, DaemonMsg::Lease(Some(by)));
+    Some(size)
+}
+
+/// Ends the lease, whoever holds it, and puts the pane back at the size the
+/// desk last asked for. Returns that size for the pty, or `None` if there was
+/// no lease.
+fn lease_end(st: &mut PaneState) -> Option<WinSize> {
+    let lease = st.lease.take()?;
+    tell_observer(st, lease.observer, DaemonMsg::Lease(None));
+    tell_controller_lease(st);
+    resize_state(st, lease.desk);
+    Some(lease.desk)
+}
+
+/// Ends the lease if `observer` holds it: it let go, or it went away.
+fn lease_release(st: &mut PaneState, observer: u64) -> Option<WinSize> {
+    if st.lease.as_ref()?.observer != observer {
+        return None;
+    }
+    lease_end(st)
+}
+
+/// The controller's resize. Under a lease it is only remembered, for when the
+/// lease ends, and echoed to the controller alone so it reflows its own grid.
+/// Returns the size for the pty if it is to change now.
+fn controller_resize(st: &mut PaneState, size: WinSize) -> Option<WinSize> {
+    if let Some(lease) = &mut st.lease {
+        lease.desk = size;
+        if let Some(sub) = &st.subscriber {
+            let _ = sub.send(DaemonMsg::Size(size));
+        }
+        return None;
+    }
+    resize_state(st, size);
+    Some(size)
+}
+
 /// One gated message to the controller and to every observer.
 ///
 /// A send error just means that client is gone; ignore it and let the next
@@ -1584,6 +1687,8 @@ impl DaemonPane {
                clipboard_write_from_spec: None,
                observers: Vec::new(),
                observer_seq: 0,
+                lease: None,
+                lease_watch: false,
                cwd: spawn.initial_cwd,
                osc_title: restored_title,
                shell: ShellState::default(),
@@ -1804,6 +1909,8 @@ impl DaemonPane {
                clipboard_write_from_spec: None,
                observers: Vec::new(),
                observer_seq: 0,
+                lease: None,
+                lease_watch: false,
                cwd: carried.cwd,
                osc_title: carried.osc_title,
                shell_spec: carried.shell_spec,
@@ -1867,6 +1974,8 @@ impl DaemonPane {
            clipboard_write_from_spec: Some(allow_remote_clipboard_write),
            observers: Vec::new(),
            observer_seq: 0,
+            lease: None,
+            lease_watch: false,
            // A native ssh pane is not running a shell of this machine's; what
            // it is, `ssh_spec` already says.
            shell_spec: None,
@@ -2239,8 +2348,51 @@ impl DaemonPane {
    }

    pub fn unobserve(&self, observer_id: u64) {
-        let mut st = self.state.lock().unwrap();
-        st.observers.retain(|obs| obs.id != observer_id);
+        let restore = {
+            let mut st = self.state.lock().unwrap();
+            let restore = lease_release(&mut st, observer_id);
+            st.observers.retain(|obs| obs.id != observer_id);
+            restore
+        };
+        if let Some(size) = restore {
+            self.resize_pty(size);
+        }
+    }
+
+    /// An observer's [`LeaseRequest`]: take the pane at its size, or let go.
+    pub fn observer_lease(&self, observer_id: u64, request: LeaseRequest) {
+        let apply = {
+            let mut st = self.state.lock().unwrap();
+            match request {
+                LeaseRequest::Take { size, by } => lease_take(&mut st, observer_id, size, by),
+                LeaseRequest::Release => lease_release(&mut st, observer_id),
+                LeaseRequest::Watch | LeaseRequest::TakeBack => None,
+            }
+        };
+        if let Some(size) = apply {
+            self.resize_pty(size);
+        }
+    }
+
+    /// The controller's [`LeaseRequest`]: hear about leases, or end one.
+    pub fn controller_lease(&self, request: LeaseRequest) {
+        let apply = {
+            let mut st = self.state.lock().unwrap();
+            match request {
+                LeaseRequest::Watch => {
+                    st.lease_watch = true;
+                    if st.lease.is_some() {
+                        tell_controller_lease(&st);
+                    }
+                    None
+                }
+                LeaseRequest::TakeBack => lease_end(&mut st),
+                LeaseRequest::Take { .. } | LeaseRequest::Release => None,
+            }
+        };
+        if let Some(size) = apply {
+            self.resize_pty(size);
+        }
    }

    pub fn controls(&self, epoch: u64) -> bool {
@@ -2305,8 +2457,15 @@ impl DaemonPane {
        }
    }

+    /// The controller's resize; see [`controller_resize`] for what a lease
+    /// makes of it.
    pub fn resize(&self, size: WinSize) {
-        resize_state(&mut self.state.lock().unwrap(), size);
+        if let Some(size) = controller_resize(&mut self.state.lock().unwrap(), size) {
+            self.resize_pty(size);
+        }
+    }
+
+    fn resize_pty(&self, size: WinSize) {
        match &self.backend {
            PaneBackend::Pty(p) => {
                if let Ok(master) = p.master.lock() {
@@ -2707,6 +2866,11 @@ impl ReplayRing {
        self.segments.back_mut().expect("ring always has a tail")
    }

+    /// The size the pane is at now: the newest segment's.
+    fn size(&self) -> WinSize {
+        self.segments.back().expect("ring always has a tail").size
+    }
+
    fn resize(&mut self, size: WinSize) {
        let tail = self.tail();
        if tail.size == size {
@@ -2928,6 +3092,7 @@ fn attach_subscriber_with_permissions(
    foreground_command: bool,
 ) -> u64 {
    st.subscriber_epoch += 1;
+    st.lease_watch = false;
    set_clipboard_permission(st, allow_remote_clipboard_write);
    replay_state(st, &subscriber, foreground_command);
    st.subscriber = Some(subscriber);
@@ -5160,6 +5325,8 @@ mod tests {
            clipboard_write_from_spec: None,
            observers: Vec::new(),
            observer_seq: 0,
+            lease: None,
+            lease_watch: false,
            shell_spec: None,
            cwd: None,
            osc_title: None,
@@ -5809,6 +5976,139 @@ mod tests {
        got
    }

+    /// A controller that watches leases, and an observer: what each hears.
+    fn leased_rig() -> (
+        PaneState,
+        mpsc::Receiver<DaemonMsg>,
+        u64,
+        mpsc::Receiver<DaemonMsg>,
+    ) {
+        let mut st = test_state(true);
+        let (controller_tx, controller_rx) = mpsc::channel();
+        attach_subscriber(&mut st, controller_tx);
+        st.lease_watch = true;
+        let (observer_tx, observer_rx) = mpsc::channel();
+        let phone = observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()), false);
+        drain(&controller_rx);
+        drain(&observer_rx);
+        (st, controller_rx, phone, observer_rx)
+    }
+
+    #[test]
+    fn a_lease_resizes_for_observers_and_only_names_the_holder_to_the_controller() {
+        let (mut st, controller_rx, phone, observer_rx) = leased_rig();
+
+        let pty = lease_take(&mut st, phone, ws(40, 30), "phone".into());
+        assert_eq!(pty, Some(ws(40, 30)));
+        assert_eq!(st.ring.size(), ws(40, 30), "a replay shows the leased size");
+        assert_eq!(
+            drain(&observer_rx),
+            vec![
+                DaemonMsg::Size(ws(40, 30)),
+                DaemonMsg::Lease(Some("phone".into()))
+            ]
+        );
+        assert_eq!(
+            drain(&controller_rx),
+            vec![DaemonMsg::Lease(Some("phone".into()))],
+            "the controller keeps its grid: no Size, or it would resize back"
+        );
+    }
+
+    #[test]
+    fn the_desk_resizing_under_a_lease_is_kept_for_later() {
+        let (mut st, controller_rx, phone, observer_rx) = leased_rig();
+        lease_take(&mut st, phone, ws(40, 30), "phone".into());
+        drain(&controller_rx);
+        drain(&observer_rx);
+
+        assert_eq!(
+            controller_resize(&mut st, ws(120, 40)),
+            None,
+            "the pty stays put"
+        );
+        assert_eq!(st.ring.size(), ws(40, 30));
+        assert_eq!(drain(&controller_rx), vec![DaemonMsg::Size(ws(120, 40))]);
+        assert!(drain(&observer_rx).is_empty());
+
+        // Letting go restores what the desk asked for last, not what it had.
+        assert_eq!(lease_release(&mut st, phone), Some(ws(120, 40)));
+        assert_eq!(st.ring.size(), ws(120, 40));
+        assert_eq!(
+            drain(&observer_rx),
+            vec![DaemonMsg::Lease(None), DaemonMsg::Size(ws(120, 40))]
+        );
+        assert_eq!(
+            drain(&controller_rx),
+            vec![DaemonMsg::Lease(None), DaemonMsg::Size(ws(120, 40))]
+        );
+        assert!(st.lease.is_none());
+    }
+
+    #[test]
+    fn taking_back_ends_the_lease_and_restores_the_desk() {
+        let (mut st, controller_rx, phone, observer_rx) = leased_rig();
+        lease_take(&mut st, phone, ws(40, 30), "phone".into());
+        drain(&controller_rx);
+        drain(&observer_rx);
+
+        assert_eq!(lease_end(&mut st), Some(ws(80, 24)));
+        assert_eq!(st.ring.size(), ws(80, 24));
+        assert_eq!(
+            drain(&observer_rx),
+            vec![DaemonMsg::Lease(None), DaemonMsg::Size(ws(80, 24))]
+        );
+        assert_eq!(lease_end(&mut st), None, "nothing left to end");
+        // The observer that lost it cannot end a lease it no longer holds.
+        assert_eq!(lease_release(&mut st, phone), None);
+    }
+
+    #[test]
+    fn a_second_observer_takes_over_and_the_desk_size_carries() {
+        let (mut st, controller_rx, phone, observer_rx) = leased_rig();
+        let (tablet_tx, tablet_rx) = mpsc::channel();
+        let tablet = observe_subscriber(&mut st, tablet_tx, Arc::new(OutputGate::new()), false);
+        drain(&tablet_rx);
+
+        lease_take(&mut st, phone, ws(40, 30), "phone".into());
+        lease_take(&mut st, tablet, ws(100, 50), "tablet".into());
+        drain(&controller_rx);
+        let heard = drain(&observer_rx);
+        assert_eq!(heard.last(), Some(&DaemonMsg::Size(ws(100, 50))));
+        assert!(
+            heard.contains(&DaemonMsg::Lease(None)),
+            "the phone is told it lost it"
+        );
+
+        assert_eq!(lease_release(&mut st, phone), None);
+        assert_eq!(lease_release(&mut st, tablet), Some(ws(80, 24)));
+    }
+
+    #[test]
+    fn a_controller_that_did_not_ask_hears_no_lease() {
+        let (mut st, controller_rx, phone, _observer_rx) = leased_rig();
+        st.lease_watch = false;
+        lease_take(&mut st, phone, ws(40, 30), "phone".into());
+        assert!(drain(&controller_rx).is_empty());
+        lease_end(&mut st);
+        assert_eq!(drain(&controller_rx), vec![DaemonMsg::Size(ws(80, 24))]);
+    }
+
+    #[test]
+    fn a_new_controller_must_ask_again() {
+        let (mut st, _controller_rx, _phone, _observer_rx) = leased_rig();
+        let (tx, _rx) = mpsc::channel();
+        attach_subscriber(&mut st, tx);
+        assert!(!st.lease_watch);
+    }
+
+    #[test]
+    fn an_unknown_observer_cannot_take_a_lease() {
+        let (mut st, _c, _phone, _o) = leased_rig();
+        assert_eq!(lease_take(&mut st, 999, ws(40, 30), "ghost".into()), None);
+        assert!(st.lease.is_none());
+    }
+
    #[test]
    fn observe_replays_state_without_displacing_the_controller() {
        let mut st = test_state(true);
@@ -39,6 +39,29 @@ pub const FEATURE_HANDOFF: &str = "handoff";
 /// and drops the route unanswered, so the client checks first and says why.
 pub const FEATURE_UPDATE_SERVER: &str = "update-server";

+/// The daemon understands [`ClientMsg::Lease`]: an observer can show a pane at
+/// its own size until it lets go or the controller takes it back. A client
+/// must see this before sending one; an older daemon cannot decode the frame
+/// and drops the connection.
+pub const FEATURE_SIZE_LEASE: &str = "size-lease";
+
+/// What a [`ClientMsg::Lease`] asks. The first two are an observer's, the last
+/// two the controller's; each side's are ignored from the other.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum LeaseRequest {
+    /// Put the pty at `size` while this observer is connected. `by` names
+    /// it to the controller ("Thomas's iPhone"). Sent again to change size.
+    Take { size: WinSize, by: String },
+    /// Put the pane back at the size the controller asked for.
+    Release,
+    /// The controller can hear [`DaemonMsg::Lease`]; tell it now and on
+    /// every change.
+    Watch,
+    /// The controller ends the lease, whoever holds it.
+    TakeBack,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct DaemonVersion {
    pub protocol: u32,
@@ -57,6 +80,7 @@ impl DaemonVersion {
            FEATURE_RESIZE_ECHO.to_string(),
            FEATURE_RESTORE_SCROLLBACK.to_string(),
            FEATURE_UPDATE_SERVER.to_string(),
+            FEATURE_SIZE_LEASE.to_string(),
        ];
        if cfg!(unix) {
            features.push(FEATURE_HANDOFF.to_string());
@@ -932,6 +956,8 @@ pub enum ClientMsg {
    },
    OnWorkspace(Box<WorkspaceRequest>),
    Version,
+    /// Only after the daemon advertised [`FEATURE_SIZE_LEASE`].
+    Lease(LeaseRequest),
 }

 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -992,6 +1018,11 @@ pub enum DaemonMsg {
    ForwardList(Vec<ManagedForward>),
    Procs(PaneProcs),
    Version(DaemonVersion),
+    /// Who holds the pane at their own size, or `None` once nobody does. Sent
+    /// to the holding observer and to a controller that asked with
+    /// [`LeaseRequest::Watch`], never to anyone else: older clients cannot
+    /// decode it.
+    Lease(Option<String>),
    Error(String),
 }

@@ -1027,6 +1058,7 @@ mod kind {
    pub const OBSERVE: u8 = 54;
    pub const SEND_INPUT: u8 = 55;
    pub const HANDOFF: u8 = 56;
+    pub const LEASE: u8 = 57;

    pub const SPAWNED: u8 = 1;
    pub const SNAPSHOT: u8 = 2;
@@ -1061,6 +1093,7 @@ mod kind {
    /// (issue #213). Compact binary selector, like `IMAGE`.
    pub const DELETE_IMAGE: u8 = 61;
    pub const CLIPBOARD_WRITE: u8 = 62;
+    pub const LEASE_HOLDER: u8 = 63;
 }

 pub fn write_frame<W: Write>(w: &mut W, kind: u8, payload: &[u8]) -> io::Result<()> {
@@ -1256,6 +1289,7 @@ impl ClientMsg {
            ClientMsg::List => write_frame(w, kind::LIST, &[]),
            ClientMsg::Shutdown => write_frame(w, kind::SHUTDOWN, &[]),
            ClientMsg::Handoff { exe } => write_frame(w, kind::HANDOFF, &to_json(exe)?),
+            ClientMsg::Lease(request) => write_frame(w, kind::LEASE, &to_json(request)?),
            ClientMsg::EnsureLoopbackForward(req) => {
                write_frame(w, kind::ENSURE_LOOPBACK_FORWARD, &to_json(req)?)
            }
@@ -1383,6 +1417,7 @@ impl ClientMsg {
            },
            kind::LIST => ClientMsg::List,
            kind::SHUTDOWN => ClientMsg::Shutdown,
+            kind::LEASE => ClientMsg::Lease(from_json(&payload)?),
            kind::HANDOFF => ClientMsg::Handoff {
                exe: from_json(&payload)?,
            },
@@ -1470,6 +1505,7 @@ impl DaemonMsg {
            DaemonMsg::Image(frame) => write_frame(w, kind::IMAGE, frame),
            DaemonMsg::DeleteImage(sel) => write_frame(w, kind::DELETE_IMAGE, sel),
            DaemonMsg::ClipboardWrite(frame) => write_frame(w, kind::CLIPBOARD_WRITE, frame),
+            DaemonMsg::Lease(by) => write_frame(w, kind::LEASE_HOLDER, &to_json(by)?),
            DaemonMsg::Cwd(path) => write_frame(w, kind::CWD, &to_json(path)?),
            DaemonMsg::Prompt {
                active,
@@ -1527,6 +1563,7 @@ impl DaemonMsg {
            kind::IMAGE => DaemonMsg::Image(payload),
            kind::DELETE_IMAGE => DaemonMsg::DeleteImage(payload),
            kind::CLIPBOARD_WRITE => DaemonMsg::ClipboardWrite(payload),
+            kind::LEASE_HOLDER => DaemonMsg::Lease(from_json(&payload)?),
            kind::CWD => DaemonMsg::Cwd(from_json(&payload)?),
            kind::PROMPT => {
                let (active, at_prompt, last_exit) = from_json(&payload)?;
@@ -1659,6 +1696,13 @@ mod tests {
    #[test]
    fn client_roundtrip() {
        let msgs = vec![
+            ClientMsg::Lease(LeaseRequest::Take {
+                size: SIZE,
+                by: "phone".into(),
+            }),
+            ClientMsg::Lease(LeaseRequest::Release),
+            ClientMsg::Lease(LeaseRequest::Watch),
+            ClientMsg::Lease(LeaseRequest::TakeBack),
            ClientMsg::Spawn {
                cwd: Some(PathBuf::from("/tmp/x")),
                size: SIZE,
@@ -1849,6 +1893,8 @@ mod tests {
    fn daemon_roundtrip() {
        let msgs = vec![
            DaemonMsg::Spawned { pane_id: 1 },
+            DaemonMsg::Lease(Some("Thomas's iPhone".into())),
+            DaemonMsg::Lease(None),
            DaemonMsg::Size(SIZE),
            DaemonMsg::Snapshot(vec![1, 2, 3, 0, 255]),
            DaemonMsg::Output((0u8..=255).collect()),
@@ -5,7 +5,7 @@ use std::sync::mpsc::{self, Receiver};
 use std::sync::{Arc, Mutex};

 use crate::daemon::pane::DaemonPane;
-use crate::daemon::protocol::{ClientMsg, DaemonMsg, DaemonVersion, RemoteKind};
+use crate::daemon::protocol::{ClientMsg, DaemonMsg, DaemonVersion, LeaseRequest, RemoteKind};
 use crate::daemon::ssh::SshConnection;
 use crate::daemon::transport::{self, Stream};

@@ -1180,7 +1180,9 @@ fn stream_observer(
    let observer_id = pane.observe(tx, gate.clone());
    let writer = spawn_writer(rx, write_stream, gate);

-    observe_loop(&mut read_stream, &refusals);
+    observe_loop(&mut read_stream, &refusals, |request| {
+        pane.observer_lease(observer_id, request)
+    });

    pane.unobserve(observer_id);
    drop(refusals);
@@ -1188,9 +1190,14 @@ fn stream_observer(
    Ok(())
 }

-fn observe_loop<R: std::io::Read>(read_stream: &mut R, refusals: &mpsc::Sender<DaemonMsg>) {
+fn observe_loop<R: std::io::Read>(
+    read_stream: &mut R,
+    refusals: &mpsc::Sender<DaemonMsg>,
+    mut lease: impl FnMut(LeaseRequest),
+) {
    loop {
        match ClientMsg::read(read_stream) {
+            Ok(ClientMsg::Lease(request)) => lease(request),
            Ok(ClientMsg::Input(_)) | Ok(ClientMsg::Resize(_)) => {
                let refused = refusals.send(DaemonMsg::Error(
                    "this connection is a read-only observer; attach to write".to_string(),
@@ -1248,6 +1255,12 @@ fn run_stream(
                    }
                    pane.resize(size);
                }
+                ClientMsg::Lease(request) => {
+                    if !pane.controls(epoch) {
+                        break 'conn;
+                    }
+                    pane.controller_lease(request);
+                }
                ClientMsg::AuthResponse {
                    request_id,
                    response,
@@ -1432,7 +1445,7 @@ mod tests {
            .unwrap();

        let (tx, rx) = std::sync::mpsc::channel();
-        observe_loop(&mut std::io::Cursor::new(wire), &tx);
+        observe_loop(&mut std::io::Cursor::new(wire), &tx, |_| {});
        drop(tx);

        assert!(
@@ -1452,7 +1465,7 @@ mod tests {
    #[test]
    fn the_observer_loop_ends_at_stream_eof() {
        let (tx, rx) = std::sync::mpsc::channel();
-        observe_loop(&mut std::io::Cursor::new(Vec::<u8>::new()), &tx);
+        observe_loop(&mut std::io::Cursor::new(Vec::<u8>::new()), &tx, |_| {});
        drop(tx);
        assert!(rx.try_recv().is_err());
    }
@@ -107,6 +107,7 @@ struct ReaderSignals {
    images: crate::terminal::images::ImageStore,
    clipboard_writes: Arc<Mutex<VecDeque<tty7_core::core::clipboard::ClipboardWrite>>>,
    clipboard_write_busy: Arc<AtomicBool>,
+    lease: Arc<Mutex<Option<String>>>,
    /// Whether this pane's pty is one a conhost renders into, and so whether
    /// the reader puts back the cursor a repaint parked. Decided per pane from
    /// its [`PtySource`], and shared rather than copied because the reader can
@@ -166,6 +167,10 @@ pub struct PaneWorkspace {
    /// `false` whenever the answer is unknown (link down, server too old to
    /// advertise), which keeps the safe reflow-at-request behavior.
    pub resize_echo: bool,
+    /// Whether that daemon knows size leases — a phone running the pane at
+    /// its own size — so this client can ask to hear about them. Read off the
+    /// same hello, `false` when unknown, like `resize_echo`.
+    pub size_lease: bool,
 }

 impl PaneWorkspace {
@@ -222,6 +227,8 @@ pub enum PaneRoute {
        /// answer everywhere a route is assigned — spawn, attach, relink —
        /// and a relink builds its route fresh, so a reconnect re-answers it.
        resize_echo: bool,
+        /// As [`PaneWorkspace::size_lease`].
+        size_lease: bool,
    },
    Unroutable(String),
 }
@@ -234,6 +241,7 @@ impl PaneRoute {
                Ok(header) => PaneRoute::Remote {
                    header: Box::new(header),
                    resize_echo: ws.resize_echo,
+                    size_lease: ws.size_lease,
                },
                Err(e) => PaneRoute::Unroutable(e.to_string()),
            },
@@ -615,6 +623,9 @@ pub struct RemoteTerminal {
    images: crate::terminal::images::ImageStore,
    clipboard_writes: Arc<Mutex<VecDeque<tty7_core::core::clipboard::ClipboardWrite>>>,
    clipboard_write_busy: Arc<AtomicBool>,
+    /// Who runs this pane at their own size — a phone, by the name it paired
+    /// under — while the daemon says so. See [`Self::take_back`].
+    lease: Arc<Mutex<Option<String>>>,
    route: PaneRoute,
    proxy: EventProxy,
    reader_thread: Option<JoinHandle<()>>,
@@ -788,6 +799,7 @@ impl RemoteTerminal {
            Self::from_stream_with(stream, size, Vec::new(), PtySource::for_route(route))?;
        term.route = route.clone();
        term.seed_cwd(spawned_in);
+        term.watch_leases();
        Ok((term, pane_id))
    }

@@ -858,6 +870,7 @@ impl RemoteTerminal {
        let mut term =
            Self::from_stream_parts(stream, size, buffered, PtySource::for_route(route), true)?;
        term.route = route.clone();
+        term.watch_leases();
        Ok(term)
    }

@@ -953,6 +966,7 @@ impl RemoteTerminal {
                images: self.images.clone(),
                clipboard_writes: self.clipboard_writes.clone(),
                clipboard_write_busy: self.clipboard_write_busy.clone(),
+                lease: self.lease.clone(),
                // Deliberately the pane's existing answer rather than one
                // rebuilt from `route`: the pty on the far side is the same pty
                // it was before the link dropped, and only this value still
@@ -967,6 +981,12 @@ impl RemoteTerminal {
        // `LinkWriter` through its `Drop`, which is what closes the old socket.
        self.link = LinkWriter::new(stream, self.input_loss())?;
        self.route = route.clone();
+        // The new link has told us nothing yet; the daemon repeats a lease
+        // that still stands when we ask again.
+        if let Ok(mut guard) = self.lease.lock() {
+            *guard = None;
+        }
+        self.watch_leases();
        self.synced_size = false;
        self.resize(size, cell_w, cell_h);
        Ok(())
@@ -1049,6 +1069,7 @@ impl RemoteTerminal {
        let images = crate::terminal::images::ImageStore::new();
        let clipboard_writes = Arc::new(Mutex::new(VecDeque::new()));
        let clipboard_write_busy = Arc::new(AtomicBool::new(false));
+        let lease = Arc::new(Mutex::new(None));

        let reader_quit = Arc::new(AtomicBool::new(false));
        let local_conpty = Arc::new(AtomicBool::new(pty == PtySource::LocalConpty));
@@ -1075,6 +1096,7 @@ impl RemoteTerminal {
                images: images.clone(),
                clipboard_writes: clipboard_writes.clone(),
                clipboard_write_busy: clipboard_write_busy.clone(),
+                lease: lease.clone(),
                local_conpty: local_conpty.clone(),
            },
        );
@@ -1113,6 +1135,7 @@ impl RemoteTerminal {
            images,
            clipboard_writes,
            clipboard_write_busy,
+            lease,
            route: PaneRoute::Local,
            proxy,
            reader_thread: Some(reader_thread),
@@ -1188,6 +1211,7 @@ impl RemoteTerminal {
                    images,
                    clipboard_writes,
                    clipboard_write_busy,
+                    lease,
                    local_conpty,
                } = signals;
                let mut awaiting_replay = awaiting_replay;
@@ -1689,6 +1713,12 @@ impl RemoteTerminal {
                                }
                                proxy.send_event(AlacEvent::Wakeup);
                            }
+                            DaemonMsg::Lease(by) => {
+                                if let Ok(mut guard) = lease.lock() {
+                                    *guard = by;
+                                }
+                                proxy.send_event(AlacEvent::Wakeup);
+                            }
                            DaemonMsg::Exited { .. } => {
                                flush_batch!();
                                child_exited.store(true, Ordering::SeqCst);
@@ -1824,6 +1854,41 @@ impl RemoteTerminal {
        }
    }

+    /// Whether the daemon behind this pane knows size leases, asked the way
+    /// [`Self::resize_echoed`] asks: never over the network from here.
+    fn leases_supported(&self) -> bool {
+        match &self.route {
+            PaneRoute::Local => crate::daemon::spawn::local_daemon_supports(
+                crate::daemon::protocol::FEATURE_SIZE_LEASE,
+            ),
+            PaneRoute::Remote { size_lease, .. } => *size_lease,
+            PaneRoute::Unroutable(_) => false,
+        }
+    }
+
+    /// Asks to be told when a phone runs this pane at its own size. Only
+    /// where the daemon knows leases: an older one drops a link that sends it.
+    fn watch_leases(&self) {
+        if self.leases_supported() {
+            self.link.send(ClientMsg::Lease(
+                crate::daemon::protocol::LeaseRequest::Watch,
+            ));
+        }
+    }
+
+    /// Who runs this pane at their own size right now, if anyone.
+    pub fn leased_by(&self) -> Option<String> {
+        self.lease.lock().ok().and_then(|guard| guard.clone())
+    }
+
+    /// Ends a phone's lease: the pane goes back to this window's size. The
+    /// daemon confirms with the lease ending, which clears [`Self::leased_by`].
+    pub fn take_back(&self) {
+        self.link.send(ClientMsg::Lease(
+            crate::daemon::protocol::LeaseRequest::TakeBack,
+        ));
+    }
+
    pub fn resize(&mut self, size: TermSize, cell_w: u16, cell_h: u16) {
        let echoed = self.resize_echoed();
        // The cell size has to be part of the early-out, not just cols/rows: it
@@ -4002,6 +4067,7 @@ mod replay_tests {
        PaneRoute::Remote {
            header: Box::new(crate::daemon::router::RouteHeader::wsl("Ubuntu-22.04")),
            resize_echo: true,
+            size_lease: false,
        }
    }

@@ -4435,6 +4501,7 @@ mod route_header_tests {
            spec: None,
            label: label.map(str::to_string),
            resize_echo: false,
+            size_lease: false,
        }
    }

@@ -4559,6 +4626,7 @@ mod parked_cursor_tests {
            let route = PaneRoute::Remote {
                header: Box::new(header),
                resize_echo: false,
+                size_lease: false,
            };
            assert_eq!(PtySource::for_route(&route), PtySource::Raw);
        }
@@ -4798,6 +4866,7 @@ mod tests {
            )),
            label: None,
            resize_echo: false,
+            size_lease: false,
        }
    }

@@ -4827,6 +4896,7 @@ mod tests {
                PaneRoute::for_workspace(Some(&ws)),
                PaneRoute::Remote {
                    resize_echo: true,
+                    size_lease: false,
                    ..
                }
            ),
@@ -4844,6 +4914,7 @@ mod tests {
            spec: None,
            label: None,
            resize_echo: false,
+            size_lease: false,
        };
        let route = PaneRoute::for_workspace(Some(&ws));
        let header = route.header().expect("WSL is routed");
@@ -4862,6 +4933,7 @@ mod tests {
            spec: None,
            label: None,
            resize_echo: false,
+            size_lease: false,
        };
        let route = PaneRoute::for_workspace(Some(&ws));
        let header = route.header().expect("a local child is routable");
@@ -4885,6 +4957,7 @@ mod tests {
            spec: None,
            label: None,
            resize_echo: false,
+            size_lease: false,
        };
        let route = PaneRoute::for_workspace(Some(&ws));
        assert!(matches!(route, PaneRoute::Unroutable(_)));
@@ -6042,6 +6115,7 @@ mod tests {
        term.route = PaneRoute::Remote {
            header: Box::new(crate::daemon::router::RouteHeader::wsl("Ubuntu-22.04")),
            resize_echo: true,
+            size_lease: false,
        };

        term.resize(TermSize::new(120, 30), 8, 17);
@@ -6106,6 +6180,7 @@ mod tests {
        term.route = PaneRoute::Remote {
            header: Box::new(crate::daemon::router::RouteHeader::wsl("Ubuntu-22.04")),
            resize_echo: true,
+            size_lease: false,
        };
        daemon_side
            .set_read_timeout(Some(std::time::Duration::from_millis(150)))
@@ -6277,6 +6352,7 @@ mod tests {
        term.route = PaneRoute::Remote {
            header: Box::new(crate::daemon::router::RouteHeader::wsl("Ubuntu-22.04")),
            resize_echo: false,
+            size_lease: false,
        };

        term.resize(TermSize::new(120, 30), 8, 17);
@@ -8552,6 +8552,7 @@ mod tests {
            }),
            label: None,
            resize_echo: false,
+            size_lease: false,
        }
    }

@@ -10200,6 +10201,7 @@ mod tests {
            spec: None,
            label: None,
            resize_echo: false,
+            size_lease: false,
        };

        let remote = ws.target.host_id();
@@ -10217,6 +10219,7 @@ mod tests {
            spec: None,
            label: None,
            resize_echo: false,
+            size_lease: false,
        };
        assert_eq!(sibling.target.host_id(), remote);
    }
@@ -14448,6 +14451,7 @@ mod gpui_tests {
            )),
            label: None,
            resize_echo: false,
+            size_lease: false,
        }));
        id
    }
@@ -14701,6 +14705,7 @@ mod gpui_tests {
                    spec: None,
                    label: Some("hummingbot".into()),
                    resize_echo: false,
+                    size_lease: false,
                }));
                assert_eq!(view.title, "hummingbot", "an untitled tab shows the name");
                view.handle_event(AlacEvent::Exit, cx);
@@ -8281,6 +8281,51 @@ impl Tty7App {
        )
    }

+    /// One pill per pane in the active tab that a phone is running at its own
+    /// size. The window keeps its grid meanwhile and shows the pane's output
+    /// laid out for the phone, so it says why, and offers the pane back.
+    fn render_lease_notices(&self, cx: &mut Context<Self>) -> Vec<gpui::AnyElement> {
+        use gpui_component::Sizable as _;
+        use gpui_component::button::ButtonVariants as _;
+        let Some(tab) = self.tabs.get(self.active) else {
+            return Vec::new();
+        };
+        let theme = cx.theme();
+        let (info, foreground, popover) = (theme.info, theme.foreground, theme.popover);
+        tab.pane
+            .terminals()
+            .into_iter()
+            .filter_map(|leaf| {
+                let by = leaf.read(cx).terminal.leased_by()?;
+                let id = leaf.entity_id().as_u64();
+                Some(
+                    crate::ui::notice::pill(info, cx)
+                        .child(
+                            div()
+                                .font_weight(gpui::FontWeight::MEDIUM)
+                                .text_color(foreground)
+                                .child(crate::ui::i18n::t_fmt(
+                                    crate::ui::i18n::L10nKey::PaneLeasedBy,
+                                    &[("by", &by)],
+                                )),
+                        )
+                        .child(
+                            gpui_component::button::Button::new(gpui::SharedString::from(format!(
+                                "lease-take-back-{id}"
+                            )))
+                            .label(crate::ui::i18n::t(
+                                crate::ui::i18n::L10nKey::RemoteActionTakeBack,
+                            ))
+                            .custom(crate::ui::theme::inverted_button(popover, cx))
+                            .small()
+                            .on_click(move |_, _, cx| leaf.read(cx).terminal.take_back()),
+                        )
+                        .into_any_element(),
+                )
+            })
+            .collect()
+    }
+
    fn render_remote_input_notice(&self, cx: &mut Context<Self>) -> Option<gpui::AnyElement> {
        if self.tabs.is_empty() {
            return None;
@@ -8658,6 +8703,7 @@ impl Render for Tty7App {
                    [self.render_remote_input_notice(cx), ssh_status]
                        .into_iter()
                        .flatten()
+                        .chain(self.render_lease_notices(cx))
                        .collect(),
                ),
                |this, el| this.child(el),
@@ -1399,6 +1399,7 @@ pub fn translate_en(key: L10nKey) -> &'static str {
        L10nKey::RemoteNoticeDisconnected => "Not connected — typing has no effect",
        L10nKey::RemoteActionRetryNow => "Retry Now",
        L10nKey::RemoteActionTakeBack => "Take Back",
+        L10nKey::PaneLeasedBy => "In use on {by}, at its screen size",
        L10nKey::RemoteActionConnect => "Connect",
        L10nKey::RemoteActionRetry => "Retry",
        L10nKey::RemoteActionRemoveEntry => "Remove entry",
@@ -1455,6 +1455,7 @@ pub fn translate_ja(key: L10nKey) -> Option<&'static str> {
        L10nKey::RemoteNoticeDisconnected => "未接続です — 入力しても反映されません",
        L10nKey::RemoteActionRetryNow => "今すぐ再試行",
        L10nKey::RemoteActionTakeBack => "取り戻す",
+        L10nKey::PaneLeasedBy => "{by} で使用中（その画面サイズで表示）",
        L10nKey::RemoteActionConnect => "接続",
        L10nKey::RemoteActionRetry => "再試行",
        L10nKey::RemoteActionRemoveEntry => "エントリを削除",
@@ -1211,6 +1211,7 @@ l10n_keys! {
    RemoteNoticeDisconnected,
    RemoteActionRetryNow,
    RemoteActionTakeBack,
+    PaneLeasedBy,
    RemoteActionConnect,
    RemoteActionRetry,
    RemoteActionRemoveEntry,
@@ -1314,6 +1314,7 @@ pub fn translate_zh(key: L10nKey) -> Option<&'static str> {
        L10nKey::RemoteNoticeDisconnected => "未连接——输入无效",
        L10nKey::RemoteActionRetryNow => "立即重试",
        L10nKey::RemoteActionTakeBack => "收回",
+        L10nKey::PaneLeasedBy => "正在 {by} 上使用，按它的屏幕尺寸显示",
        L10nKey::RemoteActionConnect => "连接",
        L10nKey::RemoteActionRetry => "重试",
        L10nKey::RemoteActionRemoveEntry => "移除条目",
@@ -1226,6 +1226,11 @@ pub(crate) fn pane_workspace_for(
        host.host_id(),
        crate::daemon::protocol::FEATURE_RESIZE_ECHO,
    );
+    let size_lease = remote_connect::HostLinks::peer_supports(
+        cx,
+        host.host_id(),
+        crate::daemon::protocol::FEATURE_SIZE_LEASE,
+    );
    // The workspace's own name where it has one, the machine's otherwise —
    // what the pane's tab falls back to when nothing has titled it, and what
    // a dead link's "— disconnected" suffix hangs off.
@@ -1241,6 +1246,7 @@ pub(crate) fn pane_workspace_for(
        spec,
        label,
        resize_echo,
+        size_lease,
    };
    if let Ok(header) = pane.route_header() {
        remote_connect::note_origin(&header.target, &pane.target);