Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions docs/architecture/peer-device-mode.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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`,
Expand Down
33 changes: 31 additions & 2 deletions src/apps/desktop/src/api/clipboard_file_api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;

Expand Down Expand Up @@ -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");
}
}
51 changes: 49 additions & 2 deletions src/apps/desktop/src/api/commands.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down
75 changes: 60 additions & 15 deletions src/apps/desktop/src/api/terminal_api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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;
}
}
Expand All @@ -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),
Expand Down Expand Up @@ -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(());
}

Expand Down Expand Up @@ -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);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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>() {
Expand Down
Loading
Loading