GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 21 Pull Requests 6 Actions Packages Projects 1 Releases Wiki Activity

修复客户端扩展索引并发竞态
Project CI / Repository checks (pull_request) Successful in 4m7s
Details
Project CI / Frontend tests (pull_request) Successful in 4m9s
Details
Project CI / Backend tests (pull_request) Successful in 8m9s
Details
Project CI / Native shell tests (pull_request) Successful in 16m4s
Details

Browse Source 
串行化扩展索引的读改写操作，避免并发更新丢失

使用扩展项 ID 记录 MCP 启动状态并隔离旧连接回调

将扩展项和来源身份纳入 MCP 连接池指纹

补充并发、旧状态回调和身份指纹回归测试

同步 DirectProject 客户端扩展技术方案
This commit is contained in:
孔令弘
2026-09-01 07:35:02 +00:00
parent 8b6e289a18
commit bb0854ec5b
3 changed files with 235 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs
+15 -13
View File
@@ -569,7 +569,7 @@ struct CodexAppServerInner {
_working_dir: tempfile::TempDir,
workspace_path: std::path::PathBuf,
workspace_mode: CodexAppServerWorkspaceMode,
client_mcp_server_names: Vec<String>,
client_mcp_server_ids_by_name: HashMap<String, String>,
client_mcp_startup_statuses: Mutex<HashMap<String, String>>,
client_mcp_startup_notify: Notify,
initial_client_mcp_startup_waited: AtomicBool,
@@ -1551,10 +1551,10 @@ impl CodexAppServerConnection {
} else {
Vec::new()
};
let client_mcp_server_names = client_mcp_servers
let client_mcp_server_ids_by_name = client_mcp_servers
.iter()
.map(|server| server.name.clone())
.collect::<Vec<_>>();
.map(|server| (server.name.clone(), server.extension_id.clone()))
.collect::<HashMap<_, _>>();
if workspace_override.is_some() && workspace_mode.allows_workspace_writes() {
trust_isolated_game_creator_codex_workspace(
&isolated_codex_home,
@@ -1701,7 +1701,7 @@ impl CodexAppServerConnection {
_working_dir: working_dir,
workspace_path,
workspace_mode,
client_mcp_server_names,
client_mcp_server_ids_by_name,
client_mcp_startup_statuses: Mutex::new(HashMap::new()),
client_mcp_startup_notify: Notify::new(),
initial_client_mcp_startup_waited: AtomicBool::new(false),
@@ -1783,7 +1783,7 @@ impl CodexAppServerConnection {
async fn wait_for_initial_client_mcp_startup(&self) {
if self.inner.workspace_mode != CodexAppServerWorkspaceMode::DirectProject
|| self.inner.client_mcp_server_names.is_empty()
|| self.inner.client_mcp_server_ids_by_name.is_empty()
|| self
.inner
.initial_client_mcp_startup_waited
@@ -1797,7 +1797,7 @@ impl CodexAppServerConnection {
loop {
let pending = {
let statuses = self.inner.client_mcp_startup_statuses.lock().await;
self.inner.client_mcp_server_names.iter().any(|name| {
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")
@@ -2645,11 +2645,7 @@ async fn read_game_creator_codex_app_server_stdout(
.pointer("/params/status")
.and_then(serde_json::Value::as_str),
) {
if inner
.client_mcp_server_names
.iter()
.any(|server_name| server_name == name)
{
if inner.client_mcp_server_ids_by_name.contains_key(name) {
inner
.client_mcp_startup_statuses
.lock()
@@ -2657,7 +2653,12 @@ async fn read_game_creator_codex_app_server_stdout(
.insert(name.to_string(), status.to_string());
inner.client_mcp_startup_notify.notify_waiters();
}
let _ = crate::client_extensions::record_client_mcp_startup_status(name, status);
if let Some(extension_id) = inner.client_mcp_server_ids_by_name.get(name) {
let _ = crate::client_extensions::record_client_mcp_startup_status(
extension_id,
status,
);
}
}
continue;
}
@@ -3851,6 +3852,7 @@ mod tests {
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).expect("create workspace");
let client_mcp = crate::client_extensions::ClientMcpRuntimeServer {
extension_id: "extension-1".to_string(),
name: "plugin-search".to_string(),
config: std::collections::BTreeMap::from([
(
apps/ai-game-creator-shell/src-tauri/src/client_extensions.rs
+218 -68
View File
@@ -3,6 +3,7 @@ use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use tauri::Manager;
use tauri_plugin_dialog::DialogExt;
use zip::ZipArchive;
@@ -12,9 +13,11 @@ const CLIENT_EXTENSIONS_INDEX_FILE_NAME: &str = "index.json";
const CLIENT_EXTENSIONS_SCHEMA_VERSION: &str = "direct-project-client-extensions.v1";
const CLIENT_EXTENSIONS_SOURCES_DIR_NAME: &str = "sources";
const RESERVED_MCP_SERVER_NAME: &str = "agc_tools";
static CLIENT_EXTENSIONS_INDEX_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct ClientMcpRuntimeServer {
pub(crate) extension_id: String,
pub(crate) name: String,
pub(crate) config: BTreeMap<String, toml::Value>,
}
@@ -137,6 +140,43 @@ fn write_index(root: &Path, index: &ClientExtensionIndex) -> Result<(), String>
crate::write_game_creator_config_atomically(&index_path(root), &format!("{content}\n"))
}
fn client_extensions_index_lock() -> &'static Mutex<()> {
CLIENT_EXTENSIONS_INDEX_LOCK.get_or_init(|| Mutex::new(()))
}
fn read_client_extension_index_locked<T>(
root: &Path,
reader: impl FnOnce(&Path, &ClientExtensionIndex) -> Result<T, String>,
) -> Result<T, String> {
let _guard = client_extensions_index_lock()
.lock()
.map_err(|_| "客户端扩展索引锁已损坏".to_string())?;
let index = read_index(root)?;
reader(root, &index)
}
fn update_client_extension_index_at<T>(
root: &Path,
updater: impl FnOnce(&Path, &mut ClientExtensionIndex) -> Result<(T, bool), String>,
) -> Result<T, String> {
let _guard = client_extensions_index_lock()
.lock()
.map_err(|_| "客户端扩展索引锁已损坏".to_string())?;
let mut index = read_index(root)?;
let (result, changed) = updater(root, &mut index)?;
if changed {
write_index(root, &index)?;
}
Ok(result)
}
fn update_client_extension_index<T>(
updater: impl FnOnce(&Path, &mut ClientExtensionIndex) -> Result<(T, bool), String>,
) -> Result<T, String> {
let root = extensions_root()?;
update_client_extension_index_at(&root, updater)
}
fn new_id(prefix: &str) -> String {
format!("{}-{}", prefix, uuid::Uuid::new_v4().simple())
}
@@ -527,6 +567,8 @@ fn client_mcp_set_fingerprint(index: &ClientExtensionIndex) -> String {
.filter(|item| item.extension_type == "mcp" && item.enabled)
.map(|item| {
(
item.id.clone(),
item.source_id.clone(),
item.name.clone(),
item.source_relative_path.clone(),
item.fingerprint.clone(),
@@ -537,8 +579,8 @@ fn client_mcp_set_fingerprint(index: &ClientExtensionIndex) -> String {
let mut digest = Sha256::new();
digest.update(b"direct-project-client-mcp.v1\0");
for (name, source_relative_path, fingerprint) in entries {
for value in [name, source_relative_path, fingerprint] {
for (id, source_id, name, source_relative_path, fingerprint) in entries {
for value in [id, source_id, name, source_relative_path, fingerprint] {
digest.update((value.len() as u64).to_le_bytes());
digest.update(value.as_bytes());
}
@@ -548,14 +590,12 @@ fn client_mcp_set_fingerprint(index: &ClientExtensionIndex) -> String {
pub(crate) fn enabled_client_skill_fingerprint() -> Result<String, String> {
let root = extensions_root()?;
let index = read_index(&root)?;
Ok(client_skill_set_fingerprint(&index))
read_client_extension_index_locked(&root, |_, index| Ok(client_skill_set_fingerprint(index)))
}
pub(crate) fn enabled_client_mcp_fingerprint() -> Result<String, String> {
let root = extensions_root()?;
let index = read_index(&root)?;
Ok(client_mcp_set_fingerprint(&index))
read_client_extension_index_locked(&root, |_, index| Ok(client_mcp_set_fingerprint(index)))
}
fn mcp_config_field<'a>(
@@ -945,13 +985,13 @@ pub(crate) fn prepare_enabled_client_skill_root(
isolated_os_home: &Path,
) -> Result<Option<PathBuf>, String> {
let root = extensions_root()?;
let index = read_index(&root)?;
prepare_client_skill_runtime_root(&root, &index, isolated_os_home)
read_client_extension_index_locked(&root, |root, index| {
prepare_client_skill_runtime_root(root, index, isolated_os_home)
})
}
pub(crate) fn prepare_enabled_client_mcp_servers() -> Result<Vec<ClientMcpRuntimeServer>, String> {
let root = extensions_root()?;
let mut index = read_index(&root)?;
update_client_extension_index(|root, index| {
let mut servers = Vec::new();
let mut index_changed = false;
@@ -987,6 +1027,7 @@ pub(crate) fn prepare_enabled_client_mcp_servers() -> Result<Vec<ClientMcpRuntim
index_changed = true;
}
servers.push(ClientMcpRuntimeServer {
extension_id: index.items[item_index].id.clone(),
name: index.items[item_index].name.clone(),
config,
});
@@ -999,36 +1040,43 @@ pub(crate) fn prepare_enabled_client_mcp_servers() -> Result<Vec<ClientMcpRuntim
}
}
}
if index_changed {
write_index(&root, &index)?;
}
Ok(servers)
Ok((servers, index_changed))
})
}
pub(crate) fn record_client_mcp_startup_status(name: &str, status: &str) -> Result<(), String> {
if name == RESERVED_MCP_SERVER_NAME {
return Ok(());
}
let root = extensions_root()?;
let mut index = read_index(&root)?;
fn apply_client_mcp_startup_status(
index: &mut ClientExtensionIndex,
extension_id: &str,
status: &str,
) -> Result<bool, String> {
let Some(item) = index
.items
.iter_mut()
.find(|item| item.extension_type == "mcp" && item.enabled && item.name == name)
.find(|item| item.extension_type == "mcp" && item.enabled && item.id == extension_id)
else {
return Ok(());
return Ok(false);
};
let last_error = match status {
"failed" => Some("MCP Server 启动失败".to_string()),
"cancelled" => Some("MCP Server 启动已取消".to_string()),
"starting" | "ready" => None,
_ => return Ok(()),
_ => return Ok(false),
};
if item.last_error == last_error {
return Ok(());
return Ok(false);
}
item.last_error = last_error;
write_index(&root, &index)
Ok(true)
}
pub(crate) fn record_client_mcp_startup_status(
extension_id: &str,
status: &str,
) -> Result<(), String> {
update_client_extension_index(|_, index| {
let changed = apply_client_mcp_startup_status(index, extension_id, status)?;
Ok(((), changed))
})
}
#[tauri::command]
@@ -1106,12 +1154,13 @@ pub(crate) async fn pick_client_extension_directory(
#[tauri::command]
pub(crate) fn list_client_extensions() -> Result<Vec<ClientExtensionItem>, String> {
let root = extensions_root()?;
let index = read_index(&root)?;
Ok(index
.items
.iter()
.map(|item| stored_item_view(&index, item))
.collect())
read_client_extension_index_locked(&root, |_, index| {
Ok(index
.items
.iter()
.map(|item| stored_item_view(index, item))
.collect())
})
}
#[tauri::command]
@@ -1137,7 +1186,6 @@ pub(crate) fn import_client_extension(
)?;
let root = extensions_root()?;
let mut index = read_index(&root)?;
let source_id = new_id("source");
let source_display_name = source_name(&source);
let source_storage_relative = format!("{}/{}", CLIENT_EXTENSIONS_SOURCES_DIR_NAME, source_id);
@@ -1172,6 +1220,7 @@ pub(crate) fn import_client_extension(
storage_path: source_storage_relative,
fingerprint: source_fingerprint,
};
update_client_extension_index_at(&root, |_, index| {
let mut used_names: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for item in &index.items {
used_names
@@ -1234,12 +1283,12 @@ pub(crate) fn import_client_extension(
));
}
index.sources.push(source_record);
write_index(&root, &index)?;
Ok(ClientExtensionImportResult {
Ok((ClientExtensionImportResult {
imported,
source_name: source_display_name,
renamed,
duplicate,
}, true))
})
})();
let _ = fs::remove_dir_all(&staging);
@@ -1251,21 +1300,17 @@ pub(crate) fn set_client_extension_enabled(
id: String,
enabled: bool,
) -> Result<ClientExtensionItem, String> {
let root = extensions_root()?;
let mut index = read_index(&root)?;
let item = index
.items
.iter_mut()
.find(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
if item.extension_type == "unknown" && enabled {
return Err("未识别扩展不能启用".to_string());
}
item.enabled = enabled;
item.last_error = None;
let view_item = item.clone();
write_index(&root, &index)?;
Ok(stored_item_view(&index, &view_item))
update_client_extension_index(|_, index| {
let item = index.items.iter_mut().find(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
if item.extension_type == "unknown" && enabled {
return Err("未识别扩展不能启用".to_string());
}
item.enabled = enabled;
item.last_error = None;
let view_item = item.clone();
Ok((stored_item_view(index, &view_item), true))
})
}
#[tauri::command]
@@ -1273,19 +1318,15 @@ pub(crate) fn rename_client_extension(
id: String,
name: String,
) -> Result<ClientExtensionItem, String> {
let root = extensions_root()?;
let mut index = read_index(&root)?;
let item_index = index
.items
.iter()
.position(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
let extension_type = index.items[item_index].extension_type.clone();
let requested = name.trim();
if requested.is_empty() {
return Err("扩展名称不能为空".to_string());
}
let normalized = native_name(requested);
update_client_extension_index(|_, index| {
let item_index = index.items.iter().position(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
let extension_type = index.items[item_index].extension_type.clone();
let conflict = (extension_type == "mcp" && normalized == RESERVED_MCP_SERVER_NAME)
|| index.items.iter().enumerate().any(|(index, item)| {
index != item_index && item.extension_type == extension_type && item.name == normalized
@@ -1306,21 +1347,18 @@ pub(crate) fn rename_client_extension(
};
index.items[item_index].name = final_name;
let view_item = index.items[item_index].clone();
write_index(&root, &index)?;
Ok(stored_item_view(&index, &view_item))
Ok((stored_item_view(index, &view_item), true))
})
}
#[tauri::command]
pub(crate) fn remove_client_extension(id: String) -> Result<(), String> {
let root = extensions_root()?;
let mut index = read_index(&root)?;
let item_index = index
.items
.iter()
.position(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
index.items.remove(item_index);
write_index(&root, &index)
update_client_extension_index(|_, index| {
let item_index = index.items.iter().position(|item| item.id == id.trim())
.ok_or_else(|| "未找到客户端扩展".to_string())?;
index.items.remove(item_index);
Ok(((), true))
})
}
#[cfg(test)]
@@ -1593,6 +1631,118 @@ mod tests {
assert_ne!(first, client_mcp_set_fingerprint(&index));
}
#[test]
fn client_mcp_fingerprint_tracks_import_identity() {
let mut index = ClientExtensionIndex::default();
index.items.push(StoredExtensionItem {
id: "mcp-1".to_string(),
source_id: "source-1".to_string(),
extension_type: "mcp".to_string(),
name: "search".to_string(),
original_name: "search".to_string(),
source_relative_path: "config.toml".to_string(),
enabled: true,
fingerprint: "content-a".to_string(),
last_error: None,
mcp_config: Some(serde_json::json!({"command": "search"})),
});
let first = client_mcp_set_fingerprint(&index);
index.items[0].id = "mcp-2".to_string();
assert_ne!(first, client_mcp_set_fingerprint(&index));
index.items[0].id = "mcp-1".to_string();
index.items[0].source_id = "source-2".to_string();
assert_ne!(first, client_mcp_set_fingerprint(&index));
}
#[test]
fn mcp_startup_status_targets_extension_id() {
let mut index = ClientExtensionIndex::default();
for (id, source_id) in [("mcp-old", "source-1"), ("mcp-new", "source-2")] {
index.items.push(StoredExtensionItem {
id: id.to_string(),
source_id: source_id.to_string(),
extension_type: "mcp".to_string(),
name: "search".to_string(),
original_name: "search".to_string(),
source_relative_path: "config.toml".to_string(),
enabled: true,
fingerprint: "content-a".to_string(),
last_error: None,
mcp_config: Some(serde_json::json!({"command": "search"})),
});
}
apply_client_mcp_startup_status(&mut index, "mcp-old", "failed")
.expect("record old extension startup failure");
assert_eq!(
index.items[0].last_error.as_deref(),
Some("MCP Server 启动失败")
);
assert_eq!(index.items[1].last_error, None);
apply_client_mcp_startup_status(&mut index, "mcp-old", "ready")
.expect("clear old extension startup failure");
assert_eq!(index.items[0].last_error, None);
assert_eq!(index.items[1].last_error, None);
}
#[test]
fn client_extension_index_updates_read_latest_snapshot() {
let directory = tempfile::tempdir().expect("temporary extension root");
let root = directory.path().join(CLIENT_EXTENSIONS_DIR_NAME);
fs::create_dir_all(root.join(CLIENT_EXTENSIONS_SOURCES_DIR_NAME))
.expect("create extension sources root");
write_index(&root, &ClientExtensionIndex::default()).expect("write initial index");
let left_root = root.clone();
let left = std::thread::spawn(move || {
update_client_extension_index_at(&left_root, |_, index| {
index.items.push(StoredExtensionItem {
id: "left".to_string(),
source_id: "source-1".to_string(),
extension_type: "skill".to_string(),
name: "left".to_string(),
original_name: "left".to_string(),
source_relative_path: "left/SKILL.md".to_string(),
enabled: true,
fingerprint: "left".to_string(),
last_error: None,
mcp_config: None,
});
Ok(((), true))
})
});
let right_root = root.clone();
let right = std::thread::spawn(move || {
update_client_extension_index_at(&right_root, |_, index| {
index.items.push(StoredExtensionItem {
id: "right".to_string(),
source_id: "source-2".to_string(),
extension_type: "mcp".to_string(),
name: "right".to_string(),
original_name: "right".to_string(),
source_relative_path: "config.toml".to_string(),
enabled: true,
fingerprint: "right".to_string(),
last_error: None,
mcp_config: Some(serde_json::json!({"command": "right"})),
});
Ok(((), true))
})
});
left.join().expect("join left update").expect("left update");
right.join().expect("join right update").expect("right update");
let index = read_index(&root).expect("read serialized index");
let ids = index
.items
.iter()
.map(|item| item.id.as_str())
.collect::<BTreeSet<_>>();
assert!(ids.contains("left"));
assert!(ids.contains("right"));
}
#[test]
fn plugin_source_discovers_skill_and_mcp_as_independent_items() {
let fixture_root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
docs/technical/【技术方案】DirectProject客户端Skill与MCP扩展导入方案-2026-08-31.md
+2
View File
@@ -330,6 +330,8 @@ Skill root 是当前启用 Skill 集合的完整投影；每次准备时先清
运行时只转换 Codex 原生能够直接使用的标准字段。STDIO 支持 `command / args / env / env_vars / cwd`；HTTP 支持 `url / bearer_token_env_var / http_headers / env_http_headers`；两类都支持原生启动超时、工具超时和工具筛选字段。客户端启用状态是运行时权威，每个第三方项固定 `enabled=true`、`required=false`，并沿用 DirectProject 无逐工具弹窗的自动批准方式。第三方 MCP 配置写入本次隔离 `CODEX_HOME/config.toml`；DirectProject 启动命令同时以 `-c mcp_optional_startup_grace_ms=120000` 传入同一值，命令行 override 是 Codex 最终生效来源。DirectProject 在首次 `turn/start` 前消费 app-server 已有的 `mcpServer/startupStatus/updated` 通知，等待每个已启用第三方 MCP 进入 `ready`、`failed` 或 `cancelled`；最长等待 `120000` 毫秒，超时后继续对话。ready 的 MCP 进入首轮，失败或超时的 MCP 被跳过并记录状态。该配置只作用于本次 DirectProject 隔离运行，不修改用户全局 Codex 配置。DirectProject 系统提示会声明客户端扩展列表中已启用的第三方 MCP；用户明确指定 Server 或工具时，模型只在当前可用工具中查找，找不到则如实说明，不伪造结果。
客户端扩展索引的所有读改写操作在进程内使用同一把互斥锁串行化；原子写文件只负责避免半截文件，不能替代这层读改写协调。每个运行时 MCP 同时保留客户端扩展项 ID，app-server 的启动状态通知按“连接建立时的 Server 名称 → 扩展项 ID”映射回写；迟到的旧连接通知找不到已删除或已重导入的新项时直接忽略。已启用 MCP 的连接池 fingerprint 包含扩展项 ID 和来源 ID，避免删除后重导入相同内容时复用旧连接。
用户导入来源中的相对 `cwd` 按该 MCP 配置文件所在目录解析；未声明 `cwd` 的 STDIO Server 默认以该目录启动，从而保留脚本参数的原生相对路径语义。客户端不替用户猜测普通文件如何启动。
`agc_tools` 为 AGC 内部保留名称，第三方扩展不得覆盖。第三方 MCP 的工具调用继续继承 DirectProject 当前既有运行策略；本功能不新增逐工具审核系统。