fix(peer): restore terminal control and file downloads (#1666)

Author: bobleer

docs/architecture/peer-device-mode.md

@@ -52,7 +52,9 @@ FS) and must not be mixed with Peer Device Mode.
 - HostInvoke on the controller is **priority-queued** (max 2 in flight). Session
   restore / session-list / dialog / workspace-startup commands outrank background
   `git_*` / `ssh_*` / `lsp_*` / `search_*` / FS / canvas / editor RPCs so hydrate
-  is not starved into relay HTTP 504s.
+  is not starved into relay HTTP 504s. Terminal commands are always interactive
+  priority, and one slot is kept free from low-priority background work so input
+  cannot be trapped behind two slow polling requests.
 - While Peer Mode is active, background noise is reduced further:
   - controller-local SSH heartbeats and remote-workspace auto-reconnect pause
   - Git / FilesPanel window-focus refresh pauses
@@ -69,7 +71,9 @@ FS) and must not be mixed with Peer Device Mode.
     return empty or no-op so hydrate does not fail.
 - Events: peer agentic projection (and other product events such as terminal /
   FS / MCP interaction) fan-out as `RemoteCommand::DeviceEvent` to attached
-  controllers; controller re-emits the same event names locally.
+  controllers; controller re-emits the same event names locally. This includes
+  SSH-backed remote PTY Ready / Data / Exit events created on B, not only B's
+  local terminal service events.
 - CLI Peer Host forwards only turns submitted through Peer Host and linked
   child turns. A background-result follow-up inherits ownership only when its
   Core-internal metadata identifies the exact tracked parent and source child
@@ -113,6 +117,17 @@ call `pickWorkspaceDirectory()`:
 Still use normal `openWorkspace` / create-workspace flows (not SSH
 `openRemoteWorkspace` / `WorkspaceKind.Remote`).
 
+## File download ownership
+
+The native save/folder dialog always selects a destination on controller A,
+while the workspace source belongs to peer B. A download is therefore a
+split-endpoint operation: B returns file bytes through the existing
+`GetFileInfo` / `ReadFileChunk` protocol and A writes those chunks through its
+local filesystem adapter. Directory downloads enumerate B recursively and
+create the corresponding tree on A. Never forward A's selected destination to
+B through `export_local_file_to_path`; paths and permissions are host-specific
+and may represent a different operating system.
+
 ## Ownership
 
 - Desktop host invoke / fan-out: `src/apps/desktop/src/api/peer_host_invoke.rs`,

src/apps/desktop/src/api/clipboard_file_api.rs

@@ -417,7 +417,7 @@ fn generate_unique_path(path: &Path) -> std::path::PathBuf {
     }
 }
 
-fn copy_directory_recursive(source: &Path, target: &Path) -> Result<(), String> {
+pub(crate) fn copy_directory_recursive(source: &Path, target: &Path) -> Result<(), String> {
     std::fs::create_dir_all(target).map_err(|e| format!("Failed to create directory: {}", e))?;
 
     for entry in
@@ -441,7 +441,8 @@ fn copy_directory_recursive(source: &Path, target: &Path) -> Result<(), String>
 #[cfg(test)]
 mod tests {
     use super::{
-        decode_file_uri, generate_unique_path, parse_clipboard_path_segments, parse_uri_list,
+        copy_directory_recursive, decode_file_uri, generate_unique_path,
+        parse_clipboard_path_segments, parse_uri_list,
     };
     use std::path::Path;
 
@@ -519,4 +520,32 @@ mod tests {
             vec!["/tmp/a.txt".to_string(), "/tmp/b.txt".to_string()]
         );
     }
+
+    #[test]
+    fn copy_directory_recursive_copies_nested_binary_files() {
+        let root = std::env::temp_dir().join(format!(
+            "bitfun-directory-copy-test-{}",
+            uuid::Uuid::new_v4()
+        ));
+        let source = root.join("source");
+        let target = root.join("target");
+        std::fs::create_dir_all(source.join("nested")).expect("create source directory");
+        std::fs::write(source.join("root.bin"), [0_u8, 255, 128]).expect("write root file");
+        std::fs::write(source.join("nested").join("child.txt"), b"child")
+            .expect("write nested file");
+
+        copy_directory_recursive(&source, &target).expect("copy directory recursively");
+
+        assert_eq!(
+            std::fs::read(target.join("root.bin")).expect("read copied root file"),
+            [0_u8, 255, 128]
+        );
+        assert_eq!(
+            std::fs::read(target.join("nested").join("child.txt"))
+                .expect("read copied nested file"),
+            b"child"
+        );
+
+        std::fs::remove_dir_all(root).expect("remove test directory");
+    }
 }

src/apps/desktop/src/api/commands.rs

@@ -2944,7 +2944,7 @@ pub async fn rename_file(
     .await
 }
 
-/// Copy a local file to another local path (binary-safe). Used for export and drag-upload into local workspaces.
+/// Copy a local file or directory to another local path (binary-safe).
 #[tauri::command]
 pub async fn export_local_file_to_path(request: ExportLocalFileRequest) -> Result<(), String> {
     let src = request.source_path;
@@ -2956,7 +2956,54 @@ pub async fn export_local_file_to_path(request: ExportLocalFileRequest) -> Resul
                 std::fs::create_dir_all(parent).map_err(|e| e.to_string())?;
             }
         }
-        std::fs::copy(&src, &dst).map_err(|e| e.to_string())?;
+        let src_path = Path::new(&src);
+        let src_metadata = std::fs::metadata(src_path).map_err(|e| {
+            format!(
+                "Failed to inspect export source '{}': {e}",
+                src_path.display()
+            )
+        })?;
+        if src_metadata.is_dir() {
+            let canonical_source = std::fs::canonicalize(src_path).map_err(|e| {
+                format!(
+                    "Failed to resolve export source '{}': {e}",
+                    src_path.display()
+                )
+            })?;
+            let destination_parent = dst_path
+                .parent()
+                .filter(|parent| !parent.as_os_str().is_empty())
+                .unwrap_or_else(|| Path::new("."));
+            let canonical_parent = std::fs::canonicalize(destination_parent).map_err(|e| {
+                format!(
+                    "Failed to resolve export destination '{}': {e}",
+                    destination_parent.display()
+                )
+            })?;
+            let resolved_destination = if dst_path.exists() {
+                std::fs::canonicalize(dst_path).map_err(|e| {
+                    format!(
+                        "Failed to resolve existing export destination '{}': {e}",
+                        dst_path.display()
+                    )
+                })?
+            } else {
+                canonical_parent.join(
+                    dst_path
+                        .file_name()
+                        .ok_or_else(|| "Export destination has no directory name".to_string())?,
+                )
+            };
+            if resolved_destination == canonical_source
+                || resolved_destination.starts_with(&canonical_source)
+            {
+                return Err("Cannot export a directory into itself".to_string());
+            }
+            super::clipboard_file_api::copy_directory_recursive(src_path, dst_path)?;
+        } else {
+            std::fs::copy(src_path, dst_path)
+                .map_err(|e| format!("Failed to copy export file: {e}"))?;
+        }
         Ok::<(), String>(())
     })
     .await

src/apps/desktop/src/api/terminal_api.rs

@@ -344,6 +344,29 @@ async fn is_remote_session(session_id: &str) -> bool {
     false
 }
 
+fn emit_terminal_event(app_handle: &AppHandle, event: &TerminalEvent) -> bool {
+    let event_name = "terminal_event";
+    let local_emit_succeeded = match app_handle.emit(event_name, event) {
+        Ok(()) => true,
+        Err(error) => {
+            warn!("Failed to emit terminal event: {}", error);
+            false
+        }
+    };
+    if let Ok(payload) = serde_json::to_value(event) {
+        super::remote_connect_api::maybe_fanout_peer_ui_event(event_name, payload);
+    }
+    local_emit_succeeded
+}
+
+fn remote_terminal_signal_bytes(signal: &str) -> Option<&'static [u8]> {
+    match signal.trim().to_ascii_uppercase().as_str() {
+        "SIGINT" | "INT" => Some(&[0x03]),
+        "SIGTSTP" | "TSTP" => Some(&[0x1a]),
+        _ => None,
+    }
+}
+
 async fn spawn_remote_pty_session(
     app: &AppHandle,
     terminal_manager: &bitfun_core::service::remote_ssh::RemoteTerminalManager,
@@ -384,8 +407,8 @@ async fn spawn_remote_pty_session(
     let app_handle = app.clone();
     let sid = session_id.clone();
     tokio::spawn(async move {
-        let _ = app_handle.emit(
-            "terminal_event",
+        emit_terminal_event(
+            &app_handle,
             &TerminalEvent::Ready {
                 session_id: sid.clone(),
                 pid: 0,
@@ -397,14 +420,13 @@ async fn spawn_remote_pty_session(
             match rx.recv().await {
                 Ok(data) => {
                     let text = String::from_utf8_lossy(&data).to_string();
-                    if let Err(e) = app_handle.emit(
-                        "terminal_event",
+                    if !emit_terminal_event(
+                        &app_handle,
                         &TerminalEvent::Data {
                             session_id: sid.clone(),
                             data: text,
                         },
                     ) {
-                        warn!("Failed to emit remote terminal event: {}", e);
                         break;
                     }
                 }
@@ -421,8 +443,8 @@ async fn spawn_remote_pty_session(
             }
         }
 
-        let _ = app_handle.emit(
-            "terminal_event",
+        emit_terminal_event(
+            &app_handle,
             &TerminalEvent::Exit {
                 session_id: sid,
                 exit_code: Some(0),
@@ -699,7 +721,22 @@ pub async fn terminal_signal(
     state: State<'_, TerminalState>,
 ) -> Result<(), String> {
     if is_remote_session(&request.session_id).await {
-        // Remote terminals don't support signal yet
+        let signal_data = remote_terminal_signal_bytes(&request.signal).ok_or_else(|| {
+            format!(
+                "Unsupported remote terminal signal: {}",
+                request.signal.trim()
+            )
+        })?;
+        let remote_manager =
+            get_remote_workspace_manager().ok_or("Remote workspace manager not available")?;
+        let terminal_manager = remote_manager
+            .get_terminal_manager()
+            .await
+            .ok_or("Remote terminal manager not available")?;
+        terminal_manager
+            .write(&request.session_id, signal_data)
+            .await
+            .map_err(|e| format!("Failed to send remote terminal signal: {}", e))?;
         return Ok(());
     }
 
@@ -913,13 +950,21 @@ pub fn start_terminal_event_loop(terminal_state: TerminalState, app_handle: AppH
         let mut rx = api.subscribe_events();
 
         while let Some(event) = rx.recv().await {
-            let event_name = "terminal_event";
-            if let Err(e) = app_handle.emit(event_name, &event) {
-                warn!("Failed to emit terminal event: {}", e);
-            }
-            if let Ok(payload) = serde_json::to_value(&event) {
-                crate::api::remote_connect_api::maybe_fanout_peer_ui_event(event_name, payload);
-            }
+            emit_terminal_event(&app_handle, &event);
         }
     });
 }
+
+#[cfg(test)]
+mod tests {
+    use super::remote_terminal_signal_bytes;
+
+    #[test]
+    fn maps_supported_remote_terminal_signals_to_control_bytes() {
+        assert_eq!(remote_terminal_signal_bytes("SIGINT"), Some(&[0x03][..]));
+        assert_eq!(remote_terminal_signal_bytes("int"), Some(&[0x03][..]));
+        assert_eq!(remote_terminal_signal_bytes("SIGTSTP"), Some(&[0x1a][..]));
+        assert_eq!(remote_terminal_signal_bytes("tstp"), Some(&[0x1a][..]));
+        assert_eq!(remote_terminal_signal_bytes("SIGTERM"), None);
+    }
+}

src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts

@@ -33,6 +33,13 @@ describe('peerInvokePriorityFor', () => {
     expect(peerInvokePriorityFor('get_system_info')).toBe('high');
   });
 
+  it('ranks all terminal commands high', () => {
+    expect(peerInvokePriorityFor('terminal_create')).toBe('high');
+    expect(peerInvokePriorityFor('terminal_write')).toBe('high');
+    expect(peerInvokePriorityFor('terminal_resize')).toBe('high');
+    expect(peerInvokePriorityFor('terminal_signal')).toBe('high');
+  });
+
   it('ranks git/ssh/editor/fs/search noise low', () => {
     expect(peerInvokePriorityFor('git_is_repository')).toBe('low');
     expect(peerInvokePriorityFor('ssh_is_connected')).toBe('low');
@@ -90,6 +97,75 @@ describe('PeerDeviceTransportAdapter queue', () => {
       'ssh_is_connected',
     ]);
   });
+
+  it('reserves one concurrency slot for terminal work', async () => {
+    const started: string[] = [];
+    const firstLowGate = createDeferred<void>();
+
+    const deviceRpc = vi.fn(async (_target: string, commandJson: string) => {
+      const parsed = JSON.parse(commandJson) as { command: string };
+      started.push(parsed.command);
+      if (parsed.command === 'git_is_repository') {
+        await firstLowGate.promise;
+      }
+      return JSON.stringify({
+        resp: 'host_invoke_result',
+        ok: true,
+        value: true,
+      });
+    });
+
+    const adapter = new PeerDeviceTransportAdapter('peer-1', deviceRpc, {}, 2);
+    await adapter.connect();
+
+    const low1 = adapter.request('git_is_repository', {
+      request: { repositoryPath: '/a' },
+    });
+    const low2 = adapter.request('ssh_is_connected', { connectionId: 'ssh-x' });
+    await Promise.resolve();
+    expect(started).toEqual(['git_is_repository']);
+
+    const terminal = adapter.request('terminal_write', {
+      request: { sessionId: 't1', data: 'pwd\r' },
+    });
+    await terminal;
+    expect(started).toEqual(['git_is_repository', 'terminal_write']);
+
+    firstLowGate.resolve();
+    await Promise.all([low1, low2]);
+    expect(started).toEqual([
+      'git_is_repository',
+      'terminal_write',
+      'ssh_is_connected',
+    ]);
+  });
+
+  it('sends split-endpoint file reads as direct peer commands', async () => {
+    const deviceRpc = vi.fn(async (_target: string, commandJson: string) => {
+      const parsed = JSON.parse(commandJson) as { cmd: string; path: string };
+      expect(parsed).toEqual({
+        cmd: 'get_file_info',
+        path: '/peer/report.bin',
+        session_id: null,
+      });
+      return JSON.stringify({
+        resp: 'file_info',
+        name: 'report.bin',
+        size: 4,
+        mime_type: 'application/octet-stream',
+      });
+    });
+    const adapter = new PeerDeviceTransportAdapter('peer-1', deviceRpc);
+
+    const response = await adapter.requestPeerCommand({
+      cmd: 'get_file_info',
+      path: '/peer/report.bin',
+      session_id: null,
+    });
+
+    expect(response.resp).toBe('file_info');
+    expect(deviceRpc).toHaveBeenCalledTimes(1);
+  });
 });
 
 function createDeferred<T>() {