mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
Make local environment optional in EnvironmentManager (#23369)
## Summary - make `EnvironmentManager` local environment/runtime paths optional - simplify constructor surface around snapshot materialization - rename local env accessors to `require_local_environment` / `try_local_environment` ## Validation - devbox Bazel build for touched crate surfaces - `//codex-rs/exec-server:exec-server-unit-tests` - `//codex-rs/app-server-client:app-server-client-unit-tests` - filtered touched `//codex-rs/core:core-unit-tests` cases
This commit is contained in:
starr-openai
committed by
GitHub
Unverified
GitHub
parent
7f4d7ae3a4
commit
5c43a64e2b
@@ -7,6 +7,7 @@ use std::sync::atomic::AtomicBool;
|
||||
use crate::attestation::app_server_attestation_provider;
|
||||
use crate::config_manager::ConfigManager;
|
||||
use crate::connection_rpc_gate::ConnectionRpcGate;
|
||||
use crate::error_code::internal_error;
|
||||
use crate::error_code::invalid_request;
|
||||
use crate::extensions::guardian_agent_spawner;
|
||||
use crate::extensions::thread_extensions;
|
||||
@@ -165,10 +166,11 @@ pub(crate) struct MessageProcessor {
|
||||
command_exec_processor: CommandExecRequestProcessor,
|
||||
process_exec_processor: ProcessExecRequestProcessor,
|
||||
config_processor: ConfigRequestProcessor,
|
||||
environment_manager: Arc<EnvironmentManager>,
|
||||
environment_processor: EnvironmentRequestProcessor,
|
||||
external_agent_config_processor: ExternalAgentConfigRequestProcessor,
|
||||
feedback_processor: FeedbackRequestProcessor,
|
||||
fs_processor: FsRequestProcessor,
|
||||
fs_processor: Option<FsRequestProcessor>,
|
||||
git_processor: GitRequestProcessor,
|
||||
initialize_processor: InitializeRequestProcessor,
|
||||
marketplace_processor: MarketplaceRequestProcessor,
|
||||
@@ -272,6 +274,23 @@ pub(crate) struct MessageProcessorArgs {
|
||||
}
|
||||
|
||||
impl MessageProcessor {
|
||||
fn fs_processor(&self) -> Result<&FsRequestProcessor, JSONRPCErrorError> {
|
||||
self.fs_processor
|
||||
.as_ref()
|
||||
.ok_or_else(|| internal_error("local filesystem is not configured"))
|
||||
}
|
||||
|
||||
fn require_local_environment(&self) -> Result<(), JSONRPCErrorError> {
|
||||
// CCA filters these local-only RPCs before they reach app-server, but
|
||||
// keep a Codex-side backstop so no-local app-server modes fail safely
|
||||
// if a client still invokes one directly.
|
||||
self.environment_manager
|
||||
.try_local_environment()
|
||||
.is_some()
|
||||
.then_some(())
|
||||
.ok_or_else(|| internal_error("local environment is not configured"))
|
||||
}
|
||||
|
||||
/// Create a new `MessageProcessor`, retaining a handle to the outgoing
|
||||
/// `Sender` so handlers can enqueue messages to be written to stdout.
|
||||
pub(crate) fn new(args: MessageProcessorArgs) -> Self {
|
||||
@@ -301,6 +320,7 @@ impl MessageProcessor {
|
||||
// affect per-thread behavior, but they must not move newly started,
|
||||
// resumed, or forked threads to a different persistence backend/root.
|
||||
let thread_store = codex_core::thread_store_from_config(config.as_ref(), state_db.clone());
|
||||
let environment_manager_for_requests = Arc::clone(&environment_manager);
|
||||
let thread_manager = Arc::new_cyclic(|thread_manager| {
|
||||
ThreadManager::new(
|
||||
config.as_ref(),
|
||||
@@ -443,7 +463,6 @@ impl MessageProcessor {
|
||||
Some(on_effective_plugins_changed),
|
||||
);
|
||||
}
|
||||
let fs_watch_manager = FsWatchManager::new(outgoing.clone());
|
||||
let config_processor = ConfigRequestProcessor::new(
|
||||
outgoing.clone(),
|
||||
config_manager.clone(),
|
||||
@@ -461,13 +480,17 @@ impl MessageProcessor {
|
||||
);
|
||||
let environment_processor =
|
||||
EnvironmentRequestProcessor::new(thread_manager.environment_manager());
|
||||
let fs_processor = FsRequestProcessor::new(
|
||||
thread_manager
|
||||
.environment_manager()
|
||||
.local_environment()
|
||||
.get_filesystem(),
|
||||
fs_watch_manager,
|
||||
);
|
||||
// `fs/*` is a local-host filesystem surface. Do not construct it when
|
||||
// the manager intentionally has no local environment.
|
||||
let fs_processor = thread_manager
|
||||
.environment_manager()
|
||||
.try_local_environment()
|
||||
.map(|environment| {
|
||||
FsRequestProcessor::new(
|
||||
environment.get_filesystem(),
|
||||
FsWatchManager::new(outgoing.clone()),
|
||||
)
|
||||
});
|
||||
let windows_sandbox_processor = WindowsSandboxRequestProcessor::new(
|
||||
outgoing.clone(),
|
||||
Arc::clone(&config),
|
||||
@@ -482,6 +505,7 @@ impl MessageProcessor {
|
||||
command_exec_processor,
|
||||
process_exec_processor,
|
||||
config_processor,
|
||||
environment_manager: environment_manager_for_requests,
|
||||
environment_processor,
|
||||
external_agent_config_processor,
|
||||
feedback_processor,
|
||||
@@ -705,7 +729,9 @@ impl MessageProcessor {
|
||||
) {
|
||||
session_state.rpc_gate.shutdown().await;
|
||||
self.outgoing.connection_closed(connection_id).await;
|
||||
self.fs_processor.connection_closed(connection_id).await;
|
||||
if let Some(fs_processor) = &self.fs_processor {
|
||||
fs_processor.connection_closed(connection_id).await;
|
||||
}
|
||||
self.command_exec_processor
|
||||
.connection_closed(connection_id)
|
||||
.await;
|
||||
@@ -911,47 +937,47 @@ impl MessageProcessor {
|
||||
self.environment_processor.environment_add(params).await
|
||||
}
|
||||
ClientRequest::FsReadFile { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.read_file(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsWriteFile { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.write_file(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsCreateDirectory { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.create_directory(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsGetMetadata { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.get_metadata(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsReadDirectory { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.read_directory(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsRemove { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.remove(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsCopy { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.copy(params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsWatch { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.watch(connection_id, params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::FsUnwatch { params, .. } => self
|
||||
.fs_processor
|
||||
.fs_processor()?
|
||||
.unwatch(connection_id, params)
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
@@ -1280,6 +1306,7 @@ impl MessageProcessor {
|
||||
.await
|
||||
.map(|response| Some(response.into())),
|
||||
ClientRequest::OneOffCommandExec { params, .. } => {
|
||||
self.require_local_environment()?;
|
||||
self.command_exec_processor
|
||||
.one_off_command_exec(&request_id, params)
|
||||
.await
|
||||
@@ -1299,11 +1326,13 @@ impl MessageProcessor {
|
||||
.command_exec_terminate(request_id.clone(), params)
|
||||
.await
|
||||
}
|
||||
ClientRequest::ProcessSpawn { params, .. } => self
|
||||
.process_exec_processor
|
||||
.process_spawn(request_id.clone(), params)
|
||||
.await
|
||||
.map(|()| None),
|
||||
ClientRequest::ProcessSpawn { params, .. } => {
|
||||
self.require_local_environment()?;
|
||||
self.process_exec_processor
|
||||
.process_spawn(request_id.clone(), params)
|
||||
.await
|
||||
.map(|()| None)
|
||||
}
|
||||
ClientRequest::ProcessWriteStdin { params, .. } => {
|
||||
self.process_exec_processor
|
||||
.process_write_stdin(request_id.clone(), params)
|
||||
|
||||
Reference in New Issue
Block a user