mirror of
https://github.com/openai/codex.git
synced 2026-10-20 00:18:49 +00:00
Expose environment status through app-server (#32920)
## What changed - Add the experimental `environment/status` request for inspecting a configured environment without starting or reconnecting it. - Report `ready`, `pending`, `disconnected`, or `unknown`, including error details for disconnected and unknown environments. - Probe ready remote environments over their existing exec-server connection. ## Testing - Add an app-server integration test covering local and remote ready, pending, disconnected, and unknown environments. GitOrigin-RevId: 397bba603aa9e0b59008ee6cb7cbde46e357652c
This commit is contained in:
committed by
copyberry
parent
80c6cd3014
commit
f96cf4d1c3
@@ -45,6 +45,7 @@ const IGNORED_DEFINITIONS: &[&str] = &["Option<()>"];
|
||||
const JSON_V1_ALLOWLIST: &[&str] = &["InitializeParams", "InitializeResponse"];
|
||||
const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] = &[
|
||||
"EnvironmentShellInfo",
|
||||
"EnvironmentStatusKind",
|
||||
"PathUri",
|
||||
"RemoteControlClient",
|
||||
"RemoteControlClientsListOrder",
|
||||
|
||||
@@ -956,6 +956,13 @@ client_request_definitions! {
|
||||
serialization: global_shared_read("environment"),
|
||||
response: v2::EnvironmentInfoResponse,
|
||||
},
|
||||
#[experimental("environment/status")]
|
||||
/// Reads the current status of a configured execution environment.
|
||||
EnvironmentStatus => "environment/status" {
|
||||
params: v2::EnvironmentStatusParams,
|
||||
serialization: global_shared_read("environment"),
|
||||
response: v2::EnvironmentStatusResponse,
|
||||
},
|
||||
|
||||
McpServerOauthLogin => "mcpServer/oauth/login" {
|
||||
params: v2::McpServerOauthLoginParams,
|
||||
|
||||
@@ -37,6 +37,51 @@ pub struct EnvironmentInfoResponse {
|
||||
pub cwd: Option<PathUri>,
|
||||
}
|
||||
|
||||
/// Parameters for reading the current status of one configured environment.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct EnvironmentStatusParams {
|
||||
/// Environment id to inspect.
|
||||
pub environment_id: String,
|
||||
}
|
||||
|
||||
/// Current status for the requested environment.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct EnvironmentStatusResponse {
|
||||
/// Current status observed without starting or recovering the environment.
|
||||
pub status: EnvironmentStatusKind,
|
||||
/// Human-readable detail for `disconnected` and `unknown`; omitted for other statuses.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
#[ts(optional)]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
/// Current status observed by app-server without starting or recovering an environment.
|
||||
///
|
||||
/// For a currently ready remote environment, app-server asks the existing
|
||||
/// exec-server connection for `environment/status` without allowing recovery.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(rename_all = "camelCase", export_to = "v2/")]
|
||||
pub enum EnvironmentStatusKind {
|
||||
/// The environment is local, or an already-connected remote exec-server answered
|
||||
/// `environment/status` over its existing initialized connection.
|
||||
Ready,
|
||||
/// The configured environment has no ready connection and no observed connection failure.
|
||||
/// This includes lazy environments that have never been started and initial startup that has
|
||||
/// not finished.
|
||||
Pending,
|
||||
/// A connection attempt, prior connection, or fail-fast `environment/status` probe observed
|
||||
/// a failure. This does not promise the failure is terminal: later normal environment use may
|
||||
/// recover it. This call does not trigger recovery; `error` contains the observed reason.
|
||||
Disconnected,
|
||||
/// The requested environment id is not configured in app-server.
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
|
||||
@@ -206,6 +206,7 @@ Example with notification opt-out:
|
||||
- `experimentalFeature/enablement/set` — patch the in-memory process-wide runtime feature enablement for currently supported feature keys. For each feature, precedence is: cloud requirements > --enable <feature_name> > config.toml > experimentalFeature/enablement/set (new) > code default. Invalid keys will be ignored.
|
||||
- `environment/add` — experimental; add or replace a named remote environment by `environmentId` and `execServerUrl` for later selection by `thread/start` or `turn/start`; optional `connectTimeoutMs` overrides the WebSocket connection timeout; returns `{}` and does not change the default environment.
|
||||
- `environment/info` — experimental; connect to a configured environment by `environmentId` and return its detected `shell` plus its default `cwd` as a canonical environment-native `file:` URI. Connection failures are returned as request errors.
|
||||
- `environment/status` — experimental; read the current status for one configured `environmentId`. Ready remote environments are probed over their existing exec-server connection without starting or reconnecting environments; the response reports `ready`, `pending`, `disconnected`, or `unknown`.
|
||||
- `collaborationMode/list` — list available collaboration mode presets (experimental, no pagination). Built-in presets do not select a model; the Plan preset selects medium reasoning effort. This response omits built-in developer instructions; clients should either pass `settings.developer_instructions: null` when setting a mode to use Codex's built-in instructions, or provide their own instructions explicitly.
|
||||
- `skills/list` — list skills for one or more `cwd` values (optional `forceReload`).
|
||||
- `skills/extraRoots/set` — replace the app-server process runtime extra standalone skill roots. The roots are not persisted; missing directories are accepted and simply load no skills.
|
||||
|
||||
@@ -966,6 +966,9 @@ impl MessageProcessor {
|
||||
ClientRequest::EnvironmentInfo { params, .. } => {
|
||||
self.environment_processor.environment_info(params).await
|
||||
}
|
||||
ClientRequest::EnvironmentStatus { params, .. } => {
|
||||
self.environment_processor.environment_status(params).await
|
||||
}
|
||||
ClientRequest::FsReadFile { params, .. } => self
|
||||
.fs_processor
|
||||
.read_file(params)
|
||||
|
||||
@@ -64,6 +64,9 @@ use codex_app_server_protocol::EnvironmentAddResponse;
|
||||
use codex_app_server_protocol::EnvironmentInfoParams;
|
||||
use codex_app_server_protocol::EnvironmentInfoResponse;
|
||||
use codex_app_server_protocol::EnvironmentShellInfo;
|
||||
use codex_app_server_protocol::EnvironmentStatusKind;
|
||||
use codex_app_server_protocol::EnvironmentStatusParams;
|
||||
use codex_app_server_protocol::EnvironmentStatusResponse;
|
||||
use codex_app_server_protocol::ExperimentalFeature as ApiExperimentalFeature;
|
||||
use codex_app_server_protocol::ExperimentalFeatureListParams;
|
||||
use codex_app_server_protocol::ExperimentalFeatureListResponse;
|
||||
@@ -368,6 +371,7 @@ use codex_core_plugins::remote::RemotePluginShareContext as RemoteCatalogPluginS
|
||||
use codex_core_plugins::remote::RemotePluginShareSummary as RemoteCatalogPluginShareSummary;
|
||||
use codex_core_plugins::remote::RemotePluginSummary as RemoteCatalogPluginSummary;
|
||||
use codex_exec_server::EnvironmentManager;
|
||||
use codex_exec_server::EnvironmentObservedStatus;
|
||||
use codex_exec_server::LOCAL_ENVIRONMENT_ID;
|
||||
use codex_exec_server::LOCAL_FS;
|
||||
use codex_features::FEATURES;
|
||||
|
||||
@@ -52,4 +52,27 @@ impl EnvironmentRequestProcessor {
|
||||
.into(),
|
||||
))
|
||||
}
|
||||
|
||||
pub(crate) async fn environment_status(
|
||||
&self,
|
||||
params: EnvironmentStatusParams,
|
||||
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
|
||||
let environment_id = params.environment_id;
|
||||
let (status, error) = match self
|
||||
.environment_manager
|
||||
.get_environment_status(&environment_id)
|
||||
.await
|
||||
{
|
||||
Some(EnvironmentObservedStatus::Ready) => (EnvironmentStatusKind::Ready, None),
|
||||
Some(EnvironmentObservedStatus::Pending) => (EnvironmentStatusKind::Pending, None),
|
||||
Some(EnvironmentObservedStatus::Disconnected { error }) => {
|
||||
(EnvironmentStatusKind::Disconnected, Some(error))
|
||||
}
|
||||
None => (
|
||||
EnvironmentStatusKind::Unknown,
|
||||
Some(format!("unknown environment id `{environment_id}`")),
|
||||
),
|
||||
};
|
||||
Ok(Some(EnvironmentStatusResponse { status, error }.into()))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Result;
|
||||
use app_test_support::TestAppServer;
|
||||
use app_test_support::to_response;
|
||||
use codex_app_server_protocol::EnvironmentAddParams;
|
||||
use codex_app_server_protocol::EnvironmentAddResponse;
|
||||
use codex_app_server_protocol::EnvironmentStatusKind;
|
||||
use codex_app_server_protocol::EnvironmentStatusParams;
|
||||
use codex_app_server_protocol::EnvironmentStatusResponse;
|
||||
use codex_app_server_protocol::JSONRPCResponse;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use futures::SinkExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio::time::sleep;
|
||||
use tokio::time::timeout;
|
||||
use tokio_tungstenite::accept_async;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
|
||||
use super::exec_server_test_support::accept_initialized_exec_server;
|
||||
use super::exec_server_test_support::read_exec_server_json;
|
||||
|
||||
const RPC_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn environment_status_reports_connection_states_with_auto_env() -> Result<()> {
|
||||
let ready_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let ready_exec_server_url = format!("ws://{}", ready_listener.local_addr()?);
|
||||
let (ready_connected_tx, ready_connected_rx) = oneshot::channel();
|
||||
let (ready_release_tx, mut ready_release_rx) = oneshot::channel();
|
||||
let ready_exec_server = tokio::spawn(async move {
|
||||
let mut websocket = accept_initialized_exec_server(ready_listener).await?;
|
||||
ready_connected_tx
|
||||
.send(())
|
||||
.map_err(|()| anyhow::anyhow!("status test stopped before exec-server was ready"))?;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = &mut ready_release_rx => return Ok::<_, anyhow::Error>(()),
|
||||
request = read_exec_server_json(&mut websocket) => {
|
||||
let request = request?;
|
||||
assert_eq!(request["method"], "environment/status");
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
json!({
|
||||
"id": request["id"],
|
||||
"result": {"status": "ready"},
|
||||
})
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let pending_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let pending_exec_server_url = format!("ws://{}", pending_listener.local_addr()?);
|
||||
let (pending_connected_tx, pending_connected_rx) = oneshot::channel();
|
||||
let (pending_release_tx, pending_release_rx) = oneshot::channel();
|
||||
let pending_exec_server = tokio::spawn(async move {
|
||||
let (stream, _) = pending_listener.accept().await?;
|
||||
let _websocket = accept_async(stream).await?;
|
||||
pending_connected_tx
|
||||
.send(())
|
||||
.map_err(|()| anyhow::anyhow!("status test stopped before pending connection"))?;
|
||||
let _ = pending_release_rx.await;
|
||||
Ok::<_, anyhow::Error>(())
|
||||
});
|
||||
|
||||
let disconnected_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let disconnected_exec_server_url = format!("ws://{}", disconnected_listener.local_addr()?);
|
||||
let (disconnected_tx, disconnected_rx) = oneshot::channel();
|
||||
let disconnected_exec_server = tokio::spawn(async move {
|
||||
let (stream, _) = disconnected_listener.accept().await?;
|
||||
let websocket = accept_async(stream).await?;
|
||||
drop(websocket);
|
||||
disconnected_tx
|
||||
.send(())
|
||||
.map_err(|()| anyhow::anyhow!("status test stopped before disconnect"))?;
|
||||
Ok::<_, anyhow::Error>(())
|
||||
});
|
||||
|
||||
let mut app_server = TestAppServer::builder().build().await?;
|
||||
timeout(RPC_TIMEOUT, app_server.initialize()).await??;
|
||||
let auto_environment_id = app_server.auto_env()?.selection().environment_id.clone();
|
||||
|
||||
add_environment(&mut app_server, "ready", &ready_exec_server_url).await?;
|
||||
add_environment(&mut app_server, "pending", &pending_exec_server_url).await?;
|
||||
add_environment(
|
||||
&mut app_server,
|
||||
"disconnected",
|
||||
&disconnected_exec_server_url,
|
||||
)
|
||||
.await?;
|
||||
timeout(RPC_TIMEOUT, ready_connected_rx).await??;
|
||||
timeout(RPC_TIMEOUT, pending_connected_rx).await??;
|
||||
timeout(RPC_TIMEOUT, disconnected_rx).await??;
|
||||
|
||||
assert_eq!(
|
||||
wait_for_status(
|
||||
&mut app_server,
|
||||
&auto_environment_id,
|
||||
EnvironmentStatusKind::Ready,
|
||||
)
|
||||
.await?,
|
||||
EnvironmentStatusResponse {
|
||||
status: EnvironmentStatusKind::Ready,
|
||||
error: None,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
wait_for_status(&mut app_server, "ready", EnvironmentStatusKind::Ready).await?,
|
||||
EnvironmentStatusResponse {
|
||||
status: EnvironmentStatusKind::Ready,
|
||||
error: None,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
read_environment_status(&mut app_server, "pending").await?,
|
||||
EnvironmentStatusResponse {
|
||||
status: EnvironmentStatusKind::Pending,
|
||||
error: None,
|
||||
}
|
||||
);
|
||||
let disconnected = wait_for_status(
|
||||
&mut app_server,
|
||||
"disconnected",
|
||||
EnvironmentStatusKind::Disconnected,
|
||||
)
|
||||
.await?;
|
||||
let disconnected_error = disconnected.error.clone();
|
||||
assert!(disconnected_error.is_some());
|
||||
assert_eq!(
|
||||
disconnected,
|
||||
EnvironmentStatusResponse {
|
||||
status: EnvironmentStatusKind::Disconnected,
|
||||
error: disconnected_error,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
read_environment_status(&mut app_server, "missing").await?,
|
||||
EnvironmentStatusResponse {
|
||||
status: EnvironmentStatusKind::Unknown,
|
||||
error: Some("unknown environment id `missing`".to_string()),
|
||||
}
|
||||
);
|
||||
|
||||
let _ = ready_release_tx.send(());
|
||||
let _ = pending_release_tx.send(());
|
||||
timeout(RPC_TIMEOUT, ready_exec_server).await???;
|
||||
timeout(RPC_TIMEOUT, pending_exec_server).await???;
|
||||
timeout(RPC_TIMEOUT, disconnected_exec_server).await???;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn add_environment(
|
||||
app_server: &mut TestAppServer,
|
||||
environment_id: &str,
|
||||
exec_server_url: &str,
|
||||
) -> Result<()> {
|
||||
let params = EnvironmentAddParams {
|
||||
environment_id: environment_id.to_string(),
|
||||
exec_server_url: exec_server_url.to_string(),
|
||||
connect_timeout_ms: None,
|
||||
};
|
||||
let add_request_id = app_server
|
||||
.send_raw_request("environment/add", Some(serde_json::to_value(params)?))
|
||||
.await?;
|
||||
let add_response: JSONRPCResponse = timeout(
|
||||
RPC_TIMEOUT,
|
||||
app_server.read_stream_until_response_message(RequestId::Integer(add_request_id)),
|
||||
)
|
||||
.await??;
|
||||
let _: EnvironmentAddResponse = to_response(add_response)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn read_environment_status(
|
||||
app_server: &mut TestAppServer,
|
||||
environment_id: &str,
|
||||
) -> Result<EnvironmentStatusResponse> {
|
||||
let params = EnvironmentStatusParams {
|
||||
environment_id: environment_id.to_string(),
|
||||
};
|
||||
let request_id = app_server
|
||||
.send_raw_request("environment/status", Some(serde_json::to_value(params)?))
|
||||
.await?;
|
||||
let response: JSONRPCResponse = timeout(
|
||||
RPC_TIMEOUT,
|
||||
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
|
||||
)
|
||||
.await??;
|
||||
to_response(response)
|
||||
}
|
||||
|
||||
async fn wait_for_status(
|
||||
app_server: &mut TestAppServer,
|
||||
environment_id: &str,
|
||||
expected: EnvironmentStatusKind,
|
||||
) -> Result<EnvironmentStatusResponse> {
|
||||
timeout(RPC_TIMEOUT, async {
|
||||
loop {
|
||||
let response = read_environment_status(app_server, environment_id).await?;
|
||||
if response.status == expected {
|
||||
return Ok(response);
|
||||
}
|
||||
sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
})
|
||||
.await?
|
||||
}
|
||||
@@ -16,6 +16,7 @@ mod current_time;
|
||||
mod dynamic_tools;
|
||||
mod environment_add;
|
||||
mod environment_info;
|
||||
mod environment_status;
|
||||
mod exec_server_test_support;
|
||||
#[cfg(not(target_os = "windows"))]
|
||||
mod executor_mcp;
|
||||
|
||||
@@ -297,19 +297,6 @@ impl EnvironmentManager {
|
||||
environment_ids
|
||||
}
|
||||
|
||||
/// Returns every configured environment id in deterministic order.
|
||||
pub fn environment_ids(&self) -> Vec<String> {
|
||||
let mut environment_ids = self
|
||||
.environments
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.keys()
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
environment_ids.sort_unstable();
|
||||
environment_ids
|
||||
}
|
||||
|
||||
/// Returns the local environment instance when one is configured.
|
||||
pub fn try_local_environment(&self) -> Option<Arc<Environment>> {
|
||||
self.local_environment.as_ref().map(Arc::clone)
|
||||
|
||||
Reference in New Issue
Block a user