修复首次MCP就绪等待丢失唤醒

调整 DirectProject MCP readiness gate 的通知注册顺序抽取 MCP 启动终态判断并增加 missed wakeup 回归测试同步客户端扩展技术方案中的等待契约
2026-09-01 08:05:09 +00:00
parent a0c72faa27
commit aed1525c1b
2 changed files with 101 additions and 36 deletions
@@ -527,6 +527,55 @@ fn should_emit_direct_codex_activity(
    true
 }

+fn is_terminal_client_mcp_startup_status(status: Option<&str>) -> bool {
+    matches!(status, Some("ready") | Some("failed") | Some("cancelled"))
+}
+
+async fn wait_for_client_mcp_startup_gate(
+    client_mcp_server_ids_by_name: &HashMap<String, String>,
+    client_mcp_startup_statuses: &Mutex<HashMap<String, String>>,
+    client_mcp_startup_notify: &Notify,
+    grace: std::time::Duration,
+) {
+    let deadline = tokio::time::Instant::now() + grace;
+    loop {
+        // Register before reading the shared state. `notify_waiters` does not retain a
+        // permit for a future created after the notification, so this ordering is part
+        // of the readiness gate's correctness contract.
+        let notified = client_mcp_startup_notify.notified();
+        tokio::pin!(notified);
+
+        let pending = {
+            let statuses = client_mcp_startup_statuses.lock().await;
+            client_mcp_server_ids_by_name.keys().any(|name| {
+                !is_terminal_client_mcp_startup_status(statuses.get(name).map(String::as_str))
+            })
+        };
+        if !pending {
+            return;
+        }
+
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            eprintln!(
+                "agent.direct_codex.client_mcp_startup_wait timed out after {}ms",
+                grace.as_millis()
+            );
+            return;
+        }
+        tokio::select! {
+            _ = &mut notified => {}
+            _ = tokio::time::sleep(remaining) => {
+                eprintln!(
+                    "agent.direct_codex.client_mcp_startup_wait timed out after {}ms",
+                    grace.as_millis()
+                );
+                return;
+            }
+        }
+    }
+}
+
 fn game_creator_codex_app_server_idle_timeout_ms(
    workspace_mode: CodexAppServerWorkspaceMode,
    request_timeout_ms: u64,
@@ -1792,41 +1841,13 @@ impl CodexAppServerConnection {
            return;
        }

-        let deadline = tokio::time::Instant::now()
-            + std::time::Duration::from_millis(DIRECT_PROJECT_MCP_OPTIONAL_STARTUP_GRACE_MS);
-        loop {
-            let pending = {
-                let statuses = self.inner.client_mcp_startup_statuses.lock().await;
-                self.inner.client_mcp_server_ids_by_name.keys().any(|name| {
-                    !matches!(
-                        statuses.get(name).map(String::as_str),
-                        Some("ready") | Some("failed") | Some("cancelled")
-                    )
-                })
-            };
-            if !pending {
-                return;
-            }
-
-            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
-            if remaining.is_zero() {
-                eprintln!(
-                    "agent.direct_codex.client_mcp_startup_wait timed out after {}ms",
-                    DIRECT_PROJECT_MCP_OPTIONAL_STARTUP_GRACE_MS
-                );
-                return;
-            }
-            tokio::select! {
-                _ = self.inner.client_mcp_startup_notify.notified() => {}
-                _ = tokio::time::sleep(remaining) => {
-                    eprintln!(
-                        "agent.direct_codex.client_mcp_startup_wait timed out after {}ms",
-                        DIRECT_PROJECT_MCP_OPTIONAL_STARTUP_GRACE_MS
-                    );
-                    return;
-                }
-            }
-        }
+        wait_for_client_mcp_startup_gate(
+            &self.inner.client_mcp_server_ids_by_name,
+            &self.inner.client_mcp_startup_statuses,
+            &self.inner.client_mcp_startup_notify,
+            std::time::Duration::from_millis(DIRECT_PROJECT_MCP_OPTIONAL_STARTUP_GRACE_MS),
+        )
+        .await;
    }

    async fn notify(&self, method: &str, params: serde_json::Value) -> Result<(), String> {
@@ -3342,6 +3363,50 @@ mod tests {
        );
    }

+    #[test]
+    fn client_mcp_startup_terminal_statuses_are_recognized() {
+        for status in ["ready", "failed", "cancelled"] {
+            assert!(is_terminal_client_mcp_startup_status(Some(status)));
+        }
+        for status in [None, Some("starting"), Some("unknown")] {
+            assert!(!is_terminal_client_mcp_startup_status(status));
+        }
+    }
+
+    #[tokio::test]
+    async fn client_mcp_startup_gate_rechecks_after_status_notification() {
+        let server_ids_by_name = Arc::new(HashMap::from([(
+            "search".to_string(),
+            "extension-search".to_string(),
+        )]));
+        let statuses = Arc::new(Mutex::new(HashMap::new()));
+        let notify = Arc::new(Notify::new());
+        let wait_server_ids_by_name = Arc::clone(&server_ids_by_name);
+        let wait_statuses = Arc::clone(&statuses);
+        let wait_notify = Arc::clone(&notify);
+        let waiter = tokio::spawn(async move {
+            wait_for_client_mcp_startup_gate(
+                &wait_server_ids_by_name,
+                &wait_statuses,
+                &wait_notify,
+                std::time::Duration::from_millis(250),
+            )
+            .await;
+        });
+
+        tokio::task::yield_now().await;
+        statuses
+            .lock()
+            .await
+            .insert("search".to_string(), "ready".to_string());
+        notify.notify_waiters();
+
+        tokio::time::timeout(std::time::Duration::from_millis(100), waiter)
+            .await
+            .expect("startup gate should wake after terminal status")
+            .expect("startup gate task should not panic");
+    }
+
    #[test]
    fn direct_home_mode_is_read_only_and_rejects_non_passive_items() {
        assert!(CodexAppServerWorkspaceMode::DirectHome.uses_direct_conversation());
@@ -349,7 +349,7 @@ Skill root 是当前启用 Skill 集合的完整投影；每次准备时先清
 - 尽量跳过失败项，保留 AGC 内置 Skill 和 `agc_tools`；
 - 不自动重试、不启动后台修复服务、不删除原始内容。

-DirectProject 首轮使用 Codex 已有的 Eager MCP 启动和 AGC 侧有界 readiness gate：所有已启用第三方 MCP 并行尝试启动，AGC 在首次 `turn/start` 前等待现有 app-server 状态通知，最多等待 `120000` 毫秒；不执行工具探测调用，也不新增外部 MCP 服务。第三方 MCP 仍保持 `required=false`，单项进入 `failed`/`cancelled` 或等待超时后，DirectProject 继续对话；后续回合可重新发现等待窗口结束后才 ready 的 MCP。
+DirectProject 首轮使用 Codex 已有的 Eager MCP 启动和 AGC 侧有界 readiness gate：所有已启用第三方 MCP 并行尝试启动，AGC 在首次 `turn/start` 前等待现有 app-server 状态通知，最多等待 `120000` 毫秒；等待实现必须先注册 `Notify` future，再读取共享状态，以免 `notify_waiters` 在状态检查和等待注册之间丢失唤醒；不执行工具探测调用，也不新增外部 MCP 服务。第三方 MCP 仍保持 `required=false`，单项进入 `failed`/`cancelled` 或等待超时后，DirectProject 继续对话；后续回合可重新发现等待窗口结束后才 ready 的 MCP。

 扩展集合的 fingerprint 纳入现有 DirectProject app-server pool key，扩展集合变化后不复用不匹配的旧运行实例。