clone/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-05-23 12:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(app-server): add runtime install progress

Browse Source
This commit is contained in:
acrognale-oai
2026-05-21 17:12:42 -04:00
parent 0b0d49f68e
commit c161795a10
21 changed files with 919 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
codex-rs/app-server-protocol/schema/json/ClientRequest.json generated
View File

@@ -5409,6 +5409,29 @@
"title": "Runtime/installRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"runtime/install/cancel"
],
"title": "Runtime/install/cancelRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Runtime/install/cancelRequest",
"type": "object"
},
{
"properties": {
"id": {

65
codex-rs/app-server-protocol/schema/json/ServerNotification.json generated
View File

@@ -2946,6 +2946,51 @@
},
"type": "object"
},
"RuntimeInstallProgressNotification": {
"properties": {
"bundleVersion": {
"type": [
"string",
"null"
]
},
"downloadedBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"phase": {
"$ref": "#/definitions/RuntimeInstallProgressPhase"
},
"totalBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
}
},
"required": [
"phase"
],
"type": "object"
},
"RuntimeInstallProgressPhase": {
"enum": [
"checking",
"downloading",
"verifying",
"extracting",
"validating",
"installed",
"configuring"
],
"type": "string"
},
"SandboxPolicy": {
"oneOf": [
{
@@ -5397,6 +5442,26 @@
"title": "Skills/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"runtime/install/progress"
],
"title": "Runtime/install/progressNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/RuntimeInstallProgressNotification"
}
},
"required": [
"method",
"params"
],
"title": "Runtime/install/progressNotification",
"type": "object"
},
{
"properties": {
"method": {

110
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json generated
View File

@@ -1333,6 +1333,29 @@
"title": "Runtime/installRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/v2/RequestId"
},
"method": {
"enum": [
"runtime/install/cancel"
],
"title": "Runtime/install/cancelRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Runtime/install/cancelRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -4091,6 +4114,26 @@
"title": "Skills/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"runtime/install/progress"
],
"title": "Runtime/install/progressNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/RuntimeInstallProgressNotification"
}
},
"required": [
"method",
"params"
],
"title": "Runtime/install/progressNotification",
"type": "object"
},
{
"properties": {
"method": {
@@ -14637,6 +14680,26 @@
}
]
},
"RuntimeInstallCancelResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"status": {
"$ref": "#/definitions/v2/RuntimeInstallCancelStatus"
}
},
"required": [
"status"
],
"title": "RuntimeInstallCancelResponse",
"type": "object"
},
"RuntimeInstallCancelStatus": {
"enum": [
"canceled",
"not-found"
],
"type": "string"
},
"RuntimeInstallManifest": {
"properties": {
"archiveName": {
@@ -14755,6 +14818,53 @@
],
"type": "object"
},
"RuntimeInstallProgressNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"bundleVersion": {
"type": [
"string",
"null"
]
},
"downloadedBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"phase": {
"$ref": "#/definitions/v2/RuntimeInstallProgressPhase"
},
"totalBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
}
},
"required": [
"phase"
],
"title": "RuntimeInstallProgressNotification",
"type": "object"
},
"RuntimeInstallProgressPhase": {
"enum": [
"checking",
"downloading",
"verifying",
"extracting",
"validating",
"installed",
"configuring"
],
"type": "string"
},
"RuntimeInstallResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {

110
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json generated
View File

@@ -2059,6 +2059,29 @@
"title": "Runtime/installRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"runtime/install/cancel"
],
"title": "Runtime/install/cancelRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Runtime/install/cancelRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -11166,6 +11189,26 @@
}
]
},
"RuntimeInstallCancelResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"status": {
"$ref": "#/definitions/RuntimeInstallCancelStatus"
}
},
"required": [
"status"
],
"title": "RuntimeInstallCancelResponse",
"type": "object"
},
"RuntimeInstallCancelStatus": {
"enum": [
"canceled",
"not-found"
],
"type": "string"
},
"RuntimeInstallManifest": {
"properties": {
"archiveName": {
@@ -11284,6 +11327,53 @@
],
"type": "object"
},
"RuntimeInstallProgressNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"bundleVersion": {
"type": [
"string",
"null"
]
},
"downloadedBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"phase": {
"$ref": "#/definitions/RuntimeInstallProgressPhase"
},
"totalBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
}
},
"required": [
"phase"
],
"title": "RuntimeInstallProgressNotification",
"type": "object"
},
"RuntimeInstallProgressPhase": {
"enum": [
"checking",
"downloading",
"verifying",
"extracting",
"validating",
"installed",
"configuring"
],
"type": "string"
},
"RuntimeInstallResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -11616,6 +11706,26 @@
"title": "Skills/changedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"runtime/install/progress"
],
"title": "Runtime/install/progressNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/RuntimeInstallProgressNotification"
}
},
"required": [
"method",
"params"
],
"title": "Runtime/install/progressNotification",
"type": "object"
},
{
"properties": {
"method": {

22
codex-rs/app-server-protocol/schema/json/v2/RuntimeInstallCancelResponse.json generated Normal file
View File

@@ -0,0 +1,22 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"RuntimeInstallCancelStatus": {
"enum": [
"canceled",
"not-found"
],
"type": "string"
}
},
"properties": {
"status": {
"$ref": "#/definitions/RuntimeInstallCancelStatus"
}
},
"required": [
"status"
],
"title": "RuntimeInstallCancelResponse",
"type": "object"
}

49
codex-rs/app-server-protocol/schema/json/v2/RuntimeInstallProgressNotification.json generated Normal file
View File

@@ -0,0 +1,49 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"RuntimeInstallProgressPhase": {
"enum": [
"checking",
"downloading",
"verifying",
"extracting",
"validating",
"installed",
"configuring"
],
"type": "string"
}
},
"properties": {
"bundleVersion": {
"type": [
"string",
"null"
]
},
"downloadedBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"phase": {
"$ref": "#/definitions/RuntimeInstallProgressPhase"
},
"totalBytes": {
"format": "uint64",
"minimum": 0.0,
"type": [
"integer",
"null"
]
}
},
"required": [
"phase"
],
"title": "RuntimeInstallProgressNotification",
"type": "object"
}

2
codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts generated
View File

File diff suppressed because one or more lines are too long

3
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts generated
View File

File diff suppressed because one or more lines are too long

6
codex-rs/app-server-protocol/schema/typescript/v2/RuntimeInstallCancelResponse.ts generated Normal file
View File

@@ -0,0 +1,6 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { RuntimeInstallCancelStatus } from "./RuntimeInstallCancelStatus";
export type RuntimeInstallCancelResponse = { status: RuntimeInstallCancelStatus, };

5
codex-rs/app-server-protocol/schema/typescript/v2/RuntimeInstallCancelStatus.ts generated Normal file
View File

@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type RuntimeInstallCancelStatus = "canceled" | "not-found";

6
codex-rs/app-server-protocol/schema/typescript/v2/RuntimeInstallProgressNotification.ts generated Normal file
View File

@@ -0,0 +1,6 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { RuntimeInstallProgressPhase } from "./RuntimeInstallProgressPhase";
export type RuntimeInstallProgressNotification = { bundleVersion: string | null, downloadedBytes: bigint | null, phase: RuntimeInstallProgressPhase, totalBytes: bigint | null, };

5
codex-rs/app-server-protocol/schema/typescript/v2/RuntimeInstallProgressPhase.ts generated Normal file
View File

@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type RuntimeInstallProgressPhase = "checking" | "downloading" | "verifying" | "extracting" | "validating" | "installed" | "configuring";

4
codex-rs/app-server-protocol/schema/typescript/v2/index.ts generated
View File

@@ -325,9 +325,13 @@ export type { ReviewDelivery } from "./ReviewDelivery";
export type { ReviewStartParams } from "./ReviewStartParams";
export type { ReviewStartResponse } from "./ReviewStartResponse";
export type { ReviewTarget } from "./ReviewTarget";
export type { RuntimeInstallCancelResponse } from "./RuntimeInstallCancelResponse";
export type { RuntimeInstallCancelStatus } from "./RuntimeInstallCancelStatus";
export type { RuntimeInstallManifest } from "./RuntimeInstallManifest";
export type { RuntimeInstallParams } from "./RuntimeInstallParams";
export type { RuntimeInstallPaths } from "./RuntimeInstallPaths";
export type { RuntimeInstallProgressNotification } from "./RuntimeInstallProgressNotification";
export type { RuntimeInstallProgressPhase } from "./RuntimeInstallProgressPhase";
export type { RuntimeInstallResponse } from "./RuntimeInstallResponse";
export type { RuntimeInstallStatus } from "./RuntimeInstallStatus";
export type { SandboxMode } from "./SandboxMode";

6
codex-rs/app-server-protocol/src/protocol/common.rs
View File

@@ -742,6 +742,11 @@ client_request_definitions! {
serialization: global("runtime-install"),
response: v2::RuntimeInstallResponse,
},
RuntimeInstallCancel => "runtime/install/cancel" {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
serialization: None,
response: v2::RuntimeInstallCancelResponse,
},
PluginUninstall => "plugin/uninstall" {
params: v2::PluginUninstallParams,
serialization: global("config"),
@@ -1480,6 +1485,7 @@ server_notification_definitions! {
ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification),
ThreadClosed => "thread/closed" (v2::ThreadClosedNotification),
SkillsChanged => "skills/changed" (v2::SkillsChangedNotification),
RuntimeInstallProgress => "runtime/install/progress" (v2::RuntimeInstallProgressNotification),
ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification),
ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification),
ThreadGoalCleared => "thread/goal/cleared" (v2::ThreadGoalClearedNotification),

38
codex-rs/app-server-protocol/src/protocol/v2/runtime.rs
View File

@@ -28,6 +28,21 @@ pub struct RuntimeInstallParams {
pub release: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "kebab-case")]
#[ts(export_to = "v2/")]
pub enum RuntimeInstallCancelStatus {
Canceled,
NotFound,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct RuntimeInstallCancelResponse {
pub status: RuntimeInstallCancelStatus,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "kebab-case")]
#[ts(export_to = "v2/")]
@@ -56,3 +71,26 @@ pub struct RuntimeInstallResponse {
pub paths: RuntimeInstallPaths,
pub status: RuntimeInstallStatus,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "kebab-case")]
#[ts(export_to = "v2/")]
pub enum RuntimeInstallProgressPhase {
Checking,
Downloading,
Verifying,
Extracting,
Validating,
Installed,
Configuring,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct RuntimeInstallProgressNotification {
pub bundle_version: Option<String>,
pub downloaded_bytes: Option<u64>,
pub phase: RuntimeInstallProgressPhase,
pub total_bytes: Option<u64>,
}

3
codex-rs/app-server/README.md
View File

@@ -212,6 +212,9 @@ Example with notification opt-out:
- `remoteControl/status/changed` — notification emitted when the remote-control status or client-visible environment id changes. `status` is one of `disabled`, `connecting`, `connected`, or `errored`; `serverName` is the local machine name used by this app-server process; `environmentId` is a string when the app-server has a current enrollment and `null` when that enrollment is cleared, invalidated, or remote control is disabled. Newly initialized app-server clients always receive the current status snapshot.
- `skills/config/write` — write user-level skill config by name or absolute path.
- `plugin/install` — install a plugin from a discovered marketplace entry, rejecting marketplace entries marked unavailable for install, install MCPs if any, and return the effective plugin auth policy plus any apps that still need auth (**under development; do not call from production clients yet**).
- `runtime/install` — install the selected primary runtime bundle and return its executable, dependency, bundled-skill, and bundled-marketplace paths after app-server config synchronization. Runtime installs are process-wide and serialized.
- `runtime/install/progress` — notification sent to the connection that requested `runtime/install` as the active install moves through checking, downloading, verifying, extracting, validating, installed, and configuring phases. Download notifications include byte counts when available.
- `runtime/install/cancel` — cancel the one active `runtime/install` request for this app-server process and report whether an install was found.
- `plugin/uninstall` — uninstall a local plugin by `pluginId` in `<plugin>@<marketplace>` form by removing its cached files and clearing its user-level config entry, or uninstall a remote ChatGPT plugin by backend `pluginId` by forwarding the uninstall to the ChatGPT plugin backend and removing any downloaded remote-plugin cache (**under development; do not call from production clients yet**).
- `mcpServer/oauth/login` — start an OAuth login for a configured MCP server; returns an `authorization_url` and later emits `mcpServer/oauthLogin/completed` once the browser flow finishes.
- `tool/requestUserInput` — prompt the user with 1–3 short questions for a tool call and return their answers (experimental).

44
codex-rs/app-server/src/message_processor.rs
View File

@@ -7,7 +7,6 @@ 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;
@@ -32,6 +31,7 @@ use crate::request_processors::McpRequestProcessor;
use crate::request_processors::PluginRequestProcessor;
use crate::request_processors::ProcessExecRequestProcessor;
use crate::request_processors::RemoteControlRequestProcessor;
use crate::request_processors::RuntimeInstallRequestProcessor;
use crate::request_processors::SearchRequestProcessor;
use crate::request_processors::ThreadGoalRequestProcessor;
use crate::request_processors::ThreadRequestProcessor;
@@ -168,7 +168,6 @@ 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,
@@ -179,8 +178,8 @@ pub(crate) struct MessageProcessor {
mcp_processor: McpRequestProcessor,
plugin_processor: PluginRequestProcessor,
remote_control_processor: RemoteControlRequestProcessor,
runtime_install_processor: RuntimeInstallRequestProcessor,
search_processor: SearchRequestProcessor,
thread_manager: Arc<ThreadManager>,
thread_goal_processor: ThreadGoalRequestProcessor,
thread_processor: ThreadRequestProcessor,
turn_processor: TurnRequestProcessor,
@@ -405,6 +404,11 @@ impl MessageProcessor {
workspace_settings_cache,
);
let remote_control_processor = RemoteControlRequestProcessor::new(remote_control_handle);
let runtime_install_processor = RuntimeInstallRequestProcessor::new(
Arc::clone(&environment_manager_for_requests),
outgoing.clone(),
Arc::clone(&thread_manager),
);
let search_processor = SearchRequestProcessor::new(outgoing.clone());
let thread_goal_processor = ThreadGoalRequestProcessor::new(
Arc::clone(&thread_manager),
@@ -491,7 +495,6 @@ impl MessageProcessor {
command_exec_processor,
process_exec_processor,
config_processor,
environment_manager: thread_manager.environment_manager(),
environment_processor,
external_agent_config_processor,
feedback_processor,
@@ -502,8 +505,8 @@ impl MessageProcessor {
mcp_processor,
plugin_processor,
remote_control_processor,
runtime_install_processor,
search_processor,
thread_manager,
thread_goal_processor,
thread_processor,
turn_processor,
@@ -974,29 +977,14 @@ impl MessageProcessor {
.await
.map(|response| Some(response.into())),
ClientRequest::RuntimeInstall { params, .. } => {
let mut params = params;
let environment = if let Some(environment_id) = params.environment_id.take() {
self.environment_manager
.get_environment(&environment_id)
.ok_or_else(|| {
invalid_request(format!(
"unknown runtime install environment id `{environment_id}`"
))
})?
} else {
self.environment_manager
.default_or_local_environment()
.ok_or_else(|| {
internal_error("runtime install environment is not configured")
})?
};
let response = environment.install_runtime(params).await?;
let response =
crate::runtime_install::finalize_runtime_install(&environment, response)
.await?;
self.thread_manager.plugins_manager().clear_cache();
self.thread_manager.skills_manager().clear_cache();
Ok(Some(response.into()))
self.runtime_install_processor
.install_runtime(connection_id, params)
.await
}
ClientRequest::RuntimeInstallCancel { .. } => {
self.runtime_install_processor
.cancel_runtime_install()
.await
}
ClientRequest::ThreadStart { params, .. } => {
self.thread_processor

2
codex-rs/app-server/src/request_processors.rs
View File

@@ -466,6 +466,7 @@ mod mcp_processor;
mod plugins;
mod process_exec_processor;
mod remote_control_processor;
mod runtime_install_processor;
mod search;
mod thread_processor;
mod token_usage_replay;
@@ -488,6 +489,7 @@ pub(crate) use mcp_processor::McpRequestProcessor;
pub(crate) use plugins::PluginRequestProcessor;
pub(crate) use process_exec_processor::ProcessExecRequestProcessor;
pub(crate) use remote_control_processor::RemoteControlRequestProcessor;
pub(crate) use runtime_install_processor::RuntimeInstallRequestProcessor;
pub(crate) use search::SearchRequestProcessor;
pub(crate) use thread_goal_processor::ThreadGoalRequestProcessor;
pub(crate) use thread_processor::ThreadRequestProcessor;

152
codex-rs/app-server/src/request_processors/runtime_install_processor.rs Normal file
View File

@@ -0,0 +1,152 @@
use super::*;
use codex_app_server_protocol::RuntimeInstallCancelResponse;
use codex_app_server_protocol::RuntimeInstallCancelStatus;
use codex_app_server_protocol::RuntimeInstallParams;
use codex_app_server_protocol::RuntimeInstallProgressNotification;
use codex_app_server_protocol::RuntimeInstallProgressPhase;
use std::sync::Mutex as StdMutex;
#[derive(Clone)]
pub(crate) struct RuntimeInstallRequestProcessor {
environment_manager: Arc<EnvironmentManager>,
outgoing: Arc<OutgoingMessageSender>,
thread_manager: Arc<ThreadManager>,
active_install: Arc<StdMutex<Option<CancellationToken>>>,
}
impl RuntimeInstallRequestProcessor {
pub(crate) fn new(
environment_manager: Arc<EnvironmentManager>,
outgoing: Arc<OutgoingMessageSender>,
thread_manager: Arc<ThreadManager>,
) -> Self {
Self {
environment_manager,
outgoing,
thread_manager,
active_install: Arc::new(StdMutex::new(None)),
}
}
pub(crate) async fn install_runtime(
&self,
connection_id: ConnectionId,
mut params: RuntimeInstallParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let (cancellation, _active_install) = self.begin_install()?;
let environment = if let Some(environment_id) = params.environment_id.take() {
self.environment_manager
.get_environment(&environment_id)
.ok_or_else(|| {
invalid_request(format!(
"unknown runtime install environment id `{environment_id}`"
))
})?
} else {
self.environment_manager
.default_or_local_environment()
.ok_or_else(|| internal_error("runtime install environment is not configured"))?
};
let (progress_tx, mut progress_rx) = tokio::sync::mpsc::unbounded_channel();
let outgoing = Arc::clone(&self.outgoing);
let progress_forwarder = tokio::spawn(async move {
while let Some(progress) = progress_rx.recv().await {
outgoing
.send_server_notification_to_connections(
&[connection_id],
ServerNotification::RuntimeInstallProgress(progress),
)
.await;
}
});
let install_result = environment
.install_runtime_with_progress(params, progress_tx, cancellation)
.await;
if let Err(error) = progress_forwarder.await {
warn!("runtime install progress forwarder failed: {error}");
}
let response = install_result?;
self.send_progress(
connection_id,
RuntimeInstallProgressNotification {
bundle_version: response.bundle_version.clone(),
downloaded_bytes: None,
phase: RuntimeInstallProgressPhase::Configuring,
total_bytes: None,
},
)
.await;
let response =
crate::runtime_install::finalize_runtime_install(&environment, response).await?;
self.thread_manager.plugins_manager().clear_cache();
self.thread_manager.skills_manager().clear_cache();
Ok(Some(response.into()))
}
pub(crate) async fn cancel_runtime_install(
&self,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let status = {
let active_install = self.active_install();
match active_install.as_ref() {
Some(cancellation) => {
cancellation.cancel();
RuntimeInstallCancelStatus::Canceled
}
None => RuntimeInstallCancelStatus::NotFound,
}
};
Ok(Some(RuntimeInstallCancelResponse { status }.into()))
}
fn begin_install(&self) -> Result<(CancellationToken, ActiveInstallGuard), JSONRPCErrorError> {
let cancellation = CancellationToken::new();
let mut active_install = self.active_install();
if active_install.is_some() {
return Err(invalid_request("runtime install is already in progress"));
}
*active_install = Some(cancellation.clone());
drop(active_install);
Ok((
cancellation,
ActiveInstallGuard {
active_install: Arc::clone(&self.active_install),
},
))
}
fn active_install(&self) -> std::sync::MutexGuard<'_, Option<CancellationToken>> {
self.active_install
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
async fn send_progress(
&self,
connection_id: ConnectionId,
progress: RuntimeInstallProgressNotification,
) {
self.outgoing
.send_server_notification_to_connections(
&[connection_id],
ServerNotification::RuntimeInstallProgress(progress),
)
.await;
}
}
struct ActiveInstallGuard {
active_install: Arc<StdMutex<Option<CancellationToken>>>,
}
impl Drop for ActiveInstallGuard {
fn drop(&mut self) {
self.active_install
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
}
}

44
codex-rs/exec-server/src/environment.rs
View File

@@ -6,6 +6,7 @@ use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RuntimeInstallParams;
use codex_app_server_protocol::RuntimeInstallResponse;
use codex_utils_absolute_path::AbsolutePathBuf;
use tokio_util::sync::CancellationToken;
use crate::ExecServerError;
use crate::ExecServerRuntimePaths;
@@ -332,15 +333,37 @@ impl RuntimeInstaller {
async fn install_runtime(
&self,
params: RuntimeInstallParams,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
self.install_runtime_with_progress(params, /*progress*/ None, CancellationToken::new())
.await
}
async fn install_runtime_with_progress(
&self,
params: RuntimeInstallParams,
progress: Option<crate::runtime_install::RuntimeInstallProgressSender>,
cancellation: CancellationToken,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
match self {
RuntimeInstaller::Local => crate::runtime_install::install_runtime(params).await,
RuntimeInstaller::Local => match progress {
Some(progress) => {
crate::runtime_install::install_runtime_with_progress(
params,
progress,
cancellation,
)
.await
}
None => crate::runtime_install::install_runtime(params).await,
},
RuntimeInstaller::Remote(client) => {
let client = client.get().await.map_err(exec_server_error_to_jsonrpc)?;
client
.runtime_install(params)
.await
.map_err(exec_server_error_to_jsonrpc)
tokio::select! {
_ = cancellation.cancelled() => Err(internal_error("runtime install canceled")),
response = client.runtime_install(params) => {
response.map_err(exec_server_error_to_jsonrpc)
}
}
}
}
}
@@ -510,6 +533,17 @@ impl Environment {
self.runtime_installer.install_runtime(params).await
}
pub async fn install_runtime_with_progress(
&self,
params: RuntimeInstallParams,
progress: crate::runtime_install::RuntimeInstallProgressSender,
cancellation: CancellationToken,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
self.runtime_installer
.install_runtime_with_progress(params, Some(progress), cancellation)
.await
}
pub async fn codex_home(&self) -> Result<AbsolutePathBuf, JSONRPCErrorError> {
if let Some(codex_home) = self.codex_home.clone() {
return Ok(codex_home);

273
codex-rs/exec-server/src/runtime_install.rs
View File

@@ -8,6 +8,8 @@ use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RuntimeInstallManifest;
use codex_app_server_protocol::RuntimeInstallParams;
use codex_app_server_protocol::RuntimeInstallPaths;
use codex_app_server_protocol::RuntimeInstallProgressNotification;
use codex_app_server_protocol::RuntimeInstallProgressPhase;
use codex_app_server_protocol::RuntimeInstallResponse;
use codex_app_server_protocol::RuntimeInstallStatus;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -19,6 +21,8 @@ use tokio::fs;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use crate::rpc::internal_error;
use crate::rpc::invalid_params;
@@ -42,6 +46,54 @@ struct InstalledRuntimeMetadata {
skills_to_remove: Option<Vec<String>>,
}
pub type RuntimeInstallProgressSender = mpsc::UnboundedSender<RuntimeInstallProgressNotification>;
#[derive(Clone)]
struct RuntimeInstallProgressReporter {
bundle_version: Option<String>,
sender: Option<RuntimeInstallProgressSender>,
}
impl RuntimeInstallProgressReporter {
fn new(bundle_version: Option<String>, sender: Option<RuntimeInstallProgressSender>) -> Self {
Self {
bundle_version,
sender,
}
}
fn phase(&self, phase: RuntimeInstallProgressPhase) {
self.send(
phase, /*downloaded_bytes*/ None, /*total_bytes*/ None,
);
}
fn download_progress(&self, downloaded_bytes: u64, total_bytes: Option<u64>) {
self.send(
RuntimeInstallProgressPhase::Downloading,
Some(downloaded_bytes),
total_bytes,
);
}
fn send(
&self,
phase: RuntimeInstallProgressPhase,
downloaded_bytes: Option<u64>,
total_bytes: Option<u64>,
) {
let Some(sender) = self.sender.as_ref() else {
return;
};
let _ = sender.send(RuntimeInstallProgressNotification {
bundle_version: self.bundle_version.clone(),
downloaded_bytes,
phase,
total_bytes,
});
}
}
pub(crate) async fn install_runtime(
params: RuntimeInstallParams,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
@@ -49,9 +101,33 @@ pub(crate) async fn install_runtime(
install_runtime_with_root(params, install_root).await
}
pub(crate) async fn install_runtime_with_progress(
params: RuntimeInstallParams,
progress: RuntimeInstallProgressSender,
cancellation: CancellationToken,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
let install_root = default_install_root()?;
install_runtime_with_root_and_control(params, install_root, Some(progress), cancellation).await
}
async fn install_runtime_with_root(
params: RuntimeInstallParams,
install_root: PathBuf,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
install_runtime_with_root_and_control(
params,
install_root,
/*progress*/ None,
CancellationToken::new(),
)
.await
}
async fn install_runtime_with_root_and_control(
params: RuntimeInstallParams,
install_root: PathBuf,
progress: Option<RuntimeInstallProgressSender>,
cancellation: CancellationToken,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
validate_manifest(&params.manifest)?;
let archive_format = runtime_archive_format(&params.manifest)?;
@@ -62,11 +138,33 @@ async fn install_runtime_with_root(
.unwrap_or_else(|| default_archive_name(archive_format).to_string());
validate_path_segment(&archive_name, "archiveName")?;
let progress =
RuntimeInstallProgressReporter::new(params.manifest.bundle_version.clone(), progress);
progress.phase(RuntimeInstallProgressPhase::Checking);
ensure_not_cancelled(&cancellation)?;
let staging_dir = make_staging_dir(&install_root).await?;
let archive_path = staging_dir.join(archive_name);
let result = async {
download_archive(&params.manifest.archive_url, &archive_path).await?;
install_runtime_from_archive(&params.manifest, &archive_path, &install_root).await
progress.download_progress(
/*downloaded_bytes*/ 0,
params.manifest.archive_size_bytes,
);
download_archive(
&params.manifest.archive_url,
&archive_path,
params.manifest.archive_size_bytes,
&progress,
&cancellation,
)
.await?;
install_runtime_from_archive_with_control(
&params.manifest,
&archive_path,
&install_root,
&progress,
&cancellation,
)
.await
}
.await;
let cleanup_result = fs::remove_dir_all(&staging_dir).await;
@@ -81,15 +179,34 @@ async fn install_runtime_with_root(
result
}
#[cfg(test)]
async fn install_runtime_from_archive(
manifest: &RuntimeInstallManifest,
archive_path: &Path,
install_root: &Path,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
install_runtime_from_archive_with_control(
manifest,
archive_path,
install_root,
&RuntimeInstallProgressReporter::new(manifest.bundle_version.clone(), None),
&CancellationToken::new(),
)
.await
}
async fn install_runtime_from_archive_with_control(
manifest: &RuntimeInstallManifest,
archive_path: &Path,
install_root: &Path,
progress: &RuntimeInstallProgressReporter,
cancellation: &CancellationToken,
) -> Result<RuntimeInstallResponse, JSONRPCErrorError> {
let runtime_root_directory_name = runtime_root_directory_name(manifest)?;
let installed_runtime_root = install_root.join(&runtime_root_directory_name);
let target_platform = target_platform();
ensure_not_cancelled(cancellation)?;
if let Some(bundle_version) = manifest.bundle_version.as_ref()
&& let Ok(Some(metadata)) = read_installed_runtime_metadata(&installed_runtime_root).await
&& metadata.bundle_version.as_ref() == Some(bundle_version)
@@ -100,6 +217,7 @@ async fn install_runtime_from_archive(
)
.await
{
progress.phase(RuntimeInstallProgressPhase::Installed);
return Ok(RuntimeInstallResponse {
bundle_version: Some(bundle_version.clone()),
paths,
@@ -111,14 +229,17 @@ async fn install_runtime_from_archive(
.await
.map_err(|err| internal_error(format!("failed to create runtime install root: {err}")))?;
progress.phase(RuntimeInstallProgressPhase::Verifying);
verify_archive_checksum(
archive_path,
&manifest.archive_sha256,
&manifest.archive_url,
cancellation,
)
.await?;
let archive_format = runtime_archive_format(manifest)?;
ensure_not_cancelled(cancellation)?;
let staging_dir = make_staging_dir(install_root).await?;
let result = async {
let extract_dir = staging_dir.join("payload");
@@ -126,17 +247,23 @@ async fn install_runtime_from_archive(
internal_error(format!("failed to create runtime extract dir: {err}"))
})?;
progress.phase(RuntimeInstallProgressPhase::Extracting);
ensure_not_cancelled(cancellation)?;
let entries = list_archive_entries(archive_format, archive_path).await?;
assert_archive_entries_stay_within_directory(&entries, &extract_dir)?;
ensure_not_cancelled(cancellation)?;
extract_archive(archive_format, archive_path, &extract_dir).await?;
let extracted_runtime_root = extract_dir.join(&runtime_root_directory_name);
progress.phase(RuntimeInstallProgressPhase::Validating);
ensure_not_cancelled(cancellation)?;
validate_runtime_root(
&extracted_runtime_root,
manifest.bundle_format_version,
target_platform,
)
.await?;
ensure_not_cancelled(cancellation)?;
let previous_runtime_root =
install_root.join(format!("{runtime_root_directory_name}.previous"));
@@ -193,6 +320,9 @@ async fn install_runtime_from_archive(
staging_dir.display()
);
}
if result.is_ok() {
progress.phase(RuntimeInstallProgressPhase::Installed);
}
result
}
@@ -301,13 +431,21 @@ fn default_archive_name(format: RuntimeArchiveFormat) -> &'static str {
}
}
async fn download_archive(url: &str, destination: &Path) -> Result<(), JSONRPCErrorError> {
let response = reqwest::Client::new()
.get(url)
.header(reqwest::header::USER_AGENT, USER_AGENT)
.send()
.await
.map_err(|err| internal_error(format!("failed to download runtime archive: {err}")))?;
async fn download_archive(
url: &str,
destination: &Path,
expected_size_bytes: Option<u64>,
progress: &RuntimeInstallProgressReporter,
cancellation: &CancellationToken,
) -> Result<(), JSONRPCErrorError> {
let response = tokio::select! {
_ = cancellation.cancelled() => return Err(runtime_install_canceled()),
response = reqwest::Client::new()
.get(url)
.header(reqwest::header::USER_AGENT, USER_AGENT)
.send() => response
}
.map_err(|err| internal_error(format!("failed to download runtime archive: {err}")))?;
if !response.status().is_success() {
return Err(internal_error(format!(
"failed to download runtime archive ({} {})",
@@ -322,14 +460,27 @@ async fn download_archive(url: &str, destination: &Path) -> Result<(), JSONRPCEr
let mut file = fs::File::create(destination)
.await
.map_err(|err| internal_error(format!("failed to create runtime archive file: {err}")))?;
let total_bytes = response.content_length().or(expected_size_bytes);
let mut downloaded_bytes = 0_u64;
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
loop {
let chunk = tokio::select! {
_ = cancellation.cancelled() => return Err(runtime_install_canceled()),
chunk = stream.next() => chunk
};
let Some(chunk) = chunk else {
break;
};
let chunk = chunk.map_err(|err| {
internal_error(format!("failed to read runtime archive bytes: {err}"))
})?;
file.write_all(&chunk)
.await
.map_err(|err| internal_error(format!("failed to write runtime archive: {err}")))?;
tokio::select! {
_ = cancellation.cancelled() => return Err(runtime_install_canceled()),
result = file.write_all(&chunk) => result
}
.map_err(|err| internal_error(format!("failed to write runtime archive: {err}")))?;
downloaded_bytes += chunk.len() as u64;
progress.download_progress(downloaded_bytes, total_bytes);
}
file.flush()
.await
@@ -341,8 +492,9 @@ async fn verify_archive_checksum(
archive_path: &Path,
expected_sha256: &str,
source_url: &str,
cancellation: &CancellationToken,
) -> Result<(), JSONRPCErrorError> {
let actual_sha256 = compute_sha256(archive_path).await?;
let actual_sha256 = compute_sha256_with_cancellation(archive_path, cancellation).await?;
if !actual_sha256.eq_ignore_ascii_case(expected_sha256) {
return Err(invalid_params(format!(
"checksum mismatch for '{source_url}': expected {expected_sha256}, got {actual_sha256}"
@@ -351,17 +503,26 @@ async fn verify_archive_checksum(
Ok(())
}
#[cfg(test)]
async fn compute_sha256(path: &Path) -> Result<String, JSONRPCErrorError> {
compute_sha256_with_cancellation(path, &CancellationToken::new()).await
}
async fn compute_sha256_with_cancellation(
path: &Path,
cancellation: &CancellationToken,
) -> Result<String, JSONRPCErrorError> {
let mut file = fs::File::open(path)
.await
.map_err(|err| internal_error(format!("failed to open runtime archive: {err}")))?;
let mut digest = Sha256::new();
let mut buffer = [0_u8; 64 * 1024];
loop {
let bytes_read = file
.read(&mut buffer)
.await
.map_err(|err| internal_error(format!("failed to read runtime archive: {err}")))?;
let bytes_read = tokio::select! {
_ = cancellation.cancelled() => return Err(runtime_install_canceled()),
bytes_read = file.read(&mut buffer) => bytes_read
}
.map_err(|err| internal_error(format!("failed to read runtime archive: {err}")))?;
if bytes_read == 0 {
break;
}
@@ -370,6 +531,18 @@ async fn compute_sha256(path: &Path) -> Result<String, JSONRPCErrorError> {
Ok(format!("{:x}", digest.finalize()))
}
fn ensure_not_cancelled(cancellation: &CancellationToken) -> Result<(), JSONRPCErrorError> {
if cancellation.is_cancelled() {
Err(runtime_install_canceled())
} else {
Ok(())
}
}
fn runtime_install_canceled() -> JSONRPCErrorError {
internal_error("runtime install canceled")
}
async fn list_archive_entries(
format: RuntimeArchiveFormat,
archive_path: &Path,
@@ -955,6 +1128,70 @@ mod tests {
assert_eq!(metadata.bundle_version.as_deref(), Some("old"));
}
#[tokio::test]
async fn install_from_archive_reports_install_progress() {
let temp_dir = tempfile::tempdir().expect("tempdir");
let payload_root = temp_dir
.path()
.join("payload")
.join(PUBLISHED_ARTIFACT_NAME);
create_runtime_root(&payload_root, "v1").await;
let archive_path = temp_dir.path().join("archive.tar.xz");
create_tar_xz(temp_dir.path().join("payload").as_path(), &archive_path).await;
let manifest = manifest_for_archive(&archive_path, "v1").await;
let (progress_tx, mut progress_rx) = mpsc::unbounded_channel();
let progress =
RuntimeInstallProgressReporter::new(manifest.bundle_version.clone(), Some(progress_tx));
install_runtime_from_archive_with_control(
&manifest,
&archive_path,
&temp_dir.path().join("install"),
&progress,
&CancellationToken::new(),
)
.await
.expect("install should succeed");
let mut phases = Vec::new();
while let Ok(notification) = progress_rx.try_recv() {
phases.push(notification.phase);
}
assert_eq!(
phases,
vec![
RuntimeInstallProgressPhase::Verifying,
RuntimeInstallProgressPhase::Extracting,
RuntimeInstallProgressPhase::Validating,
RuntimeInstallProgressPhase::Installed,
]
);
}
#[tokio::test]
async fn install_from_archive_stops_when_canceled() {
let temp_dir = tempfile::tempdir().expect("tempdir");
let archive_path = temp_dir.path().join("unused.tar.xz");
fs::write(&archive_path, b"unused")
.await
.expect("write archive");
let manifest = manifest_for_archive(&archive_path, "v1").await;
let cancellation = CancellationToken::new();
cancellation.cancel();
let error = install_runtime_from_archive_with_control(
&manifest,
&archive_path,
&temp_dir.path().join("install"),
&RuntimeInstallProgressReporter::new(manifest.bundle_version.clone(), None),
&cancellation,
)
.await
.expect_err("canceled install should fail");
assert_eq!(error.message, "runtime install canceled");
}
async fn create_runtime_root(runtime_root: &Path, bundle_version: &str) {
let node_bin = runtime_root.join("dependencies").join("node").join("bin");
let python_bin = runtime_root.join("dependencies").join("python").join("bin");