V2 FSCommandProcess协议

本文承接V2 Turn与Item协议和V2 PluginAppsMCP协议。fs/*、process/*、command/exec 和 Turn 内的 command item 都能运行程序或访问文件，但它们的 owner 不同：FS processor 调用本地 ExecutorFileSystem，process processor 管理连接范围内的独立进程，command processor 负责带 sandbox/权限配置的一次性命令。理解这几个边界，才能解释为什么某个请求没有审批、为什么输出出现在 notification 而不是最终 response，以及连接断开时哪些资源会被终止。

1. FS协议模型

源码位置：codex-rs/app-server-protocol/src/protocol/v2/fs.rs :: FsReadFileParams、FsWriteFileParams、FsReadDirectoryEntry、FsWatchParams

pub struct FsReadFileParams {
    pub path: AbsolutePathBuf,
}

pub struct FsReadFileResponse {
    pub data_base64: String,
}

pub struct FsWriteFileParams {
    pub path: AbsolutePathBuf,
    pub data_base64: String,
}

pub struct FsReadDirectoryEntry {
    pub file_name: String,
    pub is_directory: bool,
    pub is_file: bool,
}

pub struct FsWatchParams {
    pub watch_id: String,
    pub path: AbsolutePathBuf,
}

FS API 的路径类型是 AbsolutePathBuf，文件内容以 base64 传输，目录接口只返回直接子项名称和类型。watch_id 是 connection-scoped 标识，后续 fs/unwatch 和 fs/changed 都依赖它；它不是文件系统 inode，也不是全局唯一 id。

2. FS处理器与错误

源码位置：codex-rs/app-server/src/request_processors/fs_processor.rs :: file_system、read_file、write_file、map_fs_error

fn file_system(&self) -> Result<Arc<dyn ExecutorFileSystem>, JSONRPCErrorError> {
    self.environment_manager
        .try_local_environment()
        .map(|environment| environment.get_filesystem())
        .ok_or_else(|| internal_error("local filesystem is not configured"))
}

pub(crate) async fn read_file(
    &self,
    params: FsReadFileParams,
) -> Result<FsReadFileResponse, JSONRPCErrorError> {
    let path = PathUri::from_abs_path(&params.path);
    let bytes = self
        .file_system()?
        .read_file(&path, Default::default(), /*sandbox*/ None)
        .await
        .map_err(map_fs_error)?;
    Ok(FsReadFileResponse {
        data_base64: STANDARD.encode(bytes),
    })
}

pub(crate) async fn write_file(
    &self,
    params: FsWriteFileParams,
) -> Result<FsWriteFileResponse, JSONRPCErrorError> {
    let bytes = STANDARD.decode(params.data_base64).map_err(|err| {
        invalid_request(format!(
            "fs/writeFile requires valid base64 dataBase64: {err}"
        ))
    })?;
    let path = PathUri::from_abs_path(&params.path);
    self.file_system()?
        .write_file(&path, bytes, Default::default(), /*sandbox*/ None)
        .await
        .map_err(map_fs_error)?;
    Ok(FsWriteFileResponse {})
}

fn map_fs_error(err: io::Error) -> JSONRPCErrorError {
    if err.kind() == io::ErrorKind::InvalidInput {
        invalid_request(err.to_string())
    } else {
        internal_error(err.to_string())
    }
}

FS 调用要求本地 environment 已装配，但传给 ExecutorFileSystem 的 sandbox 参数是 None。这不是 Turn 文件工具的权限路径；它是 App Server host 文件抽象。base64 解码在写入前完成，非法输入属于 invalid request，底层 I/O 的其他失败则映射为 internal error。

3. FS目录与监控

源码位置：codex-rs/app-server/src/request_processors/fs_processor.rs :: read_directory、watch、unwatch、connection_closed

pub(crate) async fn read_directory(
    &self,
    params: FsReadDirectoryParams,
) -> Result<FsReadDirectoryResponse, JSONRPCErrorError> {
    let path = PathUri::from_abs_path(&params.path);
    let entries = self
        .file_system()?
        .read_directory(&path, /*sandbox*/ None)
        .await
        .map_err(map_fs_error)?;
    Ok(FsReadDirectoryResponse {
        entries: entries
            .into_iter()
            .map(|entry| FsReadDirectoryEntry {
                file_name: entry.file_name,
                is_directory: entry.is_directory,
                is_file: entry.is_file,
            })
            .collect(),
    })
}

pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) {
    self.fs_watch_manager.connection_closed(connection_id).await;
}

目录读取不会递归；需要递归复制要使用 fs/copy 并显式设置 recursive。watch 的清理由 FsWatchManager 按 connection id 执行，所以同一个字符串 watch_id 在不同连接上不会互相取消。

4. process/spawn

源码位置：codex-rs/app-server-protocol/src/protocol/v2/process.rs :: ProcessSpawnParams、ProcessWriteStdinParams、ProcessOutputDeltaNotification、ProcessExitedNotification

pub struct ProcessSpawnParams {
    pub command: Vec<String>,
    pub process_handle: String,
    pub cwd: AbsolutePathBuf,
    pub tty: bool,
    pub stream_stdin: bool,
    pub stream_stdout_stderr: bool,
    pub output_bytes_cap: Option<Option<usize>>,
    pub timeout_ms: Option<Option<i64>>,
    pub env: Option<HashMap<String, Option<String>>>,
    pub size: Option<ProcessTerminalSize>,
}

pub struct ProcessWriteStdinParams {
    pub process_handle: String,
    pub delta_base64: Option<String>,
    pub close_stdin: bool,
}

pub struct ProcessOutputDeltaNotification {
    pub process_handle: String,
    pub stream: ProcessOutputStream,
    pub delta_base64: String,
    pub cap_reached: bool,
}

pub struct ProcessExitedNotification {
    pub process_handle: String,
    pub exit_code: i32,
    pub stdout: String,
    pub stderr: String,
    pub stdout_cap_reached: bool,
    pub stderr_cap_reached: bool,
}

process/spawn 的 process_handle 是连接范围内的客户端句柄；活动句柄不能重复，进程退出后可以复用。双层 Option 让客户端区分省略默认 cap/timeout 与显式传 null 禁用 cap/timeout。TTY 会隐含 stdin 和 stdout/stderr streaming，初始 size 只有在 tty=true 时有效。

5. process控制

源码位置：codex-rs/app-server/src/request_processors/process_exec_processor.rs :: process_spawn、write_stdin、kill、resize_pty

pub(crate) async fn process_spawn(
    &self,
    request_id: ConnectionRequestId,
    params: ProcessSpawnParams,
) -> Result<(), JSONRPCErrorError> {
    self.require_local_environment()?;
    let ProcessSpawnParams {
        command,
        process_handle,
        cwd,
        tty,
        stream_stdin,
        stream_stdout_stderr,
        output_bytes_cap,
        timeout_ms,
        env: env_overrides,
        size,
    } = params;
    if command.is_empty() {
        return Err(invalid_request("command must not be empty"));
    }
    if process_handle.is_empty() {
        return Err(invalid_request("processHandle must not be empty"));
    }
    if size.is_some() && !tty {
        return Err(invalid_params("process/spawn size requires tty: true"));
    }
    let mut env = std::env::vars().collect::<HashMap<_, _>>();
    if let Some(env_overrides) = env_overrides {
        for (key, value) in env_overrides {
            match value {
                Some(value) => {
                    env.insert(key, value);
                }
                None => {
                    env.remove(&key);
                }
            }
        }
    }
    env.retain(|name, _| !is_non_inheritable_env_var(name));
    let expiration = match timeout_ms {
        Some(Some(timeout_ms)) => match u64::try_from(timeout_ms) {
            Ok(timeout_ms) => timeout_ms.into(),
            Err(_) => {
                return Err(invalid_params(format!(
                    "process/spawn timeoutMs must be non-negative, got {timeout_ms}"
                )));
            }
        },
        Some(None) => ExecExpiration::Cancellation(CancellationToken::new()),
        None => ExecExpiration::DefaultTimeout,
    };
    let output_bytes_cap = output_bytes_cap.unwrap_or(Some(DEFAULT_OUTPUT_BYTES_CAP));
    let size = size.map(terminal_size_from_protocol).transpose()?;
    self.process_exec_manager
        .start(StartProcessParams {
            outgoing: self.outgoing.clone(),
            request_id,
            process_handle,
            command,
            cwd,
            env,
            expiration,
            tty,
            stream_stdin,
            stream_stdout_stderr,
            output_bytes_cap,
            size,
        })
        .await?;
    Ok(())
}

async fn write_stdin(
    &self,
    request_id: ConnectionRequestId,
    params: ProcessWriteStdinParams,
) -> Result<ProcessWriteStdinResponse, JSONRPCErrorError> {
    if params.delta_base64.is_none() && !params.close_stdin {
        return Err(invalid_params(
            "process/writeStdin requires deltaBase64 or closeStdin",
        ));
    }
    let delta = match params.delta_base64 {
        Some(delta_base64) => STANDARD
            .decode(delta_base64)
            .map_err(|err| invalid_params(format!("invalid deltaBase64: {err}")))?,
        None => Vec::new(),
    };
    self.send_control(
        request_id.connection_id,
        params.process_handle,
        ProcessControl::Write {
            delta,
            close_stdin: params.close_stdin,
        },
    )
    .await?;
    Ok(ProcessWriteStdinResponse {})
}

process 控制操作只向连接所属的 process session 发送 control message。writeStdin 可以只关闭 stdin，也可以先写 bytes 再关闭；空请求被拒绝。process/kill 和 process/resizePty 复用同一 control channel，不会创建新的进程。

6. process清理

源码位置：codex-rs/app-server/src/request_processors/process_exec_processor.rs :: connection_closed、run_process

async fn connection_closed(&self, connection_id: ConnectionId) {
    let controls = {
        let mut sessions = self.sessions.lock().await;
        let process_handles = sessions
            .keys()
            .filter(|process_handle| process_handle.connection_id == connection_id)
            .cloned()
            .collect::<Vec<_>>();
        let mut controls = Vec::with_capacity(process_handles.len());
        for process_handle in process_handles {
            if let Some(control) = sessions.remove(&process_handle) {
                controls.push(control);
            }
        }
        controls
    };

    for control in controls {
        let _ = control
            .control_tx
            .send(ProcessControlRequest {
                control: ProcessControl::Kill,
                response_tx: None,
            })
            .await;
    }
}

连接关闭会先从 session map 移除该连接的句柄，再发送 kill control；因此 process 资源是 connection-scoped，而不是全局后台任务。启用 streaming 时，输出通过 process/outputDelta 发送，已经发送的 bytes 不会再次出现在 process/exited；未 streaming 时，退出消息保留捕获输出。

7. command/exec

源码位置：codex-rs/app-server-protocol/src/protocol/v2/command_exec.rs :: CommandExecParams、CommandExecResponse、CommandExecOutputDeltaNotification

pub struct CommandExecParams {
    pub command: Vec<String>,
    pub process_id: Option<String>,
    pub tty: bool,
    pub stream_stdin: bool,
    pub stream_stdout_stderr: bool,
    pub output_bytes_cap: Option<usize>,
    pub disable_output_cap: bool,
    pub disable_timeout: bool,
    pub timeout_ms: Option<i64>,
    pub cwd: Option<PathBuf>,
    pub env: Option<HashMap<String, Option<String>>>,
    pub size: Option<CommandExecTerminalSize>,
    pub sandbox_policy: Option<SandboxPolicy>,
    pub permission_profile: Option<String>,
}

pub struct CommandExecResponse {
    pub exit_code: i32,
    pub stdout: String,
    pub stderr: String,
}

command/exec 不创建 Thread/Turn，但使用 server sandbox；process/spawn 则明确是 host 上不带 Codex sandbox 的独立进程。command 参数还支持 sandbox_policy 或 permission_profile 二选一、输出 cap/timeout 的显式禁用和可选 cwd。

源码位置：codex-rs/app-server/src/request_processors/command_exec_processor.rs :: exec_one_off_command_inner

if params.command.is_empty() {
    return Err(invalid_request("command must not be empty"));
}

let CommandExecParams {
    command,
    process_id,
    tty,
    stream_stdin,
    stream_stdout_stderr,
    output_bytes_cap,
    disable_output_cap,
    disable_timeout,
    timeout_ms,
    cwd,
    env: env_overrides,
    size,
    sandbox_policy,
    permission_profile,
} = params;

if sandbox_policy.is_some() && permission_profile.is_some() {
    return Err(invalid_request(
        "`permissionProfile` cannot be combined with `sandboxPolicy`",
    ));
}
if size.is_some() && !tty {
    return Err(invalid_params("command/exec size requires tty: true"));
}
if disable_output_cap && output_bytes_cap.is_some() {
    return Err(invalid_params(
        "command/exec cannot set both outputBytesCap and disableOutputCap",
    ));
}
if disable_timeout && timeout_ms.is_some() {
    return Err(invalid_params(
        "command/exec cannot set both timeoutMs and disableTimeout",
    ));
}

command processor 在创建执行上下文前完成参数互斥检查，再计算 cwd、环境变量、超时、输出 capture policy 和 effective permission profile。后续响应要等进程退出，并保证所有 output delta 已经发送，所以它的 response 时序与 process/spawn 的“启动即返回”不同。

8. 测试与边界

源码位置：codex-rs/app-server/tests/suite/v2/fs.rs :: fs_methods_cover_current_fs_utils_surface、fs_write_file_rejects_invalid_base64、fs_methods_return_error_when_local_environment_is_disabled、fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifications

源码位置：codex-rs/app-server/tests/suite/v2/process_exec.rs :: process_spawn_returns_before_exit_and_emits_exit_notification、process_spawn_reports_buffered_output_cap_reached、process_kill_terminates_running_process

源码位置：codex-rs/app-server/tests/suite/v2/thread_shell_command.rs :: thread_shell_command_returns_error_when_local_environment_is_disabled、thread_shell_command_uses_existing_active_turn

cd codex-rs
cargo test -p codex-app-server fs_methods_cover_current_fs_utils_surface -- --nocapture --test-threads=1
cargo test -p codex-app-server fs_write_file_rejects_invalid_base64 -- --nocapture --test-threads=1
cargo test -p codex-app-server fs_watch_directory_reports_changed_child_paths_and_unwatch_stops_notifications -- --nocapture --test-threads=1
cargo test -p codex-app-server process_spawn_returns_before_exit_and_emits_exit_notification -- --nocapture --test-threads=1
cargo test -p codex-app-server process_spawn_reports_buffered_output_cap_reached -- --nocapture --test-threads=1
cargo test -p codex-app-server process_kill_terminates_running_process -- --nocapture --test-threads=1
cargo test -p codex-app-server thread_shell_command_returns_error_when_local_environment_is_disabled -- --nocapture --test-threads=1
cargo test -p codex-app-server thread_shell_command_uses_existing_active_turn -- --nocapture --test-threads=1

这些测试分别断言文件读写和 watch 的输入边界、独立 process 的启动/输出/终止时序，以及 Turn command 与独立命令的环境差异。它们不能证明每个操作系统的 PTY 行为、所有 sandbox backend、远程 environment 或任意进程树的终止效果。

9. 源码定位练习

遇到文件请求没有审批，先确认它是否走 FsRequestProcessor；遇到进程启动后 response 立即返回，检查是否是 process/spawn；遇到必须等待退出并接收 command output delta，检查 command/exec。连接断开时，分别沿 FsWatchManager::connection_closed 和 ProcessExecManager::connection_closed 追踪清理，就能看出 watch 与 process 的资源所有权差异。