diff --git a/docs/architecture/peer-device-mode.md b/docs/architecture/peer-device-mode.md index 5494e485a4..ef93a06e92 100644 --- a/docs/architecture/peer-device-mode.md +++ b/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`, diff --git a/src/apps/desktop/src/api/clipboard_file_api.rs b/src/apps/desktop/src/api/clipboard_file_api.rs index 9d7768eb23..47859d2c4b 100644 --- a/src/apps/desktop/src/api/clipboard_file_api.rs +++ b/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"); + } } diff --git a/src/apps/desktop/src/api/commands.rs b/src/apps/desktop/src/api/commands.rs index 4f80e412d5..8070eb70ce 100644 --- a/src/apps/desktop/src/api/commands.rs +++ b/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 diff --git a/src/apps/desktop/src/api/terminal_api.rs b/src/apps/desktop/src/api/terminal_api.rs index 63009e3447..541e31e3e2 100644 --- a/src/apps/desktop/src/api/terminal_api.rs +++ b/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); + } +} diff --git a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts index b51d78b30b..eab838d8e3 100644 --- a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts +++ b/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(); + + 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() { diff --git a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts index 5e4d04629e..cb86f33c18 100644 --- a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts +++ b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts @@ -153,7 +153,7 @@ const LOW_PRIORITY_EXACT = new Set([ ]); export function peerInvokePriorityFor(command: string): PeerInvokePriority { - if (HIGH_PRIORITY_COMMANDS.has(command)) { + if (HIGH_PRIORITY_COMMANDS.has(command) || command.startsWith('terminal_')) { return 'high'; } if ( @@ -189,6 +189,11 @@ interface HostInvokeResultEnvelope { message?: string; } +export interface PeerDeviceCommandResponse { + resp?: string; + message?: string; +} + /** Product-level HostInvoke failure (peer executed the command and returned ok:false). */ export class PeerProductCommandError extends Error { readonly isPeerProductError = true; @@ -218,6 +223,11 @@ export class PeerDeviceTransportAdapter implements ITransportAdapter { private readonly local = new TauriTransportAdapter(); private connected = false; private activeCount = 0; + private readonly activeByPriority: Record = { + high: 0, + normal: 0, + low: 0, + }; private readonly queues: Record = { high: [], normal: [], @@ -254,6 +264,21 @@ export class PeerDeviceTransportAdapter implements ITransportAdapter { return this.enqueue(priority, () => this.invokeOnPeer(action, params, timing, transportStartedAt)); } + /** + * Send an existing RemoteCommand envelope directly to the peer. This is for + * split-endpoint operations such as file download, where the peer reads the + * source but the controller owns the destination path and local write. + */ + async requestPeerCommand( + command: Record, + priority: PeerInvokePriority = 'normal', + ): Promise { + if (!this.connected) { + await this.connect(); + } + return this.enqueue(priority, () => this.invokePeerCommand(command, priority)); + } + listen(event: string, callback: (data: T) => void): () => void { return this.local.listen(event, callback); } @@ -267,6 +292,7 @@ export class PeerDeviceTransportAdapter implements ITransportAdapter { this.connected = false; for (const priority of ['high', 'normal', 'low'] as const) { this.queues[priority].length = 0; + this.activeByPriority[priority] = 0; } this.activeCount = 0; } @@ -308,8 +334,13 @@ export class PeerDeviceTransportAdapter implements ITransportAdapter { return; } this.activeCount += 1; + this.activeByPriority[next.priority] += 1; void next.run().finally(() => { - this.activeCount -= 1; + this.activeCount = Math.max(0, this.activeCount - 1); + this.activeByPriority[next.priority] = Math.max( + 0, + this.activeByPriority[next.priority] - 1, + ); this.pump(); }); } @@ -324,12 +355,47 @@ export class PeerDeviceTransportAdapter implements ITransportAdapter { if (this.queues.normal.length > 0) { return this.queues.normal.shift(); } - if (this.queues.low.length > 0) { + // Keep one transport slot available for future interactive work. Without + // this, two slow background RPCs can make terminal input appear frozen. + const lowConcurrencyLimit = this.maxConcurrent > 1 ? this.maxConcurrent - 1 : 1; + if ( + this.queues.low.length > 0 && + this.activeByPriority.low < lowConcurrencyLimit + ) { return this.queues.low.shift(); } return undefined; } + private async invokePeerCommand( + command: Record, + priority: PeerInvokePriority, + ): Promise { + const action = typeof command.cmd === 'string' ? command.cmd : 'unknown'; + try { + const raw = await this.deviceRpc(this.targetDeviceId, JSON.stringify(command)); + const envelope = JSON.parse(raw) as T; + if (envelope.resp === 'error') { + throw new PeerProductCommandError( + envelope.message || `Peer command '${action}' failed`, + ); + } + if (!envelope.resp) { + throw new Error(`Unexpected peer RPC response for '${action}'`); + } + this.hooks.onHostInvokeSuccess?.(); + return envelope; + } catch (error) { + if (error instanceof PeerProductCommandError) { + log.warn('Peer product command failed', { action, error }); + throw error; + } + log.error('Peer direct command transport failed', { action, error }); + this.hooks.onHostInvokeTransportFailure?.(error, { action, priority }); + throw error; + } + } + private async invokeOnPeer( action: string, params: unknown, diff --git a/src/web-ui/src/infrastructure/peer-device/README.md b/src/web-ui/src/infrastructure/peer-device/README.md index d024983edc..2f75ef6d5d 100644 --- a/src/web-ui/src/infrastructure/peer-device/README.md +++ b/src/web-ui/src/infrastructure/peer-device/README.md @@ -50,6 +50,18 @@ Controller-side React/transport layer for Peer Device Mode. Architecture: `createChatSession` when a live workspace exists — use `flowChatSessionConfigForCurrentWorkspace`. +10. **Download destinations stay on the controller.** Native dialogs select a + path on A. Read file chunks from B with direct Peer commands, then write + them through A's local filesystem adapter. Do not HostInvoke + `export_local_file_to_path` with A's path. Directory downloads must preserve + the tree and reject traversal-like entry names. + +11. **Terminal traffic stays interactive and observable.** All `terminal_*` + commands are high priority, low-priority polling leaves one transport slot + available, and both local and SSH-backed PTY events on B must fan out to A. + Remote `SIGINT` / `SIGTSTP` map to PTY control bytes instead of silently + succeeding without affecting the process. + ## Related account-login guards Incomplete login (cloud vs local settings choice) must not persist a session diff --git a/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.test.ts b/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.test.ts index 311e8451b6..84507a594f 100644 --- a/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.test.ts +++ b/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.test.ts @@ -1,11 +1,27 @@ import { describe, expect, it } from "vitest"; import { + decodeBase64FileChunk, + isSafePeerTransferEntryName, joinWorkspaceTargetPath, normalizeClipboardLocalPaths, resolvePasteTargetDirectory, } from "./workspaceFileTransfer"; describe("workspaceFileTransfer", () => { + it("decodes peer file chunks without corrupting binary bytes", () => { + expect(Array.from(decodeBase64FileChunk("AP+AAQI="))).toEqual([ + 0x00, 0xff, 0x80, 0x01, 0x02, + ]); + }); + + it("rejects peer directory entries that can escape the selected destination", () => { + expect(isSafePeerTransferEntryName("report.txt")).toBe(true); + expect(isSafePeerTransferEntryName("..")).toBe(false); + expect(isSafePeerTransferEntryName("nested/file.txt")).toBe(false); + expect(isSafePeerTransferEntryName("nested\\file.txt")).toBe(false); + expect(isSafePeerTransferEntryName("bad\0name")).toBe(false); + }); + it("joins remote workspace paths with POSIX separators", () => { expect( joinWorkspaceTargetPath("/home/user/project/", "file.txt", true), diff --git a/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.ts b/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.ts index c63ac7c256..46e63e1c2f 100644 --- a/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.ts +++ b/src/web-ui/src/tools/file-system/services/workspaceFileTransfer.ts @@ -5,6 +5,11 @@ import { PhysicalPosition } from "@tauri-apps/api/dpi"; import { sshApi } from "@/features/ssh-remote/sshApi"; import { workspaceAPI } from "@/infrastructure/api"; +import { getTransportAdapter } from "@/infrastructure/api/adapters"; +import { + PeerDeviceTransportAdapter, + type PeerDeviceCommandResponse, +} from "@/infrastructure/api/adapters/peer-device-adapter"; import { i18nService } from "@/infrastructure/i18n"; import { isRemoteWorkspace, type WorkspaceInfo } from "@/shared/types"; import { @@ -42,6 +47,257 @@ export interface UploadToWorkspaceOptions { isCut?: boolean; } +const PEER_FILE_CHUNK_BYTES = 1024 * 1024; + +interface PeerFileInfoResponse extends PeerDeviceCommandResponse { + resp: "file_info"; + name: string; + size: number; + mime_type: string; +} + +interface PeerFileChunkResponse extends PeerDeviceCommandResponse { + resp: "file_chunk"; + name: string; + chunk_base64: string; + offset: number; + chunk_size: number; + total_size: number; + mime_type: string; +} + +interface PeerDownloadEntry { + sourcePath: string; + destinationPath: string; + size: number; + name: string; +} + +export function decodeBase64FileChunk(value: string): Uint8Array { + const binary = atob(value); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; i += 1) { + bytes[i] = binary.charCodeAt(i); + } + return bytes; +} + +export function isSafePeerTransferEntryName(name: string): boolean { + return Boolean( + name && + name !== "." && + name !== ".." && + !name.includes("/") && + !name.includes("\\") && + !name.includes("\0"), + ); +} + +function currentPeerAdapter(): PeerDeviceTransportAdapter | null { + const adapter = getTransportAdapter(); + return adapter instanceof PeerDeviceTransportAdapter ? adapter : null; +} + +async function writeAllToLocalFile( + destinationPath: string, + chunks: AsyncIterable, + onChunkWritten: (bytes: number) => void, +): Promise { + const { open } = await import("@tauri-apps/plugin-fs"); + const file = await open(destinationPath, { + write: true, + create: true, + truncate: true, + }); + try { + for await (const chunk of chunks) { + const written = await file.write(chunk); + if (written !== chunk.byteLength) { + throw new Error( + `Incomplete local file write (${written}/${chunk.byteLength} bytes)`, + ); + } + onChunkWritten(written); + } + } finally { + await file.close(); + } +} + +async function* readPeerFileChunks( + adapter: PeerDeviceTransportAdapter, + sourcePath: string, + onFileSize: (size: number) => void, +): AsyncGenerator { + const info = await adapter.requestPeerCommand({ + cmd: "get_file_info", + path: sourcePath, + session_id: null, + }); + if (info.resp !== "file_info" || !Number.isSafeInteger(info.size) || info.size < 0) { + throw new Error(`Invalid peer file info response for '${sourcePath}'`); + } + onFileSize(info.size); + + let offset = 0; + while (offset < info.size) { + const response = await adapter.requestPeerCommand({ + cmd: "read_file_chunk", + path: sourcePath, + session_id: null, + offset, + limit: PEER_FILE_CHUNK_BYTES, + }); + if ( + response.resp !== "file_chunk" || + response.offset !== offset || + !Number.isSafeInteger(response.chunk_size) || + response.chunk_size < 0 || + !Number.isSafeInteger(response.total_size) || + response.total_size !== info.size || + response.chunk_size > Math.min(PEER_FILE_CHUNK_BYTES, info.size - offset) + ) { + throw new Error(`Invalid peer file chunk response for '${sourcePath}'`); + } + const bytes = decodeBase64FileChunk(response.chunk_base64); + if ( + bytes.byteLength !== response.chunk_size || + bytes.byteLength === 0 || + offset + bytes.byteLength > info.size + ) { + throw new Error(`Incomplete peer file chunk for '${sourcePath}' at offset ${offset}`); + } + offset += bytes.byteLength; + yield bytes; + } + return info.size; +} + +async function* readPeerSshFile( + sourcePath: string, + remoteConnectionId: string | undefined, +): AsyncGenerator { + const content = await workspaceAPI.readFileContent( + sourcePath, + "base64", + remoteConnectionId, + ); + yield decodeBase64FileChunk(content); +} + +async function collectPeerDirectoryEntries( + sourceDirectory: string, + destinationDirectory: string, +): Promise { + const { mkdir } = await import("@tauri-apps/plugin-fs"); + const pending = [{ source: sourceDirectory, destination: destinationDirectory }]; + const files: PeerDownloadEntry[] = []; + + while (pending.length > 0) { + const current = pending.shift()!; + await mkdir(current.destination, { recursive: true }); + const children = await workspaceAPI.getDirectoryChildren(current.source); + for (const child of children) { + if (!isSafePeerTransferEntryName(child.name)) { + throw new Error(`Unsafe peer file name: '${child.name}'`); + } + const destinationPath = joinWorkspaceTargetPath( + current.destination, + child.name, + false, + ); + if (child.isDirectory) { + pending.push({ source: child.path, destination: destinationPath }); + } else { + files.push({ + sourcePath: child.path, + destinationPath, + size: typeof child.size === "number" && child.size >= 0 ? child.size : 0, + name: child.name, + }); + } + } + } + + return files; +} + +async function downloadPeerWorkspacePathToDisk( + adapter: PeerDeviceTransportAdapter, + sourcePath: string, + destinationPath: string, + workspace: WorkspaceInfo | null, + isDirectory: boolean, + onProgress: (state: TransferProgressState | null) => void, +): Promise { + const entries = isDirectory + ? await collectPeerDirectoryEntries(sourcePath, destinationPath) + : [{ + sourcePath, + destinationPath, + size: 0, + name: sourcePath.split(/[/\\]/).pop() || "file", + }]; + let bytesTransferred = 0; + let bytesTotal = entries.reduce((sum, entry) => sum + entry.size, 0); + let lastTime = performance.now(); + let lastBytes = 0; + let smoothedSpeed = 0; + + for (const entry of entries) { + const entryStart = bytesTransferred; + let entryWritten = 0; + let expectedEntrySize = entry.size; + const updateExpectedEntrySize = (size: number) => { + bytesTotal += size - expectedEntrySize; + expectedEntrySize = size; + }; + const chunks = isRemoteWorkspace(workspace) + ? readPeerSshFile(entry.sourcePath, workspace?.connectionId) + : readPeerFileChunks(adapter, entry.sourcePath, updateExpectedEntrySize); + await writeAllToLocalFile(entry.destinationPath, chunks, (written) => { + entryWritten += written; + bytesTransferred = entryStart + entryWritten; + if (expectedEntrySize === 0 && entryWritten > 0) { + bytesTotal = Math.max(bytesTotal, bytesTransferred); + } + const now = performance.now(); + const elapsed = now - lastTime; + const byteDelta = bytesTransferred - lastBytes; + if (elapsed > 0 && byteDelta > 0) { + const instantSpeed = (byteDelta / elapsed) * 1000; + smoothedSpeed = smoothedSpeed === 0 + ? instantSpeed + : smoothedSpeed * 0.7 + instantSpeed * 0.3; + } + lastTime = now; + lastBytes = bytesTransferred; + onProgress({ + phase: "download", + current: bytesTransferred, + total: Math.max(bytesTotal, bytesTransferred, 1), + label: entry.name, + indeterminate: bytesTotal === 0, + bytesTransferred, + bytesTotal: Math.max(bytesTotal, bytesTransferred), + speed: smoothedSpeed, + }); + }); + bytesTransferred = entryStart + entryWritten; + } + + onProgress({ + phase: "download", + current: Math.max(bytesTransferred, 1), + total: Math.max(bytesTransferred, 1), + label: sourcePath.split(/[/\\]/).pop() || "file", + indeterminate: false, + bytesTransferred, + bytesTotal: bytesTransferred, + speed: smoothedSpeed, + }); +} + function normalizeClipboardLocalPath(path: string): string { const trimmed = path.trim(); if (!trimmed) { @@ -296,7 +552,7 @@ export async function downloadWorkspaceFileToDisk( if (picked === null) { return; } - dest = `${picked}${picked.endsWith("/") ? "" : "/"}${baseName}`; + dest = joinWorkspaceTargetPath(picked, baseName, false); } else { const { save } = await import("@tauri-apps/plugin-dialog"); dest = await save({ @@ -316,7 +572,17 @@ export async function downloadWorkspaceFileToDisk( indeterminate: true, }); try { - if (isRemoteWorkspace(workspace)) { + const peerAdapter = currentPeerAdapter(); + if (peerAdapter) { + await downloadPeerWorkspacePathToDisk( + peerAdapter, + filePath, + dest, + workspace, + Boolean(isDirectory), + onProgress, + ); + } else if (isRemoteWorkspace(workspace)) { const cid = workspace?.connectionId; if (!cid) { throw new Error(