diff --git a/frontend/package.json b/frontend/package.json index 03dcbf6..e1cc07a 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -5,6 +5,7 @@ "type": "module", "scripts": { "dev": "vite --port 5173", + "typecheck": "vue-tsc --noEmit", "build": "vite build", "preview": "vite preview" }, @@ -31,6 +32,7 @@ "postcss": "^8.4.49", "tailwindcss": "^3.4.17", "typescript": "~5.6.2", - "vite": "^6.0.5" + "vite": "^6.0.5", + "vue-tsc": "^3.3.13" } } diff --git a/frontend/pnpm-lock.yaml b/frontend/pnpm-lock.yaml index 4254797..061e576 100644 --- a/frontend/pnpm-lock.yaml +++ b/frontend/pnpm-lock.yaml @@ -72,6 +72,9 @@ importers: vite: specifier: ^6.0.5 version: 6.4.4(jiti@1.21.7) + vue-tsc: + specifier: ^3.3.13 + version: 3.3.13(typescript@5.6.3) packages: @@ -495,6 +498,20 @@ packages: vite: ^5.0.0 || ^6.0.0 vue: ^3.2.25 + '@volar/language-core@2.4.28': + resolution: {integrity: sha512-w4qhIJ8ZSitgLAkVay6AbcnC7gP3glYM3fYwKV3srj8m494E3xtrCv6E+bWviiK/8hs6e6t1ij1s2Endql7vzQ==} + + '@volar/source-map@2.4.28': + resolution: {integrity: sha512-yX2BDBqJkRXfKw8my8VarTyjv48QwxdJtvRgUpNE5erCsgEUdI2DsLbpa+rOQVAJYshY99szEcRDmyHbF10ggQ==} + + '@volar/typescript@2.4.28': + resolution: {integrity: sha512-Ja6yvWrbis2QtN4ClAKreeUZPVYMARDYZl9LMEv1iQ1QdepB6wn0jTRxA9MftYmYa4DQ4k/DaSZpFPUfxl8giw==} + peerDependencies: + typescript: '*' + peerDependenciesMeta: + typescript: + optional: true + '@vue/compiler-core@3.5.43': resolution: {integrity: sha512-zdiLhnbe1QQqgDT8xZMpNmyqZ3qlI+/Q/FHQco57Kwl/b05HhCzN6eVGN9QU9rbga4CrS0H5SYY8VGHZCt/1Hg==} @@ -507,6 +524,9 @@ packages: '@vue/compiler-ssr@3.5.43': resolution: {integrity: sha512-GF62orf7KiJX9RqrHNGrYBudsQGD0OhJ5nDs90O8UiDuS40+YMYomiXu6w6EuvtXuRDc3MSNis3EaaSKAVSWpg==} + '@vue/language-core@3.3.13': + resolution: {integrity: sha512-axA0Xemn5LuUKuKWob0/kVq7xQ2mJJM6VqA9Exvt35nM9yRC31goWVve/G/Zb2ZCLsXC8FHYjjfXlnshknSCew==} + '@vue/reactivity@3.5.43': resolution: {integrity: sha512-G/c9GyOZNI2jVaaS6OX1EF1SSFSv7H0ERqNTl4+DTFMlZmB5eVAB53aLQNam/7NL2NPtaDD7RdVrzf8uJzMuOA==} @@ -540,6 +560,9 @@ packages: '@xterm/xterm@5.5.0': resolution: {integrity: sha512-hqJHYaQb5OptNunnyAnkHyM8aCjZ1MEIDTQu1iIbbTD/xops91NB5yq1ZK/dC2JDbVWtF23zUtl9JE2NqwT87A==} + alien-signals@3.2.1: + resolution: {integrity: sha512-I8FjmltrfnDFoZedi5CG8DghVYNhzb/Ijluz7tCSJH0xpd0484Kowhbb1XDYOxfJpU1p5wnM2X54dA+IfGyD1g==} + any-promise@1.3.0: resolution: {integrity: sha512-7UvmKalWRt1wgjL1RrGxoSJW/0QZFIegpeGvZG9kjp8vrRu55XTHbwnqq2GpXm9uLbcuhxm3IqX9OB4MZR1b2A==} @@ -718,6 +741,9 @@ packages: resolution: {integrity: sha512-PXwfBhYu0hBCPw8Dn0E+WDYb7af3dSLVWKi3HGv84IdF4TyFoC0ysxFd0Goxw7nSv4T/PzEJQxsYsEiFCKo2BA==} engines: {node: '>=8.6'} + muggle-string@0.4.1: + resolution: {integrity: sha512-VNTrAak/KhO2i8dqqnqnAHOa3cYBwXEZe9h+D5h/1ZqFSTEFHdM65lR7RoIqq3tBBYavsOXV84NoHXZ0AkPyqQ==} + mz@2.7.0: resolution: {integrity: sha512-z81GNO7nnYMEhrGh9LeymoE4+Yr0Wn5McHIZMK5cfQCl+NDX08sCZgUc9/6MHni9IWuFLm1Z3HTCXu2z9fN62Q==} @@ -742,6 +768,9 @@ packages: resolution: {integrity: sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==} engines: {node: '>= 6'} + path-browserify@1.0.1: + resolution: {integrity: sha512-b7uo2UCUOYZcnF/3ID0lulOJi/bafxa1xPe7ZPsammBSpjSWQkjNxlt635YGS2MiR9GjvuXCtz2emr3jbsz98g==} + path-parse@1.0.7: resolution: {integrity: sha512-LDJzPVEEEPR+y48z93A0Ed0yXb8pAByGWo/k5YYdYgpY2/2EsOsksJrq7lOHxryrVOn1ejG6oAp8ahvOIQD8sw==} @@ -927,6 +956,15 @@ packages: yaml: optional: true + vscode-uri@3.2.0: + resolution: {integrity: sha512-m2gXo3bn0G1kT9InzMf07fTbqMbGtyckj3bH5ktLO+1Ssv+yiATZ4dhwaQv9UZWxJh6E9IFGnQyjgWVDWVBDrg==} + + vue-tsc@3.3.13: + resolution: {integrity: sha512-jE4/bXdOMSHVxEGgQCqNllJy+9N3Gx6tfSAsg0yhFK/MAN6eXqISihupnDEKSCt12VVnjN2gI3AqLYcqoh5x/A==} + hasBin: true + peerDependencies: + typescript: '>=5.0.0' + vue@3.5.43: resolution: {integrity: sha512-o5qZoksdnjIKvW1srZ3ab7pcDNYAerBjRe54D0LBLfRdCYFrSgBHVXokMas35czQc0//lmx4/tuY4ZNQ+Rf2Ng==} peerDependencies: @@ -1265,6 +1303,20 @@ snapshots: vite: 6.4.4(jiti@1.21.7) vue: 3.5.43(typescript@5.6.3) + '@volar/language-core@2.4.28': + dependencies: + '@volar/source-map': 2.4.28 + + '@volar/source-map@2.4.28': {} + + '@volar/typescript@2.4.28(typescript@5.6.3)': + dependencies: + '@volar/language-core': 2.4.28 + path-browserify: 1.0.1 + vscode-uri: 3.2.0 + optionalDependencies: + typescript: 5.6.3 + '@vue/compiler-core@3.5.43': dependencies: '@babel/parser': 7.29.9 @@ -1295,6 +1347,16 @@ snapshots: '@vue/compiler-dom': 3.5.43 '@vue/shared': 3.5.43 + '@vue/language-core@3.3.13': + dependencies: + '@volar/language-core': 2.4.28 + '@vue/compiler-dom': 3.5.43 + '@vue/shared': 3.5.43 + alien-signals: 3.2.1 + muggle-string: 0.4.1 + path-browserify: 1.0.1 + picomatch: 4.0.7 + '@vue/reactivity@3.5.43': dependencies: '@vue/shared': 3.5.43 @@ -1333,6 +1395,8 @@ snapshots: '@xterm/xterm@5.5.0': {} + alien-signals@3.2.1: {} + any-promise@1.3.0: {} anymatch@3.1.3: @@ -1516,6 +1580,8 @@ snapshots: braces: 3.0.3 picomatch: 2.3.2 + muggle-string@0.4.1: {} + mz@2.7.0: dependencies: any-promise: 1.3.0 @@ -1532,6 +1598,8 @@ snapshots: object-hash@3.0.0: {} + path-browserify@1.0.1: {} + path-parse@1.0.7: {} picocolors@1.1.1: {} @@ -1717,6 +1785,14 @@ snapshots: fsevents: 2.3.3 jiti: 1.21.7 + vscode-uri@3.2.0: {} + + vue-tsc@3.3.13(typescript@5.6.3): + dependencies: + '@volar/typescript': 2.4.28(typescript@5.6.3) + '@vue/language-core': 3.3.13 + typescript: 5.6.3 + vue@3.5.43(typescript@5.6.3): dependencies: '@vue/compiler-dom': 3.5.43 diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 415418b..3868e66 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -121,6 +121,7 @@ const loadingMcpCalls = ref(false); const mcpStatus = ref<{ exePath: string; port: number; + env: Record; claudeCodePath: string; claudeCodeSynced: boolean; codexPath: string; @@ -661,6 +662,7 @@ const displayedMcpConfigJson = computed(() => { type: 'stdio', command: mcpStatus.value?.exePath || 'vps-switch.exe', args: ['--mcp-stdio'], + env: mcpStatus.value?.env, }, }, }, diff --git a/frontend/src/components/CodeEditor.vue b/frontend/src/components/CodeEditor.vue index 3bb4b19..28380bc 100644 --- a/frontend/src/components/CodeEditor.vue +++ b/frontend/src/components/CodeEditor.vue @@ -14,7 +14,7 @@ import { nginx } from '@codemirror/legacy-modes/mode/nginx'; import { properties } from '@codemirror/legacy-modes/mode/properties'; import { toml } from '@codemirror/legacy-modes/mode/toml'; import { dockerFile } from '@codemirror/legacy-modes/mode/dockerfile'; -import { sql } from '@codemirror/legacy-modes/mode/sql'; +import { standardSQL } from '@codemirror/legacy-modes/mode/sql'; import { powerShell } from '@codemirror/legacy-modes/mode/powershell'; import { xml } from '@codemirror/legacy-modes/mode/xml'; import { diff } from '@codemirror/legacy-modes/mode/diff'; @@ -155,7 +155,7 @@ function languageFor(filename: string) { case 'toml': return StreamLanguage.define(toml); case 'sql': - return StreamLanguage.define(sql.standardSQL); + return StreamLanguage.define(standardSQL); case 'xml': case 'html': case 'htm': diff --git a/frontend/src/components/FileBrowserPanel.vue b/frontend/src/components/FileBrowserPanel.vue index 8affcf5..34f944a 100644 --- a/frontend/src/components/FileBrowserPanel.vue +++ b/frontend/src/components/FileBrowserPanel.vue @@ -124,7 +124,7 @@ let pollTimer: ReturnType | null = null; // 可点击的面包屑分段：如 /root/123 -> [/] [root] [123]；本机 Windows -> [此电脑] [C:] [Users] const breadcrumbs = computed(() => { const p = currentPath.value.trim(); - const isLocalWin = props.host.kind === 'local'; + const isLocalWin = props.host.kind === 'local' && props.host.osInfo?.os === 'windows'; if (isLocalWin) { const crumbs: BreadcrumbSegment[] = [{ label: '此电脑', path: 'drives' }]; @@ -510,7 +510,19 @@ async function saveFileEditor() { function triggerDownload(item: FileEntry) { if (item.isDir) return; - // 改走带断点续传的传输引擎（见 openInlineAction('download')），不再用浏览器直下 + if (!['127.0.0.1', 'localhost', '[::1]', '::1'].includes(window.location.hostname)) { + // 公网 Web 的本机是浏览器所在机器，交付给浏览器自身的流式下载链路。 + const params = new URLSearchParams({ hostId: props.host.id, path: item.path }); + const link = document.createElement('a'); + link.href = `/api/files/download?${params}`; + link.download = item.name; + document.body.appendChild(link); + link.click(); + link.remove(); + ensureTransferPolling(); + emit('toast', `已发起下载 ${item.name}，请在浏览器下载列表中查看`); + return; + } openInlineAction('download', item); } diff --git a/frontend/src/env.d.ts b/frontend/src/env.d.ts new file mode 100644 index 0000000..5e8aac1 --- /dev/null +++ b/frontend/src/env.d.ts @@ -0,0 +1,7 @@ +/// + +declare module '*.vue' { + import type { DefineComponent } from 'vue'; + const component: DefineComponent; + export default component; +} diff --git a/src/api.rs b/src/api.rs index cb2dabd..7935647 100644 --- a/src/api.rs +++ b/src/api.rs @@ -1,8 +1,6 @@ use crate::{ db::Database, - files::{ - self, join_remote_path, TransferManager, - }, + files::{self, join_remote_path, TransferManager}, inspector::InspectorEngine, mcp::{handle_mcp_request, sync_cli_mcp_configs, ExecTaskManager, JsonRpcRequest}, models::{HostMetaPatch, HostUpsertPayload, ReorderPayload, LOCAL_HOST_ID}, @@ -31,14 +29,13 @@ use futures::{ }; use serde::Deserialize; use serde_json::json; +#[cfg(windows)] +use std::os::windows::process::CommandExt; use std::{ collections::HashSet, sync::{Arc, Mutex}, time::Instant, }; -#[cfg(windows)] -use std::os::windows::process::CommandExt; -use tokio::io::AsyncReadExt; use tokio_util::io::StreamReader; use uuid::Uuid; @@ -60,7 +57,10 @@ pub struct AppState { pub fn build_api_router(state: Arc) -> Router { let protected_api = Router::new() .route("/api/system/about", get(api_get_about)) - .route("/api/system/autostart", get(api_get_autostart).post(api_set_autostart)) + .route( + "/api/system/autostart", + get(api_get_autostart).post(api_set_autostart), + ) .route("/api/system/check-update", post(api_check_update)) .route("/api/mcp/status", get(api_mcp_status).post(api_mcp_sync)) .route("/api/settings", get(api_get_settings)) @@ -383,9 +383,7 @@ async fn list_terminals( State(state): State>, Query(query): Query, ) -> impl IntoResponse { - let panes = state - .terminals - .list_panes(query.host_id.as_deref(), false); + let panes = state.terminals.list_panes(query.host_id.as_deref(), false); ( StatusCode::OK, Json(json!({ @@ -516,9 +514,9 @@ async fn handle_terminal_ws(socket: WebSocket, pane: Arc) { match incoming { Some(Ok(Message::Binary(data))) => { if data.first() == Some(&0x01) && data.len() > 1 { - let _ = pane.cmd_tx.send(PaneCommand::Input(data[1..].to_vec())); + let _ = pane.cmd_tx.send(PaneCommand::Input { bytes: data[1..].to_vec(), sent: None }); } else if !data.is_empty() { - let _ = pane.cmd_tx.send(PaneCommand::Input(data)); + let _ = pane.cmd_tx.send(PaneCommand::Input { bytes: data, sent: None }); } } Some(Ok(Message::Text(text))) => { @@ -707,7 +705,14 @@ async fn api_rename_item( State(state): State>, Json(p): Json, ) -> impl IntoResponse { - match files::rename_item(&state.ssh_pool, &state.db, &p.host_id, &p.old_path, &p.new_path).await + match files::rename_item( + &state.ssh_pool, + &state.db, + &p.host_id, + &p.old_path, + &p.new_path, + ) + .await { Ok(()) => (StatusCode::OK, Json(json!({ "ok": true }))).into_response(), Err(e) => err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(), @@ -840,7 +845,15 @@ async fn api_stream_upload( } } -/// 浏览器高速流式下载：直接把 64 路并发 SFTP 预读流作为 HTTP 响应输出，同时实时更新传输队列速率 +struct TemporaryDownload(std::path::PathBuf); + +impl Drop for TemporaryDownload { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// 浏览器下载沿用传输引擎，完成校验后通过 HTTP 交付，临时文件随响应关闭而清理。 async fn api_stream_download( State(state): State>, Query(q): Query, @@ -849,14 +862,6 @@ async fn api_stream_download( return err_response(StatusCode::BAD_REQUEST, "缺少下载路径").into_response(); }; - let host_name = state - .db - .get_record(&q.host_id) - .ok() - .flatten() - .map(|h| h.name) - .unwrap_or_else(|| q.host_id.clone()); - let filename = file_path .rsplit(['/', '\\']) .next() @@ -876,82 +881,60 @@ async fn api_stream_download( return build_download_response(&filename, meta.len(), Body::from_stream(stream)); } - let sftp = match state.ssh_pool.get_or_open_sftp(&state.db, &q.host_id).await { - Ok(s) => s, + let temp_dir = + std::env::temp_dir().join(format!("vps-switch-browser-download-{}", Uuid::new_v4())); + if let Err(e) = tokio::fs::create_dir(&temp_dir).await { + return err_response(StatusCode::INTERNAL_SERVER_ERROR, e).into_response(); + } + let cleanup = TemporaryDownload(temp_dir); + let local_path = cleanup.0.join("file"); + let task = match files::transfer_local_remote_file( + &state.ssh_pool, + &state.db, + &state.transfers, + &q.host_id, + "download", + &local_path.to_string_lossy(), + &file_path, + ) + .await + { + Ok(t) => t, Err(e) => return err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(), }; - - let meta = match sftp.metadata(&file_path).await { - Ok(m) => m, - Err(e) => return err_response(StatusCode::NOT_FOUND, format!("{e:#}")).into_response(), - }; - if meta.is_dir() { - return err_response(StatusCode::BAD_REQUEST, "不能直接下载文件夹").into_response(); - } - let total_size = meta.size.unwrap_or(0); - - let remote_file = match sftp.open(&file_path).await { + let file = match tokio::fs::File::open(&local_path).await { Ok(f) => f, Err(e) => return err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(), }; - - let task_id = state.transfers.create_task( - &q.host_id, - &host_name, - "download", - &filename, - None, - &file_path, - total_size, - ); - - let transfers = state.transfers.clone(); - let started = Instant::now(); - - let stream = futures::stream::unfold( - (remote_file, 0u64, task_id, transfers, false), - move |(mut rfile, mut transferred, tid, tm, done)| async move { - if done { - return None; - } - let mut buf = vec![0u8; 512 * 1024]; - match rfile.read(&mut buf).await { - Ok(0) => { - let _ = rfile.close().await; - tm.finish_task(&tid, transferred, started.elapsed().as_secs_f64()); - None - } - Ok(n) => { - buf.truncate(n); - transferred += n as u64; - tm.update_progress( - &tid, - transferred, - total_size, - started.elapsed().as_secs_f64(), - ); - Some(( - Ok::(bytes::Bytes::from(buf)), - (rfile, transferred, tid, tm, false), - )) - } - Err(err) => { - tm.fail_task(&tid, &err); - Some((Err(err), (rfile, transferred, tid, tm, true))) - } - } - }, - ); - - build_download_response(&filename, total_size, Body::from_stream(stream)) + let reader = tokio_util::io::ReaderStream::with_capacity(file, 512 * 1024); + let stream = futures::stream::unfold((reader, cleanup), |(mut reader, cleanup)| async move { + reader.next().await.map(|bytes| (bytes, (reader, cleanup))) + }); + build_download_response(&filename, task.transferred_bytes, Body::from_stream(stream)) } fn build_download_response(filename: &str, size: u64, body: Body) -> Response { let safe_ascii: String = filename .chars() - .map(|c| if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { c } else { '_' }) + .map(|c| { + if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { + c + } else { + '_' + } + }) .collect(); - let disposition = format!("attachment; filename=\"{safe_ascii}\""); + let encoded: String = filename + .bytes() + .map(|b| { + if b.is_ascii_alphanumeric() || b"-._~".contains(&b) { + (b as char).to_string() + } else { + format!("%{b:02X}") + } + }) + .collect(); + let disposition = format!("attachment; filename=\"{safe_ascii}\"; filename*=UTF-8''{encoded}"); Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -1013,8 +996,11 @@ async fn api_create_local_remote_transfer( Json(p): Json, ) -> impl IntoResponse { if p.direction != "upload" && p.direction != "download" { - return err_response(StatusCode::BAD_REQUEST, "direction 只能是 upload 或 download") - .into_response(); + return err_response( + StatusCode::BAD_REQUEST, + "direction 只能是 upload 或 download", + ) + .into_response(); } if state.db.get_record(&p.host_id).ok().flatten().is_none() { return err_response(StatusCode::BAD_REQUEST, "主机不存在").into_response(); @@ -1071,7 +1057,10 @@ async fn api_clear_transfers(State(state): State>) -> impl IntoRes const SESSION_COOKIE_NAME: &str = "vps_switch_session"; fn extract_session_token(headers: &HeaderMap) -> Option { - if let Some(auth) = headers.get(header::AUTHORIZATION).and_then(|v| v.to_str().ok()) { + if let Some(auth) = headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + { if let Some(tok) = auth.strip_prefix("Bearer ") { if !tok.trim().is_empty() { return Some(tok.trim().to_string()); @@ -1149,12 +1138,13 @@ async fn api_auth_login( State(state): State>, Json(p): Json, ) -> Response { - match state - .db - .verify_admin_credentials(&p.username, &p.password) - { + match state.db.verify_admin_credentials(&p.username, &p.password) { Ok(true) => { - let token = format!("sess_{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let token = format!( + "sess_{}{}", + Uuid::new_v4().simple(), + Uuid::new_v4().simple() + ); state.sessions.lock().unwrap().insert(token.clone()); let cookie = format!( "{SESSION_COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000" @@ -1178,10 +1168,7 @@ async fn api_auth_login( } } -async fn api_auth_logout( - State(state): State>, - headers: HeaderMap, -) -> Response { +async fn api_auth_logout(State(state): State>, headers: HeaderMap) -> Response { if let Some(tok) = extract_session_token(&headers) { state.sessions.lock().unwrap().remove(&tok); } @@ -1200,7 +1187,12 @@ fn check_autostart_enabled() -> bool { #[cfg(windows)] { std::process::Command::new("reg") - .args(["query", "HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run", "/v", "VPSSwitch"]) + .args([ + "query", + "HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run", + "/v", + "VPSSwitch", + ]) .creation_flags(CREATE_NO_WINDOW) .output() .map(|o| o.status.success()) @@ -1327,62 +1319,72 @@ async fn api_get_about(State(state): State>) -> impl IntoResponse async fn api_check_update() -> impl IntoResponse { let current_version = env!("CARGO_PKG_VERSION"); - + let started = Instant::now(); + let mut command = tokio::process::Command::new("gh"); + command.args([ + "release", + "view", + "--repo", + "weidu12123/vps-switch", + "--json", + "tagName,name,publishedAt,body,url", + ]); #[cfg(windows)] - let gh_res = std::process::Command::new("gh") - .args([ - "release", - "view", - "--repo", - "weidu12123/vps-switch", - "--json", - "tagName,name,publishedAt,body,url", - ]) - .creation_flags(CREATE_NO_WINDOW) - .output(); - - #[cfg(not(windows))] - let gh_res = std::process::Command::new("gh") - .args([ - "release", - "view", - "--repo", - "weidu12123/vps-switch", - "--json", - "tagName,name,publishedAt,body,url", - ]) - .output(); - - if let Ok(out) = gh_res { - if out.status.success() { - if let Ok(info) = serde_json::from_slice::(&out.stdout) { - let tag = info["tagName"].as_str().unwrap_or("").trim_start_matches('v'); - let has_update = !tag.is_empty() && tag != current_version; - return ( - StatusCode::OK, - Json(json!({ - "ok": true, - "currentVersion": format!("v{current_version}"), - "latestVersion": info["tagName"].as_str().unwrap_or(&format!("v{current_version}")), - "hasUpdate": has_update, - "publishedAt": info["publishedAt"].as_str(), - "releaseNotes": info["body"].as_str(), - "releaseUrl": info["url"].as_str(), - })), - ); + command.creation_flags(CREATE_NO_WINDOW); + let mut stdout = String::new(); + let mut stderr = String::new(); + let mut exit_code = None; + let error = match command.output().await { + Ok(out) => { + stdout = String::from_utf8_lossy(&out.stdout).into_owned(); + stderr = String::from_utf8_lossy(&out.stderr).into_owned(); + exit_code = out.status.code(); + if out.status.success() { + match serde_json::from_slice::(&out.stdout) { + Ok(info) + if info["tagName"] + .as_str() + .is_some_and(|s| !s.trim().is_empty()) => + { + let tag = info["tagName"].as_str().unwrap().trim_start_matches('v'); + let has_update = !tag.is_empty() && tag != current_version; + return ( + StatusCode::OK, + Json(json!({ + "ok": true, + "currentVersion": format!("v{current_version}"), + "latestVersion": info["tagName"].as_str().unwrap_or(&format!("v{current_version}")), + "hasUpdate": has_update, + "publishedAt": info["publishedAt"].as_str(), + "releaseNotes": info["body"].as_str(), + "releaseUrl": info["url"].as_str(), + "stdout": stdout, + "stderr": stderr, + "exitCode": exit_code, + "durationMs": started.elapsed().as_millis() as u64, + })), + ); + } + Ok(_) => "更新响应缺少版本号".to_string(), + Err(e) => format!("无法解析更新响应: {e}"), + } + } else { + format!("更新查询失败（退出码 {:?}）: {}", exit_code, stderr.trim()) } } - } + Err(e) => format!("无法运行 gh 查询更新: {e}"), + }; ( - StatusCode::OK, + StatusCode::BAD_GATEWAY, Json(json!({ - "ok": true, + "ok": false, "currentVersion": format!("v{current_version}"), - "latestVersion": format!("v{current_version}"), - "hasUpdate": false, - "message": "当前已是最新版本", - "releaseUrl": "https://github.com/weidu12123/vps-switch", + "error": error, + "stdout": stdout, + "stderr": stderr, + "exitCode": exit_code, + "durationMs": started.elapsed().as_millis() as u64, })), ) } @@ -1432,7 +1434,11 @@ async fn api_update_auth_settings( { Ok(settings) => { // 若刚开启了登录保护，自动为当前管理员签发会话 Cookie，避免改完设置立刻被踢出 - let token = format!("sess_{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let token = format!( + "sess_{}{}", + Uuid::new_v4().simple(), + Uuid::new_v4().simple() + ); state.sessions.lock().unwrap().insert(token.clone()); let cookie = format!( "{SESSION_COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000" @@ -1467,11 +1473,10 @@ async fn api_update_mcp_settings( State(state): State>, Json(p): Json, ) -> impl IntoResponse { - match state.db.update_mcp_settings( - p.mcp_enabled, - p.mcp_require_api_key, - p.regenerate_key, - ) { + match state + .db + .update_mcp_settings(p.mcp_enabled, p.mcp_require_api_key, p.regenerate_key) + { Ok(settings) => ( StatusCode::OK, Json(json!({ @@ -1519,7 +1524,10 @@ async fn api_update_guard( } fn extract_mcp_api_key(headers: &HeaderMap) -> Option { - if let Some(auth) = headers.get(header::AUTHORIZATION).and_then(|v| v.to_str().ok()) { + if let Some(auth) = headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + { if let Some(k) = auth.strip_prefix("Bearer ") { if !k.trim().is_empty() { return Some(k.trim().to_string()); @@ -1544,7 +1552,8 @@ async fn api_mcp_rpc( }; if !settings.mcp_enabled { - return err_response(StatusCode::FORBIDDEN, "MCP Server 当前已在设置中关闭").into_response(); + return err_response(StatusCode::FORBIDDEN, "MCP Server 当前已在设置中关闭") + .into_response(); } // API Key 只在**非回环绑定**（真的对外部署）时才校验。绑在本机回环时，端点只有本机 @@ -1626,4 +1635,3 @@ async fn api_sse_events(State(state): State>) -> impl IntoResponse }); Sse::new(stream).keep_alive(KeepAlive::new().interval(std::time::Duration::from_secs(15))) } - diff --git a/src/db.rs b/src/db.rs index b18220e..3bf64c5 100644 --- a/src/db.rs +++ b/src/db.rs @@ -161,10 +161,10 @@ impl Database { .or_else(|_| std::env::var("USER")) .unwrap_or_else(|_| "local".to_string()); let os_info = OsInfo { - os: Some("windows".to_string()), - distro_id: Some("windows".to_string()), - version_id: Some("11".to_string()), - pretty_name: Some("Windows 11 Local".to_string()), + os: Some(std::env::consts::OS.to_string()), + distro_id: Some(std::env::consts::OS.to_string()), + version_id: None, + pretty_name: Some(format!("{} Local", std::env::consts::OS)), arch: Some(std::env::consts::ARCH.to_string()), kernel: None, cpu_model: None, @@ -191,17 +191,35 @@ impl Database { now ], )?; + } else if !cfg!(windows) { + let raw: Option = conn.query_row( + "SELECT os_json FROM hosts WHERE id = ?1", + params![LOCAL_HOST_ID], + |row| row.get(0), + )?; + if let Some(mut os) = raw.and_then(|s| serde_json::from_str::(&s).ok()) { + if os.os.as_deref() == Some("windows") { + os.os = Some(std::env::consts::OS.to_string()); + os.distro_id = Some(std::env::consts::OS.to_string()); + os.version_id = None; + os.pretty_name = Some(format!("{} Local", std::env::consts::OS)); + os.arch = Some(std::env::consts::ARCH.to_string()); + conn.execute( + "UPDATE hosts SET os_json = ?1 WHERE id = ?2", + params![serde_json::to_string(&os)?, LOCAL_HOST_ID], + )?; + } + } } Ok(()) } pub fn count_ssh_hosts(&self) -> Result { let conn = self.conn.lock().map_err(|_| anyhow!("数据库锁异常"))?; - let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM hosts WHERE kind = 'ssh'", - [], - |row| row.get(0), - )?; + let count: i64 = + conn.query_row("SELECT COUNT(*) FROM hosts WHERE kind = 'ssh'", [], |row| { + row.get(0) + })?; Ok(count as usize) } @@ -212,9 +230,7 @@ impl Database { let tags: Vec = serde_json::from_str(&tags_raw).unwrap_or_default(); let links: Vec = serde_json::from_str(&links_raw).unwrap_or_default(); - let os_info: Option = os_raw - .as_deref() - .and_then(|s| serde_json::from_str(s).ok()); + let os_info: Option = os_raw.as_deref().and_then(|s| serde_json::from_str(s).ok()); Ok(HostRecord { id: row.get("id")?, @@ -284,9 +300,13 @@ impl Database { pub fn get_record(&self, id: &str) -> Result> { let conn = self.conn.lock().map_err(|_| anyhow!("数据库锁异常"))?; - conn.query_row("SELECT * FROM hosts WHERE id = ?1", params![id], Self::row_to_record) - .optional() - .map_err(Into::into) + conn.query_row( + "SELECT * FROM hosts WHERE id = ?1", + params![id], + Self::row_to_record, + ) + .optional() + .map_err(Into::into) } pub fn get_public_host(&self, id: &str) -> Result> { @@ -624,8 +644,9 @@ impl Database { /// 从本机旧版 1Shell (%APPDATA%\1Shell\data\1shell.db) 导入全部主机与偏好 pub fn import_from_1shell(&self) -> Result { - let db_path = find_1shell_db_path() - .ok_or_else(|| anyhow!("未找到本机 1Shell 数据库 (%APPDATA%\\1Shell\\data\\1shell.db)"))?; + let db_path = find_1shell_db_path().ok_or_else(|| { + anyhow!("未找到本机 1Shell 数据库 (%APPDATA%\\1Shell\\data\\1shell.db)") + })?; let src_conn = Connection::open_with_flags(&db_path, OpenFlags::SQLITE_OPEN_READ_ONLY) .with_context(|| format!("只读打开 1Shell 数据库失败: {}", db_path.display()))?; @@ -817,8 +838,12 @@ impl Database { [], )?; conn.execute( - "INSERT OR IGNORE INTO settings (key, value) VALUES ('auth_enabled', 'false')", - [], + "INSERT OR IGNORE INTO settings (key, value) VALUES ('auth_enabled', ?1)", + params![if default_auth_enabled() { + "true" + } else { + "false" + }], )?; conn.execute( "INSERT OR IGNORE INTO settings (key, value) VALUES ('admin_username', 'admin')", @@ -902,17 +927,17 @@ impl Database { } pub fn get_app_settings(&self) -> Result { - let is_vps_mode = !cfg!(windows) - || std::env::var("HOST") - .map(|h| h == "0.0.0.0") - .unwrap_or(false); + let is_vps_mode = default_auth_enabled(); let auth_enabled = match self.get_setting("auth_enabled")? { Some(v) => v == "true", None => { // Windows 本机默认关闭，VPS/Linux 部署（或绑定 0.0.0.0）默认开启 let default_enabled = is_vps_mode; - let _ = self.set_setting("auth_enabled", if default_enabled { "true" } else { "false" }); + self.set_setting( + "auth_enabled", + if default_enabled { "true" } else { "false" }, + )?; default_enabled } }; @@ -931,8 +956,8 @@ impl Database { _ => { // 若是 VPS 部署且从未设置过密码，自动初始化一个初始密码 "vps-switch" if is_vps_mode { - if let Ok(Some(enc)) = self.master_key.encrypt("vps-switch") { - let _ = self.set_setting("admin_password_enc", &enc); + if let Some(enc) = self.master_key.encrypt("vps-switch")? { + self.set_setting("admin_password_enc", &enc)?; true } else { false @@ -1004,7 +1029,10 @@ impl Database { } self.set_setting("admin_username", clean_user)?; - self.set_setting("auth_enabled", if final_auth_enabled { "true" } else { "false" })?; + self.set_setting( + "auth_enabled", + if final_auth_enabled { "true" } else { "false" }, + )?; self.get_app_settings() } @@ -1052,6 +1080,15 @@ pub struct AppSettings { pub mcp_api_key: String, } +fn default_auth_enabled() -> bool { + !cfg!(windows) + || std::env::var("HOST") + .ok() + .and_then(|host| host.parse::().ok()) + .map(|ip| !ip.is_loopback()) + .unwrap_or(false) +} + fn generate_api_key() -> String { format!( "vpsk_{}{}", @@ -1062,7 +1099,10 @@ fn generate_api_key() -> String { fn find_1shell_db_path() -> Option { if let Ok(appdata) = std::env::var("APPDATA") { - let candidate = PathBuf::from(appdata).join("1Shell").join("data").join("1shell.db"); + let candidate = PathBuf::from(appdata) + .join("1Shell") + .join("data") + .join("1shell.db"); if candidate.exists() { return Some(candidate); } @@ -1101,3 +1141,68 @@ fn normalize_tags(tags: Vec) -> Vec { } out } + +#[cfg(test)] +mod tests { + use super::*; + + fn open_test_database() -> (PathBuf, Database) { + let dir = std::env::temp_dir().join(format!("vps-switch-db-test-{}", Uuid::new_v4())); + let key = MasterKey::load_or_create(&dir).unwrap(); + let db = Database::open(&dir, key).unwrap(); + (dir, db) + } + + #[test] + fn first_start_auth_and_local_metadata_match_platform() { + let (dir, db) = open_test_database(); + let settings = db.get_app_settings().unwrap(); + if !cfg!(windows) { + assert!(settings.auth_enabled); + assert!(settings.has_admin_password); + assert!(db.verify_admin_credentials("admin", "vps-switch").unwrap()); + } else if std::env::var("HOST").is_err() { + assert!(!settings.auth_enabled); + } + let local = db.get_record(LOCAL_HOST_ID).unwrap().unwrap(); + assert_eq!( + local.os_info.unwrap().os.as_deref(), + Some(std::env::consts::OS) + ); + drop(db); + std::fs::remove_dir_all(dir).unwrap(); + } + + #[test] + fn existing_auth_choice_survives_restart() { + let (dir, db) = open_test_database(); + db.set_setting("auth_enabled", "false").unwrap(); + drop(db); + let db = Database::open(&dir, MasterKey::load_or_create(&dir).unwrap()).unwrap(); + assert!(!db.get_app_settings().unwrap().auth_enabled); + drop(db); + std::fs::remove_dir_all(dir).unwrap(); + } + + #[cfg(not(windows))] + #[test] + fn linux_reopen_repairs_legacy_windows_local_metadata() { + let (dir, db) = open_test_database(); + db.conn.lock().unwrap().execute( + "UPDATE hosts SET os_json = ?1 WHERE id = ?2", + params![r#"{"os":"windows","distroId":"windows","versionId":"11","prettyName":"Windows 11 Local"}"#, LOCAL_HOST_ID] + ).unwrap(); + drop(db); + let db = Database::open(&dir, MasterKey::load_or_create(&dir).unwrap()).unwrap(); + let os = db + .get_record(LOCAL_HOST_ID) + .unwrap() + .unwrap() + .os_info + .unwrap(); + assert_eq!(os.os.as_deref(), Some("linux")); + assert_eq!(os.version_id, None); + drop(db); + std::fs::remove_dir_all(dir).unwrap(); + } +} diff --git a/src/files.rs b/src/files.rs index f3d0520..8568de9 100644 --- a/src/files.rs +++ b/src/files.rs @@ -249,7 +249,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result { let raw = dir_path.unwrap_or("").trim(); // Windows 顶层虚拟根目录 "drives"（此电脑）：列出 C:/ D:/ E:/ 等全部可用盘符 - if raw.eq_ignore_ascii_case("drives") || raw == "/" || raw == "\\" { + if cfg!(windows) && (raw.eq_ignore_ascii_case("drives") || raw == "/" || raw == "\\") { let mut drives = Vec::new(); for letter in b'A'..=b'Z' { let drive_str = format!("{}:/", letter as char); @@ -276,7 +276,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result { std::env::var("USERPROFILE") .or_else(|_| std::env::var("HOME")) .map(PathBuf::from) - .unwrap_or_else(|_| PathBuf::from("C:/")) + .unwrap_or_else(|_| PathBuf::from(if cfg!(windows) { "C:/" } else { "/" })) } else if raw.len() == 2 && raw.as_bytes()[1] == b':' { PathBuf::from(format!("{raw}/")) } else { @@ -326,7 +326,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result { .parent() .map(|p| p.to_string_lossy().replace('\\', "/")) .filter(|p| !p.is_empty() && p != &norm_path) - .or_else(|| Some("drives".to_string())) + .or_else(|| cfg!(windows).then(|| "drives".to_string())) }; Ok(DirListing { diff --git a/src/mcp.rs b/src/mcp.rs index 6c5b37c..e18119a 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -84,6 +84,8 @@ pub struct ExecTaskSnapshot { pub duration_ms: u64, pub stdout: String, pub stderr: String, + pub stdout_truncated: bool, + pub stderr_truncated: bool, pub hint: Option, pub started_at: String, } @@ -120,6 +122,8 @@ impl ExecTaskInner { duration_ms: dur, stdout: String::from_utf8_lossy(&self.stdout).to_string(), stderr: String::from_utf8_lossy(&self.stderr).to_string(), + stdout_truncated: self.stdout.len() > MAX_OUTPUT_BYTES, + stderr_truncated: self.stderr.len() > MAX_OUTPUT_BYTES, hint: self.hint.clone(), started_at: self.started_at.clone(), } @@ -137,7 +141,7 @@ const MAX_OUTPUT_BYTES: usize = 1024 * 1024; const MAX_TASKS: usize = 60; fn append_capped(buf: &mut Vec, chunk: &[u8]) { - if buf.len() >= MAX_OUTPUT_BYTES { + if buf.len() > MAX_OUTPUT_BYTES || chunk.is_empty() { return; } let room = MAX_OUTPUT_BYTES - buf.len(); @@ -187,17 +191,38 @@ impl ExecTaskManager { map.values().map(|t| t.lock().unwrap().snapshot()).collect() } - pub fn cancel_task(&self, task_id: &str) -> Option { - let arc = self.tasks.lock().unwrap().get(task_id).cloned()?; - let mut g = arc.lock().unwrap(); - if let Some(tx) = g.cancel_tx.take() { - let _ = tx.send(()); + pub async fn cancel_task(&self, task_id: &str) -> Result { + let arc = self + .tasks + .lock() + .unwrap() + .get(task_id) + .cloned() + .ok_or_else(|| anyhow!("未找到可取消的命令任务: {task_id}"))?; + { + let mut g = arc.lock().unwrap(); + if g.state != "running" { + return Ok(g.snapshot()); + } + if let Some(tx) = g.cancel_tx.take() { + let _ = tx.send(()); + } } - if g.state == "running" { - g.state = "cancelled".to_string(); - g.ended_ms = Some(g.started_inst.elapsed().as_millis() as u64); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + { + let g = arc.lock().unwrap(); + if g.state != "running" { + return Ok(g.snapshot()); + } + } + if Instant::now() >= deadline { + return Err(anyhow!( + "取消请求已发送，但尚未确认命令停止（taskId: {task_id}）" + )); + } + tokio::time::sleep(Duration::from_millis(50)).await; } - Some(g.snapshot()) } pub async fn exec_with_wait( @@ -369,7 +394,8 @@ async fn run_local_exec_task( } tokio::select! { _ = &mut *cancel_rx => { - let _ = child.kill().await; + let killed = child.kill().await; + finish_exec_cancellation(&inner, killed.err().map(|e| format!("终止本机进程失败: {e}"))); return; } res = stdout.read(&mut out_buf), if !out_eof => { @@ -488,7 +514,14 @@ async fn run_ssh_exec_task( cancel_rx: &mut oneshot::Receiver<()>, ) { // 并发执行闸：permit 持有到本任务结束，保证同一主机上的 session channel 不超过 MaxSessions - let _exec_permit = match state.ssh_pool.acquire_exec_slot(&host.id).await { + let slot = tokio::select! { + _ = &mut *cancel_rx => { + finish_exec_cancellation(&inner, None); + return; + } + result = state.ssh_pool.acquire_exec_slot(&host.id) => result, + }; + let _exec_permit = match slot { Ok(p) => p, Err(e) => { let mut g = inner.lock().unwrap(); @@ -499,7 +532,14 @@ async fn run_ssh_exec_task( } }; - let mut channel = match open_host_channel(&state.ssh_pool, &state.db, &host.id).await { + let opened = tokio::select! { + _ = &mut *cancel_rx => { + finish_exec_cancellation(&inner, None); + return; + } + result = open_host_channel(&state.ssh_pool, &state.db, &host.id) => result, + }; + let mut channel = match opened { Ok((_, ch)) => ch, Err(e) => { let mut g = inner.lock().unwrap(); @@ -518,12 +558,24 @@ async fn run_ssh_exec_task( .map(|s| s.eq_ignore_ascii_case("windows")) .unwrap_or(false); + let pid_prefix = format!("\u{1e}vps-exec:{}:", inner.lock().unwrap().id); let final_cmd = if is_windows { wrap_windows_remote_command(cmd) } else { - cmd.to_string() + // 记录这一条 exec 的进程组，不改用户命令的 shell、输出通道或退出码。 + format!( + "printf '\\036vps-exec:{}:%s\\037' \"$$\" >&2; eval '{}'", + inner.lock().unwrap().id, + cmd.replace('\'', "'\\''") + ) }; + if cancel_rx.try_recv().is_ok() { + let _ = channel.close().await; + finish_exec_cancellation(&inner, None); + return; + } + if let Err(e) = channel.exec(true, final_cmd.as_str()).await { let mut g = inner.lock().unwrap(); g.state = "failed".to_string(); @@ -534,11 +586,30 @@ async fn run_ssh_exec_task( } let mut exit_code: Option = None; + let mut exit_signal = None; + let mut cancel_deadline = None; let mut connection_lost = false; + let mut remote_pid = None; + let mut pid_pending = !is_windows; + let mut pid_buffer = Vec::new(); + let mut group_kill_sent = false; + let mut cancel_error = None; loop { tokio::select! { - _ = &mut *cancel_rx => { + _ = &mut *cancel_rx, if cancel_deadline.is_none() => { + // CHANNEL_CLOSE 不会终止远端进程。发送 SSH signal 请求，随后等服务端确认退出。 + if let Err(e) = channel.signal(russh::Sig::KILL).await { + let _ = channel.close().await; + finish_exec_cancellation(&inner, Some(format!("发送远端终止信号失败，无法确认命令停止: {e}"))); + return; + } + cancel_deadline = Some(Instant::now() + Duration::from_secs(5)); + } + _ = tokio::time::sleep_until(tokio::time::Instant::from_std( + cancel_deadline.unwrap_or_else(|| Instant::now() + Duration::from_secs(86400)) + )), if cancel_deadline.is_some() => { let _ = channel.close().await; + finish_exec_cancellation(&inner, Some(cancel_error.unwrap_or_else(|| "远端未确认终止信号，无法保证命令已停止".to_string()))); return; } msg_opt = channel.wait() => { @@ -548,18 +619,23 @@ async fn run_ssh_exec_task( g.last_output_inst = Instant::now(); } Some(ChannelMsg::ExtendedData { data, .. }) => { - let mut g = inner.lock().unwrap(); let cleaned = if is_windows { clean_clixml_stderr(&data) + } else if pid_pending { + consume_remote_pid(&mut pid_buffer, &mut pid_pending, &mut remote_pid, &pid_prefix, &data) } else { data.to_vec() }; + let mut g = inner.lock().unwrap(); append_capped(&mut g.stderr, &cleaned); g.last_output_inst = Instant::now(); } Some(ChannelMsg::ExitStatus { exit_status }) => { exit_code = Some(exit_status as i32); } + Some(ChannelMsg::ExitSignal { signal_name, .. }) => { + exit_signal = Some(signal_name); + } // EOF 只表示"输出结束"：OpenSSH 的 exit-status 常在其之后到达， // 必须读到 Close 才收尾，否则高频丢退出码（旧构建靠谎报成功掩盖了它） Some(ChannelMsg::Eof) => {} @@ -573,6 +649,24 @@ async fn run_ssh_exec_task( } } } + if cancel_deadline.is_some() && !group_kill_sent { + if let Some(pid) = remote_pid { + group_kill_sent = true; + // 某些 OpenSSH 忽略 signal 请求；用同一连接的短控制 channel 终止已确认的进程组。 + if let Err(e) = kill_remote_exec_group(state, &host.id, pid).await { + cancel_error = Some(format!("终止远端进程组失败，无法确认命令停止: {e:#}")); + } + } + } + } + + if !pid_buffer.is_empty() { + append_capped(&mut inner.lock().unwrap().stderr, &pid_buffer); + } + + if cancel_deadline.is_some() && matches!(exit_signal, Some(russh::Sig::KILL)) { + finish_exec_cancellation(&inner, None); + return; } let mut g = inner.lock().unwrap(); @@ -613,6 +707,90 @@ async fn run_ssh_exec_task( } } +fn consume_remote_pid( + buffer: &mut Vec, + pending: &mut bool, + pid: &mut Option, + prefix: &str, + data: &[u8], +) -> Vec { + buffer.extend_from_slice(data); + if let Some(start) = buffer + .windows(prefix.len()) + .position(|w| w == prefix.as_bytes()) + { + let value_start = start + prefix.len(); + if let Some(end_rel) = buffer[value_start..].iter().position(|b| *b == 0x1f) { + let end = value_start + end_rel; + if let Some(value) = std::str::from_utf8(&buffer[value_start..end]) + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|pid| *pid > 1 && *pid <= i32::MAX as u32) + { + *pid = Some(value); + buffer.drain(start..=end); + *pending = false; + return std::mem::take(buffer); + } + } + } + if buffer.len() > 4096 { + *pending = false; + return std::mem::take(buffer); + } + Vec::new() +} + +async fn kill_remote_exec_group(state: &Arc, host_id: &str, pid: u32) -> Result<()> { + let (_, mut channel) = tokio::time::timeout( + Duration::from_secs(3), + open_host_channel(&state.ssh_pool, &state.db, host_id), + ) + .await??; + // /proc 的进程组必须仍是这条 SSH exec 自己的组，不能对共享会话发送组信号。 + let command = format!("IFS= read -r s < /proc/{pid}/stat || exit 1; s=${{s##*) }}; set -- $s; [ \"$3\" = \"{pid}\" ] || exit 2; kill -KILL -- -{pid}"); + channel.exec(true, command.as_str()).await?; + let mut exit_code = None; + let mut output = Vec::new(); + let result = tokio::time::timeout(Duration::from_secs(3), async { + while let Some(msg) = channel.wait().await { + match msg { + ChannelMsg::ExitStatus { exit_status } => exit_code = Some(exit_status), + ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => { + output.extend_from_slice(&data) + } + ChannelMsg::Close => break, + _ => {} + } + } + }) + .await; + let _ = channel.close().await; + result?; + if exit_code != Some(0) { + return Err(anyhow!( + "控制命令退出码 {:?}: {}", + exit_code, + String::from_utf8_lossy(&output) + )); + } + Ok(()) +} + +fn finish_exec_cancellation(inner: &Arc>, error: Option) { + let mut g = inner.lock().unwrap(); + if g.state == "running" { + g.state = if error.is_some() { + "failed" + } else { + "cancelled" + } + .to_string(); + g.hint = error; + g.ended_ms = Some(g.started_inst.elapsed().as_millis() as u64); + } +} + // ==================== MCP JSON-RPC 2.0 协议与 7 大精锐工具定义 ==================== #[derive(Debug, Deserialize)] @@ -1030,8 +1208,15 @@ async fn dispatch_tool_call( } async fn tool_vps_list(state: &Arc, args: Value) -> Result { - let refresh = args.get("refresh").and_then(|v| v.as_bool()).unwrap_or(false); - let role_filter = args.get("role").and_then(|v| v.as_str()).unwrap_or("").trim(); + let refresh = args + .get("refresh") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let role_filter = args + .get("role") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim(); let search_q = args .get("search") .and_then(|v| v.as_str()) @@ -1167,7 +1352,7 @@ async fn tool_vps_inspect(state: &Arc, args: Value) -> Result { .await; Ok(json!({ - "ok": true, + "ok": results.iter().all(|r| r["ok"] == true), "results": results, })) } @@ -1210,7 +1395,9 @@ async fn tool_vps_exec( } } if host_queries.is_empty() { - return Err(anyhow!("请提供 host 或 hosts 参数（直接传主机名，如 'vip-hk'）")); + return Err(anyhow!( + "请提供 host 或 hosts 参数（直接传主机名，如 'vip-hk'）" + )); } let mut targets = Vec::new(); @@ -1248,6 +1435,8 @@ async fn tool_vps_exec( duration_ms: 0, stdout: String::new(), stderr: format!("{e:#}"), + stdout_truncated: false, + stderr_truncated: false, hint: None, started_at: Utc::now().to_rfc3339(), }) @@ -1288,10 +1477,22 @@ async fn tool_vps_terminal(state: &Arc, args: Value) -> Result let wait_ms = args.get("wait_ms").and_then(|v| v.as_u64()).unwrap_or(6000); let pane_id_arg = args.get("pane_id").and_then(|v| v.as_str()); + let specified_pane = pane_id_arg + .map(|pid| { + let pane = state + .terminals + .get_pane(pid) + .ok_or_else(|| anyhow!("指定的终端 pane 不存在: {pid}"))?; + if pane.host_id != host.id { + return Err(anyhow!("终端 {pid} 不属于目标主机「{}」", host.name)); + } + Ok(pane) + }) + .transpose()?; if action == "close" { - if let Some(pid) = pane_id_arg { - state.terminals.close_pane(&state.ssh_pool, pid).await; + if let Some(pane) = specified_pane { + state.terminals.close_pane(&state.ssh_pool, &pane.id).await; } else if let Some(pane) = state.terminals.find_active_pane_for_host(&host.id) { state.terminals.close_pane(&state.ssh_pool, &pane.id).await; } @@ -1299,11 +1500,8 @@ async fn tool_vps_terminal(state: &Arc, args: Value) -> Result } // 查找或自动创建该主机的终端会话（默认 headless=true 无头运行，除非 reveal_ui=true） - let pane = if let Some(pid) = pane_id_arg { - state - .terminals - .get_pane(pid) - .ok_or_else(|| anyhow!("指定的终端 pane 不存在: {pid}"))? + let pane = if let Some(pane) = specified_pane { + pane } else if let Some(existing) = state.terminals.find_active_pane_for_host(&host.id) { existing } else { @@ -1323,8 +1521,6 @@ async fn tool_vps_terminal(state: &Arc, args: Value) -> Result .terminals .get_pane(&created.id) .ok_or_else(|| anyhow!("创建终端后获取句柄失败"))?; - // 等待远端初始 shell 提示符就绪 - tokio::time::sleep(Duration::from_millis(450)).await; handle }; @@ -1346,7 +1542,7 @@ async fn tool_vps_terminal(state: &Arc, args: Value) -> Result if submit { bytes.push(b'\r'); } - pane.send_and_wait(bytes, max_lines, wait_ms).await + pane.send_and_wait(bytes, max_lines, wait_ms).await? } "key" => { let raw_input = args @@ -1375,7 +1571,7 @@ async fn tool_vps_terminal(state: &Arc, args: Value) -> Result "left" => b"\x1b[D".to_vec(), other => other.as_bytes().to_vec(), }; - pane.send_and_wait(bytes, max_lines, wait_ms).await + pane.send_and_wait(bytes, max_lines, wait_ms).await? } other => return Err(anyhow!("未知的 vps_terminal action: {other}")), }; @@ -1402,7 +1598,8 @@ async fn tool_vps_fs(state: &Arc, args: Value) -> Result { match action { "list" => { - let listing = files::list_dir(&state.ssh_pool, &state.db, &host.id, Some(path), false).await?; + let listing = + files::list_dir(&state.ssh_pool, &state.db, &host.id, Some(path), false).await?; Ok(json!({ "ok": true, "host": host.name, @@ -1448,7 +1645,10 @@ async fn tool_vps_fs(state: &Arc, args: Value) -> Result { Ok(json!({ "ok": true, "host": host.name, "oldPath": path, "newPath": new_path })) } "remove" => { - let is_dir = args.get("is_dir").and_then(|v| v.as_bool()).unwrap_or(false); + let is_dir = args + .get("is_dir") + .and_then(|v| v.as_bool()) + .unwrap_or(false); files::remove_item(&state.ssh_pool, &state.db, &host.id, path, is_dir).await?; Ok(json!({ "ok": true, "host": host.name, "removed": path })) } @@ -1591,7 +1791,10 @@ async fn tool_vps_task( exec_tasks: &Arc, args: Value, ) -> Result { - let action = args.get("action").and_then(|v| v.as_str()).unwrap_or("list"); + let action = args + .get("action") + .and_then(|v| v.as_str()) + .unwrap_or("list"); let task_id = args.get("task_id").and_then(|v| v.as_str()).unwrap_or(""); // 阻塞等待秒数：>0 时等到任务结束或超时再返回，**一次调用**即拿到最终结果， // 取代"查询→running→再查询"的轮询（旧设计里 vps_task 无阻塞模式，是死循环的另一半）。 @@ -1625,7 +1828,7 @@ async fn tool_vps_task( } } } - Ok(json!({ "ok": true, "task": snap })) + Ok(json!({ "ok": snap.state == "completed", "task": snap })) } else { let mut t = state .transfers @@ -1656,14 +1859,12 @@ async fn tool_vps_task( } } } - Ok(json!({ "ok": true, "transfer": t })) + Ok(json!({ "ok": t.state == "completed", "transfer": t })) } } "cancel" => { - let t = exec_tasks - .cancel_task(task_id) - .ok_or_else(|| anyhow!("未找到可取消的命令任务: {task_id}"))?; - Ok(json!({ "ok": true, "task": t })) + let t = exec_tasks.cancel_task(task_id).await?; + Ok(json!({ "ok": t.state == "cancelled" || t.state == "completed", "task": t })) } _ => Ok(json!({ "ok": true, @@ -1675,7 +1876,11 @@ async fn tool_vps_task( /// `--mcp-stdio` 本地模式：若常驻主进程正在监听则直接桥接过去（与桌面 UI 共享连接池、探针缓存和终端唤起事件）， /// 若主进程未运行则在当前进程内直接处理。 -pub async fn run_stdio_bridge(state: Arc, exec_tasks: Arc, port: u16) -> Result<()> { +pub async fn run_stdio_bridge( + state: Arc, + exec_tasks: Arc, + port: u16, +) -> Result<()> { let stdin = tokio::io::stdin(); let mut lines = tokio::io::BufReader::new(stdin).lines(); @@ -1766,6 +1971,7 @@ async fn try_forward_to_daemon(port: u16, api_key: &str, raw_json: &str) -> Opti pub struct CliSyncStatus { pub exe_path: String, pub port: u16, + pub env: std::collections::BTreeMap, pub claude_code_path: String, pub claude_code_synced: bool, pub codex_path: String, @@ -1792,12 +1998,22 @@ pub fn sync_cli_mcp_configs(port: u16) -> CliSyncStatus { let claude_path = home.join(".claude.json"); let codex_path = home.join(".codex").join("config.toml"); - let claude_ok = sync_claude_json(&claude_path, &exe_path).unwrap_or(false); - let codex_ok = sync_codex_toml(&codex_path, &exe_path).unwrap_or(false); + let data_dir = std::path::absolute(crate::resolve_data_dir()) + .unwrap_or_else(|_| crate::resolve_data_dir()); + let env = std::collections::BTreeMap::from([ + ("PORT".to_string(), port.to_string()), + ( + "DATA_DIR".to_string(), + data_dir.to_string_lossy().to_string(), + ), + ]); + let claude_ok = sync_claude_json(&claude_path, &exe_path, &env).unwrap_or(false); + let codex_ok = sync_codex_toml(&codex_path, &exe_path, &env).unwrap_or(false); CliSyncStatus { exe_path, port, + env, claude_code_path: claude_path.to_string_lossy().to_string(), claude_code_synced: claude_ok, codex_path: codex_path.to_string_lossy().to_string(), @@ -1805,7 +2021,11 @@ pub fn sync_cli_mcp_configs(port: u16) -> CliSyncStatus { } } -fn sync_claude_json(claude_path: &std::path::Path, exe_path: &str) -> Result { +fn sync_claude_json( + claude_path: &std::path::Path, + exe_path: &str, + env: &std::collections::BTreeMap, +) -> Result { if !claude_path.exists() { return Ok(false); } @@ -1818,12 +2038,11 @@ fn sync_claude_json(claude_path: &std::path::Path, exe_path: &str) -> Result Result Result { +fn sync_codex_toml( + codex_path: &std::path::Path, + exe_path: &str, + env: &std::collections::BTreeMap, +) -> Result { if !codex_path.exists() { return Ok(false); } let raw = std::fs::read_to_string(codex_path)?; let posix_exe = exe_path.replace('\\', "/"); let desired_block = format!( - "[mcp_servers.vps-switch]\ncommand = \"{}\"\nargs = [\"--mcp-stdio\"]", - posix_exe + "[mcp_servers.vps-switch]\ncommand = {}\nargs = [\"--mcp-stdio\"]\nenv = {{ DATA_DIR = {}, PORT = {} }}", + serde_json::to_string(&posix_exe)?, + serde_json::to_string(&env["DATA_DIR"] )?, + serde_json::to_string(&env["PORT"] )?, ); - if raw.contains(&desired_block) { - return Ok(true); - } - let updated = if let Some(start_idx) = raw.find("[mcp_servers.vps-switch]") { let after_header = start_idx + "[mcp_servers.vps-switch]".len(); let next_section_rel = raw[after_header..].find("\n["); @@ -1874,7 +2095,104 @@ fn sync_codex_toml(codex_path: &std::path::Path, exe_path: &str) -> Result }; let tmp_path = codex_path.with_extension("toml.vps-switch-tmp"); + if updated == std::fs::read_to_string(codex_path)? { + return Ok(true); + } std::fs::write(&tmp_path, updated)?; std::fs::rename(&tmp_path, codex_path)?; Ok(true) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn remote_pid_header_handles_split_frames_and_preserves_command_stderr() { + let mut buffer = Vec::new(); + let mut pending = true; + let mut pid = None; + let prefix = "\u{1e}vps-exec:test-id:"; + assert!(consume_remote_pid( + &mut buffer, + &mut pending, + &mut pid, + prefix, + b"startup warning\n\x1evps-exec:test-id:12" + ) + .is_empty()); + let output = consume_remote_pid( + &mut buffer, + &mut pending, + &mut pid, + prefix, + b"34\x1fcommand stderr\n", + ); + assert_eq!(pid, Some(1234)); + assert!(!pending); + assert_eq!(output, b"startup warning\ncommand stderr\n"); + } + + #[test] + fn invalid_remote_pid_does_not_remove_stderr_or_target_a_process_group() { + let mut buffer = Vec::new(); + let mut pending = true; + let mut pid = None; + let prefix = "\u{1e}vps-exec:test-id:"; + let mut raw = b"\x1evps-exec:test-id:0\x1f".to_vec(); + raw.extend(vec![b'A'; 4096]); + let output = consume_remote_pid(&mut buffer, &mut pending, &mut pid, prefix, &raw); + assert_eq!(output, raw); + assert!(pid.is_none()); + assert!(!pending); + } + + #[test] + fn output_limit_reports_loss_at_exact_chunk_boundary() { + let mut output = vec![b'A'; MAX_OUTPUT_BYTES]; + append_capped(&mut output, &[]); + assert_eq!(output.len(), MAX_OUTPUT_BYTES); + append_capped(&mut output, b"lost bytes"); + assert!(String::from_utf8_lossy(&output).ends_with("[输出超过 1 MiB，已截断]")); + let marked_size = output.len(); + append_capped(&mut output, b"more lost bytes"); + assert_eq!(output.len(), marked_size); + } + + #[test] + fn output_limit_marks_crossing_chunk_and_preserves_short_output() { + let mut output = "完整 output".as_bytes().to_vec(); + append_capped(&mut output, b"\n"); + assert_eq!(String::from_utf8(output).unwrap(), "完整 output\n"); + let mut output = vec![b'A'; MAX_OUTPUT_BYTES - 2]; + append_capped(&mut output, b"BCDE"); + assert_eq!(&output[MAX_OUTPUT_BYTES - 2..MAX_OUTPUT_BYTES], b"BC"); + assert!(output.len() > MAX_OUTPUT_BYTES); + } + + #[test] + fn cli_configs_keep_daemon_environment_and_neighbor_entries() { + let dir = std::env::temp_dir().join(format!("vps-switch-config-test-{}", Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let claude = dir.join("claude.json"); + let codex = dir.join("codex.toml"); + std::fs::write(&claude, r#"{"mcpServers":{"other":{"command":"keep"}}}"#).unwrap(); + std::fs::write(&codex, "model = \"keep\"\n[mcp_servers.vps-switch]\ncommand = \"old\"\nargs = []\n[mcp_servers.other]\ncommand = \"keep\"\n").unwrap(); + let env = std::collections::BTreeMap::from([ + ("PORT".to_string(), "18881".to_string()), + ("DATA_DIR".to_string(), "C:\\数据\\quoted\"dir".to_string()), + ]); + sync_claude_json(&claude, "C:\\tools\\vps-switch.exe", &env).unwrap(); + sync_codex_toml(&codex, "C:\\tools\\vps-switch.exe", &env).unwrap(); + let json: Value = serde_json::from_str(&std::fs::read_to_string(&claude).unwrap()).unwrap(); + assert_eq!(json["mcpServers"]["other"]["command"], "keep"); + assert_eq!(json["mcpServers"]["vps-switch"]["env"], json!(env)); + let first = std::fs::read_to_string(&codex).unwrap(); + assert!(first.contains("PORT = \"18881\"")); + assert!(first.contains(&serde_json::to_string(&env["DATA_DIR"]).unwrap())); + assert!(first.contains("[mcp_servers.other]\ncommand = \"keep\"")); + sync_codex_toml(&codex, "C:\\tools\\vps-switch.exe", &env).unwrap(); + assert_eq!(std::fs::read_to_string(&codex).unwrap(), first); + std::fs::remove_dir_all(&dir).unwrap(); + } +} diff --git a/src/terminal.rs b/src/terminal.rs index b746e37..9467e5c 100644 --- a/src/terminal.rs +++ b/src/terminal.rs @@ -20,7 +20,7 @@ use std::{ sync::{Arc, Mutex}, time::{Duration, Instant}, }; -use tokio::sync::{broadcast, mpsc}; +use tokio::sync::{broadcast, mpsc, oneshot}; use uuid::Uuid; const RING_BUFFER_CAPACITY: usize = 2 * 1024 * 1024; @@ -95,7 +95,10 @@ pub enum PaneBroadcast { } pub enum PaneCommand { - Input(Vec), + Input { + bytes: Vec, + sent: Option>>, + }, Paste(String), Resize(u16, u16), Close, @@ -214,8 +217,8 @@ impl PaneInternalState { (trimmed_screen, cursor_line) } - - fn scan_osc_sequences(&mut self, chunk: &[u8]) -> bool { let mut changed = false; + fn scan_osc_sequences(&mut self, chunk: &[u8]) -> bool { + let mut changed = false; self.osc_tail.extend_from_slice(chunk); if self.osc_tail.len() > 4096 { let keep = 1024; @@ -500,6 +503,7 @@ pub struct PaneHandle { pub state: Arc>, pub cmd_tx: mpsc::UnboundedSender, pub broadcast_tx: broadcast::Sender, + input_lock: tokio::sync::Mutex<()>, } impl PaneHandle { @@ -562,14 +566,22 @@ impl PaneHandle { bytes: Vec, max_lines: usize, wait_ms: u64, - ) -> ScreenSnapshot { - let seq_before = self.state.lock().unwrap().output_seq; - let _ = self.cmd_tx.send(PaneCommand::Input(bytes)); + ) -> Result { + let _input_guard = self.input_lock.lock().await; + let (sent, received) = oneshot::channel(); + self.cmd_tx + .send(PaneCommand::Input { + bytes, + sent: Some(sent), + }) + .map_err(|_| anyhow::anyhow!("终端输入通道已关闭"))?; + // 基线来自真正写入 PTY 的时刻：初始化/注入输出不能冒充本次输入的响应。 + let seq_before = received.await.context("终端在发送输入前已关闭")??; let deadline = Instant::now() + Duration::from_millis(wait_ms.clamp(200, 60_000)); loop { tokio::time::sleep(Duration::from_millis(80)).await; - let (exited, seq_now, quiet_ms, running, cursor_line) = { + let (exited, seq_now, quiet_ms, running, integrated, cursor_line) = { let g = self.state.lock().unwrap(); let (_, cl) = g.read_screen_grid(5); ( @@ -577,6 +589,7 @@ impl PaneHandle { g.output_seq, g.last_output_at.elapsed().as_millis() as u64, g.running, + g.shell_kind.is_some(), cl, ) }; @@ -588,7 +601,9 @@ impl PaneHandle { if looks_like_interactive_prompt(&cursor_line) && quiet_ms >= 550 { break; } - if (!running || looks_like_shell_prompt(&cursor_line)) && quiet_ms >= 350 { + if (!running || (!integrated && looks_like_shell_prompt(&cursor_line))) + && quiet_ms >= 350 + { break; } } @@ -597,7 +612,7 @@ impl PaneHandle { } } - self.read_screen_snapshot(max_lines) + Ok(self.read_screen_snapshot(max_lines)) } } @@ -641,7 +656,11 @@ impl TerminalManager { }); } - pub fn list_panes(&self, host_id_filter: Option<&str>, include_headless: bool) -> Vec { + pub fn list_panes( + &self, + host_id_filter: Option<&str>, + include_headless: bool, + ) -> Vec { let guard = self.panes.lock().unwrap(); guard .values() @@ -700,9 +719,9 @@ impl TerminalManager { /// 该主机是否还有活着的终端（pane 自己退出也算没了）。 fn has_live_pane(&self, host_id: &str) -> bool { let guard = self.panes.lock().unwrap(); - guard.values().any(|p| { - p.host_id == host_id && !p.state.lock().unwrap().exited - }) + guard + .values() + .any(|p| p.host_id == host_id && !p.state.lock().unwrap().exited) } pub async fn create_pane( @@ -809,7 +828,10 @@ fn spawn_local_conpty_pane( let cmd = { let mut c = CommandBuilder::new("powershell.exe"); let init_hook = r#"$global:__vps_orig_prompt = $function:prompt; function global:prompt { $ec = if ($?) { 0 } else { if ($LASTEXITCODE) { $LASTEXITCODE } else { 1 } }; $p = ($executionContext.SessionState.Path.CurrentLocation.Path -replace '\\','/'); [Console]::Write("$([char]27)]133;D;$ec$([char]7)$([char]27)]7;file://localhost/$p$([char]7)$([char]27)]133;A$([char]7)"); & $global:__vps_orig_prompt }"#; - c.args(["-NoLogo", "-NoExit", "-Command", init_hook]); + c.args(["-NoLogo", "-NoExit"]); + if headless { + c.args(["-Command", init_hook]); + } c.env("TERM", "xterm-256color"); c.env("TERM_PROGRAM", "vps-switch"); if let Some(ref cwd) = initial_cwd { @@ -862,6 +884,7 @@ fn spawn_local_conpty_pane( state: state.clone(), cmd_tx, broadcast_tx: broadcast_tx.clone(), + input_lock: tokio::sync::Mutex::new(()), }); let state_read = state.clone(); @@ -904,15 +927,32 @@ fn spawn_local_conpty_pane( let state_cmd = state.clone(); let bcast_cmd = broadcast_tx.clone(); tokio::spawn(async move { + let mut initial_input = true; while let Some(cmd) = cmd_rx.recv().await { match cmd { - PaneCommand::Input(bytes) => { + PaneCommand::Input { bytes, sent } => { + if initial_input && sent.is_some() { + if !wait_state(&state_cmd, Duration::from_secs(6), |g| { + prompt_settled(g, 250) + }) + .await + { + let _ = sent + .unwrap() + .send(Err(anyhow::anyhow!("终端初始提示符尚未就绪"))); + continue; + } + } + initial_input = false; + let seq_before = state_cmd.lock().unwrap().output_seq; if bytes.contains(&b'\r') || bytes.contains(&b'\n') { let mut guard = state_cmd.lock().unwrap(); guard.running = true; } - let _ = writer.write_all(&bytes); - let _ = writer.flush(); + let result = writer.write_all(&bytes).and_then(|_| writer.flush()); + if let Some(sent) = sent { + let _ = sent.send(result.map(|_| seq_before).map_err(Into::into)); + } } PaneCommand::Paste(text) => { let payload = encode_safe_paste(&state_cmd, &text); @@ -1048,13 +1088,22 @@ async fn inject_shell_integration( state: &Arc>, ) { /// 还回回显。注入的每一条退出路径都要经过这里。 - async fn restore_echo(write: &ChannelWriteHalf) { - let _ = write.data(&b"stty echo 2>/dev/null\n"[..]).await; + async fn restore_echo( + write: &ChannelWriteHalf, + state: &Arc>, + ) { + let seq_before = state.lock().unwrap().output_seq; + if write.data(&b"stty echo 2>/dev/null\n"[..]).await.is_ok() { + let _ = wait_state(state, Duration::from_secs(3), |g| { + g.output_seq > seq_before && prompt_settled(g, 250) + }) + .await; + } } // P0：等首个提示符。失败（超时/Windows 原生 shell）就把回显还回去、原地退出。 if !wait_state(state, Duration::from_secs(6), |g| prompt_settled(g, 300)).await { - restore_echo(write).await; + restore_echo(write, state).await; return; } let windows_shell = { @@ -1063,11 +1112,12 @@ async fn inject_shell_integration( prompt_is_windows_style(&cursor) }; if windows_shell { - restore_echo(write).await; + restore_echo(write, state).await; return; } // P1：探测 shell 类型（回显已关，这一行在屏幕上不可见） + let seq_before = state.lock().unwrap().output_seq; if write .data(format!("{SHELL_PROBE_LINE}\n").as_bytes()) .await @@ -1075,23 +1125,22 @@ async fn inject_shell_integration( { return; } - let seq_before = state.lock().unwrap().output_seq; if !wait_state(state, Duration::from_millis(2500), move |g| { g.output_seq > seq_before && g.shell_kind.is_some() }) .await { - restore_echo(write).await; + restore_echo(write, state).await; return; } let Some(shell) = state.lock().unwrap().shell_kind else { - restore_echo(write).await; + restore_echo(write, state).await; return; }; // P2：写入钩子。等待条件带 output_seq 护栏——回显已关，上一阶段的提示符仍在 // 光标行上，没有护栏会在注入尚未执行时就误判"已完成"。 let Some(inject_line) = injection_line_for(shell) else { - restore_echo(write).await; + restore_echo(write, state).await; return; }; let seq_before_inject = state.lock().unwrap().output_seq; @@ -1109,7 +1158,7 @@ async fn inject_shell_integration( // P3：还回回显。注入疑似失败（例如 shell 语法报错）也照样还——报错上下文留在屏幕上 // 由用户自己看，绝不为了"擦干净"去动光标。 - restore_echo(write).await; + restore_echo(write, state).await; } async fn spawn_ssh_pty_pane( @@ -1150,6 +1199,7 @@ async fn spawn_ssh_pty_pane( state: state.clone(), cmd_tx, broadcast_tx: broadcast_tx.clone(), + input_lock: tokio::sync::Mutex::new(()), }); let (mut read_half, write_half) = channel.split(); @@ -1201,7 +1251,11 @@ async fn spawn_ssh_pty_pane( // 写半边：请求 pty（注入前先把远端回显关掉，注入写进去的每一行都不会留痕）→ 请求 shell // → 静默注入 shell integration → 进入输入/控制循环。 tokio::spawn(async move { - let modes: &[(Pty, u32)] = if skip_injection { &[] } else { &[(Pty::ECHO, 0)] }; + let modes: &[(Pty, u32)] = if skip_injection { + &[] + } else { + &[(Pty::ECHO, 0)] + }; if write_half .request_pty( false, @@ -1226,11 +1280,17 @@ async fn spawn_ssh_pty_pane( while let Some(cmd) = cmd_rx.recv().await { match cmd { - PaneCommand::Input(bytes) => { + PaneCommand::Input { bytes, sent } => { + let seq_before = state.lock().unwrap().output_seq; if bytes.contains(&b'\r') || bytes.contains(&b'\n') { state.lock().unwrap().running = true; } - if write_half.data(&bytes[..]).await.is_err() { + let result = write_half.data(&bytes[..]).await; + let failed = result.is_err(); + if let Some(sent) = sent { + let _ = sent.send(result.map(|_| seq_before).map_err(Into::into)); + } + if failed { break; } } @@ -1264,10 +1324,7 @@ async fn spawn_ssh_pty_pane( fn encode_safe_paste(state: &Arc>, text: &str) -> Vec { let bracketed = state.lock().unwrap().bracketed_paste_enabled(); - let sanitized: String = text - .chars() - .filter(|&c| c != '\x1b' && c != '\0') - .collect(); + let sanitized: String = text.chars().filter(|&c| c != '\x1b' && c != '\0').collect(); if bracketed { let mut out = Vec::with_capacity(sanitized.len() + 12); out.extend_from_slice(b"\x1b[200~");