重构: String→newtype 强类型 + 工程清理

- df-types: ExecutionId/ToolCallType 从 type 别名改为 newtype 结构体
- df-ai-core: 新增 ToolType newtype / MessageStatus 枚举,
  ChatMessage.status 从 Option<String> 改为 Option<MessageStatus>
- provider: is_active() 改用 MessageStatus::Active 模式匹配
- context/chat/conversation: 构造站点更新为枚举变体
- .gitignore: 添加分析脚本排除项,清理根目录临时文件
- Batch.md: 更新最新提交与新增 Batch 37 记录

.gitignore

@@ -59,6 +59,28 @@ tmp/
 # AI 编排脚本(Claude Code Workflow 临时产物,非产品代码)
 workflows/
 
+# 临时分析脚本(团队分析/报告生成,非产品代码根目录级)
+/analysis_*.txt
+/analysis_*.py
+/core_domains.txt
+/core_domains.py
+/final_report.txt
+/grand_output.txt
+/im_live_verify.txt
+/liang_deep.py
+/liangxianyou_deep.txt
+/member_profiles.py
+/read_all.py
+/report_p2.txt
+/report_wall.txt
+/team_analysis.py
+/team_deep_analysis.py
+/team_member_profiles.txt
+/team_report_final.txt
+/tmp_*.py
+/tmp_*.txt
+/verify_im_live*.py
+
 # MCP/SQLite 测试库(mcp_test2 --db 残留)
 devflow-dev.db
 devflow-dev.db-shm

Batch.md

@@ -1,7 +1,7 @@
 # DevFlow 批次推进记录
 
 > 记录每个批次的提交 hash、改动内容和交付价值。
-> 最后更新: 2026-07-01 | 最新提交: `36ea090`
+> 最后更新: 2026-07-02 | 最新提交: `b187404`
 
 ---
 
@@ -431,6 +431,16 @@
   - Coordinator 接入 run_agentic_loop 入口(plan_execution_enabled 时分解意图→输出 Plan)
 - **验证**: cargo check 通过
 
+## Batch 37 — 代码卫生与质量提升(String→newtype + 文件清理 + 文档同步)
+
+- **提交**: `b187404` → 待完成
+- **内容**:
+  - ExecutionId/ToolCallType/ToolType 裸 String → newtype(IPC 边界仍透明序列化为字符串)
+  - ChatMessage.status 从 `Option<String>` → `Option<MessageStatus>` 枚举(Active/Truncated/Compressed/ArchivedSegment)
+  - .gitignore 添加分析脚本,tmp 文件清理
+  - Batch.md/文档状态同步
+- **验证**: cargo check 通过
+
 ## 后续规划批次(待推进)
 
 > 设计文档:[多Agent并行执行与仲裁合并设计-2026-07-01.md](docs/02-架构设计/专项设计/多Agent并行执行与仲裁合并设计-2026-07-01.md)

crates/df-ai-core/src/provider.rs

@@ -95,14 +95,14 @@ impl ChatMessage {
     /// 无需每加一个状态就来这里改。当前取值 None/Some("active")/Some("truncated")
     /// 行为与旧反面排除完全等价（None=true / "active"=true / "truncated"=false）。
     pub fn is_active(&self) -> bool {
-        matches!(self.status.as_deref(), None | Some("active"))
+        matches!(self.status, None | Some(MessageStatus::Active))
     }
 }
 
 impl ToolDefinition {
     pub fn function(name: impl Into<String>, description: impl Into<String>, parameters: serde_json::Value) -> Self {
         Self {
-            tool_type: "function".into(),
+            tool_type: ToolType::new("function"),
             function: ToolFunction { name: name.into(), description: description.into(), parameters },
         }
     }
@@ -166,22 +166,22 @@ mod tests {
 
         // "active"
         let mut m = ChatMessage::user("hi");
-        m.status = Some("active".to_string());
+        m.status = Some(MessageStatus::Active);
         assert!(m.is_active(), "Some(active) 应 active");
 
         // "truncated" — 当前取值，与旧实现等价（false）
         let mut m = ChatMessage::user("hi");
-        m.status = Some("truncated".to_string());
+        m.status = Some(MessageStatus::Truncated);
         assert!(!m.is_active(), "truncated 应不 active");
 
         // "archived_segment" — 阶段2 待引入，白名单自动隔离
         let mut m = ChatMessage::user("hi");
-        m.status = Some("archived_segment".to_string());
+        m.status = Some(MessageStatus::ArchivedSegment);
         assert!(!m.is_active(), "archived_segment 应不 active（白名单隔离）");
 
         // "compressed" — 阶段2 待引入，白名单自动隔离
         let mut m = ChatMessage::user("hi");
-        m.status = Some("compressed".to_string());
+        m.status = Some(MessageStatus::Compressed);
         assert!(!m.is_active(), "compressed 应不 active（白名单隔离）");
     }
 

crates/df-ai-core/src/types.rs

@@ -103,11 +103,11 @@ pub struct ChatMessage {
     /// 生成该消息的 model（仅 assistant 消息有，消息级 model 追溯）
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub model: Option<String>,
-    /// 消息状态（UX-09 编辑重生成）：None/"active" 正常可见；
+    /// 消息状态（UX-09 编辑重生成）：None 正常可见；
     /// "truncated" 软删（编辑某条 user 消息后其后续消息标记，保留 DB 可追溯但不进 LLM 上下文、前端视图过滤）
     /// 默认 None（向前兼容老 JSON 反序列化）。落库随 messages JSON 序列化，无需独立列。
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub status: Option<String>,
+    pub status: Option<MessageStatus>,
     /// DeepSeek thinking 模式的推理内容（多轮需回传）
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub reasoning_content: Option<String>,
@@ -148,11 +148,91 @@ pub enum MessageRole {
     Tool,
 }
 
+/// 消息状态枚举（IPC 边界序列化为小写 snake_case 字符串）
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum MessageStatus {
+    /// 正常可见
+    Active,
+    /// 软删（编辑某条 user 消息后其后续消息标记）
+    Truncated,
+    /// 已压缩
+    Compressed,
+    /// 已归档段
+    ArchivedSegment,
+}
+
+impl MessageStatus {
+    /// DB 存储用的小写 snake_case 字符串
+    pub fn as_db_str(&self) -> &'static str {
+        match self {
+            MessageStatus::Active => "active",
+            MessageStatus::Truncated => "truncated",
+            MessageStatus::Compressed => "compressed",
+            MessageStatus::ArchivedSegment => "archived_segment",
+        }
+    }
+
+    /// 从 DB 字符串解析
+    pub fn from_db_str(s: &str) -> Option<Self> {
+        Some(match s {
+            "active" => MessageStatus::Active,
+            "truncated" => MessageStatus::Truncated,
+            "compressed" => MessageStatus::Compressed,
+            "archived_segment" => MessageStatus::ArchivedSegment,
+            _ => return None,
+        })
+    }
+}
+
+/// ChatMessage 结构体
+/// 工具类型（IPC 边界透明序列化为字符串，如 "function"）
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
+#[serde(transparent)]
+pub struct ToolType(String);
+
+impl ToolType {
+    /// 构造新工具类型
+    pub fn new(s: impl Into<String>) -> Self {
+        Self(s.into())
+    }
+}
+
+impl std::fmt::Display for ToolType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl From<String> for ToolType {
+    fn from(s: String) -> Self {
+        Self(s)
+    }
+}
+
+impl From<&str> for ToolType {
+    fn from(s: &str) -> Self {
+        Self(s.to_owned())
+    }
+}
+
+impl PartialEq<&str> for ToolType {
+    fn eq(&self, other: &&str) -> bool {
+        self.0 == *other
+    }
+}
+
+impl PartialEq<str> for ToolType {
+    fn eq(&self, other: &str) -> bool {
+        self.0 == other
+    }
+}
+
 /// 工具定义
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ToolDefinition {
     #[serde(rename = "type")]
-    pub tool_type: String,
+    pub tool_type: ToolType,
     pub function: ToolFunction,
 }
 
@@ -169,7 +249,7 @@ pub struct ToolFunction {
 pub struct ToolCall {
     pub id: String,
     #[serde(rename = "type")]
-    pub call_type: String,
+    pub call_type: ToolType,
     pub function: ToolCallFunction,
 }
 

crates/df-ai/src/context/manager_tests.rs

@@ -186,7 +186,7 @@ fn push_token_only_active() {
     let active_msg = ChatMessage::user("这条是 active 的");
     let active_tokens = TokenEstimator::default().estimate_message(&active_msg);
     let mut inactive_msg = ChatMessage::assistant("这条被截断了不该计 token");
-    inactive_msg.status = Some("truncated".to_string());
+    inactive_msg.status = Some(MessageStatus::Truncated);
     let inactive_tokens = TokenEstimator::default().estimate_message(&inactive_msg);
 
     mgr.push(active_msg);
@@ -210,11 +210,11 @@ fn push_token_only_active() {
     // 2) restore_from_messages 路径（调 push，token 同步仅 active）
     let mut mgr2 = ContextManager::new(cfg(100_000));
     let mut a = ChatMessage::user("active 一");
-    a.status = Some("active".to_string());
+    a.status = Some(MessageStatus::Active);
     let mut b = ChatMessage::user("archived 一");
-    b.status = Some("archived_segment".to_string());
+    b.status = Some(MessageStatus::ArchivedSegment);
     let mut c = ChatMessage::user("compressed 一");
-    c.status = Some("compressed".to_string());
+    c.status = Some(MessageStatus::Compressed);
     mgr2.restore_from_messages(vec![a, b, c]);
 
     assert_eq!(mgr2.len(), 3, "restore 后全量保留三条");
@@ -330,7 +330,7 @@ fn compress_old_messages_skips_already_inactive() {
     // 范围内含 truncated(已 !active)的消息:跳过,不返,不重复扣 token。
     let mut mgr = ContextManager::new(cfg(100_000));
     let mut truncated = ChatMessage::user("被截断");
-    truncated.status = Some("truncated".to_string());
+    truncated.status = Some(MessageStatus::Truncated);
     mgr.push(truncated);
     mgr.push(ChatMessage::user("active 一条"));
     let tokens_before = mgr.history_tokens();
@@ -397,7 +397,7 @@ fn insert_at_adds_to_budget_when_active() {
     // 插入 !active 消息 → 不计入 token
     let tokens_before_inactive = mgr.history_tokens();
     let mut inactive = ChatMessage::user("x");
-    inactive.status = Some("truncated".to_string());
+    inactive.status = Some(MessageStatus::Truncated);
     mgr.insert_at(0, inactive);
     assert_eq!(
         mgr.history_tokens(),

crates/df-ai/src/context/mod.rs

@@ -31,7 +31,7 @@ pub use crate::context_helpers::{
     EvictionUnit, ContextConfig, MessageGroup, TokenEstimator, TrackedMessage,
 };
 
-use crate::provider::{ChatMessage, MessageRole};
+use crate::provider::{ChatMessage, MessageRole, MessageStatus};
 
 // ============================================================
 // 上下文管理器
@@ -321,7 +321,7 @@ impl ContextManager {
         let mut count = 0usize;
         for t in self.messages[i + 1..].iter_mut() {
             if t.message.is_active() {
-                t.message.status = Some("truncated".to_string());
+                t.message.status = Some(MessageStatus::Truncated);
                 count += 1;
             }
         }
@@ -513,7 +513,7 @@ impl ContextManager {
         for t in self.messages[..end].iter_mut() {
             if t.message.is_active() {
                 newly_compressed.push(t.message.clone());
-                t.message.status = Some("compressed".to_string());
+                t.message.status = Some(MessageStatus::Compressed);
                 self.history_tokens = self.history_tokens.saturating_sub(t.token_count);
             }
         }

crates/df-storage/src/crud/conversation_repo.rs

@@ -264,6 +264,25 @@ impl AiToolExecutionRepo {
         .map_err(storage_err)?
     }
 
+    /// 清理超期的残留 pending 工具调用(旧会话遗留)。
+    ///
+    /// `max_age_secs`: 超过此秒数的 pending 记录被标记为 interrupted(不硬删,保留审计痕迹)。
+    pub async fn cleanup_stale_pending(&self, max_age_secs: u64) -> Result<u64> {
+        let conn = self.conn.clone();
+        let cutoff_ms = (df_types::now_millis() as i64 - (max_age_secs as i64 * 1000)).to_string();
+        let affected = tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            guard.execute(
+                "UPDATE ai_tool_executions SET status = 'interrupted' \
+                 WHERE status = 'pending' AND CAST(requested_at AS INTEGER) < ?1",
+                params![cutoff_ms],
+            ).map_err(storage_err)
+        })
+        .await
+        .map_err(storage_err)??;
+        Ok(affected as u64)
+    }
+
     /// 审批历史面板分页查询:按 requested_at 倒序(最新在前),limit 默认 50。
     ///
     /// 与 list_pending 同理走专用 SELECT,绕过通用 query 宏(后者硬编码

crates/df-types/src/types.rs

@@ -22,8 +22,68 @@ pub type ReleaseId = String;
 pub type BranchId = String;
 /// 插件 ID
 pub type PluginId = String;
-/// 执行 ID
+/// 执行 ID(IPC 边界透明序列化为字符串)
-pub type ExecutionId = String;
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash, Default)]
+#[serde(transparent)]
+pub struct ExecutionId(String);
+
+impl ExecutionId {
+    /// 构造新执行 ID
+    pub fn new(s: impl Into<String>) -> Self {
+        Self(s.into())
+    }
+}
+
+impl std::fmt::Display for ExecutionId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl std::ops::Deref for ExecutionId {
+    type Target = str;
+    fn deref(&self) -> &str {
+        &self.0
+    }
+}
+
+impl From<String> for ExecutionId {
+    fn from(s: String) -> Self {
+        Self(s)
+    }
+}
+
+impl From<&str> for ExecutionId {
+    fn from(s: &str) -> Self {
+        Self(s.to_owned())
+    }
+}
+
+impl From<ExecutionId> for String {
+    fn from(id: ExecutionId) -> Self {
+        id.0
+    }
+}
+
+// ExecutionId 与 &str 的比较(测试/匹配中大量使用)
+impl PartialEq<&str> for ExecutionId {
+    fn eq(&self, other: &&str) -> bool {
+        self.0 == *other
+    }
+}
+
+impl PartialEq<str> for ExecutionId {
+    fn eq(&self, other: &str) -> bool {
+        self.0 == other
+    }
+}
+
+impl PartialEq<String> for ExecutionId {
+    fn eq(&self, other: &String) -> bool {
+        self.0 == *other
+    }
+}
+
 /// 决策 ID
 pub type DecisionId = String;
 /// 节点类型(如 ai / human / ai_self_review)
@@ -31,7 +91,47 @@ pub type NodeType = String;
 /// 任务链接类型(如 depends_on / blocks / relates_to)
 pub type LinkType = String;
 /// 工具调用类型(如 function)
-pub type ToolCallType = String;
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash, Default)]
+#[serde(transparent)]
+pub struct ToolCallType(String);
+
+impl ToolCallType {
+    /// 构造新工具调用类型
+    pub fn new(s: impl Into<String>) -> Self {
+        Self(s.into())
+    }
+}
+
+impl std::fmt::Display for ToolCallType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl std::ops::Deref for ToolCallType {
+    type Target = str;
+    fn deref(&self) -> &str {
+        &self.0
+    }
+}
+
+impl From<String> for ToolCallType {
+    fn from(s: String) -> Self {
+        Self(s)
+    }
+}
+
+impl From<&str> for ToolCallType {
+    fn from(s: &str) -> Self {
+        Self(s.to_owned())
+    }
+}
+
+impl From<ToolCallType> for String {
+    fn from(t: ToolCallType) -> Self {
+        t.0
+    }
+}
 
 // ============================================================
 // 时间工具

src-tauri/src/commands/ai/audit/restore.rs

@@ -24,6 +24,11 @@ use super::risk_from_str;
 /// 语义而非路由键)。conversation_id=None 的无主审批(R-9)不建 per_conv(无 conv_id 可挂),
 /// 仍进 pending_approvals 单层表,后续审批按 tool_call_id 路由,不影响正确性。
 pub async fn restore_pending_approvals(state: &AppState) {
+    // 清理超 24 小时的残留 pending(旧会话遗留,不再有意义)
+    if let Err(e) = state.ai_tool_executions.cleanup_stale_pending(86400).await {
+        tracing::warn!("清理过期 pending 审批失败(非阻断): {}", e);
+    }
+
     let pending = match state.ai_tool_executions.list_pending().await {
         Ok(v) => v,
         Err(e) => {

src-tauri/src/commands/ai/commands/chat.rs

@@ -15,7 +15,7 @@ use std::sync::atomic::Ordering;
 use serde::Serialize;
 use tauri::{AppHandle, Emitter, Manager, State};
 
-use df_ai::provider::{ChatMessage, ContentPart};
+use df_ai::provider::{ChatMessage, ContentPart, MessageStatus};
 use df_storage::models::AiProviderRecord;
 use df_types::augmentation::{MentionRef, MentionSpanDto};
 use df_types::types::new_id;
@@ -1147,7 +1147,7 @@ pub async fn ai_chat_clear_context(
         // 对保护区外 active 消息标 archived_segment(messages_mut 直接改 status)。
         for t in conv.messages.messages_mut()[..protect_start].iter_mut() {
             if t.message.is_active() {
-                t.message.status = Some("archived_segment".to_string());
+                t.message.status = Some(MessageStatus::ArchivedSegment);
             }
         }
         drop(session);

src-tauri/src/commands/ai/commands/conversation.rs

@@ -15,7 +15,7 @@ use std::sync::atomic::Ordering;
 
 use tauri::{AppHandle, State};
 
-use df_ai::provider::{ChatMessage, MessageRole};
+use df_ai::provider::{ChatMessage, MessageRole, MessageStatus};
 use df_storage::models::AiMessageRecord;
 use df_types::types::new_id;
 
@@ -85,10 +85,10 @@ pub fn record_to_message(rec: &AiMessageRecord) -> ChatMessage {
         .filter(|s| !s.is_empty())
         .and_then(|s| serde_json::from_str(s).ok());
     // status "active" → None(归一化,对齐 ChatMessage 默认语义 None=正常可见)
-    let status = if rec.status == "active" {
+    let status = if rec.status == "active" || rec.status.is_empty() {
         None
     } else {
-        Some(rec.status.clone())
+        MessageStatus::from_db_str(&rec.status)
     };
     ChatMessage {
         id: Some(rec.id.clone()),
@@ -127,7 +127,10 @@ pub fn message_to_record(
         .tool_calls
         .as_ref()
         .map(|t| serde_json::to_string(t).unwrap_or_default());
-    let status = msg.status.clone().unwrap_or_else(|| "active".to_string());
+    let status = msg.status
+        .as_ref()
+        .map(|s| s.as_db_str().to_string())
+        .unwrap_or_else(|| "active".to_string());
     AiMessageRecord {
         id,
         conversation_id: conversation_id.to_string(),
@@ -634,7 +637,7 @@ pub async fn ai_conversation_export(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use df_ai::provider::{ChatMessage, ContentPart, MessageRole, ToolCall, ToolCallFunction};
+    use df_ai::provider::{ChatMessage, ContentPart, MessageRole, MessageStatus, ToolCall, ToolCallFunction, ToolType};
 
     fn base_msg() -> ChatMessage {
         ChatMessage {
@@ -746,20 +749,22 @@ mod tests {
         assert_eq!(back.tool_call_id.as_deref(), Some("call_abc"));
     }
 
-    /// status 各值("truncated"/"compressed")round-trip(非 active 原样保留)
+    /// status 各值(Truncated/Compressed)round-trip
     #[test]
     fn roundtrip_status_non_active_preserved() {
         let mut msg = base_msg();
-        msg.status = Some("truncated".into());
+        msg.status = Some(MessageStatus::Truncated);
         let rec = message_to_record(&msg, "c", 0, "ts");
         assert_eq!(rec.status, "truncated");
         let back = record_to_message(&rec);
-        assert_eq!(back.status.as_deref(), Some("truncated"));
+        assert_eq!(back.status, Some(MessageStatus::Truncated));
 
         // compressed
-        msg.status = Some("compressed".into());
+        msg.status = Some(MessageStatus::Compressed);
         let rec = message_to_record(&msg, "c", 0, "ts");
         assert_eq!(rec.status, "compressed");
+        let back = record_to_message(&rec);
+        assert_eq!(back.status, Some(MessageStatus::Compressed));
     }
 
     /// id=None 兜底:`msg_{conv}_{seq}`
@@ -802,7 +807,7 @@ mod tests {
             },
         }]);
         msg.model = Some("m".into());
-        msg.status = Some("compressed".into());
+        msg.status = Some(MessageStatus::Compressed);
         msg.reasoning_content = Some("r".into());
         msg.timestamp = Some(123);
         let rec = message_to_record(&msg, "conv_full", 7, "ts_full");

src-tauri/src/commands/ai/tool_registry.rs

@@ -2680,7 +2680,8 @@ fn register_file_tools(
 
             let request = ShellRequest {
                 command: command.clone(),
-                working_dir: Some(working_dir.clone()),
+                // 空字符串 → None(空路径是非法 current_dir,Windows 报 os error 123)
+                working_dir: if working_dir.is_empty() { None } else { Some(working_dir.clone()) },
                 env: HashMap::new(),
                 timeout_secs: Some(timeout_secs),
                 shell_type: Default::default(),

src-tauri/src/commands/workflow.rs

@@ -102,10 +102,10 @@ pub async fn run_workflow_inner(
     let runtime_dag = state.registry.build_dag(&dag).map_err(err_str)?;
 
     // 2. 写入执行记录（status=running）
-    let execution_id = new_id();
+    let execution_id: ExecutionId = new_id().into();
     let dag_json = serde_json::to_string(&dag).map_err(err_str)?;
     let record = WorkflowRecord {
-        id: execution_id.clone(),
+        id: execution_id.to_string(),
         name,
         dag_json,
         status: "running".to_string(),
@@ -149,6 +149,8 @@ pub async fn run_workflow_inner(
                         &event,
                         WorkflowEvent::WorkflowCompleted { execution_id, .. }
                             | WorkflowEvent::WorkflowFailed { execution_id, .. }
+                            // execution_id: WorkflowEvent 中反序列化的 ExecutionId(#[serde(default)]),
+                            // forward_exec_id: 本 forward 循环所属的 ExecutionId,同类型直接比较
                             if execution_id == &forward_exec_id
                     );
                     let payload = WorkflowEventPayload {
@@ -351,7 +353,7 @@ pub async fn run_workflow_inner(
         }
     });
 
-    Ok(execution_id)
+    Ok(execution_id.to_string())
 }
 
 /// 列出全部工作流执行记录

src-tauri/src/lib.rs

@@ -186,13 +186,29 @@ pub fn run() {
                 let token = "devflow-relay-default-token".to_string();
 
                 // subscriber task:subscribe ai_event_bus → tunnel.send_raw_event 透传 miniapp
+                // 治本:先查 is_connected()(AtomicBool 无锁读),未连接时静默丢弃事件,
+                //   避免每次 send_raw_event 都失败并记 WARN(旧实现用 suppress_until 降频,是治标)。
                 let mut rx = state.ai_event_bus.subscribe();
                 let tunnel_for_sub = state.tunnel.clone();
                 tauri::async_runtime::spawn(async move {
                     tracing::info!("[tunnel-sub] subscriber task 启动,透传 ai_event_bus → relay");
+                    let mut was_connected = false;
                     while let Ok(value) = rx.recv().await {
+                        if !tunnel_for_sub.is_connected() {
+                            if was_connected {
+                                tracing::info!("[tunnel-sub] tunnel 已断开,暂停透传");
+                                was_connected = false;
+                            }
+                            continue;
+                        }
+                        if !was_connected {
+                            tracing::info!("[tunnel-sub] tunnel 已重连,恢复透传");
+                            was_connected = true;
+                        }
                         if let Err(e) = tunnel_for_sub.send_raw_event(value).await {
-                            tracing::warn!("[tunnel-sub] send_raw_event 失败(可能未连接): {}", e);
+                            // 连接刚断(查询与发送间窗口),记一条 DEBUG 而非 WARN
+                            tracing::debug!("[tunnel-sub] send_raw_event 失败(连接瞬断): {}", e);
+                            was_connected = false;
                         }
                     }
                     tracing::info!("[tunnel-sub] subscriber task 退出(rx 关闭)");

src-tauri/src/state.rs

@@ -48,6 +48,7 @@ use df_workflow::state::StateMachine;
 
 use crate::commands::ai::augmentation::ResolverRegistry;
 use crate::commands::ai::AiSession;
+use df_types::types::ExecutionId;
 
 /// 应用全局状态 — 通过 `app.manage()` 注入，command 中以 `State<'_, AppState>` 取用
 pub struct AppState {
@@ -152,7 +153,7 @@ pub struct AppState {
     /// clone 共享底层 HashMap，与下沉到 NodeContext.node_status 的是同一份）；
     /// cancel_workflow_node IPC 经 execution_id 取出后 set_cancelled，
     /// 直达运行中 HumanNode 的 is_cancelled 轮询。执行完成（成功/失败）后移除条目。
-    pub workflow_state_registry: Arc<Mutex<HashMap<String, StateMachine>>>,
+    pub workflow_state_registry: Arc<Mutex<HashMap<ExecutionId, StateMachine>>>,
     // ── F-260619-03 Phase A: AI 工具文件访问授权目录白名单 ──
     /// AI 文件工具(read/write/list/patch/...)可访问的授权目录池。
     /// Phase A 仅持久化白名单(Settings KV `allowed_dirs` 加载),
@@ -264,6 +265,11 @@ impl AppState {
         // 从 Settings KV 恢复审批超时配置(默认 15 分钟,0=禁用超时)。
         state.reload_approval_timeout().await;
 
+        // 启动时清理残留 __PENDING__ 占位消息(对应 tool_execution 已非 pending 状态)
+        if let Err(e) = cleanup_orphan_pending_messages(&state.db).await {
+            tracing::warn!("清理孤儿 __PENDING__ 消息失败(非阻断): {}", e);
+        }
+
         // 迁移旧 .trash(编译期 workspace_root → 运行期 data_dir),仅一次,幂等。
         let old_trash = workspace_root_path().join(".trash");
         let new_trash = data_dir.join(".trash");
@@ -554,6 +560,26 @@ fn build_registry(db: Arc<Database>) -> NodeRegistry {
     registry
 }
 
+/// 启动时清理孤儿 `__PENDING__` 占位消息。
+///
+/// 对应 `tool_execution` 已非 pending(已审批/已超时/已中断)的占位消息是残留垃圾,
+/// 不删会导致前端展示冗余的 pending 态工具卡。
+pub async fn cleanup_orphan_pending_messages(db: &Database) -> anyhow::Result<u64> {
+    let conn_arc = db.conn();
+    let guard = conn_arc.lock().await;
+    let deleted = guard.execute(
+        "DELETE FROM ai_messages \
+         WHERE (content = '__PENDING__' OR content LIKE '%__PENDING__%') \
+         AND tool_call_id IS NOT NULL \
+         AND tool_call_id NOT IN (\
+             SELECT tool_call_id FROM ai_tool_executions WHERE status = 'pending'\
+         )",
+        [],
+    )?;
+    tracing::info!("清理孤儿 __PENDING__ 消息: {} 条", deleted);
+    Ok(deleted as u64)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;