fix(channels): 飞书频道断线重连改为指数退避无限重试，网络恢复后自愈

feat(tools): 新增持久 PTY 会话工具替代管道式交互式 shell
- 新增 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 条目
2026-08-21 11:49:17 +08:00 · 2026-08-21 08:30:44 +08:00
26 changed files with 1480 additions and 785 deletions
--- 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",
--- 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 = [
--- a/src/channels/feishu.rs
+++ b/src/channels/feishu.rs
@ -2308,8 +2308,9 @@ impl Channel for FeishuChannel {
        let channel = self.clone();
        let bus = bus.clone();
        tokio::spawn(async move {
-            let mut consecutive_failures = 0;
+            let mut consecutive_failures: u32 = 0;
-            let max_failures = 3;
+            let base_retry_secs: u64 = 5;
            let max_retry_secs: u64 = 60;
            loop {
                if !*channel.running.read().await {
@ -2319,15 +2320,12 @@ impl Channel for FeishuChannel {
                let shutdown_rx = shutdown_tx.subscribe();
                match channel.run_ws_loop(bus.clone(), shutdown_rx).await {
                    Ok(_) => {
                        consecutive_failures = 0;
                        tracing::info!("Feishu WebSocket disconnected");
                    }
                    Err(e) => {
-                        consecutive_failures += 1;
+                        consecutive_failures = consecutive_failures.saturating_add(1);
                        tracing::error!(attempt = consecutive_failures, error = %e, "Feishu WebSocket error");
                        if consecutive_failures >= max_failures {
                            tracing::error!("Feishu channel: max failures reached, stopping");
                            break;
                        }
                    }
                }
@ -2335,8 +2333,15 @@ impl Channel for FeishuChannel {
                    break;
                }
-                tracing::info!("Feishu channel retrying in 5s...");
+                let exponent = consecutive_failures.saturating_sub(1).min(6);
-                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+                let retry_secs = (base_retry_secs * (1u64 << exponent)).min(max_retry_secs);
                tracing::info!("Feishu channel retrying in {}s...", retry_secs);
                let mut remaining = retry_secs;
                while remaining > 0 && *channel.running.read().await {
                    let step = remaining.min(base_retry_secs);
                    tokio::time::sleep(tokio::time::Duration::from_secs(step)).await;
                    remaining -= step;
                }
            }
            *channel.running.write().await = false;
--- a/src/command/handlers/delete_topic.rs
+++ b/src/command/handlers/delete_topic.rs
@ -81,8 +81,7 @@ async fn handle_delete_topic(
        .map_err(|e| CommandError::new("DELETE_TOPIC_ERROR", e.to_string()))?;
    // 查询更新后的话题列表，返回给前端刷新侧边栏
-    let topic_summaries =
+    let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking(
        crate::command::handlers::list_topics::list_topic_summaries_blocking(
        handler.store.clone(),
        session_id,
    )
--- 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<Vec<TopicSummary>, CommandError> {
    let session_id_bg = session_id.to_string();
-    tokio::task::spawn_blocking(
+    tokio::task::spawn_blocking(move || -> Result<Vec<TopicSummary>, CommandError> {
        move || -> Result<Vec<TopicSummary>, 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<CommandResponse, CommandError> {
-    let summaries =
+    let summaries = list_topic_summaries_blocking(handler.store.clone(), &session_id).await?;
        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()))?;
--- a/src/command/handlers/rename_topic.rs
+++ b/src/command/handlers/rename_topic.rs
@ -83,8 +83,7 @@ async fn handle_rename_topic(
    // 标题未变化时直接返回当前列表，避免无意义写入
    if old_display == trimmed_title {
-        let topic_summaries =
+        let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking(
            crate::command::handlers::list_topics::list_topic_summaries_blocking(
            handler.store.clone(),
            session_id,
        )
@ -110,8 +109,7 @@ async fn handle_rename_topic(
        .map_err(|e| CommandError::new("RENAME_TOPIC_ERROR", e.to_string()))?;
    // 查询更新后的话题列表，返回给前端刷新侧边栏
-    let topic_summaries =
+    let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking(
        crate::command::handlers::list_topics::list_topic_summaries_blocking(
        handler.store.clone(),
        session_id,
    )
--- a/src/command/handlers/session.rs
+++ b/src/command/handlers/session.rs
@ -103,8 +103,7 @@ async fn handle_create_session(
    }
    // Query the full topic list so the frontend sidebar can update
-    let topic_summaries =
+    let topic_summaries = crate::command::handlers::list_topics::list_topic_summaries_blocking(
        crate::command::handlers::list_topics::list_topic_summaries_blocking(
        handler.store.clone(),
        session_id,
    )
--- 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 {
+    LoadOlderMessages { topic_id: String, before_seq: i64 },
        topic_id: String,
        before_seq: i64,
    },
    /// 删除指定话题
    DeleteTopic { topic_id: String },
    /// 重命名指定话题
--- 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 {
--- 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<metrics_exporter_prometheus::PrometheusHandle>,
    /// PTY 会话管理器（与 ToolRegistryFactory 内注入 PtyTool 的同一实例），
    /// gateway 关停/重启时统一终止全部 PTY 会话，防止子进程泄漏。
    pub pty_sessions: Arc<PtySessionManager>,
 }
 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(());
--- 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
+                            error_metadata.insert("error_kind".to_string(), "panic".to_string());
                                .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 {
--- 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,
+    DefaultSubAgentRuntime, InMemoryTaskRepository, NoopSessionMessageSender, PtySessionManager,
-    SubAgentRuntime, SubAgentRuntimeConfig, SubagentCatalog, TaskTool, ToolRegistry,
+    SessionMessageSender, SubAgentRuntime, SubAgentRuntimeConfig, SubagentCatalog, TaskTool,
    ToolRegistry,
 };
 use super::agent_factory::AgentFactory;
@ -69,6 +70,7 @@ pub(crate) fn build_session_manager(
        Arc<ModelSelectionStore>,
        Arc<ModelSelectionStore>,
        Option<Arc<dyn SubAgentRuntime>>,
        Arc<PtySessionManager>,
    ),
    AgentError,
 > {
@ -123,6 +125,7 @@ pub(crate) fn build_session_manager_with_sender(
        Arc<ModelSelectionStore>,
        Arc<ModelSelectionStore>,
        Option<Arc<dyn SubAgentRuntime>>,
        Arc<PtySessionManager>,
    ),
    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<RwLock<HashMap<String, Vec<TodoItem>>>> =
        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,
    ))
 }
--- 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<ToolRegistry> {
--- 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 工具使用说明
+/// pty 工具使用说明
-const SHELL_TOOLS_INSTRUCTIONS: &str = r#"# Shell 交互终端
+const PTY_TOOLS_INSTRUCTIONS: &str = r#"# PTY 持久终端（pty 工具）
- 当 shell 工具返回包含 `__PICOBOT_PENDING_USER_ACTION__` 和 `[session_id: xxx]` 的结果时，表示进程正在等待输入
+## 何时使用
- 阅读已输出的内容，理解提示含义（如确认提示 Y/N、输入密码、选择选项等）
+- 程序需要真实终端（TTY）：交互式向导、确认提示（Y/N）、输入密码/验证码、TUI 程序
- 使用 `session_id` 和 `stdin_input` 参数回复交互内容，例如：`{"command": "echo test", "session_id": "xxx", "stdin_input": "Y"}`
+- 长运行服务/REPL 需要跨多轮工具调用保持状态并持续交互
- 常见场景：确认提示输入 Y/N、输入密码/验证码、选择选项、Read-Host 等"#;
+- 普通一次性命令优先使用 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#"# 定时任务
--- 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,
+    MemoryManageTool, MemorySearchTool, PtySessionManager, PtyTool, SchedulerManageTool,
-    ShellSessionManager, SkillActivateTool, SkillManageTool, SubAgentRuntime, TaskTool, TimeTool,
+    SessionMessageSender, SessionSendTool, SkillActivateTool, SkillManageTool, SubAgentRuntime,
-    TodoReadTool, TodoWriteTool, ToolRegistry, WaitForSubagentsTool, WebFetchTool,
+    TaskTool, TimeTool, TodoReadTool, TodoWriteTool, ToolRegistry, WaitForSubagentsTool,
    WebFetchTool,
 };
 pub(crate) struct ToolRegistryFactory {
@ -31,7 +32,7 @@ pub(crate) struct ToolRegistryFactory {
    subagent_runtime: Option<Arc<dyn SubAgentRuntime>>,
    mcp_manager: Option<Arc<McpClientManager>>,
    todo_state: Option<Arc<RwLock<HashMap<String, Vec<TodoItem>>>>>,
-    shell_session_manager: Arc<ShellSessionManager>,
+    pty_session_manager: Arc<PtySessionManager>,
 }
 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<PtySessionManager> {
        self.pty_session_manager.clone()
    }
    pub(crate) fn with_todo_state(
        mut self,
        state: Arc<RwLock<HashMap<String, Vec<TodoItem>>>>,
@ -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(
--- 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(
--- a/src/platform/mod.rs
+++ b/src/platform/mod.rs
@ -136,199 +136,6 @@ pub fn dangerous_command_patterns_for_platform(platform: Platform) -> Vec<String
    patterns
 }
 /// Check whether a child process is blocked waiting for stdin input.
 ///
 /// Uses platform-specific mechanisms to determine if the process is genuinely
 /// waiting for user input (as opposed to computing, sleeping, or doing I/O).
 ///
 /// Returns `None` when the platform does not support this check or the
 /// information cannot be read.
 pub fn is_process_waiting_on_stdin(pid: u32) -> Option<bool> {
    #[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::<libc::proc_taskinfo>() 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<bool> {
    // 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<u8> = 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
--- 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| {
+            let rows = stmt.query_map(topic_params.as_slice(), |row| row.get::<_, String>(0))?;
                row.get::<_, String>(0)
            })?;
            for tid in rows {
                affected_topics.push(tid?);
            }
-            let sql = format!(
+            let sql =
-                "DELETE FROM messages WHERE session_id = ? AND id IN ({placeholders})"
+                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<String> = {
            let conn = self.pool.get()?;
            let mut stmt = conn.prepare("SELECT id FROM topics WHERE session_id = ?1")?;
--- 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<String> = 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
+    assert!(
        !store
            .batch_topic_token_stats(&[topic.id.as_str()])
            .unwrap()
-        .is_empty());
+            .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(
--- 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<regex::Regex>,
    shell: ShellKind,
    session_manager: Arc<ShellSessionManager>,
 }
 impl BashTool {
-    pub fn new(session_manager: Arc<ShellSessionManager>) -> 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<String, String> {
        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,7 +420,6 @@ impl BashTool {
                        None => std::future::pending().await,
                    }
                } => {
                    {
                    let mut buf = if is_stderr {
                        stderr_buf.lock().await
                    } else {
@ -551,36 +428,6 @@ impl BashTool {
                    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));
                    }
                }
                _ = sleep_until(deadline) => {
                    // 严格硬超时：deadline 到达一律 kill 并返回超时错误，不再转 pending 会话。
                    if let Some(rx_ref) = rx.as_mut() {
@ -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<R>(stream: R, is_stderr: bool, tx: mpsc::UnboundedSender<(bool, String)>)
+pub(crate) async fn read_stream<R>(
-where
+    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();
--- 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::{
--- a/src/tools/pty.rs
+++ 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<PtySessionManager>,
    /// 危险命令拦截正则：构造时预编译（与 BashTool 同源规则）。
    deny_patterns: Vec<regex::Regex>,
 }
 impl PtyTool {
    pub fn new(manager: Arc<PtySessionManager>) -> 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<String> {
        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<ToolResult> {
    Ok(ToolResult {
        success: false,
        output: String::new(),
        error: Some(message),
    })
 }
 fn manager_result(result: Result<String, String>) -> anyhow::Result<ToolResult> {
    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<ToolResult> {
        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"));
        }
    }
 }
--- a/src/tools/pty_session.rs
+++ 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<regex::Regex> = 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<bool> = 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<Vec<u8>>,
    pending: Vec<u8>,
 }
 impl AsyncRead for ChannelReader {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        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<dyn MasterPty + Send>,
    writer: Arc<StdMutex<Option<Box<dyn Write + Send>>>>,
    /// child 由会话持有（watcher 用 try_wait 轮询，不独占 child）。
    /// kill 走完整 `Child::kill()` 语义（Unix：SIGHUP → 宽限 → SIGKILL；
    /// Windows：TerminateProcess），优于 `clone_killer()` 的简化实现
    /// （Unix 仅发 SIGHUP，对捕获/忽略 SIGHUP 的进程无效）。
    child: Arc<StdMutex<Option<Box<dyn portable_pty::Child + Send + Sync>>>>,
    output_buf: Arc<Mutex<String>>,
    /// read 增量游标（字节偏移，指向 output_buf 的字符边界）。
    read_offset: Arc<StdMutex<usize>>,
    /// 进程退出状态描述；None 表示仍在运行。
    exit_status: Arc<StdMutex<Option<String>>>,
    notify: Arc<Notify>,
    stream_task: tokio::task::JoinHandle<()>,
    drain_task: tokio::task::JoinHandle<()>,
    watcher_task: tokio::task::JoinHandle<()>,
    created_at: Instant,
    last_activity: Arc<StdMutex<Instant>>,
 }
 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<PtySessionManager>`.
 pub struct PtySessionManager {
    sessions: Mutex<HashMap<String, PtySession>>,
 }
 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<String, String> {
        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<StdMutex<Option<String>>> = 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<StdMutex<Option<Box<dyn portable_pty::Child + Send + Sync>>>> =
            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::<Vec<u8>>();
        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<String, String> {
        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<String, String> {
        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<String, String> {
        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::<String>()
        };
        // 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<String> = {
            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<String> = {
            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.");
    }
 }
--- a/src/tools/shell_session.rs
+++ b/src/tools/shell_session.rs
@ -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<ChildStdin>,
    stdout_buf: Arc<Mutex<String>>,
    stderr_buf: Arc<Mutex<String>>,
    /// 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<ShellSessionManager>`.
 pub struct ShellSessionManager {
    sessions: Mutex<HashMap<String, ShellSession>>,
 }
 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<String, String> {
        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<String, String> {
        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<String, String> {
        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<String> = {
            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<i32>) -> 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()
    }
 }
--- 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<string, string> = {
  edit: 'Edit',
  write: 'Write',
  bash: 'Bash',
  pty: 'PTY',
  http_request: 'HTTP Request',
  web_fetch: 'Web Fetch',
  memory_search: 'Memory Search',
--- 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: '搜索记忆' },