From 4ce89f8d82684e3d29cecf4f8c96fecb3870bb6a Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Fri, 21 Aug 2026 08:30:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E6=96=B0=E5=A2=9E=E6=8C=81?= =?UTF-8?q?=E4=B9=85=20PTY=20=E4=BC=9A=E8=AF=9D=E5=B7=A5=E5=85=B7=E6=9B=BF?= =?UTF-8?q?=E4=BB=A3=E7=AE=A1=E9=81=93=E5=BC=8F=E4=BA=A4=E4=BA=92=E5=BC=8F?= =?UTF-8?q?=20shell?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 PtyTool：spawn/write/read/kill/list 五操作管理真实伪终端会话（portable-pty） - PtySessionManager：活动续期 TTL 惰性回收、ANSI 剥离、增量读取游标 - kill 走完整 Child::kill() 语义（Unix SIGHUP→宽限→SIGKILL），spawn_blocking 执行 - 修复：write_input 不再持管理器锁跨阻塞 IO；kill 前 drain 管道尾部输出；watcher 改 try_lock - 移除 shell_session.rs 管道式实现；前端 Tools/Subagents 页补充 PTY 条目 --- Cargo.lock | 182 ++++- Cargo.toml | 5 +- src/command/handlers/delete_topic.rs | 11 +- src/command/handlers/list_topics.rs | 17 +- src/command/handlers/rename_topic.rs | 22 +- src/command/handlers/session.rs | 11 +- src/command/mod.rs | 5 +- src/gateway/http.rs | 5 +- src/gateway/mod.rs | 11 + src/gateway/processor.rs | 7 +- src/gateway/runtime.rs | 12 +- src/gateway/session.rs | 2 +- src/gateway/tool_prompt_provider.rs | 25 +- src/gateway/tool_registry_factory.rs | 26 +- src/gateway/ws.rs | 3 +- src/platform/mod.rs | 193 ----- src/storage/mod.rs | 14 +- src/storage/tests.rs | 32 +- src/tools/bash.rs | 209 +----- src/tools/mod.rs | 6 +- src/tools/pty.rs | 468 ++++++++++++ src/tools/pty_session.rs | 672 ++++++++++++++++++ src/tools/shell_session.rs | 302 -------- .../components/Settings/tabs/SubagentsTab.tsx | 1 + web/src/components/Settings/tabs/ToolsTab.tsx | 1 + 25 files changed, 1466 insertions(+), 776 deletions(-) create mode 100644 src/tools/pty.rs create mode 100644 src/tools/pty_session.rs delete mode 100644 src/tools/shell_session.rs diff --git a/Cargo.lock b/Cargo.lock index 0a658d9..a834323 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -223,6 +223,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.11.1" @@ -609,6 +615,12 @@ version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f" +[[package]] +name = "downcast-rs" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" + [[package]] name = "dunce" version = "1.0.5" @@ -679,6 +691,17 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "filedescriptor" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e40758ed24c9b2eeb76c35fb0aebc66c626084edd827e07e1552279814c6682d" +dependencies = [ + "libc", + "thiserror 1.0.69", + "winapi", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -1278,6 +1301,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "ioctl-rs" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7970510895cee30b3e9128319f2cefd4bde883a39f38baa279567ba3a7eb97d" +dependencies = [ + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -1478,6 +1510,15 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "memoffset" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aa361d4faea93603064a027415f07bd8e1d5c88c9fbf68bf56a285428fd79ce" +dependencies = [ + "autocfg", +] + [[package]] name = "metrics" version = "0.23.1" @@ -1587,13 +1628,27 @@ dependencies = [ "pxfm", ] +[[package]] +name = "nix" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f346ff70e7dbfd675fe90590b92d59ef2de15a8779ae305ebcbfd3f0caf59be4" +dependencies = [ + "autocfg", + "bitflags 1.3.2", + "cfg-if", + "libc", + "memoffset", + "pin-utils", +] + [[package]] name = "nix" version = "0.31.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cfg-if", "cfg_aliases", "libc", @@ -1746,12 +1801,12 @@ dependencies = [ "http", "iana-time-zone", "image", - "libc", "metrics", "metrics-exporter-prometheus", "meval", "mime_guess", "parking_lot", + "portable-pty", "prost", "r2d2", "r2d2_sqlite", @@ -1786,6 +1841,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "pkg-config" version = "0.3.33" @@ -1798,7 +1859,7 @@ version = "0.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60769b8b31b2a9f263dae2776c37b1b28ae246943cf719eb6946a1db05128a61" dependencies = [ - "bitflags", + "bitflags 2.11.1", "crc32fast", "fdeflate", "flate2", @@ -1811,6 +1872,27 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" +[[package]] +name = "portable-pty" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "806ee80c2a03dbe1a9fb9534f8d19e4c0546b790cde8fd1fea9d6390644cb0be" +dependencies = [ + "anyhow", + "bitflags 1.3.2", + "downcast-rs", + "filedescriptor", + "lazy_static", + "libc", + "log", + "nix 0.25.1", + "serial", + "shared_library", + "shell-words", + "winapi", + "winreg", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -1862,7 +1944,7 @@ checksum = "2e842efad9119158434d193c6682e2ebee4b44d6ad801d7b349623b3f57cdf55" dependencies = [ "futures", "indexmap", - "nix", + "nix 0.31.3", "tokio", "tracing", "windows", @@ -2079,7 +2161,7 @@ version = "11.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186" dependencies = [ - "bitflags", + "bitflags 2.11.1", ] [[package]] @@ -2088,7 +2170,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.11.1", ] [[package]] @@ -2273,7 +2355,7 @@ version = "0.39.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a0d2b0146dd9661bf67bb107c0bb2a55064d556eeb3fc314151b957f313bcd4e" dependencies = [ - "bitflags", + "bitflags 2.11.1", "fallible-iterator", "fallible-streaming-iterator", "hashlink 0.11.1", @@ -2337,7 +2419,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags", + "bitflags 2.11.1", "errno", "libc", "linux-raw-sys", @@ -2498,7 +2580,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags", + "bitflags 2.11.1", "core-foundation", "core-foundation-sys", "libc", @@ -2622,6 +2704,48 @@ dependencies = [ "unsafe-libyaml", ] +[[package]] +name = "serial" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1237a96570fc377c13baa1b88c7589ab66edced652e43ffb17088f003db3e86" +dependencies = [ + "serial-core", + "serial-unix", + "serial-windows", +] + +[[package]] +name = "serial-core" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f46209b345401737ae2125fe5b19a77acce90cd53e1658cda928e4fe9a64581" +dependencies = [ + "libc", +] + +[[package]] +name = "serial-unix" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f03fbca4c9d866e24a459cbca71283f545a37f8e3e002ad8c70593871453cab7" +dependencies = [ + "ioctl-rs", + "libc", + "serial-core", + "termios", +] + +[[package]] +name = "serial-windows" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15c6d3b776267a75d31bbdfd5d36c0ca051251caafc285827052bc53bcdc8162" +dependencies = [ + "libc", + "serial-core", +] + [[package]] name = "sha1" version = "0.10.6" @@ -2653,6 +2777,22 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shared_library" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a9e7e0f2bfae24d8a5b5a66c5b257a83c7412304311512a0c054cd5e619da11" +dependencies = [ + "lazy_static", + "libc", +] + +[[package]] +name = "shell-words" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc6fe69c597f9c37bfeeeeeb33da3530379845f10be461a66d16d03eca2ded77" + [[package]] name = "shlex" version = "1.3.0" @@ -2818,6 +2958,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "termios" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5d9cf598a6d7ce700a4e6a9199da127e6819a61e64b68609683cc9a01b5683a" +dependencies = [ + "libc", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -3024,7 +3173,7 @@ version = "0.6.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68d6fdd9f81c2819c9a8b0e0cd91660e7746a8e6ea2ba7c6b2b057985f6bcb51" dependencies = [ - "bitflags", + "bitflags 2.11.1", "bytes", "futures-core", "futures-util", @@ -3404,7 +3553,7 @@ version = "0.244.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ - "bitflags", + "bitflags 2.11.1", "hashbrown 0.15.5", "indexmap", "semver", @@ -3716,6 +3865,15 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winreg" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80d0f4e272c85def139476380b12f9ac60926689dd2e01d4923222f40580869d" +dependencies = [ + "winapi", +] + [[package]] name = "wit-bindgen" version = "0.51.0" @@ -3780,7 +3938,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", - "bitflags", + "bitflags 2.11.1", "indexmap", "log", "serde", diff --git a/Cargo.toml b/Cargo.toml index 4de191b..c81112d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,7 +61,6 @@ subtle = "2.6" parking_lot = "0.12" wechatbot = { path = "vendor/wechatbot" } encoding_rs = "0.8" -libc = "0.2" gray_matter = { version = "0.2", default-features = false, features = ["yaml"] } # MCP (Model Context Protocol) support rmcp = { version = "1.7", features = [ @@ -74,6 +73,10 @@ schemars = "1.0" http = "1" tower-http = { version = "0.6", features = ["fs", "cors"] } rust-embed = "8" +# PTY support for the pty tool (Unix pty + Windows ConPTY). +# 锁定 0.8 系：0.9 在 Windows 启用 PSEUDOCONSOLE_INHERIT_CURSOR， +# ConPTY 启动时发 DSR 查询并阻塞子进程输出，直到宿主应答（已知回归）。 +portable-pty = "0.8" [target.'cfg(windows)'.dependencies] windows-sys = { version = "0.59", features = [ diff --git a/src/command/handlers/delete_topic.rs b/src/command/handlers/delete_topic.rs index 6a07b02..8842236 100644 --- a/src/command/handlers/delete_topic.rs +++ b/src/command/handlers/delete_topic.rs @@ -81,12 +81,11 @@ async fn handle_delete_topic( .map_err(|e| CommandError::new("DELETE_TOPIC_ERROR", e.to_string()))?; // 查询更新后的话题列表，返回给前端刷新侧边栏 - let topic_summaries = - crate::command::handlers::list_topics::list_topic_summaries_blocking( - handler.store.clone(), - session_id, - ) - .await?; + let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking( + handler.store.clone(), + session_id, + ) + .await?; let topics_json = serde_json::to_string(&topic_summaries) .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; diff --git a/src/command/handlers/list_topics.rs b/src/command/handlers/list_topics.rs index 7fefd80..afb1cdd 100644 --- a/src/command/handlers/list_topics.rs +++ b/src/command/handlers/list_topics.rs @@ -107,14 +107,12 @@ pub async fn list_topic_summaries_blocking( session_id: &str, ) -> Result, CommandError> { let session_id_bg = session_id.to_string(); - tokio::task::spawn_blocking( - move || -> Result, CommandError> { - let topics = store - .list_topics(&session_id_bg) - .map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; - build_topic_summaries(store.as_ref(), topics) - }, - ) + tokio::task::spawn_blocking(move || -> Result, CommandError> { + let topics = store + .list_topics(&session_id_bg) + .map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; + build_topic_summaries(store.as_ref(), topics) + }) .await .map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))? } @@ -161,8 +159,7 @@ async fn handle_list_topics( session_id: String, ctx: CommandContext, ) -> Result { - let summaries = - list_topic_summaries_blocking(handler.store.clone(), &session_id).await?; + let summaries = list_topic_summaries_blocking(handler.store.clone(), &session_id).await?; let topics_json = serde_json::to_string(&summaries) .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; diff --git a/src/command/handlers/rename_topic.rs b/src/command/handlers/rename_topic.rs index fe19a07..0b3ad99 100644 --- a/src/command/handlers/rename_topic.rs +++ b/src/command/handlers/rename_topic.rs @@ -83,12 +83,11 @@ async fn handle_rename_topic( // 标题未变化时直接返回当前列表，避免无意义写入 if old_display == trimmed_title { - let topic_summaries = - crate::command::handlers::list_topics::list_topic_summaries_blocking( - handler.store.clone(), - session_id, - ) - .await?; + let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking( + handler.store.clone(), + session_id, + ) + .await?; let topic_summaries_json = serde_json::to_string(&topic_summaries) .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; @@ -110,12 +109,11 @@ async fn handle_rename_topic( .map_err(|e| CommandError::new("RENAME_TOPIC_ERROR", e.to_string()))?; // 查询更新后的话题列表，返回给前端刷新侧边栏 - let topic_summaries = - crate::command::handlers::list_topics::list_topic_summaries_blocking( - handler.store.clone(), - session_id, - ) - .await?; + let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking( + handler.store.clone(), + session_id, + ) + .await?; let topic_summaries_json = serde_json::to_string(&topic_summaries) .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; diff --git a/src/command/handlers/session.rs b/src/command/handlers/session.rs index 5829e3e..2a066d2 100644 --- a/src/command/handlers/session.rs +++ b/src/command/handlers/session.rs @@ -103,12 +103,11 @@ async fn handle_create_session( } // Query the full topic list so the frontend sidebar can update - let topic_summaries = - crate::command::handlers::list_topics::list_topic_summaries_blocking( - handler.store.clone(), - session_id, - ) - .await?; + let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking( + handler.store.clone(), + session_id, + ) + .await?; let topics_json = serde_json::to_string(&topic_summaries) .map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?; diff --git a/src/command/mod.rs b/src/command/mod.rs index ec8f10d..4e9377c 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -50,10 +50,7 @@ pub enum Command { /// 加载指定 channel + chat_id 的对话消息 LoadChatMessages { channel: String, chat_id: String }, /// 分页加载话题更早的历史消息（用户向上滚动触发） - LoadOlderMessages { - topic_id: String, - before_seq: i64, - }, + LoadOlderMessages { topic_id: String, before_seq: i64 }, /// 删除指定话题 DeleteTopic { topic_id: String }, /// 重命名指定话题 diff --git a/src/gateway/http.rs b/src/gateway/http.rs index 41dadc3..bbb8028 100644 --- a/src/gateway/http.rs +++ b/src/gateway/http.rs @@ -1315,7 +1315,10 @@ pub async fn topic_select_model( }) .await }; - if let Err(e) = update_result.map_err(|e| e.to_string()).and_then(|r| r.map_err(|e| e.to_string())) { + if let Err(e) = update_result + .map_err(|e| e.to_string()) + .and_then(|r| r.map_err(|e| e.to_string())) + { return ( StatusCode::INTERNAL_SERVER_ERROR, Json(SelectModelResponse { diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 700b7a0..834e35c 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -48,6 +48,7 @@ use crate::config::LLMProviderConfig; use crate::logging; use crate::scheduler::Scheduler; use crate::skills::SkillRuntime; +use crate::tools::PtySessionManager; use crate::tools::task::repository::TaskRepository; use crate::tools::task::runtime::SubagentRuntime; use agent_task_executor::{AgentTaskExecutor, SchedulerMaintenanceService}; @@ -82,6 +83,9 @@ pub struct GatewayState { /// Prometheus metrics handle（/metrics 端点渲染用）。 /// None 表示 recorder 安装失败；热重启时从 OnceLock 缓存复用，不会因重复安装而变为 None。 pub prometheus_handle: Option, + /// PTY 会话管理器（与 ToolRegistryFactory 内注入 PtyTool 的同一实例）， + /// gateway 关停/重启时统一终止全部 PTY 会话，防止子进程泄漏。 + pub pty_sessions: Arc, } impl GatewayState { @@ -120,6 +124,7 @@ impl GatewayState { model_selections, topic_model_selections, subagent_executor, + pty_sessions, ) = build_session_manager_with_sender( agent_prompt_reinject_every, show_tool_results, @@ -168,6 +173,7 @@ impl GatewayState { model_selections, topic_model_selections, prometheus_handle, + pty_sessions, }) } @@ -545,6 +551,7 @@ pub async fn run( let channel_manager = state.channel_manager.clone(); let cancel_manager = state.cancel_manager.clone(); let mcp_manager = state.mcp_manager.clone(); + let pty_sessions = state.pty_sessions.clone(); // Spawn ctrl_c / restart handler tokio::spawn(async move { @@ -557,6 +564,8 @@ pub async fn run( tracing::info!("Shutting down MCP servers before shutdown"); let _ = mgr.shutdown_all().await; } + tracing::info!("Terminating PTY sessions before shutdown"); + pty_sessions.shutdown().await; let _ = channel_manager.stop_all().await; let _ = result_tx.send(false); let _ = shutdown_tx.send(()); @@ -570,6 +579,8 @@ pub async fn run( tracing::info!("Shutting down MCP servers before restart"); let _ = mgr.shutdown_all().await; } + tracing::info!("Terminating PTY sessions before restart"); + pty_sessions.shutdown().await; let _ = channel_manager.stop_all().await; let _ = result_tx.send(true); let _ = shutdown_tx.send(()); diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 388a6a3..720a188 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -233,8 +233,7 @@ impl InboundProcessor { } let mut error_metadata = panic_forwarded_metadata.clone(); - error_metadata - .insert("error_kind".to_string(), "panic".to_string()); + error_metadata.insert("error_kind".to_string(), "panic".to_string()); if let Err(publish_error) = processor .bus .publish_outbound( @@ -718,7 +717,9 @@ impl InboundProcessor { // 僵尸清理：先剔除"DB=running 但执行任务已消失"的僵尸记录， // 否则它们会让 pending 永远非空 → ExecutionCompleted 永远被跳过 // → 前端 loading 永不停止（用户视角"卡死"）。 - if let (Some(ref topic_id), Some(ref runtime)) = (current_topic.as_ref(), self.subagent_executor.as_ref()) { + if let (Some(ref topic_id), Some(ref runtime)) = + (current_topic.as_ref(), self.subagent_executor.as_ref()) + { runtime.reap_orphan_subagents(topic_id).await; } let has_pending_subagents = if let Some(ref topic_id) = current_topic { diff --git a/src/gateway/runtime.rs b/src/gateway/runtime.rs index cf5203b..8079658 100644 --- a/src/gateway/runtime.rs +++ b/src/gateway/runtime.rs @@ -25,8 +25,9 @@ use crate::tools::task::repository::TaskRepository; use crate::tools::task::runtime::SubagentRuntime; use crate::tools::todo_write::TodoItem; use crate::tools::{ - DefaultSubAgentRuntime, InMemoryTaskRepository, NoopSessionMessageSender, SessionMessageSender, - SubAgentRuntime, SubAgentRuntimeConfig, SubagentCatalog, TaskTool, ToolRegistry, + DefaultSubAgentRuntime, InMemoryTaskRepository, NoopSessionMessageSender, PtySessionManager, + SessionMessageSender, SubAgentRuntime, SubAgentRuntimeConfig, SubagentCatalog, TaskTool, + ToolRegistry, }; use super::agent_factory::AgentFactory; @@ -69,6 +70,7 @@ pub(crate) fn build_session_manager( Arc, Arc, Option>, + Arc, ), AgentError, > { @@ -123,6 +125,7 @@ pub(crate) fn build_session_manager_with_sender( Arc, Arc, Option>, + Arc, ), AgentError, > { @@ -179,6 +182,10 @@ pub(crate) fn build_session_manager_with_sender( task_config.clone(), ); + // PTY 会话管理器在工厂内部创建（主 agent 与子代理共享同一实例）， + // 此处取句柄向上透传，供 gateway 关停时统一 shutdown。 + let pty_sessions = factory.pty_session_manager(); + // Create shared todo state for TodoWriteTool let todo_state: Arc>>> = Arc::new(RwLock::new(HashMap::new())); @@ -399,5 +406,6 @@ pub(crate) fn build_session_manager_with_sender( model_selections, topic_model_selections, subagent_executor, + pty_sessions, )) } diff --git a/src/gateway/session.rs b/src/gateway/session.rs index cb188ae..2c0bcf8 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -1074,7 +1074,7 @@ impl SessionManager { model_resolver, crate::config::CompactionConfig::default(), ) - .map(|(session_manager, _, _, _, _, _, _)| session_manager) + .map(|(session_manager, _, _, _, _, _, _, _)| session_manager) } pub fn tools(&self) -> Arc { diff --git a/src/gateway/tool_prompt_provider.rs b/src/gateway/tool_prompt_provider.rs index cf692d5..b8a0acf 100644 --- a/src/gateway/tool_prompt_provider.rs +++ b/src/gateway/tool_prompt_provider.rs @@ -31,7 +31,7 @@ impl SystemPromptProvider for ToolPromptProvider { MEMORY_TOOLS_INSTRUCTIONS, SKILL_TOOLS_INSTRUCTIONS, TODO_WRITE_INSTRUCTIONS, - SHELL_TOOLS_INSTRUCTIONS, + PTY_TOOLS_INSTRUCTIONS, SCHEDULER_TOOLS_INSTRUCTIONS, ), context: Some("tools".to_string()), @@ -86,13 +86,24 @@ const TODO_WRITE_INSTRUCTIONS: &str = r#"# TodoWrite 工具 ## 查询当前列表 - 对话开始时、不确定状态时、完成一项后，应主动调用 `todo_read` 查看任务列表"#; -/// shell / bash 工具使用说明 -const SHELL_TOOLS_INSTRUCTIONS: &str = r#"# Shell 交互终端 +/// pty 工具使用说明 +const PTY_TOOLS_INSTRUCTIONS: &str = r#"# PTY 持久终端（pty 工具） -- 当 shell 工具返回包含 `__PICOBOT_PENDING_USER_ACTION__` 和 `[session_id: xxx]` 的结果时，表示进程正在等待输入 -- 阅读已输出的内容，理解提示含义（如确认提示 Y/N、输入密码、选择选项等） -- 使用 `session_id` 和 `stdin_input` 参数回复交互内容，例如：`{"command": "echo test", "session_id": "xxx", "stdin_input": "Y"}` -- 常见场景：确认提示输入 Y/N、输入密码/验证码、选择选项、Read-Host 等"#; +## 何时使用 +- 程序需要真实终端（TTY）：交互式向导、确认提示（Y/N）、输入密码/验证码、TUI 程序 +- 长运行服务/REPL 需要跨多轮工具调用保持状态并持续交互 +- 普通一次性命令优先使用 bash/shell 工具 + +## 操作流程 +1. `{"action": "spawn", "command": "..."}` 创建会话，返回 `session_id` +2. `{"action": "read", "session_id": "...", "wait_ms": 2000}` 读取新增输出（增量，只返回未读过的内容），重复调用直到看到预期提示 +3. `{"action": "write", "session_id": "...", "input": "Y"}` 发送输入（自动补换行） +4. 不再需要时 `{"action": "kill", "session_id": "..."}` 终止会话 + +## 注意 +- 发送 Ctrl+C 等控制字符：`{"action": "write", "session_id": "...", "input": "\u0003", "raw": true}` +- `list` 可查看所有会话；会话在 30 分钟无任何 read/write 后自动过期 +- read 返回 `[Process status: exited (...)]` 表示进程已退出"#; /// silent_agent_task 工具使用说明 const SCHEDULER_TOOLS_INSTRUCTIONS: &str = r#"# 定时任务 diff --git a/src/gateway/tool_registry_factory.rs b/src/gateway/tool_registry_factory.rs index 48a4f2e..de1fec8 100644 --- a/src/gateway/tool_registry_factory.rs +++ b/src/gateway/tool_registry_factory.rs @@ -12,9 +12,10 @@ use crate::storage::{ use crate::tools::todo_write::TodoItem; use crate::tools::{ BashTool, CalculatorTool, FileEditTool, FileReadTool, FileWriteTool, HttpRequestTool, - MemoryManageTool, MemorySearchTool, SchedulerManageTool, SessionMessageSender, SessionSendTool, - ShellSessionManager, SkillActivateTool, SkillManageTool, SubAgentRuntime, TaskTool, TimeTool, - TodoReadTool, TodoWriteTool, ToolRegistry, WaitForSubagentsTool, WebFetchTool, + MemoryManageTool, MemorySearchTool, PtySessionManager, PtyTool, SchedulerManageTool, + SessionMessageSender, SessionSendTool, SkillActivateTool, SkillManageTool, SubAgentRuntime, + TaskTool, TimeTool, TodoReadTool, TodoWriteTool, ToolRegistry, WaitForSubagentsTool, + WebFetchTool, }; pub(crate) struct ToolRegistryFactory { @@ -31,7 +32,7 @@ pub(crate) struct ToolRegistryFactory { subagent_runtime: Option>, mcp_manager: Option>, todo_state: Option>>>>, - shell_session_manager: Arc, + pty_session_manager: Arc, } impl ToolRegistryFactory { @@ -61,10 +62,15 @@ impl ToolRegistryFactory { subagent_runtime: None, mcp_manager: None, todo_state: None, - shell_session_manager: Arc::new(ShellSessionManager::new()), + pty_session_manager: Arc::new(PtySessionManager::new()), } } + /// PTY 会话管理器句柄（供 gateway 在关停时统一 shutdown）。 + pub(crate) fn pty_session_manager(&self) -> Arc { + self.pty_session_manager.clone() + } + pub(crate) fn with_todo_state( mut self, state: Arc>>>, @@ -142,7 +148,10 @@ impl ToolRegistryFactory { registry.register(SkillManageTool::new(self.skills.clone())); } if self.is_enabled("bash") { - registry.register(BashTool::new(self.shell_session_manager.clone())); + registry.register(BashTool::new()); + } + if self.is_enabled("pty") { + registry.register(PtyTool::new(self.pty_session_manager.clone())); } if self.is_enabled("http_request") { registry.register(HttpRequestTool::new( @@ -198,7 +207,10 @@ impl ToolRegistryFactory { registry.register(FileEditTool::new()); } if self.is_enabled("bash") { - registry.register(BashTool::new(self.shell_session_manager.clone())); + registry.register(BashTool::new()); + } + if self.is_enabled("pty") { + registry.register(PtyTool::new(self.pty_session_manager.clone())); } if self.is_enabled("http_request") { registry.register(HttpRequestTool::new( diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index f2e7d6b..fffc969 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -590,7 +590,8 @@ async fn handle_inbound( router.register(Box::new(LoadChatMessagesCommandHandler::new())); // 注册 load_older_messages 处理器（历史分页） router.register(Box::new( - crate::command::handlers::load_chat_messages::LoadOlderMessagesCommandHandler::new(), + crate::command::handlers::load_chat_messages::LoadOlderMessagesCommandHandler::new( + ), )); // 注册 stop_execution 处理器 router.register(Box::new(StopExecutionCommandHandler::new( diff --git a/src/platform/mod.rs b/src/platform/mod.rs index 55fb2e8..a402fae 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -136,199 +136,6 @@ pub fn dangerous_command_patterns_for_platform(platform: Platform) -> Vec Option { - #[cfg(target_os = "linux")] - { - let wchan = std::fs::read_to_string(format!("/proc/{}/wchan", pid)).ok()?; - let wchan = wchan.trim(); - if wchan.is_empty() { - return None; - } - Some(wchan.contains("tty_read") || wchan.contains("n_tty_read") || wchan == "pipe_wait") - } - #[cfg(target_os = "macos")] - { - use std::mem; - let mut task_info: libc::proc_taskinfo = unsafe { mem::zeroed() }; - let size = mem::size_of::() as i32; - let ret = unsafe { - libc::proc_pidinfo( - pid as i32, - libc::PROC_PIDTASKINFO, - 0, - &mut task_info as *mut _ as *mut libc::c_void, - size, - ) - }; - if ret <= 0 { - return None; - } - // pti_numrunning == 0 means no thread is actively on CPU. - // Combined with output silence this strongly suggests the process - // is blocked on I/O (likely a stdin read). - Some(task_info.pti_numrunning == 0) - } - #[cfg(target_os = "windows")] - { - windows_is_process_waiting_on_stdin(pid) - } - #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] - { - let _ = pid; - None - } -} - -/// Windows implementation: check if a process is waiting for stdin input. -/// -/// Uses NtQuerySystemInformation to enumerate process threads and check if -/// all threads are in Wait state with Executive wait reason, which indicates -/// the process is blocked on I/O (likely console input). -#[cfg(target_os = "windows")] -fn windows_is_process_waiting_on_stdin(pid: u32) -> Option { - // SystemProcessInformation = 5 - const SYSTEM_PROCESS_INFORMATION: u32 = 5; - const STATUS_INFO_LENGTH_MISMATCH: i32 = -1073741820; // 0xC0000004 - - #[repr(C)] - #[allow(non_snake_case)] - struct SystemProcessInfo { - next_entry_offset: u32, - number_of_threads: u32, - working_set_private_size: i64, - hard_fault_count: u32, - number_of_threads_high_watermark: u32, - cycle_time: u64, - create_time: i64, - user_time: i64, - kernel_time: i64, - image_name_length: u16, - image_name_max_length: u16, - image_name: *const u16, - base_priority: i32, - unique_process_id: *mut std::ffi::c_void, - inherited_from_unique_process_id: *mut std::ffi::c_void, - handle_count: u32, - session_id: u32, - unique_process_key: usize, - peak_virtual_size: usize, - virtual_size: usize, - page_fault_count: u32, - peak_working_set_size: usize, - working_set_size: usize, - quota_peak_paged_pool_usage: usize, - quota_paged_pool_usage: usize, - quota_peak_non_paged_pool_usage: usize, - quota_non_paged_pool_usage: usize, - pagefile_usage: usize, - peak_pagefile_usage: usize, - private_page_count: usize, - read_operation_count: i64, - write_operation_count: i64, - other_operation_count: i64, - read_transfer_count: i64, - write_transfer_count: i64, - other_transfer_count: i64, - // SYSTEM_THREAD_INFORMATION[1] follows in memory - threads: [SystemThreadInfo; 1], - } - - #[repr(C)] - #[derive(Clone, Copy)] - #[allow(non_snake_case)] - struct SystemThreadInfo { - kernel_time: i64, - user_time: i64, - create_time: i64, - wait_time: u32, - start_address: *mut std::ffi::c_void, - client_id_unique_process: *mut std::ffi::c_void, - client_id_unique_thread: *mut std::ffi::c_void, - priority: i32, - base_priority: i32, - context_switches: u32, - thread_state: u32, - wait_reason: u32, - } - - #[allow(non_snake_case)] - unsafe extern "system" { - fn NtQuerySystemInformation( - system_information_class: u32, - system_information: *mut u8, - system_information_length: u32, - return_length: *mut u32, - ) -> i32; - } - - unsafe { - // Query buffer size first - let mut buf_len: u32 = 0; - let status = NtQuerySystemInformation( - SYSTEM_PROCESS_INFORMATION, - std::ptr::null_mut(), - 0, - &mut buf_len, - ); - if status != STATUS_INFO_LENGTH_MISMATCH || buf_len == 0 { - return None; - } - - // Allocate buffer with extra space (processes may be created between calls) - buf_len = buf_len.saturating_mul(2).max(65536); - let mut buffer: Vec = vec![0u8; buf_len as usize]; - - let status = NtQuerySystemInformation( - SYSTEM_PROCESS_INFORMATION, - buffer.as_mut_ptr(), - buf_len, - &mut buf_len, - ); - if status < 0 { - return None; - } - - // Walk the linked list of SYSTEM_PROCESS_INFORMATION - let mut offset: usize = 0; - loop { - let info = &*(buffer.as_ptr().add(offset) as *const SystemProcessInfo); - let proc_id = info.unique_process_id as u32; - - if proc_id == pid { - let thread_count = info.number_of_threads as usize; - if thread_count == 0 { - return Some(false); - } - - // Thread states: Running=2, Waiting=5 - // Wait reasons: Executive=0, FreePage=1, PageIn=2, PoolAllocation=3, - // DelayExecution=4, Suspended=5, UserRequest=6, ... - // Executive wait + all threads waiting = likely blocked on I/O - let all_waiting = (0..thread_count).all(|i| { - let thread = &*info.threads.as_ptr().add(i); - thread.thread_state == 5 && thread.wait_reason == 0 - }); - - return Some(all_waiting); - } - - if info.next_entry_offset == 0 { - break; - } - offset += info.next_entry_offset as usize; - } - - None - } -} - /// Disable Windows console QuickEdit mode. /// /// With QuickEdit enabled (the default), clicking the console window enters diff --git a/src/storage/mod.rs b/src/storage/mod.rs index d674f6f..75a00ab 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -665,16 +665,13 @@ impl SessionStore { topic_params.push(id); } let mut stmt = conn.prepare(&topic_sql)?; - let rows = stmt.query_map(topic_params.as_slice(), |row| { - row.get::<_, String>(0) - })?; + let rows = stmt.query_map(topic_params.as_slice(), |row| row.get::<_, String>(0))?; for tid in rows { affected_topics.push(tid?); } - let sql = format!( - "DELETE FROM messages WHERE session_id = ? AND id IN ({placeholders})" - ); + let sql = + format!("DELETE FROM messages WHERE session_id = ? AND id IN ({placeholders})"); let mut params_vec: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(chunk.len() + 1); params_vec.push(&session_id); for id in chunk { @@ -2306,7 +2303,10 @@ impl SessionStore { /// /// 用于整会话级变更路径（replace_active_history / compact_active_history）： /// 这些路径会删除或重组整个 session 的消息行，所有 topic 的统计都可能漂移。 - pub fn recompute_session_topic_usage_stats(&self, session_id: &str) -> Result<(), StorageError> { + pub fn recompute_session_topic_usage_stats( + &self, + session_id: &str, + ) -> Result<(), StorageError> { let ids: Vec = { let conn = self.pool.get()?; let mut stmt = conn.prepare("SELECT id FROM topics WHERE session_id = ?1")?; diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 089bd03..8b75f43 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -965,7 +965,9 @@ fn test_delete_messages_by_ids_removes_only_target_rows() { // 删除中间 2 条（sanitize 回写场景：只删被清理的消息） let to_delete: Vec = all[1..3].iter().map(|m| m.id.clone()).collect(); - let deleted = store.delete_messages_by_ids(&session.id, &to_delete).unwrap(); + let deleted = store + .delete_messages_by_ids(&session.id, &to_delete) + .unwrap(); assert_eq!(deleted, 2); let remaining = store.load_messages_for_topic(&topic.id, None).unwrap(); @@ -1009,7 +1011,9 @@ fn assistant_with_usage( fn test_topic_token_stats_incremental_maintenance() { let store = SessionStore::in_memory().unwrap(); let session = store.create_cli_session(Some("stats")).unwrap(); - let topic = store.create_topic(&session.id, "topic-stats", None).unwrap(); + let topic = store + .create_topic(&session.id, "topic-stats", None) + .unwrap(); // 无 usage 数据时不返回条目（前端不显示 token 标签） store @@ -1132,10 +1136,12 @@ fn test_topic_token_stats_recompute_after_delete_and_clear() { &assistant_with_usage("a3", 300, 100, 0, None), ) .unwrap(); - assert!(!store - .batch_topic_token_stats(&[topic.id.as_str()]) - .unwrap() - .is_empty()); + assert!( + !store + .batch_topic_token_stats(&[topic.id.as_str()]) + .unwrap() + .is_empty() + ); store.clear_messages(&session.id).unwrap(); let stats = store.batch_topic_token_stats(&[topic.id.as_str()]).unwrap(); assert!(stats.is_empty()); @@ -1268,7 +1274,10 @@ fn test_foreign_keys_enforced_on_pool_connections() { |row| row.get(0), ) .unwrap(); - assert_eq!(dangling, 0, "SET NULL 未生效：删除话题后消息仍持悬空 topic_id"); + assert_eq!( + dangling, 0, + "SET NULL 未生效：删除话题后消息仍持悬空 topic_id" + ); let remaining: i64 = conn .query_row( "SELECT COUNT(*) FROM messages WHERE session_id = ?1", @@ -1289,14 +1298,19 @@ fn test_foreign_keys_enforced_on_pool_connections() { |row| row.get(0), ) .unwrap(); - assert_eq!(orphan_topics, 0, "CASCADE 未生效：删除会话后 topics 孤儿残留"); + assert_eq!( + orphan_topics, 0, + "CASCADE 未生效：删除会话后 topics 孤儿残留" + ); } #[test] fn test_clear_messages_resets_topic_message_count() { let store = SessionStore::in_memory().unwrap(); let session = store.create_cli_session(Some("clear-count")).unwrap(); - let topic = store.create_topic(&session.id, "topic-clear", None).unwrap(); + let topic = store + .create_topic(&session.id, "topic-clear", None) + .unwrap(); for i in 1..=3 { store .append_message_with_topic( diff --git a/src/tools/bash.rs b/src/tools/bash.rs index 357d961..4c12486 100644 --- a/src/tools/bash.rs +++ b/src/tools/bash.rs @@ -11,85 +11,20 @@ use tokio::sync::{Mutex, mpsc}; use tokio::time::{Instant, sleep_until}; use crate::platform::{ShellInfo, dangerous_command_patterns}; -use crate::tools::shell_session::ShellSessionManager; use crate::tools::traits::{Tool, ToolResult}; use crate::tools::{check_null_args, extract_u64}; const MAX_TIMEOUT_SECS: u64 = 600; const MAX_OUTPUT_CHARS: usize = 50_000; /// 运行时单流输出缓冲上限（字节）：超出后保留头尾、丢弃中段， -/// 防止长输出命令（或交互式会话的 drain 任务）无限增长吃满内存。 +/// 防止长输出命令无限增长吃满内存。 const MAX_RUNTIME_BUFFER_BYTES: usize = 1024 * 1024; -/// pending 短语增量检测的尾部窗口（字节）：交互提示只出现在输出尾部， -/// 只需"新 chunk + 尾部窗口"即可捕获（窗口 ≥ 最长短语长度，覆盖跨 chunk 边界）。 -const PENDING_WINDOW_BYTES: usize = 2048; /// 子进程退出后，等待 read_stream 把管道残余输出排空的最长时间。 /// /// 不能无界等待 EOF：若子进程派生了继承 stdout 管道的守护进程 /// （如 `adb start-server` 启动的 adb daemon），EOF 永不到达， /// 会永久阻塞并绕过 deadline（select! 已由 child.wait() 分支退出）。 const STREAM_DRAIN_MS: u64 = 1000; -const PENDING_USER_ACTION_MARKER: &str = "__PICOBOT_PENDING_USER_ACTION__"; -const INTERACTIVE_HINT: &str = - "进程正在等待输入。请使用 session_id 和 stdin_input 参数回复交互内容。"; -const NON_INTERACTIVE_HINT: &str = - "该命令正在等待你完成外部操作。完成后请告诉我继续，或重新运行后续检查命令。"; - -/// "等待用户操作"检测短语（全部小写；检测前对输出做 to_lowercase）。 -/// 新增短语时保持小写，并确保长度不超过 PENDING_WINDOW_BYTES。 -const PENDING_USER_ACTION_PHRASES: &[&str] = &[ - // 中文 — 原有 - "等待用户授权", - "等待授权", - "等待你授权", - "在浏览器中打开以下链接进行认证", - // 中文 — 新增（lark-cli 等工具的常见提示） - "请在浏览器中", - "请打开以下链接", - "打开以下链接", - "打开链接", - "访问以下", - "访问此链接", - "复制链接", - "输入验证码", - "输入授权码", - "完成认证", - "完成授权", - "请登录", - "正在等待", - "等待用户", - "手动授权", - // 英文 — 原有 - "open the following link", - "waiting for authorization", - "waiting for user authorization", - "waiting for approval", - "device/verify", - "user_code=", - // 英文 — 新增 - "visit the following url", - "visit this url", - "open the following url", - "browser to authenticate", - "browser to complete", - "enter the code", - "enter code", - "verification code", - "authorization code", - "one-time code", - "device code", - "oauth", - "go to the following", - "navigate to the following", - "paste the code", -]; - -/// 在小写化文本中检测 pending 短语（调用方需先 to_lowercase）。 -fn contains_pending_phrase(lowercase_text: &str) -> bool { - PENDING_USER_ACTION_PHRASES - .iter() - .any(|phrase| lowercase_text.contains(phrase)) -} /// 缓冲超限时保留头尾（头 1/2 + 尾 1/4），在字符边界处截断。 /// 头尾之外的中段对最终 truncate_output（头+尾各 25K 字符）已无贡献。 @@ -204,11 +139,10 @@ pub struct BashTool { /// 避免每次执行命令都重新编译全部正则。 deny_patterns: Vec, shell: ShellKind, - session_manager: Arc, } impl BashTool { - pub fn new(session_manager: Arc) -> Self { + pub fn new() -> Self { Self { timeout_secs: 60, working_dir: None, @@ -223,7 +157,6 @@ impl BashTool { }) .collect(), shell: ShellKind::detect(), - session_manager, } } @@ -274,26 +207,6 @@ impl BashTool { tail ) } - - fn pending_output(&self, output: &str, session_id: Option<&str>) -> String { - let hint = if session_id.is_some() { - INTERACTIVE_HINT - } else { - NON_INTERACTIVE_HINT - }; - let session_line = session_id - .map(|id| format!("[session_id: {}]\n", id)) - .unwrap_or_default(); - let output_section = if output.trim().is_empty() { - "(进程尚未输出内容。进程正在等待输入，请使用 session_id 和 stdin_input 参数发送输入内容。)" - } else { - &self.truncate_output(output.trim()) - }; - format!( - "{}\n{}{}\n\n{}", - PENDING_USER_ACTION_MARKER, session_line, hint, output_section - ) - } } async fn drain_available_chunks( @@ -316,7 +229,7 @@ async fn drain_available_chunks( impl Default for BashTool { fn default() -> Self { - Self::new(Arc::new(ShellSessionManager::new())) + Self::new() } } @@ -343,18 +256,6 @@ impl Tool for BashTool { "description": format!("Timeout in seconds (default {}, max {})", self.timeout_secs, MAX_TIMEOUT_SECS), "minimum": 1, "maximum": MAX_TIMEOUT_SECS - }, - "interactive": { - "type": "boolean", - "description": "Whether this command may enter a wait-for-user-action flow such as browser/device authentication" - }, - "session_id": { - "type": "string", - "description": "Continue a previous interactive session by providing its session_id" - }, - "stdin_input": { - "type": "string", - "description": "Input text to send to the process stdin (used with session_id)" } }, "required": ["command"] @@ -370,26 +271,6 @@ impl Tool for BashTool { return Ok(result); } - // Handle session continuation first - if let Some(session_id) = args.get("session_id").and_then(|v| v.as_str()) { - let input = args - .get("stdin_input") - .and_then(|v| v.as_str()) - .unwrap_or(""); - return match self.session_manager.send_input(session_id, input).await { - Ok(output) => Ok(ToolResult { - success: true, - output, - error: None, - }), - Err(e) => Ok(ToolResult { - success: false, - output: String::new(), - error: Some(e), - }), - }; - } - let command = match args.get("command").and_then(|v| v.as_str()) { Some(c) => c, None => { @@ -462,7 +343,9 @@ impl BashTool { ) -> Result { let mut cmd = Command::new(self.shell.executable()); cmd.args(self.shell.command_args(command)) - .stdin(Stdio::piped()) + // 一次性命令不支持交互：stdin 直接 EOF，等待输入的程序会立即得到 EOF + // 而不是挂起至超时。需要交互式会话请使用 pty 工具。 + .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) // 外部取消安全：run_command future 被 drop 时（/stop 的 select! 竞速、 @@ -474,8 +357,6 @@ impl BashTool { let mut child = cmd.spawn().map_err(|e| format!("Failed to spawn: {}", e))?; - // Take stdin writer before stdout/stderr - let child_stdin = child.stdin.take(); let stdout = child.stdout.take(); let stderr = child.stderr.take(); let (tx, rx_inner) = mpsc::unbounded_channel::<(bool, String)>(); @@ -492,9 +373,6 @@ impl BashTool { let stdout_buf = Arc::new(Mutex::new(String::new())); let stderr_buf = Arc::new(Mutex::new(String::new())); - // pending 短语增量检测窗口：仅保留最近输出的小写化尾部， - // 每个 chunk 只扫描"窗口 + 新 chunk"（O(chunk)），不再全量重扫（O(累计输出)）。 - let mut pending_window = String::new(); let deadline = Instant::now() + Duration::from_secs(timeout_secs); loop { @@ -542,44 +420,13 @@ impl BashTool { None => std::future::pending().await, } } => { - { - let mut buf = if is_stderr { - stderr_buf.lock().await - } else { - stdout_buf.lock().await - }; - buf.push_str(&chunk); - cap_output_buffer(&mut buf); - } - - // 增量 pending 检测：交互提示只出现在输出尾部，只扫描 - // "尾部窗口 + 新 chunk"，避免对全量输出做 O(n²) 重扫 - pending_window.push_str(&chunk.to_lowercase()); - if pending_window.len() > PENDING_WINDOW_BYTES * 2 { - let cut = pending_window - .floor_char_boundary(pending_window.len() - PENDING_WINDOW_BYTES); - pending_window.drain(..cut); - } - if contains_pending_phrase(&pending_window) { - let mut rx_val = rx.take().unwrap(); - drain_available_chunks(&mut rx_val, &stdout_buf, &stderr_buf).await; - let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None); - // 保存为交互式会话：read_stream 任务作为会话生产者，必须继续运行，不 abort。 - if let Some(stdin) = child_stdin { - let session_id = self.session_manager.save_session( - child, stdin, rx_val, - stdout_buf.lock().await.clone(), - stderr_buf.lock().await.clone(), - ).await; - return Ok(self.pending_output(&combined, Some(&session_id))); - } - // 无 stdin 可存会话：硬终止并回收，终止 read_stream 任务避免泄漏。 - Self::kill_and_reap(&mut child).await; - for t in &read_tasks { - t.abort(); - } - return Ok(self.pending_output(&combined, None)); - } + let mut buf = if is_stderr { + stderr_buf.lock().await + } else { + stdout_buf.lock().await + }; + buf.push_str(&chunk); + cap_output_buffer(&mut buf); } _ = sleep_until(deadline) => { // 严格硬超时：deadline 到达一律 kill 并返回超时错误，不再转 pending 会话。 @@ -610,8 +457,11 @@ impl BashTool { /// without a trailing newline are still visible to the detection logic. const STREAM_FLUSH_MS: u64 = 500; -async fn read_stream(stream: R, is_stderr: bool, tx: mpsc::UnboundedSender<(bool, String)>) -where +pub(crate) async fn read_stream( + stream: R, + is_stderr: bool, + tx: mpsc::UnboundedSender<(bool, String)>, +) where R: AsyncRead + Unpin + Send + 'static, { let mut reader = BufReader::new(stream); @@ -659,7 +509,7 @@ where } /// 尝试 UTF-8 解码，失败则尝试 GBK 解码 -fn decode_bytes(bytes: &[u8]) -> String { +pub(crate) fn decode_bytes(bytes: &[u8]) -> String { // 首先尝试 UTF-8 if let Ok(s) = std::str::from_utf8(bytes) { return s.to_string(); @@ -833,27 +683,6 @@ mod tests { assert!(result.error.unwrap().contains("timed out")); } - #[tokio::test] - async fn test_pending_user_action_detection() { - let tool = BashTool::default(); - let command = if cfg!(target_os = "windows") { - "Write-Host 'waiting for authorization'; Start-Sleep -Seconds 10" - } else { - "printf 'waiting for authorization'; sleep 10" - }; - let result = tool - .execute(json!({ - "command": command, - "timeout": 5, - "interactive": true - })) - .await - .unwrap(); - - assert!(result.success); - assert!(result.output.contains(PENDING_USER_ACTION_MARKER)); - } - #[test] fn test_truncate_output_handles_utf8_char_boundaries() { let tool = BashTool::default(); diff --git a/src/tools/mod.rs b/src/tools/mod.rs index de2752f..1a768fa 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -6,11 +6,12 @@ pub mod file_write; pub mod http_request; pub mod memory_manage; pub mod memory_search; +pub mod pty; +pub mod pty_session; pub mod registry; pub mod scheduler_manage; pub mod schema; pub mod session_send; -pub mod shell_session; pub mod skill_activate; pub mod skill_manage; pub mod task; @@ -29,6 +30,8 @@ pub use file_write::FileWriteTool; pub use http_request::HttpRequestTool; pub use memory_manage::MemoryManageTool; pub use memory_search::MemorySearchTool; +pub use pty::PtyTool; +pub use pty_session::PtySessionManager; pub use registry::ToolRegistry; pub use scheduler_manage::SchedulerManageTool; pub use schema::{CleaningStrategy, SchemaCleanr}; @@ -36,7 +39,6 @@ pub use session_send::{ NoopSessionMessageSender, SessionMessageSender, SessionSendOutcome, SessionSendRequest, SessionSendTool, }; -pub use shell_session::ShellSessionManager; pub use skill_activate::SkillActivateTool; pub use skill_manage::SkillManageTool; pub use task::{ diff --git a/src/tools/pty.rs b/src/tools/pty.rs new file mode 100644 index 0000000..0db39f7 --- /dev/null +++ b/src/tools/pty.rs @@ -0,0 +1,468 @@ +//! The `pty` tool: manage persistent pseudo-terminal (PTY) sessions. +//! +//! A single tool with an `action` parameter (spawn/write/read/kill/list), +//! backed by `PtySessionManager`. Interactive programs, long-running +//! services and multi-step commands keep their terminal state across tool +//! calls; output is read incrementally. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::json; + +use crate::platform::dangerous_command_patterns; +use crate::tools::pty_session::{DEFAULT_PTY_COLS, DEFAULT_PTY_ROWS, PtySessionManager}; +use crate::tools::traits::{Tool, ToolResult}; +use crate::tools::{check_null_args, extract_bool, extract_string, extract_u64, require_string}; + +const MAX_PTY_COLS: u64 = 500; +const MAX_PTY_ROWS: u64 = 200; + +pub struct PtyTool { + manager: Arc, + /// 危险命令拦截正则：构造时预编译（与 BashTool 同源规则）。 + deny_patterns: Vec, +} + +impl PtyTool { + pub fn new(manager: Arc) -> Self { + Self { + manager, + deny_patterns: dangerous_command_patterns() + .into_iter() + .filter_map(|p| match regex::Regex::new(&p) { + Ok(re) => Some(re), + Err(e) => { + tracing::warn!(pattern = %p, error = %e, "Invalid deny pattern skipped"); + None + } + }) + .collect(), + } + } + + fn guard_command(&self, command: &str) -> Option { + let lower = command.to_lowercase(); + for re in &self.deny_patterns { + if re.is_match(&lower) { + return Some(format!( + "Command blocked by safety guard (dangerous pattern: {})", + re.as_str() + )); + } + } + None + } +} + +impl Default for PtyTool { + fn default() -> Self { + Self::new(Arc::new(PtySessionManager::new())) + } +} + +fn error_result(message: String) -> anyhow::Result { + Ok(ToolResult { + success: false, + output: String::new(), + error: Some(message), + }) +} + +fn manager_result(result: Result) -> anyhow::Result { + match result { + Ok(output) => Ok(ToolResult { + success: true, + output, + error: None, + }), + Err(e) => error_result(e), + } +} + +#[async_trait] +impl Tool for PtyTool { + fn name(&self) -> &str { + "pty" + } + + fn description(&self) -> &str { + "Manage persistent pseudo-terminal (PTY) sessions for interactive programs, long-running services and multi-step commands that must keep terminal state across tool calls. Actions: spawn (create a session running a command), write (send input), read (read new output), kill (terminate a session), list (show all sessions). Prefer this tool over bash/shell when the program needs a real terminal (TTY) or when you must keep interacting with it over multiple calls." + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["spawn", "write", "read", "kill", "list"], + "description": "Operation to perform: spawn=create a session, write=send input to a session, read=read new output, kill=terminate a session, list=show all sessions" + }, + "command": { + "type": "string", + "description": "spawn: the command to run inside the PTY" + }, + "session_id": { + "type": "string", + "description": "Target session for write/read/kill (returned by spawn)" + }, + "input": { + "type": "string", + "description": "write: text to send to the PTY; control characters can be embedded (e.g. \\u0003 for Ctrl+C)" + }, + "raw": { + "type": "boolean", + "description": "write: send input exactly as-is without appending a newline (use for control sequences)" + }, + "wait_ms": { + "type": "integer", + "description": "read: how long to wait for new output in milliseconds (default 500, max 10000)" + }, + "cwd": { + "type": "string", + "description": "spawn: working directory for the command" + }, + "cols": { + "type": "integer", + "description": format!("spawn: terminal columns (default {})", DEFAULT_PTY_COLS), + "minimum": 1, + "maximum": MAX_PTY_COLS + }, + "rows": { + "type": "integer", + "description": format!("spawn: terminal rows (default {})", DEFAULT_PTY_ROWS), + "minimum": 1, + "maximum": MAX_PTY_ROWS + }, + "shell": { + "type": "boolean", + "description": "spawn: run the command via the platform shell (default true); false executes it directly (whitespace-split)" + }, + "env": { + "type": "object", + "description": "spawn: additional environment variables as string key/value pairs" + } + }, + "required": ["action"] + }) + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + if let Some(result) = check_null_args(&args, "pty") { + return Ok(result); + } + + let action = match require_string(&args, "action") { + Ok(a) => a.to_lowercase(), + Err(e) => return error_result(e), + }; + + match action.as_str() { + "spawn" => { + let command = match require_string(&args, "command") { + Ok(c) => c, + Err(e) => return error_result(e), + }; + if let Some(blocked) = self.guard_command(&command) { + return error_result(blocked); + } + let cwd = extract_string(&args, "cwd"); + let cols = extract_u64(&args, "cols") + .map(|v| v.clamp(1, MAX_PTY_COLS) as u16) + .unwrap_or(DEFAULT_PTY_COLS); + let rows = extract_u64(&args, "rows") + .map(|v| v.clamp(1, MAX_PTY_ROWS) as u16) + .unwrap_or(DEFAULT_PTY_ROWS); + let use_shell = extract_bool(&args, "shell").unwrap_or(true); + let env: Vec<(String, String)> = args + .get("env") + .and_then(|v| v.as_object()) + .map(|obj| { + obj.iter() + .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string()))) + .collect() + }) + .unwrap_or_default(); + + match self + .manager + .spawn(&command, cwd.as_deref(), &env, cols, rows, use_shell) + .await + { + Ok(session_id) => Ok(ToolResult { + success: true, + output: format!( + "PTY session started.\n[session_id: {}]\nCommand: {}\nUse action=read with this session_id to fetch output, action=write to send input, action=kill to terminate.", + session_id, command + ), + error: None, + }), + Err(e) => error_result(e), + } + } + "write" => { + let session_id = match require_string(&args, "session_id") { + Ok(s) => s, + Err(e) => return error_result(e), + }; + let input = match require_string(&args, "input") { + Ok(i) => i, + Err(e) => return error_result(e), + }; + let raw = extract_bool(&args, "raw").unwrap_or(false); + manager_result(self.manager.write_input(&session_id, &input, raw).await) + } + "read" => { + let session_id = match require_string(&args, "session_id") { + Ok(s) => s, + Err(e) => return error_result(e), + }; + let wait_ms = extract_u64(&args, "wait_ms").unwrap_or(0); + manager_result(self.manager.read_output(&session_id, wait_ms).await) + } + "kill" => { + let session_id = match require_string(&args, "session_id") { + Ok(s) => s, + Err(e) => return error_result(e), + }; + manager_result(self.manager.kill_session(&session_id).await) + } + "list" => manager_result(Ok(self.manager.list_sessions().await)), + other => error_result(format!( + "Unknown action: {}. Expected one of spawn, write, read, kill, list", + other + )), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::tools::pty_session::pty_available; + + /// 从 spawn 结果中提取 [session_id: xxx] + fn extract_session_id(output: &str) -> String { + let marker = "[session_id: "; + let start = output + .find(marker) + .expect("spawn output must contain session_id") + + marker.len(); + let end = output[start..] + .find(']') + .expect("session_id must be terminated") + + start; + output[start..end].to_string() + } + + /// 轮询 read 直到输出包含 expected 或耗尽尝试次数（PTY 输出到达时机不确定）。 + async fn read_until(tool: &PtyTool, session_id: &str, expected: &str, attempts: u32) -> String { + let mut accumulated = String::new(); + for _ in 0..attempts { + let result = tool + .execute(json!({ "action": "read", "session_id": session_id, "wait_ms": 2000 })) + .await + .unwrap(); + assert!(result.success, "read failed: {:?}", result.error); + accumulated.push_str(&result.output); + if accumulated.contains(expected) { + return accumulated; + } + } + accumulated + } + + #[tokio::test] + async fn test_spawn_read_exit() { + if !pty_available() { + // 无控制台受限环境（如部分 CI 沙箱）不支持 ConPTY，跳过。 + return; + } + let tool = PtyTool::default(); + let command = if cfg!(target_os = "windows") { + "Write-Output 'PTY_HELLO'; exit 0" + } else { + "echo PTY_HELLO; exit 0" + }; + + let spawned = tool + .execute(json!({ "action": "spawn", "command": command })) + .await + .unwrap(); + assert!(spawned.success, "spawn failed: {:?}", spawned.error); + let session_id = extract_session_id(&spawned.output); + + let output = read_until(&tool, &session_id, "PTY_HELLO", 5).await; + assert!(output.contains("PTY_HELLO"), "output was: {}", output); + + // 进程退出后 read 仍能返回状态 + let final_read = tool + .execute(json!({ "action": "read", "session_id": session_id, "wait_ms": 2000 })) + .await + .unwrap(); + assert!(final_read.success); + assert!(final_read.output.contains("[Process status:")); + + let killed = tool + .execute(json!({ "action": "kill", "session_id": session_id })) + .await + .unwrap(); + assert!(killed.success); + } + + #[tokio::test] + async fn test_write_input_interactive() { + if !pty_available() { + return; + } + let tool = PtyTool::default(); + let command = if cfg!(target_os = "windows") { + "$v = Read-Host 'Q'; Write-Output \"got:$v\"" + } else { + "read -r -p 'Q: ' x && echo got:$x" + }; + + let spawned = tool + .execute(json!({ "action": "spawn", "command": command })) + .await + .unwrap(); + assert!(spawned.success, "spawn failed: {:?}", spawned.error); + let session_id = extract_session_id(&spawned.output); + + // 等待提示出现 + let _ = read_until(&tool, &session_id, "Q", 5).await; + + let written = tool + .execute(json!({ "action": "write", "session_id": session_id, "input": "hello_pty" })) + .await + .unwrap(); + assert!(written.success, "write failed: {:?}", written.error); + + let output = read_until(&tool, &session_id, "got:hello_pty", 5).await; + assert!(output.contains("got:hello_pty"), "output was: {}", output); + + let killed = tool + .execute(json!({ "action": "kill", "session_id": session_id })) + .await + .unwrap(); + assert!(killed.success); + } + + #[tokio::test] + async fn test_read_is_incremental() { + if !pty_available() { + return; + } + let tool = PtyTool::default(); + let command = if cfg!(target_os = "windows") { + "Write-Output 'FIRST_LINE'; Start-Sleep -Seconds 30" + } else { + "echo FIRST_LINE; sleep 30" + }; + + let spawned = tool + .execute(json!({ "action": "spawn", "command": command })) + .await + .unwrap(); + let session_id = extract_session_id(&spawned.output); + + let first = read_until(&tool, &session_id, "FIRST_LINE", 5).await; + assert!(first.contains("FIRST_LINE")); + + // 无新输出时第二次 read 不应重复返回旧内容 + let second = tool + .execute(json!({ "action": "read", "session_id": session_id, "wait_ms": 300 })) + .await + .unwrap(); + assert!(second.success); + assert!( + !second.output.contains("FIRST_LINE"), + "second read repeated old output: {}", + second.output + ); + assert!(second.output.contains("(no new output)")); + + let killed = tool + .execute(json!({ "action": "kill", "session_id": session_id })) + .await + .unwrap(); + assert!(killed.success); + } + + #[tokio::test] + async fn test_kill_long_running() { + if !pty_available() { + return; + } + let tool = PtyTool::default(); + let command = if cfg!(target_os = "windows") { + "Start-Sleep -Seconds 60" + } else { + "sleep 60" + }; + + let spawned = tool + .execute(json!({ "action": "spawn", "command": command })) + .await + .unwrap(); + let session_id = extract_session_id(&spawned.output); + + let listed = tool.execute(json!({ "action": "list" })).await.unwrap(); + assert!(listed.success); + assert!(listed.output.contains(&session_id)); + + let killed = tool + .execute(json!({ "action": "kill", "session_id": session_id })) + .await + .unwrap(); + assert!(killed.success); + assert!(killed.output.contains("terminated")); + + let listed_after = tool.execute(json!({ "action": "list" })).await.unwrap(); + assert!(!listed_after.output.contains(&session_id)); + } + + #[tokio::test] + async fn test_unknown_session_errors() { + let tool = PtyTool::default(); + for action in ["write", "read", "kill"] { + let mut args = json!({ "action": action, "session_id": "no-such-session" }); + if action == "write" { + args["input"] = json!("x"); + } + let result = tool.execute(args).await.unwrap(); + assert!(!result.success, "action {} should fail", action); + assert!(result.error.unwrap().contains("Session not found")); + } + } + + #[tokio::test] + async fn test_invalid_action_and_missing_params() { + let tool = PtyTool::default(); + + let result = tool.execute(json!({ "action": "explode" })).await.unwrap(); + assert!(!result.success); + assert!(result.error.unwrap().contains("Unknown action")); + + let result = tool.execute(json!({})).await.unwrap(); + assert!(!result.success); + assert!(result.error.unwrap().contains("action")); + + let result = tool.execute(json!({ "action": "spawn" })).await.unwrap(); + assert!(!result.success); + assert!(result.error.unwrap().contains("command")); + + let result = tool + .execute(json!({ "action": "spawn", "command": "rm -rf /some/path" })) + .await + .unwrap(); + if cfg!(target_os = "windows") { + // Windows 平台无 rm 拦截规则，命令本身允许 spawn（此处仅验证未被误拦） + assert!(result.success || !result.error.unwrap().contains("blocked")); + } else { + assert!(!result.success); + assert!(result.error.unwrap().contains("blocked")); + } + } +} diff --git a/src/tools/pty_session.rs b/src/tools/pty_session.rs new file mode 100644 index 0000000..640b055 --- /dev/null +++ b/src/tools/pty_session.rs @@ -0,0 +1,672 @@ +//! Persistent PTY (pseudo-terminal) session management. +//! +//! Provides `PtySessionManager`, an independent service that keeps child +//! processes alive in real pseudo-terminals between tool calls. Unlike the +//! removed pipe-based interactive shell sessions, a PTY makes the child see +//! an actual TTY, which is required by TUI programs, prompts that only +//! appear on a terminal, and long-running services with terminal state. +//! +//! The manager is created at the gateway layer and injected into `PtyTool` +//! via `Arc`. It does NOT start background tasks on its own; expired +//! sessions are swept lazily on each public call, and final cleanup is +//! driven externally via `shutdown()`. + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::pin::Pin; +use std::sync::{Arc, LazyLock, Mutex as StdMutex}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use portable_pty::{CommandBuilder, MasterPty, PtySize}; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::sync::{Mutex, Notify, mpsc}; +use tokio::time::Instant; +use uuid::Uuid; + +use crate::tools::bash::{ShellKind, cap_output_buffer, read_stream}; + +/// 会话过期时长（秒）：基于最后活动时间（读/写/创建），非创建时间。 +/// 长运行服务只要定期 read/write 即不会被回收。 +const SESSION_TTL_SECS: u64 = 1800; +/// read 等待新输出的默认/上限毫秒数。 +const DEFAULT_READ_WAIT_MS: u64 = 500; +const MAX_READ_WAIT_MS: u64 = 10_000; +/// kill 时返回的尾部输出字符数。 +const KILL_TAIL_CHARS: usize = 2000; + +pub const DEFAULT_PTY_COLS: u16 = 120; +pub const DEFAULT_PTY_ROWS: u16 = 30; + +/// ANSI 转义序列剥离：CSI（光标/颜色控制）、OSC（标题等，BEL 或 ST 结尾）、 +/// 字符集选择与双字节 Fe 转义。PTY 输出含大量控制序列，剥离后更适合 LLM 阅读。 +static ANSI_RE: LazyLock = LazyLock::new(|| { + regex::Regex::new( + r"\x1b(?:\[[0-9;?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)?|[()#][0-9A-Za-z]|[@-Z\\-_])", + ) + .expect("ANSI escape regex must compile") +}); + +fn strip_ansi(text: &str) -> String { + let cleaned = ANSI_RE.replace_all(text, ""); + let normalized = cleaned.replace("\r\n", "\n"); + // 回车覆盖语义：每行只保留最后一次 \r 之后的片段（进度条等场景）， + // 避免 LLM 看到大量被覆盖的中间状态。 + let mut result = String::with_capacity(normalized.len()); + for (i, line) in normalized.split('\n').enumerate() { + if i > 0 { + result.push('\n'); + } + match line.rfind('\r') { + Some(pos) => result.push_str(&line[pos + 1..]), + None => result.push_str(line), + } + } + result +} + +/// 探测当前环境是否支持 PTY。 +/// +/// Windows ConPTY 需要可创建控制台的会话；无控制台的受限环境 +/// （如某些 CI 沙箱/服务会话）openpty 会失败。Unix 恒为 true。 +/// 测试用它在不支持的环境中提前跳过，产品代码用它给出清晰的降级提示。 +pub fn pty_available() -> bool { + static AVAILABLE: std::sync::OnceLock = std::sync::OnceLock::new(); + *AVAILABLE.get_or_init(|| { + portable_pty::native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .is_ok() + }) +} + +/// 将阻塞式 PTY reader 的字节块通过 mpsc 通道转接为 AsyncRead， +/// 从而复用 bash 工具的 `read_stream`（按行即时发送 + 500ms 空闲 flush +/// 无换行提示 + UTF-8→GBK→lossy 解码）。 +struct ChannelReader { + rx: mpsc::UnboundedReceiver>, + pending: Vec, +} + +impl AsyncRead for ChannelReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + if this.pending.is_empty() { + match this.rx.poll_recv(cx) { + Poll::Ready(Some(chunk)) => this.pending = chunk, + Poll::Ready(None) => return Poll::Ready(Ok(())), + Poll::Pending => return Poll::Pending, + } + } + let n = this.pending.len().min(buf.remaining()); + buf.put_slice(&this.pending[..n]); + this.pending.drain(..n); + Poll::Ready(Ok(())) + } +} + +struct PtySession { + command: String, + /// 持有 master 以保持 PTY 打开；从不直接读取，kill 时随会话 drop 而关闭。 + _master: Box, + writer: Arc>>>, + /// child 由会话持有（watcher 用 try_wait 轮询，不独占 child）。 + /// kill 走完整 `Child::kill()` 语义（Unix：SIGHUP → 宽限 → SIGKILL； + /// Windows：TerminateProcess），优于 `clone_killer()` 的简化实现 + /// （Unix 仅发 SIGHUP，对捕获/忽略 SIGHUP 的进程无效）。 + child: Arc>>>, + output_buf: Arc>, + /// read 增量游标（字节偏移，指向 output_buf 的字符边界）。 + read_offset: Arc>, + /// 进程退出状态描述；None 表示仍在运行。 + exit_status: Arc>>, + notify: Arc, + stream_task: tokio::task::JoinHandle<()>, + drain_task: tokio::task::JoinHandle<()>, + watcher_task: tokio::task::JoinHandle<()>, + created_at: Instant, + last_activity: Arc>, +} + +impl PtySession { + fn touch(&self) { + *self.last_activity.lock().unwrap() = Instant::now(); + } + + fn is_expired(&self) -> bool { + self.last_activity.lock().unwrap().elapsed() > Duration::from_secs(SESSION_TTL_SECS) + } + + fn status_text(&self) -> String { + self.exit_status + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| "running".to_string()) + } +} + +/// Independent service for managing persistent PTY sessions. +/// +/// Thread-safe — designed to be shared via `Arc`. +pub struct PtySessionManager { + sessions: Mutex>, +} + +impl PtySessionManager { + /// Create a new, empty session manager. + pub fn new() -> Self { + Self { + sessions: Mutex::new(HashMap::new()), + } + } + + /// Spawn a command in a new PTY and register it as a session. + /// + /// Returns the new `session_id` for subsequent write/read/kill calls. + pub async fn spawn( + &self, + command: &str, + cwd: Option<&str>, + env: &[(String, String)], + cols: u16, + rows: u16, + use_shell: bool, + ) -> Result { + self.cleanup_expired().await; + + let pty_system = portable_pty::native_pty_system(); + let pair = pty_system + .openpty(PtySize { + rows, + cols, + pixel_width: 0, + pixel_height: 0, + }) + .map_err(|e| { + let mut msg = format!("Failed to open PTY: {}", e); + if cfg!(windows) { + // ConPTY（CreatePseudoConsole）要求进程处于可创建控制台的会话； + // 无控制台的受限环境会返回 E_HANDLE，给出可操作的降级提示。 + msg.push_str( + " (Windows ConPTY requires a console-capable session; \ + run picobot from a real terminal, or use the bash/shell tool instead)", + ); + } + msg + })?; + + let shell = ShellKind::detect(); + let mut cb = if use_shell { + let mut cb = CommandBuilder::new(shell.executable()); + for arg in shell.command_args(command) { + cb.arg(arg); + } + cb + } else { + let mut parts = command.split_whitespace(); + let program = parts.next().ok_or_else(|| "Empty command".to_string())?; + let mut cb = CommandBuilder::new(program); + for arg in parts { + cb.arg(arg); + } + cb + }; + if let Some(dir) = cwd { + cb.cwd(dir); + } + for (key, value) in env { + cb.env(key, value); + } + + // Windows ConPTY spawn 与 openpty 均已完成；直接 spawn。 + // 先取 reader 再 spawn（ConPTY 安全顺序，避免错过启动输出）。 + let mut reader = pair + .master + .try_clone_reader() + .map_err(|e| format!("Failed to clone PTY reader: {}", e))?; + let writer = pair + .master + .take_writer() + .map_err(|e| format!("Failed to take PTY writer: {}", e))?; + + let child = pair + .slave + .spawn_command(cb) + .map_err(|e| format!("Failed to spawn command in PTY: {}", e))?; + + let exit_status: Arc>> = Arc::new(StdMutex::new(None)); + let notify = Arc::new(Notify::new()); + + // watcher：try_wait 非阻塞轮询（100ms）回填退出状态。 + // 不采用阻塞式 child.wait() 独占 child，以便 kill 路径能使用 + // 完整的 Child::kill() 语义（见 PtySession.child 字段注释）。 + let child: Arc>>> = + Arc::new(StdMutex::new(Some(child))); + let child_w = child.clone(); + let es_clone = exit_status.clone(); + let n_clone = notify.clone(); + let watcher_task = tokio::spawn(async move { + loop { + // try_lock 而非阻塞 lock：kill 宽限期内（Unix ~250ms） + // child 锁被 kill 路径持有，阻塞式 lock 会卡住 worker 线程。 + // 注意：含 Guard 的值不得跨 await（非 Send），先在无 await 的 + // 两段式 match 中解出结果。 + let poll_result = match child_w.try_lock() { + Ok(mut guard) => match guard.as_mut() { + Some(c) => Some(c.try_wait()), + None => break, + }, + Err(_) => None, + }; + let poll_result = match poll_result { + Some(r) => r, + None => { + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + }; + match poll_result { + Ok(Some(status)) => { + *es_clone.lock().unwrap() = Some(format!("exited ({})", status)); + n_clone.notify_waiters(); + break; + } + Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await, + Err(e) => { + *es_clone.lock().unwrap() = Some(format!("wait error: {}", e)); + n_clone.notify_waiters(); + break; + } + } + } + }); + + // reader 线程：阻塞读 PTY 输出，字节块经通道转给异步 read_stream。 + let (raw_tx, raw_rx) = mpsc::unbounded_channel::>(); + std::thread::spawn(move || { + let mut chunk = [0u8; 4096]; + loop { + match reader.read(&mut chunk) { + Ok(0) => break, + Ok(n) => { + if raw_tx.send(chunk[..n].to_vec()).is_err() { + break; + } + } + Err(_) => break, + } + } + }); + + // stream/drain 任务：复用 bash 的按行解码逻辑，汇入带缓冲上限的输出缓冲。 + let (line_tx, mut line_rx) = mpsc::unbounded_channel::<(bool, String)>(); + let stream_task = tokio::spawn(read_stream( + ChannelReader { + rx: raw_rx, + pending: Vec::new(), + }, + false, + line_tx, + )); + let output_buf = Arc::new(Mutex::new(String::new())); + let buf_clone = output_buf.clone(); + let n_drain = notify.clone(); + let drain_task = tokio::spawn(async move { + while let Some((_, chunk)) = line_rx.recv().await { + let mut buf = buf_clone.lock().await; + buf.push_str(&chunk); + cap_output_buffer(&mut buf); + drop(buf); + n_drain.notify_waiters(); + } + }); + + let session_id = Uuid::new_v4().to_string(); + let session = PtySession { + command: command.to_string(), + _master: pair.master, + writer: Arc::new(StdMutex::new(Some(writer))), + child, + output_buf, + read_offset: Arc::new(StdMutex::new(0)), + exit_status, + notify, + stream_task, + drain_task, + watcher_task, + created_at: Instant::now(), + last_activity: Arc::new(StdMutex::new(Instant::now())), + }; + + self.sessions + .lock() + .await + .insert(session_id.clone(), session); + Ok(session_id) + } + + /// Write input to a session's PTY. + /// + /// Unless `raw` is set, a platform-appropriate line terminator is + /// appended when missing (ConPTY needs `\r\n` for Enter). With `raw` + /// the input is sent as-is so control characters (e.g. `\u0003` for + /// Ctrl+C) can be delivered without a trailing newline. + pub async fn write_input( + &self, + session_id: &str, + input: &str, + raw: bool, + ) -> Result { + self.cleanup_expired().await; + + // 仅取 Arc 句柄即释放 sessions 锁：PTY 写是阻塞式句柄，若子进程 + // 停止消费输入导致缓冲满，write_all 会长期阻塞——绝不能在持有 + // 管理器锁的情况下阻塞，否则一个卡写的会话会冻结所有会话。 + let (writer, exit_status, last_activity) = { + let sessions = self.sessions.lock().await; + let session = sessions + .get(session_id) + .ok_or_else(|| format!("Session not found: {}", session_id))?; + ( + session.writer.clone(), + session.exit_status.clone(), + session.last_activity.clone(), + ) + }; + + let data = if raw { + input.to_string() + } else { + let mut s = input.replace("\r\n", "\n"); + if !(s.ends_with('\n') || s.ends_with('\r')) { + s.push('\n'); + } + if cfg!(windows) { + s = s.replace('\n', "\r\n"); + } + s + }; + + // PTY 写端是阻塞式句柄；输入量小，直接同步写入（与 ShellSessionManager 一致）。 + let result = { + let mut writer_guard = writer.lock().unwrap(); + match writer_guard.as_mut() { + Some(w) => w + .write_all(data.as_bytes()) + .and_then(|_| w.flush()) + .map_err(|e| format!("Failed to write to PTY: {}", e)), + None => Err("Session input is closed".to_string()), + } + }; + result?; + *last_activity.lock().unwrap() = Instant::now(); + let status = exit_status + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| "running".to_string()); + Ok(format!( + "Input sent to session {}.\n[Process status: {}]", + session_id, status + )) + } + + /// Read new output since the last read (incremental cursor semantics). + /// + /// Waits up to `wait_ms` for new output to arrive. Returns ANSI-stripped + /// output plus the process status. When the output buffer was trimmed by + /// the size cap the cursor may become invalid; in that case the whole + /// retained buffer is returned once. + pub async fn read_output(&self, session_id: &str, wait_ms: u64) -> Result { + self.cleanup_expired().await; + + let wait_ms = if wait_ms == 0 { + DEFAULT_READ_WAIT_MS + } else { + wait_ms.min(MAX_READ_WAIT_MS) + }; + let deadline = Instant::now() + Duration::from_millis(wait_ms); + + // 仅取出跨 await 所需的句柄：会话内含非 Sync 的 MasterPty， + // 不能持有会话引用跨越等待点（会破坏 async_trait 的 Send 约束）。 + let (output_buf, read_offset, exit_status, notify, last_activity) = { + let sessions = self.sessions.lock().await; + let session = sessions + .get(session_id) + .ok_or_else(|| format!("Session not found: {}", session_id))?; + ( + session.output_buf.clone(), + session.read_offset.clone(), + session.exit_status.clone(), + session.notify.clone(), + session.last_activity.clone(), + ) + }; + + loop { + // 先武装 notify 再检查缓冲，避免检查与等待之间丢失唤醒。 + let notified = notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + + let (new_raw, status) = { + let buf = output_buf.lock().await; + let offset = *read_offset.lock().unwrap(); + // 缓冲被上限截断后旧偏移失效：退化为返回保留的全部内容。 + let new_raw = match buf.get(offset..) { + Some(slice) => slice.to_string(), + None => buf.clone(), + }; + *read_offset.lock().unwrap() = buf.len(); + let status = exit_status.lock().unwrap().clone(); + (new_raw, status) + }; + + if !new_raw.is_empty() { + *last_activity.lock().unwrap() = Instant::now(); + return Ok(format_read_output(&strip_ansi(&new_raw), status.as_deref())); + } + if let Some(ref st) = status { + *last_activity.lock().unwrap() = Instant::now(); + return Ok(format_read_output("", Some(st.as_str()))); + } + if Instant::now() >= deadline { + return Ok(format_read_output("", None)); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + let _ = tokio::time::timeout(remaining, notified).await; + } + } + + /// Kill a session's process, tear down the PTY and remove the session. + /// + /// Returns the tail of the final output plus the exit status. + pub async fn kill_session(&self, session_id: &str) -> Result { + let mut sessions = self.sessions.lock().await; + let mut session = sessions + .remove(session_id) + .ok_or_else(|| format!("Session not found: {}", session_id))?; + drop(sessions); + + // 完整 `Child::kill()` 语义（Unix: SIGHUP → 宽限 → SIGKILL； + // Windows: TerminateProcess）。Unix 实现内部含 ~250ms 宽限等待， + // 放入 blocking 池执行以免阻塞 async worker；期间 watcher 以 + // try_lock 规避锁等待。失败不阻断清理（如进程恰已退出的竞态）。 + { + let child_arc = session.child.clone(); + let _ = tokio::task::spawn_blocking(move || { + let mut guard = child_arc.lock().unwrap(); + if let Some(c) = guard.as_mut() { + let _ = c.kill(); + } + }) + .await; + } + + // 等待 watcher 回填退出状态（kill 后通常立即到达），带超时兜底。 + let notified = session.notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if session.exit_status.lock().unwrap().is_none() { + let _ = tokio::time::timeout(Duration::from_secs(3), notified).await; + } + + // 先停 stream（其持有 line_tx，abort 后 drop → line 管道关闭）， + // 等 drain 自然排空管道中残留的行后再取尾部输出，避免 abort + // 丢失 kill 前最后一刻到达的数据；1s 超时兜底防 hang。 + session.stream_task.abort(); + if tokio::time::timeout(Duration::from_secs(1), &mut session.drain_task) + .await + .is_err() + { + session.drain_task.abort(); + } + session.watcher_task.abort(); + + let status = session.status_text(); + let tail = { + let buf = session.output_buf.lock().await; + let total = buf.chars().count(); + let skip = total.saturating_sub(KILL_TAIL_CHARS); + buf.chars().skip(skip).collect::() + }; + // drop(session) 在此发生：master 关闭，reader 线程读到 EOF/错误后退出。 + let cleaned = strip_ansi(&tail); + let mut result = format!( + "Session {} terminated.\n[Process status: {}]", + session_id, status + ); + if !cleaned.trim().is_empty() { + result.push_str(&format!("\nFinal output (tail):\n{}", cleaned.trim_end())); + } + Ok(result) + } + + /// List all sessions with their command, status and activity info. + pub async fn list_sessions(&self) -> String { + self.cleanup_expired().await; + + let sessions = self.sessions.lock().await; + if sessions.is_empty() { + return "No active PTY sessions.".to_string(); + } + + let now = Instant::now(); + let mut entries: Vec<(&String, &PtySession)> = sessions.iter().collect(); + entries.sort_by_key(|(_, s)| s.created_at); + + let mut lines = vec![format!("{} active PTY session(s):", entries.len())]; + for (id, session) in entries { + let buffered = session + .output_buf + .try_lock() + .map(|guard| guard.len()) + .unwrap_or(0); + lines.push(format!( + "- [session_id: {}]\n command: {}\n status: {}\n created: {}s ago, last activity: {}s ago, buffered: {} bytes", + id, + session.command, + session.status_text(), + now.duration_since(session.created_at).as_secs(), + now.duration_since(*session.last_activity.lock().unwrap()).as_secs(), + buffered, + )); + } + lines.join("\n") + } + + /// Remove and kill all sessions idle longer than `SESSION_TTL_SECS`. + /// + /// Called lazily at the entry of each public operation; the manager + /// never starts background tasks on its own. + pub async fn cleanup_expired(&self) { + let expired: Vec = { + let sessions = self.sessions.lock().await; + sessions + .iter() + .filter(|(_, session)| session.is_expired()) + .map(|(id, _)| id.clone()) + .collect() + }; + for id in expired { + let _ = self.kill_session(&id).await; + } + } + + /// Gracefully kill and remove all active sessions (gateway shutdown). + pub async fn shutdown(&self) { + let ids: Vec = { + let sessions = self.sessions.lock().await; + sessions.keys().cloned().collect() + }; + for id in ids { + let _ = self.kill_session(&id).await; + } + } + + /// Number of currently registered sessions. + pub async fn active_count(&self) -> usize { + self.sessions.lock().await.len() + } +} + +impl Default for PtySessionManager { + fn default() -> Self { + Self::new() + } +} + +fn format_read_output(new_output: &str, status: Option<&str>) -> String { + let mut result = if new_output.trim().is_empty() { + "(no new output)".to_string() + } else { + new_output.trim_end().to_string() + }; + result.push_str(&format!( + "\n[Process status: {}]", + status.unwrap_or("running") + )); + result +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_strip_ansi_removes_color_and_cursor_sequences() { + let input = "\x1b[31mred\x1b[0m plain \x1b[2J\x1b[H\x1b]0;title\x07end"; + assert_eq!(strip_ansi(input), "red plain end"); + } + + #[test] + fn test_strip_ansi_normalizes_line_endings() { + assert_eq!(strip_ansi("a\r\nb\r\nc"), "a\nb\nc"); + assert_eq!( + strip_ansi("progress 50%\rprogress 100%\n"), + "progress 100%\n" + ); + } + + #[tokio::test] + async fn test_spawn_unknown_session_read() { + let manager = PtySessionManager::new(); + let err = manager.read_output("no-such-session", 0).await.unwrap_err(); + assert!(err.contains("Session not found")); + } + + #[tokio::test] + async fn test_list_empty() { + let manager = PtySessionManager::new(); + assert_eq!(manager.list_sessions().await, "No active PTY sessions."); + } +} diff --git a/src/tools/shell_session.rs b/src/tools/shell_session.rs deleted file mode 100644 index 50c8d0c..0000000 --- a/src/tools/shell_session.rs +++ /dev/null @@ -1,302 +0,0 @@ -//! Interactive shell session management. -//! -//! Provides `ShellSessionManager`, an independent service that keeps child -//! processes alive between tool calls so the Agent can interact with -//! stdin-waiting prompts (e.g. `Read-Host`, `Confirm (Y/N)`). -//! -//! The manager is created at the gateway/bootstrap layer and injected into -//! `BashTool` via `Arc`. It does NOT start background tasks on its own; -//! cleanup is driven externally via `cleanup_expired()` or `shutdown()`. - -use std::collections::HashMap; -use std::sync::Arc; -use std::time::Duration; - -use tokio::io::AsyncWriteExt; -use tokio::process::{Child, ChildStdin}; -use tokio::sync::{Mutex, mpsc}; -use tokio::time::Instant; - -use uuid::Uuid; - -use crate::tools::bash::cap_output_buffer; - -const SESSION_TIMEOUT_SECS: u64 = 300; // 5 minutes -const OUTPUT_WAIT_MS: u64 = 2000; - -/// A single interactive shell session backed by a live child process. -struct ShellSession { - child: Child, - stdin_writer: Option, - stdout_buf: Arc>, - stderr_buf: Arc>, - /// Background task handle that drains the output channel into buffers. - _drain_task: tokio::task::JoinHandle<()>, - created_at: Instant, -} - -/// Independent service for managing interactive shell sessions. -/// -/// Thread-safe — designed to be shared via `Arc`. -pub struct ShellSessionManager { - sessions: Mutex>, -} - -impl ShellSessionManager { - /// Create a new, empty session manager. - pub fn new() -> Self { - Self { - sessions: Mutex::new(HashMap::new()), - } - } - - /// Save a child process as an interactive session. - /// - /// The caller provides: - /// - `child`: the spawned child process (with piped stdin/stdout/stderr) - /// - `stdin_writer`: the write-half of the piped stdin - /// - `rx`: the output channel receiver (produced by `read_stream` tasks) - /// - `initial_stdout` / `initial_stderr`: output already collected before - /// the session was created - /// - /// Returns a unique `session_id` that can be used for subsequent - /// `send_input` / `get_output` / `close_session` calls. - pub async fn save_session( - &self, - mut child: Child, - stdin_writer: ChildStdin, - mut rx: mpsc::UnboundedReceiver<(bool, String)>, - initial_stdout: String, - initial_stderr: String, - ) -> String { - let session_id = Uuid::new_v4().to_string(); - let stdout_buf = Arc::new(Mutex::new(initial_stdout)); - let stderr_buf = Arc::new(Mutex::new(initial_stderr)); - - // Spawn a background task that drains the channel into buffers. - // Buffers are capped in size (head+tail retained) so a long-lived - // chatty process cannot grow memory without bound during the session TTL. - let stdout_clone = stdout_buf.clone(); - let stderr_clone = stderr_buf.clone(); - let drain_task = tokio::spawn(async move { - while let Some((is_stderr, chunk)) = rx.recv().await { - if is_stderr { - let mut buf = stderr_clone.lock().await; - buf.push_str(&chunk); - cap_output_buffer(&mut buf); - } else { - let mut buf = stdout_clone.lock().await; - buf.push_str(&chunk); - cap_output_buffer(&mut buf); - } - } - }); - - // Kill the child's inherited stdin to prevent blocking on close. - // The actual stdin writing is done via stdin_writer. - let _ = child.stdin.take(); - - let session = ShellSession { - child, - stdin_writer: Some(stdin_writer), - stdout_buf, - stderr_buf, - _drain_task: drain_task, - created_at: Instant::now(), - }; - - self.sessions - .lock() - .await - .insert(session_id.clone(), session); - session_id - } - - /// Send input to a session's stdin and return new output. - /// - /// After writing, waits up to `OUTPUT_WAIT_MS` for new output to arrive. - /// If the child process exits during the wait, returns the final output. - pub async fn send_input(&self, session_id: &str, input: &str) -> Result { - let mut sessions = self.sessions.lock().await; - - let session = sessions - .get_mut(session_id) - .ok_or_else(|| format!("Session not found: {}", session_id))?; - - // Write input to stdin - if let Some(writer) = &mut session.stdin_writer { - let data = if input.ends_with('\n') { - input.to_string() - } else { - format!("{}\n", input) - }; - writer - .write_all(data.as_bytes()) - .await - .map_err(|e| format!("Failed to write stdin: {}", e))?; - writer - .flush() - .await - .map_err(|e| format!("Failed to flush stdin: {}", e))?; - } else { - return Err("Session stdin is closed".to_string()); - } - - // Record output length before wait (byte offsets — buffers only grow via - // push_str, so a previous end is always a valid char boundary, unless a - // buffer-cap trim happened in between, which is handled below). - let prev_stdout_len = session.stdout_buf.lock().await.len(); - let prev_stderr_len = session.stderr_buf.lock().await.len(); - - // Wait for new output or process exit - let deadline = Instant::now() + Duration::from_millis(OUTPUT_WAIT_MS); - tokio::select! { - status = session.child.wait() => { - // Process exited — collect final output - let stdout = session.stdout_buf.lock().await.clone(); - let stderr = session.stderr_buf.lock().await.clone(); - let code = status.ok().and_then(|s| s.code()); - drop(sessions); - return Ok(Self::format_output(&stdout, &stderr, code)); - } - _ = tokio::time::sleep_until(deadline) => { - // Timeout — fall through to return current output - } - } - - // 按字节偏移在锁内直接切片取新增输出：避免对整个缓冲做全量克隆， - // 也修复了旧实现"字节长度当字符数 skip"导致多字节输出丢失的问题。 - let new_stdout = { - let buf = session.stdout_buf.lock().await; - match buf.get(prev_stdout_len..) { - Some(new_part) => new_part.to_string(), - // 偏移失效（缓冲被上限截断过）：退化为返回截断后的全部内容 - None => buf.clone(), - } - }; - let new_stderr = { - let buf = session.stderr_buf.lock().await; - match buf.get(prev_stderr_len..) { - Some(new_part) => new_part.to_string(), - None => buf.clone(), - } - }; - - let mut result = String::new(); - if !new_stdout.is_empty() { - result.push_str(&new_stdout); - } - if !new_stderr.trim().is_empty() { - if !result.is_empty() { - result.push('\n'); - } - result.push_str("STDERR:\n"); - result.push_str(&new_stderr); - } - if result.is_empty() { - result.push_str("(No new output after input. Session still active.)"); - } - - Ok(result) - } - - /// Get the full accumulated output of a session. - pub async fn get_output(&self, session_id: &str) -> Result { - let sessions = self.sessions.lock().await; - let session = sessions - .get(session_id) - .ok_or_else(|| format!("Session not found: {}", session_id))?; - - let stdout = session.stdout_buf.lock().await.clone(); - let stderr = session.stderr_buf.lock().await.clone(); - Ok(Self::format_output(&stdout, &stderr, None)) - } - - /// Close a session: kill the child process and return final output. - pub async fn close_session(&self, session_id: &str) -> Result { - let mut sessions = self.sessions.lock().await; - let mut session = sessions - .remove(session_id) - .ok_or_else(|| format!("Session not found: {}", session_id))?; - drop(sessions); // Release lock before awaiting - - // Close stdin to signal EOF - if let Some(mut writer) = session.stdin_writer.take() { - let _ = writer.shutdown().await; - } - - // Kill and wait - let _ = session.child.start_kill(); - let status = session.child.wait().await.ok(); - let code = status.and_then(|s| s.code()); - - let stdout = session.stdout_buf.lock().await.clone(); - let stderr = session.stderr_buf.lock().await.clone(); - - // Abort drain task - session._drain_task.abort(); - - Ok(Self::format_output(&stdout, &stderr, code)) - } - - /// Remove and kill all expired sessions (older than `SESSION_TIMEOUT_SECS`). - /// - /// Designed to be called periodically from the owner (e.g. gateway). - pub async fn cleanup_expired(&self) { - let timeout = Duration::from_secs(SESSION_TIMEOUT_SECS); - let mut expired_ids = Vec::new(); - - { - let sessions = self.sessions.lock().await; - for (id, session) in sessions.iter() { - if session.created_at.elapsed() > timeout { - expired_ids.push(id.clone()); - } - } - } - - for id in expired_ids { - let _ = self.close_session(&id).await; - } - } - - /// Gracefully shut down all active sessions. - pub async fn shutdown(&self) { - let ids: Vec = { - let sessions = self.sessions.lock().await; - sessions.keys().cloned().collect() - }; - for id in ids { - let _ = self.close_session(&id).await; - } - } - - /// Number of currently active sessions. - pub async fn active_count(&self) -> usize { - self.sessions.lock().await.len() - } - - fn format_output(stdout: &str, stderr: &str, exit_code: Option) -> String { - let mut output = String::new(); - if !stdout.is_empty() { - output.push_str(stdout); - } - if !stderr.trim().is_empty() { - if !output.is_empty() { - output.push('\n'); - } - output.push_str("STDERR:\n"); - output.push_str(stderr); - } - if let Some(code) = exit_code { - output.push_str(&format!("\nExit code: {}", code)); - } - output - } -} - -impl Default for ShellSessionManager { - fn default() -> Self { - Self::new() - } -} diff --git a/web/src/components/Settings/tabs/SubagentsTab.tsx b/web/src/components/Settings/tabs/SubagentsTab.tsx index 3b16212..ed26368 100644 --- a/web/src/components/Settings/tabs/SubagentsTab.tsx +++ b/web/src/components/Settings/tabs/SubagentsTab.tsx @@ -26,6 +26,7 @@ const KNOWN_TOOL_LABELS: Record = { edit: 'Edit', write: 'Write', bash: 'Bash', + pty: 'PTY', http_request: 'HTTP Request', web_fetch: 'Web Fetch', memory_search: 'Memory Search', diff --git a/web/src/components/Settings/tabs/ToolsTab.tsx b/web/src/components/Settings/tabs/ToolsTab.tsx index 70fd200..ee79fe9 100644 --- a/web/src/components/Settings/tabs/ToolsTab.tsx +++ b/web/src/components/Settings/tabs/ToolsTab.tsx @@ -9,6 +9,7 @@ const TASK_KNOWN_TOOLS: KnownSource[] = [ { key: 'edit', label: 'Edit', description: '编辑文件' }, { key: 'write', label: 'Write', description: '写入文件' }, { key: 'bash', label: 'Bash', description: '执行 Shell 命令' }, + { key: 'pty', label: 'PTY', description: '持久伪终端会话' }, { key: 'http_request', label: 'HTTP Request', description: '发送 HTTP 请求' }, { key: 'web_fetch', label: 'Web Fetch', description: '抓取网页内容' }, { key: 'memory_search', label: 'Memory Search', description: '搜索记忆' },