Fix SFTP upload error handling and clean up /tmp staging files

dartssh2's SftpFileWriter leaks chunk write failures as unhandled
exceptions and never completes when the server rejects data (e.g. a
full /tmp), so deploys hung and spammed the console. Upload in
windowed writeBytes batches instead so errors propagate, remove the
partial file, and report free space on the target directory.

Every deploy flow now removes its staging files in a finally block
(success, failure or cancel) and sweeps peachdeploy-*/peachweb-*
leftovers older than an hour from earlier crashed runs.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

lib/Peach/comms/Comms.dart

@@ -257,6 +257,8 @@ class SshService implements ConnectComms {
       ('cleanup', 'Finalize'),
       ('logs', 'Fetch logs'),
     ]);
+    final staging =
+        '/tmp/peachdeploy-${DateTime.now().millisecondsSinceEpoch}.jar';
     try {
       steps.active('connect');
       await _connect(server);
@@ -265,8 +267,6 @@ class SshService implements ConnectComms {
       final target = service.remotePath(jarLabel);
       final backup = '$target.bak';
       final dir = target.substring(0, target.lastIndexOf('/'));
-      final staging =
-          '/tmp/peachdeploy-${DateTime.now().millisecondsSinceEpoch}.jar';
 
       await _run('mkdir -p ${_shQuote(dir)}', sudo: sudo, password: pw);
 
@@ -415,6 +415,8 @@ class SshService implements ConnectComms {
       }
       _log(LogLevel.error, 'Deploy failed: $e');
       return DeployResult(false, _canRevert, 'Error: $e');
+    } finally {
+      await _cleanStaging(server, [staging]);
     }
   }
 
@@ -632,6 +634,8 @@ class SshService implements ConnectComms {
       }
       _log(LogLevel.error, 'Web deploy failed: $e');
       return DeployResult(false, _canRevert, 'Error: $e');
+    } finally {
+      await _cleanStaging(server, [archive, staging]);
     }
   }
 
@@ -923,6 +927,8 @@ class SshService implements ConnectComms {
       steps.failRemaining();
       _log(LogLevel.error, 'Unit install failed: $e');
       return DeployResult(false, false, 'Error: $e');
+    } finally {
+      await _cleanStaging(server, [tmp]);
     }
   }
 
@@ -2170,27 +2176,84 @@ class SshService implements ConnectComms {
     final total = bytes.length;
     var sent = 0;
     const chunk = 32 * 1024;
+    // Chunks in flight per batch (~1 MB) — keeps the pipe full without
+    // dartssh2's SftpFileWriter, which leaks write errors as unhandled
+    // exceptions and never completes its future when the server rejects data.
+    const window = 32;
     final meter = _SpeedMeter();
-    final stream =
-        Stream<Uint8List>.fromIterable([
-          for (var i = 0; i < total; i += chunk)
-            Uint8List.sublistView(
-              bytes,
-              i,
-              i + chunk > total ? total : i + chunk,
-            ),
-        ]).map((c) {
-          sent += c.length;
-          if (meter.tick(sent)) {
-            _emitTransfer(sent, total, meter, uploading: true);
-          }
-          return c;
-        });
-    await file.write(stream);
+    try {
+      for (var i = 0; i < total; i += chunk * window) {
+        final batch = <Future<void>>[];
+        for (var j = i; j < total && j < i + chunk * window; j += chunk) {
+          final end = j + chunk > total ? total : j + chunk;
+          batch.add(
+            file.writeBytes(Uint8List.sublistView(bytes, j, end), offset: j),
+          );
+        }
+        await Future.wait(batch);
+        sent = (i + chunk * window) > total ? total : i + chunk * window;
+        if (meter.tick(sent)) {
+          _emitTransfer(sent, total, meter, uploading: true);
+        }
+      }
+    } on SftpStatusError catch (e) {
+      try {
+        await file.close();
+      } catch (_) {}
+      try {
+        await sftp.remove(remotePath);
+      } catch (_) {}
+      throw Exception(
+        'SFTP write to $remotePath failed (${e.message}, code ${e.code})'
+        '${await _freeSpaceHint(sftp, remotePath, total)}',
+      );
+    }
     await file.close();
     _emitTransfer(total, total, meter, uploading: true, force: true);
   }
 
+  /// Removes this run's /tmp staging files (no-op if they were already moved
+  /// into place) plus stale ones left by earlier crashed or cancelled runs.
+  /// Best-effort: never throws, so it can't mask the deploy's own result.
+  Future<void> _cleanStaging(ServerProfile s, List<String> paths) async {
+    try {
+      await _connect(s).timeout(const Duration(seconds: 20));
+      await _run(
+        'rm -rf ${paths.map(_shQuote).join(' ')}; '
+        "find /tmp -maxdepth 1 \\( -name 'peachdeploy-*' -o -name 'peachweb-*' \\) "
+        '-mmin +60 -exec rm -rf {} + 2>/dev/null; true',
+        sudo: s.useSudo,
+        password: s.password,
+      ).timeout(const Duration(seconds: 30));
+    } catch (e) {
+      _log(LogLevel.warn, 'Could not clean staging files: $e');
+    }
+  }
+
+  /// Best-effort explanation for a generic SFTP write failure — usually the
+  /// target filesystem (often /tmp) is full or over quota.
+  Future<String> _freeSpaceHint(
+    SftpClient sftp,
+    String remotePath,
+    int needed,
+  ) async {
+    try {
+      final slash = remotePath.lastIndexOf('/');
+      final dir = slash > 0 ? remotePath.substring(0, slash) : '/';
+      final vfs = await sftp.statvfs(dir);
+      if (vfs.isReadOnly) return ' — $dir is mounted read-only.';
+      final unit = vfs.fundamentalBlockSize > 0
+          ? vfs.fundamentalBlockSize
+          : vfs.blockSize;
+      final free = unit * vfs.freeBlocksForNonRoot;
+      return ' — $dir has ${_fmtBytes(free)} free, upload needs '
+          '${_fmtBytes(needed)}. Free up space on the server.';
+    } catch (_) {
+      return ' — the remote filesystem rejected the data; check free space '
+          'and quota on the server (df -h).';
+    }
+  }
+
   // ---- lifecycle ------------------------------------------------------------
 
   @override