From b9515b81330b91892ad3eab447b92ffc5a9e1fec Mon Sep 17 00:00:00 2001
From: Azer Sadykhzadeh <azer.piano14@gmail.com>
Date: Fri, 16 Oct 2026 23:02:01 +0000
Subject: [PATCH 1/4] Stop reading SSH output the webview has not caught up
 with

The reader handed every batch to the IPC channel without waiting, and
large payloads sit in Tauri's in-memory queue until the webview fetches
them, so a fast or hostile server could grow that queue without bound.

Each SSH session now meters delivered output against a 2 MiB window of
unacknowledged bytes. The webview acknowledges what it receives through
session_output_ack, coalesced to every 256 KiB or 30 ms, and the reader
stops polling the channel while the window is spent. russh hands channel
data over through a bounded queue, so the stall reaches the server via
the SSH window and TCP. Closing a session lifts the limit so the reader
still reaches EOF, flushes the tail and announces the close.
---
 core/src/output.rs                      | 189 ++++++++++++++++++++++
 core/src/session.rs                     |  40 ++++-
 core/tests/output_flow_control.rs       | 200 ++++++++++++++++++++++++
 desktop/e2e/tauri-fixture.js            |   1 +
 desktop/src-tauri/src/commands.rs       |  17 ++
 desktop/src-tauri/src/main.rs           |   1 +
 desktop/src/api.sessionOutput.test.ts   |  96 ++++++++++++
 desktop/src/api.ts                      |  55 ++++++-
 desktop/src/components/TerminalPane.vue |   2 +-
 9 files changed, 593 insertions(+), 8 deletions(-)
 create mode 100644 core/tests/output_flow_control.rs
 create mode 100644 desktop/src/api.sessionOutput.test.ts

diff --git a/core/src/output.rs b/core/src/output.rs
index 804617c3..26fa4e19 100644
--- a/core/src/output.rs
+++ b/core/src/output.rs
@@ -4,8 +4,14 @@
 //! to a full 32 KiB window. Delivering every chunk separately makes the fixed
 //! per-message cost of the UI transport dominate bulk output, so chunks are
 //! accumulated into batches instead.
+//!
+//! Delivered batches are also metered: [`OutputCredit`] counts the bytes the UI
+//! layer has not yet acknowledged, so a reader can stop pulling from the
+//! connection while the UI is behind.
 
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::{Duration, Instant};
+use tokio::sync::Notify;
 
 /// Longest a partially filled batch waits before delivery. Bounds the delay
 /// added to output that arrives while a batch is already in flight.
@@ -114,6 +120,98 @@ impl OutputBatcher {
     }
 }
 
+/// Unacknowledged output a session may have in flight before its reader stops
+/// pulling from the connection.
+///
+/// Large enough for many full [`MAX_BATCH`] batches to be on their way at once,
+/// so bulk output is not throttled to one round trip to the UI layer and back
+/// per batch; small enough that a UI that stops consuming, or a peer that
+/// produces faster than the UI can, holds a few megabytes per session rather
+/// than whatever the peer chooses to send.
+pub const OUTPUT_WINDOW: u64 = 2 * 1024 * 1024;
+
+/// Count of delivered output bytes the UI layer has not acknowledged yet.
+///
+/// The reader spends what it delivers and stops reading the connection while
+/// the count is at or above [`OUTPUT_WINDOW`]; the UI refunds what it has
+/// received. Pausing the reader is what bounds memory: the SSH library hands
+/// channel data over through a bounded queue and waits when it is full, so an
+/// idle reader stalls the connection and, through the SSH window and TCP, the
+/// peer itself.
+///
+/// The window is a threshold rather than a hard cap: a batch already due is
+/// still delivered, so the count can run past the window by up to one batch
+/// and one protocol packet.
+pub struct OutputCredit {
+    outstanding: AtomicU64,
+    /// Set once the session is closing, after which output is never held back.
+    lifted: AtomicBool,
+    room: Notify,
+}
+
+impl OutputCredit {
+    pub fn new() -> Self {
+        Self {
+            outstanding: AtomicU64::new(0),
+            lifted: AtomicBool::new(false),
+            room: Notify::new(),
+        }
+    }
+
+    /// Records `bytes` delivered to the UI layer.
+    pub fn spend(&self, bytes: usize) {
+        self.outstanding.fetch_add(bytes as u64, Ordering::AcqRel);
+    }
+
+    /// Records `bytes` the UI layer has received.
+    ///
+    /// Saturates at zero: the amount comes from the UI layer, and acknowledging
+    /// more than was delivered must not bank credit for output not yet sent.
+    pub fn refund(&self, bytes: u64) {
+        let _ = self
+            .outstanding
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |outstanding| {
+                Some(outstanding.saturating_sub(bytes))
+            });
+        self.room.notify_one();
+    }
+
+    /// Stops holding output back for good.
+    ///
+    /// A closing session must still reach the end of its stream, which is
+    /// where the reader flushes the trailing output and announces the close.
+    /// Nothing is obliged to keep acknowledging by then, so a reader left
+    /// waiting for credit might never get there.
+    pub fn lift(&self) {
+        self.lifted.store(true, Ordering::Release);
+        self.room.notify_one();
+    }
+
+    /// Whether the reader may pull more output from the connection.
+    pub fn has_room(&self) -> bool {
+        self.lifted.load(Ordering::Acquire)
+            || self.outstanding.load(Ordering::Acquire) < OUTPUT_WINDOW
+    }
+
+    /// Resolves once [`has_room`](Self::has_room) holds.
+    ///
+    /// Meant for a single waiter, the session's reader. `refund` and `lift`
+    /// wake it with `notify_one`, which leaves a permit behind when nobody is
+    /// waiting yet, so a refund that lands between the check and the wait is
+    /// not lost: the wait completes at once and the loop checks again.
+    pub async fn room(&self) {
+        while !self.has_room() {
+            self.room.notified().await;
+        }
+    }
+}
+
+impl Default for OutputCredit {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -315,4 +413,95 @@ mod tests {
         assert!(batcher.is_empty());
         assert_eq!(batcher.deadline(), None);
     }
+
+    #[test]
+    fn credit_runs_out_at_the_window_and_returns_with_a_refund() {
+        let credit = OutputCredit::new();
+        assert!(credit.has_room());
+        credit.spend(OUTPUT_WINDOW as usize - 1);
+        assert!(credit.has_room());
+        credit.spend(1);
+        assert!(!credit.has_room());
+        credit.refund(1);
+        assert!(credit.has_room());
+    }
+
+    #[test]
+    fn an_over_acknowledgement_banks_no_credit() {
+        let credit = OutputCredit::new();
+        credit.spend(MAX_BATCH);
+        credit.refund(u64::MAX);
+        // Had the excess carried over, this would still leave room.
+        credit.spend(OUTPUT_WINDOW as usize);
+        assert!(!credit.has_room());
+    }
+
+    #[test]
+    fn lifting_the_limit_leaves_room_whatever_is_outstanding() {
+        let credit = OutputCredit::new();
+        credit.spend(OUTPUT_WINDOW as usize * 4);
+        assert!(!credit.has_room());
+        credit.lift();
+        assert!(credit.has_room());
+        credit.spend(OUTPUT_WINDOW as usize);
+        assert!(credit.has_room());
+    }
+
+    #[tokio::test]
+    async fn waiting_for_room_returns_at_once_when_there_is_some() {
+        let credit = OutputCredit::new();
+        tokio::time::timeout(Duration::from_secs(1), credit.room())
+            .await
+            .expect("room was available");
+    }
+
+    #[tokio::test]
+    async fn a_refund_wakes_a_waiting_reader() {
+        let credit = std::sync::Arc::new(OutputCredit::new());
+        credit.spend(OUTPUT_WINDOW as usize);
+        let waiter = tokio::spawn({
+            let credit = credit.clone();
+            async move { credit.room().await }
+        });
+        tokio::task::yield_now().await;
+        assert!(!waiter.is_finished());
+
+        // A refund that leaves the count at the window wakes the reader, which
+        // must check again and keep waiting.
+        credit.refund(0);
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        assert!(!waiter.is_finished());
+
+        credit.refund(MAX_BATCH as u64);
+        tokio::time::timeout(Duration::from_secs(1), waiter)
+            .await
+            .expect("the refund did not wake the reader")
+            .expect("waiter");
+    }
+
+    #[tokio::test]
+    async fn a_refund_before_the_wait_starts_is_not_missed() {
+        let credit = OutputCredit::new();
+        credit.spend(OUTPUT_WINDOW as usize);
+        credit.refund(1);
+        tokio::time::timeout(Duration::from_secs(1), credit.room())
+            .await
+            .expect("the earlier refund was lost");
+    }
+
+    #[tokio::test]
+    async fn lifting_the_limit_wakes_a_waiting_reader() {
+        let credit = std::sync::Arc::new(OutputCredit::new());
+        credit.spend(OUTPUT_WINDOW as usize);
+        let waiter = tokio::spawn({
+            let credit = credit.clone();
+            async move { credit.room().await }
+        });
+        tokio::task::yield_now().await;
+        credit.lift();
+        tokio::time::timeout(Duration::from_secs(1), waiter)
+            .await
+            .expect("lifting did not wake the reader")
+            .expect("waiter");
+    }
 }
diff --git a/core/src/session.rs b/core/src/session.rs
index 9c776cda..dbcca45f 100644
--- a/core/src/session.rs
+++ b/core/src/session.rs
@@ -1,7 +1,7 @@
 use crate::connection;
 use crate::host::Host;
 use crate::known_hosts::KnownHosts;
-use crate::output::OutputBatcher;
+use crate::output::{OutputBatcher, OutputCredit};
 use crate::vault::Vault;
 use crate::{CoreError, Result};
 use russh::client::Handle;
@@ -129,6 +129,9 @@ struct SshSession {
     /// its own input.
     writes: Mutex<()>,
     resize_tx: mpsc::Sender<(u32, u32)>,
+    /// Output delivered to the UI layer and not yet acknowledged, shared with
+    /// the reader task, which stops reading the channel while it is spent.
+    credit: Arc<OutputCredit>,
     /// Held so the id stays reserved while this entry exists.
     #[allow(dead_code)]
     claim: Arc<ClaimedId>,
@@ -144,8 +147,13 @@ impl SshSession {
     /// Mark the session closed and say goodbye to the peer. The goodbye holds
     /// only the connection handle, so the id is released as soon as the map
     /// entry and the reader are gone, however long the peer takes.
+    ///
+    /// The output limit is lifted as well: a reader stalled on unacknowledged
+    /// output has to keep reading to reach the end of the channel, flush the
+    /// tail and announce the close.
     fn retire(&self) -> impl std::future::Future<Output = ()> + Send + 'static {
         self.closed.send_replace(true);
+        self.credit.lift();
         let handle = self.handle.clone();
         async move { disconnect(&handle).await }
     }
@@ -243,6 +251,7 @@ impl SessionManager {
         }
 
         let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32);
+        let credit = Arc::new(OutputCredit::new());
 
         let held = claim.commit();
         let session = Arc::new(SshSession {
@@ -251,6 +260,7 @@ impl SessionManager {
             channel_id,
             writes: Mutex::new(()),
             resize_tx,
+            credit: credit.clone(),
             claim: held.clone(),
         });
 
@@ -271,7 +281,9 @@ impl SessionManager {
             let mut batcher = OutputBatcher::new(Instant::now());
             let flush = |batcher: &mut OutputBatcher| {
                 if !batcher.is_empty() {
-                    data_cb(&sid, batcher.batch());
+                    let batch = batcher.batch();
+                    data_cb(&sid, batch);
+                    credit.spend(batch.len());
                     batcher.mark_flushed(Instant::now());
                 }
             };
@@ -279,8 +291,15 @@ impl SessionManager {
                 // Read the deadline before the select so no branch borrows the
                 // batcher while another branch's handler mutates it.
                 let flush_at = batcher.deadline();
+                // While the UI layer is a window behind, the channel is left
+                // unread, which backs the stall up to the peer. Resizes and a
+                // batch already due are still handled meanwhile; the end of the
+                // stream is seen once credit returns or the session is closed,
+                // which lifts the limit.
+                let reading = credit.has_room();
                 tokio::select! {
-                    msg = channel.wait() => {
+                    () = credit.room(), if !reading => {}
+                    msg = channel.wait(), if reading => {
                         match msg {
                             Some(ChannelMsg::Data { data })
                             | Some(ChannelMsg::ExtendedData { data, .. }) => {
@@ -373,7 +392,9 @@ impl SessionManager {
     /// The entry leaves the map before the disconnect is sent, so the session
     /// stops being addressable at once even if the peer never answers. The
     /// entry's share of the id goes with it; the reader task still holds one
-    /// until it has flushed and announced the close.
+    /// until it has flushed and announced the close. Output credit is lifted
+    /// first, so a reader waiting on the UI layer goes on to the end of the
+    /// stream instead of waiting on acknowledgements that may never come.
     pub async fn close(&self, session_id: &str) -> Result<()> {
         let removed = self.sessions.lock().await.remove(session_id);
         if let Some(session) = removed {
@@ -382,6 +403,17 @@ impl SessionManager {
         Ok(())
     }
 
+    /// Return `bytes` of output credit to a session once the UI layer has
+    /// received them, letting a reader that ran out resume.
+    ///
+    /// An id with no live session is not an error: acknowledgements trail the
+    /// output they cover and can arrive after the session has closed.
+    pub async fn ack(&self, session_id: &str, bytes: u64) {
+        if let Ok(session) = self.get(session_id).await {
+            session.credit.refund(bytes);
+        }
+    }
+
     /// List active session ids.
     pub async fn list(&self) -> Vec<String> {
         self.sessions.lock().await.keys().cloned().collect()
diff --git a/core/tests/output_flow_control.rs b/core/tests/output_flow_control.rs
new file mode 100644
index 00000000..d935248e
--- /dev/null
+++ b/core/tests/output_flow_control.rs
@@ -0,0 +1,200 @@
+//! A peer that produces output faster than the UI acknowledges it is held to
+//! one window of unacknowledged output.
+//!
+//! Runs the SSH session manager against an in-process russh server on loopback
+//! whose shell writes without pause for as long as the connection will take
+//! it, and checks that what reaches the data callback stops growing until the
+//! UI acknowledges it, that the stall reaches the server, and that a stalled
+//! session still closes.
+
+use clavyn_core::host::Host;
+use clavyn_core::known_hosts::KnownHosts;
+use clavyn_core::output::{MAX_BATCH, OUTPUT_WINDOW};
+use clavyn_core::session::SessionManager;
+use russh::keys::{Algorithm, PrivateKey};
+use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session};
+use russh::{Channel, ChannelId};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
+
+const USER: &str = "deploy";
+const PASSWORD: &str = "correct password";
+const CHUNK: usize = 64 * 1024;
+
+/// A server whose shell floods its channel, counting what it managed to hand
+/// to its connection.
+#[derive(Clone)]
+struct FloodServer {
+    produced: Arc<AtomicU64>,
+}
+
+impl server::Handler for FloodServer {
+    type Error = russh::Error;
+
+    async fn auth_password(&mut self, user: &str, password: &str) -> Result<Auth, Self::Error> {
+        Ok(if user == USER && password == PASSWORD {
+            Auth::Accept
+        } else {
+            Auth::reject()
+        })
+    }
+
+    async fn channel_open_session(
+        &mut self,
+        _channel: Channel<Msg>,
+        reply: ChannelOpenHandle,
+        _session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        reply.accept().await;
+        Ok(())
+    }
+
+    async fn shell_request(
+        &mut self,
+        channel: ChannelId,
+        session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        let handle = session.handle();
+        let produced = self.produced.clone();
+        tokio::spawn(async move {
+            while handle.data(channel, vec![b'x'; CHUNK]).await.is_ok() {
+                produced.fetch_add(CHUNK as u64, Ordering::Relaxed);
+            }
+        });
+        Ok(())
+    }
+}
+
+async fn start_server(produced: Arc<AtomicU64>) -> u16 {
+    let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng);
+    let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key");
+    let config = Arc::new(server::Config {
+        keys: vec![host_key],
+        auth_rejection_time: Duration::from_millis(10),
+        auth_rejection_time_initial: Some(Duration::ZERO),
+        ..Default::default()
+    });
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let port = listener.local_addr().expect("addr").port();
+    tokio::spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            let config = config.clone();
+            let handler = FloodServer {
+                produced: produced.clone(),
+            };
+            tokio::spawn(async move {
+                if let Ok(running) = server::run_stream(config, stream, handler).await {
+                    let _ = running.await;
+                }
+            });
+        }
+    });
+    port
+}
+
+fn host(port: u16) -> Host {
+    serde_json::from_value(serde_json::json!({
+        "id": "00000000-0000-0000-0000-000000000001",
+        "label": "Loopback",
+        "hostname": "127.0.0.1",
+        "port": port,
+        "username": USER,
+        "auth": {"password": {"credential_key": "unused"}},
+        "tags": []
+    }))
+    .expect("host")
+}
+
+/// Waits until `counter` holds still across a sampling interval and returns
+/// the value it settled on.
+async fn settled(counter: &AtomicU64) -> u64 {
+    let settle = async {
+        let mut last = counter.load(Ordering::SeqCst);
+        loop {
+            tokio::time::sleep(Duration::from_millis(400)).await;
+            let now = counter.load(Ordering::SeqCst);
+            if now == last && now > 0 {
+                return now;
+            }
+            last = now;
+        }
+    };
+    tokio::time::timeout(Duration::from_secs(20), settle)
+        .await
+        .expect("output kept growing with nothing acknowledged")
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let produced = Arc::new(AtomicU64::new(0));
+    let port = start_server(produced.clone()).await;
+
+    let delivered = Arc::new(AtomicU64::new(0));
+    let (closed_tx, mut closed_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+    let sessions = SessionManager::new(
+        {
+            let delivered = delivered.clone();
+            Arc::new(move |_, data: &[u8]| {
+                delivered.fetch_add(data.len() as u64, Ordering::SeqCst);
+            })
+        },
+        Arc::new(move |id: &str, _| {
+            let _ = closed_tx.send(id.to_string());
+        }),
+    );
+    let known_hosts = Arc::new(Mutex::new(
+        KnownHosts::load(dir.path().join("known_hosts.json")).expect("known hosts"),
+    ));
+
+    let claim = sessions.claim("flood").expect("claim");
+    sessions
+        .create_ssh_session(&claim, &host(port), None, known_hosts, None, None, Some(PASSWORD), 80, 24)
+        .await
+        .expect("ssh session");
+    drop(claim);
+
+    // A due batch is still delivered after the window is spent, and the
+    // batcher can hold one protocol packet past a full batch.
+    let ceiling = OUTPUT_WINDOW + MAX_BATCH as u64 + CHUNK as u64;
+
+    let first = settled(&delivered).await;
+    assert!(first >= OUTPUT_WINDOW, "stalled at {first} bytes, short of the window");
+    assert!(first <= ceiling, "{first} bytes delivered past a window of {OUTPUT_WINDOW}");
+
+    // The stall reaches the peer: it stops getting anything onto the
+    // connection rather than having it buffered on our side without limit.
+    let stalled_at = settled(&produced).await;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+    assert_eq!(produced.load(Ordering::SeqCst), stalled_at, "the server never stalled");
+    assert_eq!(delivered.load(Ordering::SeqCst), first);
+
+    // Acknowledging what arrived lets about one more window through, and no
+    // more than that.
+    sessions.ack("flood", first).await;
+    let second = settled(&delivered).await;
+    assert!(second > first, "acknowledged output did not resume the session");
+    assert!(
+        second - first <= ceiling,
+        "{} bytes delivered past an acknowledgement of {first}",
+        second - first
+    );
+
+    // A stalled session still closes, and announces it, with nothing more
+    // acknowledged.
+    tokio::time::timeout(Duration::from_secs(10), sessions.close("flood"))
+        .await
+        .expect("close waited on the stalled reader")
+        .expect("close");
+    let closed = tokio::time::timeout(Duration::from_secs(15), closed_rx.recv())
+        .await
+        .expect("a stalled session never announced its close");
+    assert_eq!(closed.as_deref(), Some("flood"));
+
+    // Acknowledgements trailing a closed session are ignored.
+    sessions.ack("flood", 1).await;
+    sessions.ack("never-existed", 1).await;
+}
diff --git a/desktop/e2e/tauri-fixture.js b/desktop/e2e/tauri-fixture.js
index f9b6a836..c4a8d592 100644
--- a/desktop/e2e/tauri-fixture.js
+++ b/desktop/e2e/tauri-fixture.js
@@ -111,6 +111,7 @@
         return;
       }
       if (command === "session_resize") return;
+      if (command === "session_output_ack") return;
       if (command === "session_write") {
         const text = new TextDecoder().decode(new Uint8Array(args.data));
         state.writes.push({ id: args.sessionId, text });
diff --git a/desktop/src-tauri/src/commands.rs b/desktop/src-tauri/src/commands.rs
index 1d80e9b6..53acc926 100644
--- a/desktop/src-tauri/src/commands.rs
+++ b/desktop/src-tauri/src/commands.rs
@@ -952,6 +952,23 @@ pub async fn session_resize(
     Err("session not found".into())
 }
 
+/// Acknowledge `bytes` of a session's output as received by the webview.
+///
+/// An SSH session stops reading from its connection once a window of output is
+/// unacknowledged, so a webview that falls behind slows the peer down instead
+/// of letting undelivered output pile up in the IPC layer. Local terminals are
+/// not metered, and an id with no live SSH session — a local terminal, or a
+/// session that has just closed — is accepted and ignored.
+#[tauri::command]
+pub async fn session_output_ack(
+    state: State<'_, Arc<AppState>>,
+    session_id: String,
+    bytes: u64,
+) -> ApiResult<()> {
+    state.sessions.ack(&session_id, bytes).await;
+    Ok(())
+}
+
 #[tauri::command]
 pub async fn close_session(state: State<'_, Arc<AppState>>, session_id: String) -> ApiResult<()> {
     // The sink stays registered: tearing the session down makes its task reach
diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs
index 9024747e..bb9534e1 100644
--- a/desktop/src-tauri/src/main.rs
+++ b/desktop/src-tauri/src/main.rs
@@ -194,6 +194,7 @@ fn main() {
             commands::create_local_terminal,
             commands::session_write,
             commands::session_resize,
+            commands::session_output_ack,
             commands::close_session,
             commands::list_sessions,
             commands::sftp_connect,
diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts
new file mode 100644
index 00000000..78378d51
--- /dev/null
+++ b/desktop/src/api.sessionOutput.test.ts
@@ -0,0 +1,96 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { OUTPUT_ACK_DELAY_MS, OUTPUT_ACK_THRESHOLD, sessionOutput } from "./api";
+import { getInvokeMock, setInvokeHandler } from "./test/setup";
+
+const acks = () =>
+  getInvokeMock().mock.calls
+    .filter(([command]) => command === "session_output_ack")
+    .map(([, args]) => args);
+
+const frame = (length: number) => new Uint8Array(length).fill(120).buffer;
+
+function open(sessionId = "s1") {
+  const onData = vi.fn();
+  const onEnd = vi.fn();
+  const channel = sessionOutput(sessionId, onData, onEnd);
+  const deliver = (length: number) => channel.onmessage(frame(length));
+  return { onData, onEnd, deliver };
+}
+
+describe("session output acknowledgement", () => {
+  beforeEach(() => vi.useFakeTimers());
+  afterEach(() => vi.useRealTimers());
+
+  it("coalesces small batches into one acknowledgement after a short delay", () => {
+    const { onData, deliver } = open();
+    deliver(10);
+    deliver(2000);
+    deliver(5);
+    expect(onData).toHaveBeenCalledTimes(3);
+    expect(acks()).toEqual([]);
+
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS);
+    expect(acks()).toEqual([{ sessionId: "s1", bytes: 2015 }]);
+
+    // Nothing pending, so the timer does not fire an empty acknowledgement.
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10);
+    expect(acks()).toHaveLength(1);
+  });
+
+  it("acknowledges at once when the pending total reaches the threshold", () => {
+    const { deliver } = open();
+    deliver(OUTPUT_ACK_THRESHOLD - 1);
+    expect(acks()).toEqual([]);
+    deliver(64 * 1024);
+    expect(acks()).toEqual([{ sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD - 1 + 64 * 1024 }]);
+
+    // The delayed acknowledgement was folded into that one.
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10);
+    expect(acks()).toHaveLength(1);
+
+    // Later output starts a fresh count.
+    deliver(7);
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS);
+    expect(acks()).toEqual([
+      { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD - 1 + 64 * 1024 },
+      { sessionId: "s1", bytes: 7 },
+    ]);
+  });
+
+  it("acknowledges each session's output under its own id", () => {
+    const first = open("first");
+    const second = open("second");
+    first.deliver(3);
+    second.deliver(4);
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS);
+    expect(acks()).toEqual(
+      expect.arrayContaining([
+        { sessionId: "first", bytes: 3 },
+        { sessionId: "second", bytes: 4 },
+      ]),
+    );
+    expect(acks()).toHaveLength(2);
+  });
+
+  it("never acknowledges the end-of-stream frame or anything after it", () => {
+    const { onData, onEnd, deliver } = open();
+    deliver(100);
+    deliver(0);
+    expect(onEnd).toHaveBeenCalledTimes(1);
+    expect(onData).toHaveBeenCalledTimes(1);
+
+    vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10);
+    expect(acks()).toEqual([]);
+  });
+
+  it("swallows a failed acknowledgement", async () => {
+    setInvokeHandler("session_output_ack", () => Promise.reject(new Error("backend gone")));
+    const { onData, deliver } = open();
+    deliver(OUTPUT_ACK_THRESHOLD);
+    expect(acks()).toHaveLength(1);
+    // Let the rejection settle; an unhandled one fails the run.
+    await vi.runAllTimersAsync();
+    deliver(1);
+    expect(onData).toHaveBeenCalledTimes(2);
+  });
+});
diff --git a/desktop/src/api.ts b/desktop/src/api.ts
index ae31007d..5e2b301e 100644
--- a/desktop/src/api.ts
+++ b/desktop/src/api.ts
@@ -66,6 +66,20 @@ export const pickKeyFile = () => invoke<string | null>("pick_key_file");
  */
 export type SessionOutput = Channel<ArrayBuffer>;
 
+/** Received output that is acknowledged at once rather than on the timer. */
+export const OUTPUT_ACK_THRESHOLD = 256 * 1024;
+/** Longest received output waits to be acknowledged below the threshold. */
+export const OUTPUT_ACK_DELAY_MS = 30;
+
+/**
+ * Reports output bytes as received. An SSH session stops reading from its
+ * connection while too much of its output is unacknowledged, so a webview that
+ * falls behind slows the remote end down rather than letting undelivered
+ * output pile up in the IPC layer. Local terminals ignore it.
+ */
+export const sessionOutputAck = (sessionId: string, bytes: number) =>
+  invoke<void>("session_output_ack", { sessionId, bytes });
+
 /**
  * Opens a sink for one session.
  *
@@ -75,13 +89,48 @@ export type SessionOutput = Channel<ArrayBuffer>;
  * to take the asynchronous transport route. `onEnd` is therefore the only
  * signal that says no more output is coming; the `session-closed` event carries
  * no index and can overtake output that is still in flight.
+ *
+ * Every batch is acknowledged on receipt, whether or not the pane still wants
+ * it: the backend meters what the IPC layer holds, not what the emulator
+ * renders. Acknowledgements are coalesced, going out once
+ * `OUTPUT_ACK_THRESHOLD` bytes are pending or `OUTPUT_ACK_DELAY_MS` after the
+ * first unacknowledged batch, so a trickle of output never leaves credit
+ * stranded. Nothing is acknowledged once the stream has ended; the session is
+ * gone by then.
  */
-export function sessionOutput(onData: (bytes: Uint8Array) => void, onEnd: () => void): SessionOutput {
+export function sessionOutput(
+  sessionId: string,
+  onData: (bytes: Uint8Array) => void,
+  onEnd: () => void,
+): SessionOutput {
   const channel: SessionOutput = new Channel();
+  let pending = 0;
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  const stopTimer = () => {
+    if (timer !== null) clearTimeout(timer);
+    timer = null;
+  };
+  const acknowledge = () => {
+    stopTimer();
+    const bytes = pending;
+    pending = 0;
+    // A lost acknowledgement leaves the session paused at worst; it is not
+    // something the pane can act on, so it must not surface as an unhandled
+    // rejection.
+    if (bytes > 0) sessionOutputAck(sessionId, bytes).catch(() => {});
+  };
   channel.onmessage = (message) => {
     const bytes = new Uint8Array(message);
-    if (bytes.length === 0) onEnd();
-    else onData(bytes);
+    if (bytes.length === 0) {
+      stopTimer();
+      pending = 0;
+      onEnd();
+      return;
+    }
+    pending += bytes.length;
+    if (pending >= OUTPUT_ACK_THRESHOLD) acknowledge();
+    else if (timer === null) timer = setTimeout(acknowledge, OUTPUT_ACK_DELAY_MS);
+    onData(bytes);
   };
   return channel;
 }
diff --git a/desktop/src/components/TerminalPane.vue b/desktop/src/components/TerminalPane.vue
index 1b403d05..eedfe4f5 100644
--- a/desktop/src/components/TerminalPane.vue
+++ b/desktop/src/components/TerminalPane.vue
@@ -265,7 +265,7 @@ async function connectSession() {
   // never reach the emulator even before its session is torn down. Bytes keep
   // being accepted until the sink itself reports the end of the stream: the
   // close notification does not travel with them and can arrive first.
-  const outputSink = api.sessionOutput(data => {
+  const outputSink = api.sessionOutput(sessionId, data => {
     if (disposed || props.pane.closing || streamEnded || currentSessionId !== sessionId) return;
     if (sessionEnded) tailOutput?.push(data);
     else if (pendingOutput) pendingOutput.push(data);

From b91ca46dbd0b8de41404e5b04adc1aaf97a05832 Mon Sep 17 00:00:00 2001
From: Azer Sadykhzadeh <azer.piano14@gmail.com>
Date: Sat, 17 Oct 2026 06:56:58 +0000
Subject: [PATCH 2/4] Spend output credit before handing the batch over

The reader handed a batch to the UI layer and only then counted it
against the window. An acknowledgement that overtook the spend would be
clamped against nothing outstanding and lost, and enough of those would
leave the window spent for good.
---
 core/src/session.rs | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/core/src/session.rs b/core/src/session.rs
index dbcca45f..ff104c7f 100644
--- a/core/src/session.rs
+++ b/core/src/session.rs
@@ -282,8 +282,11 @@ impl SessionManager {
             let flush = |batcher: &mut OutputBatcher| {
                 if !batcher.is_empty() {
                     let batch = batcher.batch();
-                    data_cb(&sid, batch);
+                    // Spent before the batch is handed over, so an
+                    // acknowledgement can never arrive ahead of the spend
+                    // it pays back.
                     credit.spend(batch.len());
+                    data_cb(&sid, batch);
                     batcher.mark_flushed(Instant::now());
                 }
             };

From d8e758bf934bfac1f114d17b7af3e16ac850de1b Mon Sep 17 00:00:00 2001
From: Azer Sadykhzadeh <azer.piano14@gmail.com>
Date: Sat, 17 Oct 2026 08:24:55 +0000
Subject: [PATCH 3/4] Retry a failed output acknowledgement until the stream
 ends

The acknowledgement was dropped on failure. A reader that had used up its
window then waited for credit that no further output would ever trigger,
so one lost IPC call could freeze a busy session for good.
---
 core/src/session.rs                   |  5 +-
 core/src/sftp.rs                      | 99 ++++++++++-----------------
 desktop/src/api.sessionOutput.test.ts | 30 ++++++++
 desktop/src/api.ts                    | 27 ++++++--
 4 files changed, 91 insertions(+), 70 deletions(-)

diff --git a/core/src/session.rs b/core/src/session.rs
index ff104c7f..dbcca45f 100644
--- a/core/src/session.rs
+++ b/core/src/session.rs
@@ -282,11 +282,8 @@ impl SessionManager {
             let flush = |batcher: &mut OutputBatcher| {
                 if !batcher.is_empty() {
                     let batch = batcher.batch();
-                    // Spent before the batch is handed over, so an
-                    // acknowledgement can never arrive ahead of the spend
-                    // it pays back.
-                    credit.spend(batch.len());
                     data_cb(&sid, batch);
+                    credit.spend(batch.len());
                     batcher.mark_flushed(Instant::now());
                 }
             };
diff --git a/core/src/sftp.rs b/core/src/sftp.rs
index bda7c62b..228ce79c 100644
--- a/core/src/sftp.rs
+++ b/core/src/sftp.rs
@@ -48,10 +48,9 @@ pub struct SftpManager {
 struct SftpConnection {
     handle: Handle<connection::SshHandler>,
     sftp: SftpSession,
-    /// Flipped once the connection is closed. Every operation watches it and
-    /// stops, so one that is stuck behind a peer that stopped reading, or
-    /// simply still running, cannot outlive the close or report success
-    /// after it.
+    /// Flipped once the connection is closed. Listings and transfers watch it
+    /// and stop, so one that is stuck behind a peer that stopped reading, or
+    /// simply still running, cannot outlive the close.
     closed: tokio::sync::watch::Sender<bool>,
 }
 
@@ -70,8 +69,8 @@ impl SftpConnection {
         }
     }
 
-    /// Stop every request, listing and transfer on this connection, end the
-    /// SFTP channel, then the connection. Bounded, so an unresponsive peer cannot
+    /// Stop every listing and transfer on this connection, end the SFTP
+    /// channel, then the connection. Bounded, so an unresponsive peer cannot
     /// hold the caller.
     async fn close(&self) {
         self.closed.send_replace(true);
@@ -225,13 +224,10 @@ impl SftpManager {
     pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result<String> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .canonicalize(path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}")))
-        })
-        .await
+        conn.sftp
+            .canonicalize(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}")))
     }
 
     /// Read a file's contents into a byte vector.
@@ -241,13 +237,10 @@ impl SftpManager {
     pub async fn read_file(&self, session_id: &str, path: &str) -> Result<Vec<u8>> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .read(path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("read file: {e}")))
-        })
-        .await
+        conn.sftp
+            .read(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("read file: {e}")))
     }
 
     /// Write data to a remote file (creates or overwrites).
@@ -257,13 +250,10 @@ impl SftpManager {
     pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .write(path, data)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("write file: {e}")))
-        })
-        .await
+        conn.sftp
+            .write(path, data)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("write file: {e}")))
     }
 
     /// Stream a remote file directly to a local path. An existing local file is
@@ -336,39 +326,30 @@ impl SftpManager {
     pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .create_dir(path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("mkdir: {e}")))
-        })
-        .await
+        conn.sftp
+            .create_dir(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("mkdir: {e}")))
     }
 
     /// Remove a file.
     pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .remove_file(path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("remove file: {e}")))
-        })
-        .await
+        conn.sftp
+            .remove_file(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("remove file: {e}")))
     }
 
     /// Remove a directory.
     pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .remove_dir(path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("remove dir: {e}")))
-        })
-        .await
+        conn.sftp
+            .remove_dir(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("remove dir: {e}")))
     }
 
     /// Rename a file or directory.
@@ -380,13 +361,10 @@ impl SftpManager {
     ) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.until_closed(async {
-            conn.sftp
-                .rename(old_path, new_path)
-                .await
-                .map_err(|e| CoreError::Ssh(format!("rename: {e}")))
-        })
-        .await
+        conn.sftp
+            .rename(old_path, new_path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("rename: {e}")))
     }
 
     /// Get file metadata (size, permissions, modified time).
@@ -394,13 +372,10 @@ impl SftpManager {
         let conn = self.get(session_id).await?;
 
         let attrs = conn
-            .until_closed(async {
-                conn.sftp
-                    .metadata(path)
-                    .await
-                    .map_err(|e| CoreError::Ssh(format!("stat: {e}")))
-            })
-            .await?;
+            .sftp
+            .metadata(path)
+            .await
+            .map_err(|e| CoreError::Ssh(format!("stat: {e}")))?;
 
         let name = path.rsplit('/').next().unwrap_or(path).to_string();
         let modified = attrs
diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts
index 78378d51..b5ab7200 100644
--- a/desktop/src/api.sessionOutput.test.ts
+++ b/desktop/src/api.sessionOutput.test.ts
@@ -57,6 +57,36 @@ describe("session output acknowledgement", () => {
     ]);
   });
 
+  it("retries an acknowledgement that failed, so the reader is not left waiting", async () => {
+    let fail = true;
+    setInvokeHandler("session_output_ack", () => {
+      if (fail) throw new Error("ipc unavailable");
+    });
+    const { deliver } = open();
+    deliver(OUTPUT_ACK_THRESHOLD);
+    await vi.advanceTimersByTimeAsync(0);
+    expect(acks()).toEqual([{ sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD }]);
+
+    // No further output arrives, so only a retry can hand the credit back.
+    fail = false;
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(acks()).toEqual([
+      { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD },
+      { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD },
+    ]);
+  });
+
+  it("stops retrying once the stream has ended", async () => {
+    setInvokeHandler("session_output_ack", () => {
+      throw new Error("session closed");
+    });
+    const { deliver } = open();
+    deliver(OUTPUT_ACK_THRESHOLD);
+    deliver(0);
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(acks()).toHaveLength(1);
+  });
+
   it("acknowledges each session's output under its own id", () => {
     const first = open("first");
     const second = open("second");
diff --git a/desktop/src/api.ts b/desktop/src/api.ts
index 5e2b301e..8e11c389 100644
--- a/desktop/src/api.ts
+++ b/desktop/src/api.ts
@@ -70,6 +70,8 @@ export type SessionOutput = Channel<ArrayBuffer>;
 export const OUTPUT_ACK_THRESHOLD = 256 * 1024;
 /** Longest received output waits to be acknowledged below the threshold. */
 export const OUTPUT_ACK_DELAY_MS = 30;
+/** How many times one failed acknowledgement is retried before giving up. */
+export const OUTPUT_ACK_MAX_RETRIES = 8;
 
 /**
  * Reports output bytes as received. An SSH session stops reading from its
@@ -105,6 +107,8 @@ export function sessionOutput(
 ): SessionOutput {
   const channel: SessionOutput = new Channel();
   let pending = 0;
+  let failures = 0;
+  let ended = false;
   let timer: ReturnType<typeof setTimeout> | null = null;
   const stopTimer = () => {
     if (timer !== null) clearTimeout(timer);
@@ -114,14 +118,29 @@ export function sessionOutput(
     stopTimer();
     const bytes = pending;
     pending = 0;
-    // A lost acknowledgement leaves the session paused at worst; it is not
-    // something the pane can act on, so it must not surface as an unhandled
-    // rejection.
-    if (bytes > 0) sessionOutputAck(sessionId, bytes).catch(() => {});
+    if (bytes === 0) return;
+    sessionOutputAck(sessionId, bytes).then(
+      () => {
+        failures = 0;
+      },
+      () => {
+        // Credit that never reaches the backend is never handed back, and a
+        // reader that has run out waits for an acknowledgement no further
+        // output will trigger. The bytes go back into the count and are
+        // retried, backing off, until the stream ends.
+        if (ended) return;
+        pending += bytes;
+        failures += 1;
+        if (failures > OUTPUT_ACK_MAX_RETRIES) return;
+        stopTimer();
+        timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, 2000));
+      },
+    );
   };
   channel.onmessage = (message) => {
     const bytes = new Uint8Array(message);
     if (bytes.length === 0) {
+      ended = true;
       stopTimer();
       pending = 0;
       onEnd();

From 09e30729230d050d953180aa440daf04c63776ee Mon Sep 17 00:00:00 2001
From: Azer Sadykhzadeh <azer.piano14@gmail.com>
Date: Sat, 17 Oct 2026 08:37:54 +0000
Subject: [PATCH 4/4] Restore the SFTP close signal and the spend ordering, and
 retry acknowledgements for as long as the stream is open

The previous commit picked up stale copies of sftp.rs and session.rs, which
undid the close signal on single SFTP requests and moved the credit spend
back behind the batch hand-off. A capped retry could also give up during a
long outage and leave the session paused for good.
---
 core/src/session.rs                   |  5 +-
 core/src/sftp.rs                      | 99 +++++++++++++++++----------
 desktop/src/api.sessionOutput.test.ts | 23 ++++++-
 desktop/src/api.ts                    | 11 +--
 4 files changed, 93 insertions(+), 45 deletions(-)

diff --git a/core/src/session.rs b/core/src/session.rs
index dbcca45f..ff104c7f 100644
--- a/core/src/session.rs
+++ b/core/src/session.rs
@@ -282,8 +282,11 @@ impl SessionManager {
             let flush = |batcher: &mut OutputBatcher| {
                 if !batcher.is_empty() {
                     let batch = batcher.batch();
-                    data_cb(&sid, batch);
+                    // Spent before the batch is handed over, so an
+                    // acknowledgement can never arrive ahead of the spend
+                    // it pays back.
                     credit.spend(batch.len());
+                    data_cb(&sid, batch);
                     batcher.mark_flushed(Instant::now());
                 }
             };
diff --git a/core/src/sftp.rs b/core/src/sftp.rs
index 228ce79c..bda7c62b 100644
--- a/core/src/sftp.rs
+++ b/core/src/sftp.rs
@@ -48,9 +48,10 @@ pub struct SftpManager {
 struct SftpConnection {
     handle: Handle<connection::SshHandler>,
     sftp: SftpSession,
-    /// Flipped once the connection is closed. Listings and transfers watch it
-    /// and stop, so one that is stuck behind a peer that stopped reading, or
-    /// simply still running, cannot outlive the close.
+    /// Flipped once the connection is closed. Every operation watches it and
+    /// stops, so one that is stuck behind a peer that stopped reading, or
+    /// simply still running, cannot outlive the close or report success
+    /// after it.
     closed: tokio::sync::watch::Sender<bool>,
 }
 
@@ -69,8 +70,8 @@ impl SftpConnection {
         }
     }
 
-    /// Stop every listing and transfer on this connection, end the SFTP
-    /// channel, then the connection. Bounded, so an unresponsive peer cannot
+    /// Stop every request, listing and transfer on this connection, end the
+    /// SFTP channel, then the connection. Bounded, so an unresponsive peer cannot
     /// hold the caller.
     async fn close(&self) {
         self.closed.send_replace(true);
@@ -224,10 +225,13 @@ impl SftpManager {
     pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result<String> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .canonicalize(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .canonicalize(path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}")))
+        })
+        .await
     }
 
     /// Read a file's contents into a byte vector.
@@ -237,10 +241,13 @@ impl SftpManager {
     pub async fn read_file(&self, session_id: &str, path: &str) -> Result<Vec<u8>> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .read(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("read file: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .read(path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("read file: {e}")))
+        })
+        .await
     }
 
     /// Write data to a remote file (creates or overwrites).
@@ -250,10 +257,13 @@ impl SftpManager {
     pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .write(path, data)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("write file: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .write(path, data)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("write file: {e}")))
+        })
+        .await
     }
 
     /// Stream a remote file directly to a local path. An existing local file is
@@ -326,30 +336,39 @@ impl SftpManager {
     pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .create_dir(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("mkdir: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .create_dir(path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("mkdir: {e}")))
+        })
+        .await
     }
 
     /// Remove a file.
     pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .remove_file(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("remove file: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .remove_file(path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("remove file: {e}")))
+        })
+        .await
     }
 
     /// Remove a directory.
     pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .remove_dir(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("remove dir: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .remove_dir(path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("remove dir: {e}")))
+        })
+        .await
     }
 
     /// Rename a file or directory.
@@ -361,10 +380,13 @@ impl SftpManager {
     ) -> Result<()> {
         let conn = self.get(session_id).await?;
 
-        conn.sftp
-            .rename(old_path, new_path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("rename: {e}")))
+        conn.until_closed(async {
+            conn.sftp
+                .rename(old_path, new_path)
+                .await
+                .map_err(|e| CoreError::Ssh(format!("rename: {e}")))
+        })
+        .await
     }
 
     /// Get file metadata (size, permissions, modified time).
@@ -372,10 +394,13 @@ impl SftpManager {
         let conn = self.get(session_id).await?;
 
         let attrs = conn
-            .sftp
-            .metadata(path)
-            .await
-            .map_err(|e| CoreError::Ssh(format!("stat: {e}")))?;
+            .until_closed(async {
+                conn.sftp
+                    .metadata(path)
+                    .await
+                    .map_err(|e| CoreError::Ssh(format!("stat: {e}")))
+            })
+            .await?;
 
         let name = path.rsplit('/').next().unwrap_or(path).to_string();
         let modified = attrs
diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts
index b5ab7200..0aebae4d 100644
--- a/desktop/src/api.sessionOutput.test.ts
+++ b/desktop/src/api.sessionOutput.test.ts
@@ -76,6 +76,25 @@ describe("session output acknowledgement", () => {
     ]);
   });
 
+  it("keeps retrying through a long outage", async () => {
+    let fail = true;
+    setInvokeHandler("session_output_ack", () => {
+      if (fail) throw new Error("ipc unavailable");
+    });
+    const { deliver } = open();
+    deliver(OUTPUT_ACK_THRESHOLD);
+    await vi.advanceTimersByTimeAsync(60_000);
+    const attempts = acks().length;
+    expect(attempts).toBeGreaterThan(10);
+
+    fail = false;
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(acks().length).toBe(attempts + 1);
+    // Once it went through nothing is left to send.
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(acks().length).toBe(attempts + 1);
+  });
+
   it("stops retrying once the stream has ended", async () => {
     setInvokeHandler("session_output_ack", () => {
       throw new Error("session closed");
@@ -118,8 +137,8 @@ describe("session output acknowledgement", () => {
     const { onData, deliver } = open();
     deliver(OUTPUT_ACK_THRESHOLD);
     expect(acks()).toHaveLength(1);
-    // Let the rejection settle; an unhandled one fails the run.
-    await vi.runAllTimersAsync();
+    // Let the rejection and a few retries settle; an unhandled one fails the run.
+    await vi.advanceTimersByTimeAsync(5000);
     deliver(1);
     expect(onData).toHaveBeenCalledTimes(2);
   });
diff --git a/desktop/src/api.ts b/desktop/src/api.ts
index 8e11c389..9be661d1 100644
--- a/desktop/src/api.ts
+++ b/desktop/src/api.ts
@@ -70,8 +70,8 @@ export type SessionOutput = Channel<ArrayBuffer>;
 export const OUTPUT_ACK_THRESHOLD = 256 * 1024;
 /** Longest received output waits to be acknowledged below the threshold. */
 export const OUTPUT_ACK_DELAY_MS = 30;
-/** How many times one failed acknowledgement is retried before giving up. */
-export const OUTPUT_ACK_MAX_RETRIES = 8;
+/** Longest wait between retries of an acknowledgement that failed. */
+export const OUTPUT_ACK_MAX_RETRY_MS = 2000;
 
 /**
  * Reports output bytes as received. An SSH session stops reading from its
@@ -127,13 +127,14 @@ export function sessionOutput(
         // Credit that never reaches the backend is never handed back, and a
         // reader that has run out waits for an acknowledgement no further
         // output will trigger. The bytes go back into the count and are
-        // retried, backing off, until the stream ends.
+        // retried, backing off to OUTPUT_ACK_MAX_RETRY_MS, for as long as the
+        // stream is open: giving up at any point would leave the session
+        // paused for good.
         if (ended) return;
         pending += bytes;
         failures += 1;
-        if (failures > OUTPUT_ACK_MAX_RETRIES) return;
         stopTimer();
-        timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, 2000));
+        timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, OUTPUT_ACK_MAX_RETRY_MS));
       },
     );
   };