codex_exec_server/
capability_discovery.rs

1use std::collections::HashSet;
2use std::io;
3
4use codex_exec_server_protocol::CapabilityRootDiscoverRequest;
5use codex_exec_server_protocol::CapabilityRootDiscovery;
6use codex_exec_server_protocol::CapabilityRootsDiscoverParams;
7use codex_exec_server_protocol::CapabilityRootsDiscoverResponse;
8use codex_exec_server_protocol::CapabilityTextFile;
9use codex_exec_server_protocol::DISCOVERABLE_PLUGIN_MANIFEST_PATHS;
10use codex_exec_server_protocol::DiscoveredPluginFiles;
11use codex_exec_server_protocol::DiscoveredSkillFiles;
12use codex_file_system::ExecutorFileSystem;
13use codex_file_system::WalkEntryKind;
14use codex_file_system::WalkOptions;
15use codex_utils_path_uri::PathUri;
16use futures::StreamExt;
17use serde::Deserialize;
18use serde_json::Value;
19
20const MAX_ROOTS_PER_REQUEST: usize = 128;
21const MAX_SCAN_DEPTH: usize = 6;
22const MAX_DIRECTORIES_PER_ROOT: usize = 2_000;
23const MAX_ENTRIES_PER_ROOT: usize = 20_000;
24const MAX_FILE_BYTES: usize = 1024 * 1024;
25const MAX_BUNDLE_BYTES_PER_ROOT: usize = 16 * 1024 * 1024;
26const MAX_CONCURRENT_ROOTS: usize = 8;
27const SKILL_FILE_NAME: &str = "SKILL.md";
28const SKILL_METADATA_PATH: &str = "agents/openai.yaml";
29const DEFAULT_MCP_CONFIG_PATH: &str = ".mcp.json";
30
31#[derive(Debug, thiserror::Error)]
32pub enum CapabilityDiscoveryError {
33    #[error("capability root discovery accepts at most {MAX_ROOTS_PER_REQUEST} roots")]
34    TooManyRoots,
35}
36
37/// Discovers and materializes capability manifests using one executor-local filesystem.
38///
39/// Product parsing and policy intentionally remain with the caller. This operation owns the
40/// filesystem-expensive portion: bounded traversal, recognized-file selection, and reads.
41#[tracing::instrument(
42    name = "capability_roots.discover_v1",
43    skip_all,
44    fields(root_count = params.roots.len())
45)]
46pub async fn discover_capability_roots(
47    file_system: &dyn ExecutorFileSystem,
48    params: CapabilityRootsDiscoverParams,
49) -> Result<CapabilityRootsDiscoverResponse, CapabilityDiscoveryError> {
50    if params.roots.len() > MAX_ROOTS_PER_REQUEST {
51        return Err(CapabilityDiscoveryError::TooManyRoots);
52    }
53
54    let roots = futures::stream::iter(params.roots)
55        .map(|root| discover_root(file_system, root))
56        .buffered(MAX_CONCURRENT_ROOTS)
57        .collect()
58        .await;
59    Ok(CapabilityRootsDiscoverResponse { roots })
60}
61
62async fn discover_root(
63    file_system: &dyn ExecutorFileSystem,
64    request: CapabilityRootDiscoverRequest,
65) -> CapabilityRootDiscovery {
66    let CapabilityRootDiscoverRequest { id, path } = request;
67    let mut discovery = CapabilityRootDiscovery {
68        id,
69        path: path.clone(),
70        plugin: None,
71        skills: Vec::new(),
72        namespace_manifests: Vec::new(),
73        warnings: Vec::new(),
74        error: None,
75    };
76
77    match file_system.get_metadata(&path, /*sandbox*/ None).await {
78        Ok(metadata) if metadata.is_directory => {}
79        Ok(_) => {
80            discovery.error = Some(format!("capability root {path} is not a directory"));
81            return discovery;
82        }
83        Err(error) => {
84            discovery.error = Some(format!("failed to inspect capability root {path}: {error}"));
85            return discovery;
86        }
87    }
88
89    let walk = match file_system
90        .walk(
91            &path,
92            WalkOptions {
93                max_depth: MAX_SCAN_DEPTH,
94                max_directories: MAX_DIRECTORIES_PER_ROOT,
95                max_entries: MAX_ENTRIES_PER_ROOT,
96                follow_directory_symlinks: true,
97                prune_hidden_directories: false,
98            },
99            /*sandbox*/ None,
100        )
101        .await
102    {
103        Ok(walk) => walk,
104        Err(error) => {
105            discovery.error = Some(format!("failed to scan capability root {path}: {error}"));
106            return discovery;
107        }
108    };
109    discovery
110        .warnings
111        .extend(walk.errors.into_iter().map(|error| {
112            format!(
113                "failed to scan capability path {}: {}",
114                error.path, error.message
115            )
116        }));
117    if walk.truncated {
118        discovery.warnings.push(format!(
119            "capability scan reached its traversal limit (root: {path})"
120        ));
121    }
122
123    let mut skill_paths = Vec::new();
124    let mut namespace_manifest_paths = Vec::new();
125    for entry in walk.entries {
126        if entry.kind != WalkEntryKind::File {
127            continue;
128        }
129        if entry.path.basename().as_deref() == Some(SKILL_FILE_NAME) {
130            skill_paths.push(entry.path.clone());
131        }
132        if is_plugin_manifest_path(&entry.path) {
133            namespace_manifest_paths.push(entry.path);
134        }
135    }
136    skill_paths.sort_unstable_by_key(PathUri::to_string);
137    namespace_manifest_paths.sort_unstable_by(|left, right| {
138        let left_root = plugin_root_for_manifest(left).map(|path| path.to_string());
139        let right_root = plugin_root_for_manifest(right).map(|path| path.to_string());
140        left_root
141            .cmp(&right_root)
142            .then_with(|| plugin_manifest_priority(left).cmp(&plugin_manifest_priority(right)))
143    });
144
145    let mut budget = BundleBudget::default();
146    let root_manifest =
147        read_first_plugin_manifest(file_system, &path, &mut budget, &mut discovery.warnings).await;
148
149    let inherited_manifest = match root_manifest.as_ref() {
150        Some(manifest) => Some(manifest.clone()),
151        None => {
152            read_nearest_ancestor_manifest(file_system, &path, &mut budget, &mut discovery.warnings)
153                .await
154        }
155    };
156    let mut seen_namespace_roots = HashSet::new();
157    if let Some(manifest) = inherited_manifest {
158        if let Some(plugin_root) = plugin_root_for_manifest(&manifest.path) {
159            seen_namespace_roots.insert(plugin_root);
160        }
161        discovery.namespace_manifests.push(manifest);
162    }
163    for manifest_path in namespace_manifest_paths {
164        let Some(plugin_root) = plugin_root_for_manifest(&manifest_path) else {
165            continue;
166        };
167        if !seen_namespace_roots.insert(plugin_root) {
168            continue;
169        }
170        if let Some(manifest) = read_optional_text_file(
171            file_system,
172            manifest_path,
173            &mut budget,
174            &mut discovery.warnings,
175        )
176        .await
177        {
178            discovery.namespace_manifests.push(manifest);
179        }
180    }
181
182    if let Some(manifest) = root_manifest {
183        let declarations = plugin_declaration_paths(&path, &manifest, &mut discovery.warnings);
184        let mcp_path = if declarations.mcp_inline {
185            None
186        } else {
187            declarations
188                .mcp_config
189                .or_else(|| path.join(DEFAULT_MCP_CONFIG_PATH).ok())
190        };
191        let mcp_config = match mcp_path {
192            Some(path) => {
193                read_optional_text_file(file_system, path, &mut budget, &mut discovery.warnings)
194                    .await
195            }
196            None => None,
197        };
198        let apps_config = match declarations.apps_config {
199            Some(path) => {
200                read_optional_text_file(file_system, path, &mut budget, &mut discovery.warnings)
201                    .await
202            }
203            None => None,
204        };
205        discovery.plugin = Some(DiscoveredPluginFiles {
206            manifest,
207            mcp_config,
208            apps_config,
209        });
210    }
211
212    for skill_path in skill_paths {
213        let Some(instructions) = read_optional_text_file(
214            file_system,
215            skill_path.clone(),
216            &mut budget,
217            &mut discovery.warnings,
218        )
219        .await
220        else {
221            continue;
222        };
223        let metadata = match skill_path
224            .parent()
225            .and_then(|skill_dir| skill_dir.join(SKILL_METADATA_PATH).ok())
226        {
227            Some(metadata_path) => {
228                read_optional_text_file(
229                    file_system,
230                    metadata_path,
231                    &mut budget,
232                    &mut discovery.warnings,
233                )
234                .await
235            }
236            None => None,
237        };
238        discovery.skills.push(DiscoveredSkillFiles {
239            instructions,
240            metadata,
241        });
242    }
243
244    discovery
245}
246
247async fn read_first_plugin_manifest(
248    file_system: &dyn ExecutorFileSystem,
249    root: &PathUri,
250    budget: &mut BundleBudget,
251    warnings: &mut Vec<String>,
252) -> Option<CapabilityTextFile> {
253    for relative_path in DISCOVERABLE_PLUGIN_MANIFEST_PATHS {
254        let Ok(path) = root.join(relative_path) else {
255            continue;
256        };
257        if let Some(manifest) = read_optional_text_file(file_system, path, budget, warnings).await {
258            return Some(manifest);
259        }
260    }
261    None
262}
263
264async fn read_nearest_ancestor_manifest(
265    file_system: &dyn ExecutorFileSystem,
266    root: &PathUri,
267    budget: &mut BundleBudget,
268    warnings: &mut Vec<String>,
269) -> Option<CapabilityTextFile> {
270    let mut ancestor = root.parent();
271    while let Some(path) = ancestor {
272        if let Some(manifest) =
273            read_first_plugin_manifest(file_system, &path, budget, warnings).await
274        {
275            return Some(manifest);
276        }
277        ancestor = path.parent();
278    }
279    None
280}
281
282async fn read_optional_text_file(
283    file_system: &dyn ExecutorFileSystem,
284    path: PathUri,
285    budget: &mut BundleBudget,
286    warnings: &mut Vec<String>,
287) -> Option<CapabilityTextFile> {
288    let metadata = match file_system.get_metadata(&path, /*sandbox*/ None).await {
289        Ok(metadata) if metadata.is_file => metadata,
290        Ok(_) => return None,
291        Err(error) if error.kind() == io::ErrorKind::NotFound => return None,
292        Err(error) => {
293            warnings.push(format!("failed to inspect capability file {path}: {error}"));
294            return None;
295        }
296    };
297    let Ok(size) = usize::try_from(metadata.size) else {
298        warnings.push(format!("capability file {path} is too large"));
299        return None;
300    };
301    if size > MAX_FILE_BYTES {
302        warnings.push(format!(
303            "capability file {path} exceeds the {MAX_FILE_BYTES}-byte limit"
304        ));
305        return None;
306    }
307    if !budget.can_add(size) {
308        warnings.push(format!(
309            "capability root bundle exceeds the {MAX_BUNDLE_BYTES_PER_ROOT}-byte limit"
310        ));
311        return None;
312    }
313    let mut stream = match file_system.read_file_stream(&path, /*sandbox*/ None).await {
314        Ok(stream) => stream,
315        Err(error) => {
316            warnings.push(format!("failed to read capability file {path}: {error}"));
317            return None;
318        }
319    };
320    let mut contents = Vec::with_capacity(size);
321    while let Some(chunk) = stream.next().await {
322        let chunk = match chunk {
323            Ok(chunk) => chunk,
324            Err(error) => {
325                warnings.push(format!("failed to read capability file {path}: {error}"));
326                return None;
327            }
328        };
329        let Some(new_len) = contents.len().checked_add(chunk.len()) else {
330            warnings.push(format!("capability file {path} exceeded its read limit"));
331            return None;
332        };
333        if new_len > MAX_FILE_BYTES || !budget.can_add(new_len) {
334            warnings.push(format!("capability file {path} exceeded its read limit"));
335            return None;
336        }
337        contents.extend_from_slice(&chunk);
338    }
339    let contents = match String::from_utf8(contents) {
340        Ok(contents) => contents,
341        Err(error) => {
342            warnings.push(format!("capability file {path} is not UTF-8: {error}"));
343            return None;
344        }
345    };
346    budget.add(contents.len());
347    Some(CapabilityTextFile { path, contents })
348}
349
350fn is_plugin_manifest_path(path: &PathUri) -> bool {
351    plugin_manifest_priority(path).is_some()
352}
353
354fn plugin_manifest_priority(path: &PathUri) -> Option<usize> {
355    if path.basename().as_deref() != Some("plugin.json") {
356        return None;
357    }
358    let manifest_directory = path.parent()?.basename()?;
359    DISCOVERABLE_PLUGIN_MANIFEST_PATHS
360        .iter()
361        .position(|relative_path| {
362            relative_path.strip_suffix("/plugin.json") == Some(manifest_directory.as_str())
363        })
364}
365
366fn plugin_root_for_manifest(path: &PathUri) -> Option<PathUri> {
367    path.parent()?.parent()
368}
369
370#[derive(Default)]
371struct BundleBudget {
372    bytes: usize,
373}
374
375impl BundleBudget {
376    fn can_add(&self, bytes: usize) -> bool {
377        self.bytes
378            .checked_add(bytes)
379            .is_some_and(|total| total <= MAX_BUNDLE_BYTES_PER_ROOT)
380    }
381
382    fn add(&mut self, bytes: usize) {
383        self.bytes += bytes;
384    }
385}
386
387#[derive(Default)]
388struct PluginDeclarationPaths {
389    mcp_config: Option<PathUri>,
390    mcp_inline: bool,
391    apps_config: Option<PathUri>,
392}
393
394#[derive(Deserialize)]
395#[serde(rename_all = "camelCase")]
396struct RawPluginDeclarations {
397    #[serde(default)]
398    mcp_servers: Option<Value>,
399    #[serde(default)]
400    apps: Option<Value>,
401}
402
403fn plugin_declaration_paths(
404    root: &PathUri,
405    manifest: &CapabilityTextFile,
406    warnings: &mut Vec<String>,
407) -> PluginDeclarationPaths {
408    let declarations = match serde_json::from_str::<RawPluginDeclarations>(&manifest.contents) {
409        Ok(declarations) => declarations,
410        Err(_) => return PluginDeclarationPaths::default(),
411    };
412    PluginDeclarationPaths {
413        mcp_config: declarations.mcp_servers.as_ref().and_then(|value| {
414            declared_file_path(root, "mcpServers", value, &manifest.path, warnings)
415        }),
416        mcp_inline: declarations
417            .mcp_servers
418            .as_ref()
419            .is_some_and(Value::is_object),
420        apps_config: declarations
421            .apps
422            .as_ref()
423            .and_then(|value| declared_file_path(root, "apps", value, &manifest.path, warnings)),
424    }
425}
426
427fn declared_file_path(
428    root: &PathUri,
429    field: &str,
430    value: &Value,
431    manifest_path: &PathUri,
432    warnings: &mut Vec<String>,
433) -> Option<PathUri> {
434    let Value::String(path) = value else {
435        return None;
436    };
437    let Some(relative_path) = path.strip_prefix("./") else {
438        warnings.push(format!(
439            "ignoring {field} in {manifest_path}: path must start with `./`"
440        ));
441        return None;
442    };
443    if relative_path.is_empty()
444        || relative_path
445            .split(['/', '\\'])
446            .any(|component| component == "..")
447    {
448        warnings.push(format!(
449            "ignoring {field} in {manifest_path}: path must remain below the capability root"
450        ));
451        return None;
452    }
453    match root.join(relative_path) {
454        Ok(path) if path.starts_with(root) => Some(path),
455        Ok(_) | Err(_) => {
456            warnings.push(format!(
457                "ignoring {field} in {manifest_path}: path must remain below the capability root"
458            ));
459            None
460        }
461    }
462}
codex_exec_server/capability_discovery.rs

codex_exec_server/
capability_discovery.rs
