mirror of
https://github.com/openai/codex.git
synced 2026-11-11 21:10:39 +00:00
Add an explicit app-server daemon update command (#43562)
## What changed Add `codex app-server daemon update` to check the latest stable release once, even when automatic updates are disabled. The command requires an installer-owned latest-channel standalone installation and reports `updated`, `noUpdate`, or `unsupported` as JSON, including installed and running versions. Route manual requests through the updater so scheduled and manual installs share one owner and accepted updates continue if the CLI exits. Restart a running managed daemon when its binary or version differs from the selected installation; this may interrupt active or queued work. Return installer failures as command errors and retry requests across updater replacement. ## Testing Add coverage for manual updates with automatic updates disabled, queued requests, same-version binary replacement, updater handoff retries, and unsupported installations or unmanaged servers. GitOrigin-RevId: 240b63c79bae3a64a80768131a30da70900a4d62
This commit is contained in:
Eric Traut
@@ -3,41 +3,26 @@ use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
|
||||
use pretty_assertions::assert_eq;
|
||||
#[cfg(unix)]
|
||||
use tempfile::TempDir;
|
||||
|
||||
use super::INSTALL_URL;
|
||||
use super::InstallerHttp;
|
||||
use super::InstallerResponse;
|
||||
use super::fetch_installer_script;
|
||||
use super::update_modes_for_identities;
|
||||
use crate::RestartMode;
|
||||
use crate::UpdaterRefreshMode;
|
||||
#[cfg(unix)]
|
||||
use super::manual_update::run as manual_update_once;
|
||||
#[cfg(unix)]
|
||||
use crate::Daemon;
|
||||
#[cfg(unix)]
|
||||
use crate::UpdateOutput;
|
||||
#[cfg(unix)]
|
||||
use crate::UpdateStatus;
|
||||
#[cfg(unix)]
|
||||
use crate::managed_install::executable_identity;
|
||||
#[cfg(unix)]
|
||||
use crate::managed_install::executable_identity_from_bytes;
|
||||
|
||||
#[test]
|
||||
fn unchanged_updater_uses_version_based_restart() {
|
||||
assert_eq!(
|
||||
update_modes_for_identities(
|
||||
&executable_identity_from_bytes(b"same"),
|
||||
&executable_identity_from_bytes(b"same"),
|
||||
),
|
||||
(RestartMode::IfVersionChanged, UpdaterRefreshMode::None)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn changed_updater_forces_refresh_even_when_version_may_match() {
|
||||
assert_eq!(
|
||||
update_modes_for_identities(
|
||||
&executable_identity_from_bytes(b"old"),
|
||||
&executable_identity_from_bytes(b"new"),
|
||||
),
|
||||
(
|
||||
RestartMode::Always,
|
||||
UpdaterRefreshMode::ReexecIfManagedBinaryChanged,
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn installer_fetch_uses_exact_url_and_preserves_bytes() {
|
||||
let script = b"#!/bin/sh\nprintf 'update bytes'\n".to_vec();
|
||||
@@ -137,6 +122,369 @@ async fn cancelling_installer_stops_children_and_releases_fallback_lock() {
|
||||
assert!(!delayed.exists());
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn manual_update_daemon(home: &TempDir) -> (Daemon, String) {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
|
||||
let target = if cfg!(target_os = "macos") {
|
||||
format!("{}-apple-darwin", std::env::consts::ARCH)
|
||||
} else {
|
||||
format!("{}-unknown-linux-musl", std::env::consts::ARCH)
|
||||
};
|
||||
let release = format!("1.0.0-{target}");
|
||||
let standalone = home.path().join("packages/standalone");
|
||||
let bin = standalone.join("releases").join(&release).join("codex");
|
||||
std::fs::create_dir_all(bin.parent().expect("binary parent")).expect("release directory");
|
||||
std::fs::write(
|
||||
&bin,
|
||||
b"#!/bin/sh\nif [ \"$1\" = '--version' ]; then echo codex 1.0.0; else exec sleep 30; fi\n",
|
||||
)
|
||||
.expect("managed binary");
|
||||
std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755))
|
||||
.expect("executable binary");
|
||||
std::os::unix::fs::symlink(format!("releases/{release}"), standalone.join("current"))
|
||||
.expect("current release");
|
||||
std::fs::write(standalone.join("auto-update-version"), &release).expect("latest marker");
|
||||
let state = home.path().join("app-server-daemon");
|
||||
(
|
||||
Daemon {
|
||||
socket_path: home.path().join("app-server-control/server.sock"),
|
||||
pid_file: state.join("app-server.pid"),
|
||||
update_pid_file: state.join("app-server-updater.pid"),
|
||||
operation_lock_file: state.join("daemon.lock"),
|
||||
settings_file: state.join("settings.json"),
|
||||
managed_codex_bin: standalone.join("current/codex"),
|
||||
},
|
||||
release,
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn test_terminate() -> tokio::signal::unix::Signal {
|
||||
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
||||
.expect("install test signal handler")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn manual_request_retries_after_updater_replacement() {
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let home = TempDir::new().expect("home");
|
||||
let (daemon, _) = manual_update_daemon(&home);
|
||||
let socket_path = daemon.manual_update_socket_path();
|
||||
codex_uds::prepare_private_socket_directory(socket_path.parent().expect("socket parent"))
|
||||
.await
|
||||
.expect("socket directory");
|
||||
let mut listener = codex_uds::UnixListener::bind(&socket_path)
|
||||
.await
|
||||
.expect("old updater socket");
|
||||
let expected = UpdateOutput {
|
||||
status: UpdateStatus::NoUpdate,
|
||||
managed_codex_path: daemon.managed_codex_bin.clone(),
|
||||
installed_version: None,
|
||||
running_version: None,
|
||||
message: "already current".to_string(),
|
||||
};
|
||||
let reply = expected.clone();
|
||||
let server = tokio::spawn(async move {
|
||||
let mut old = listener.accept().await.expect("first connection");
|
||||
let mut request = [0; 7];
|
||||
old.read_exact(&mut request).await.expect("first request");
|
||||
drop(old);
|
||||
drop(listener);
|
||||
tokio::fs::remove_file(&socket_path)
|
||||
.await
|
||||
.expect("remove old socket");
|
||||
let mut successor = codex_uds::UnixListener::bind(&socket_path)
|
||||
.await
|
||||
.expect("successor socket");
|
||||
let mut connection = successor.accept().await.expect("retried connection");
|
||||
connection
|
||||
.read_exact(&mut request)
|
||||
.await
|
||||
.expect("retried request");
|
||||
connection
|
||||
.write_all(&serde_json::to_vec(&Ok::<_, String>(reply)).expect("serialize response"))
|
||||
.await
|
||||
.expect("send response");
|
||||
});
|
||||
assert_eq!(
|
||||
super::manual_update::request(&daemon)
|
||||
.await
|
||||
.expect("request survives handoff"),
|
||||
expected
|
||||
);
|
||||
server.await.expect("replacement task");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn manual_request_recovers_when_one_shot_updater_exits() {
|
||||
use tokio::io::AsyncReadExt;
|
||||
|
||||
let home = TempDir::new().expect("home");
|
||||
let (daemon, _) = manual_update_daemon(&home);
|
||||
let socket_path = daemon.manual_update_socket_path();
|
||||
codex_uds::prepare_private_socket_directory(socket_path.parent().expect("socket parent"))
|
||||
.await
|
||||
.expect("socket directory");
|
||||
let mut listener = codex_uds::UnixListener::bind(&socket_path)
|
||||
.await
|
||||
.expect("one-shot updater socket");
|
||||
let server = tokio::spawn(async move {
|
||||
let mut connection = listener.accept().await.expect("request connection");
|
||||
let mut request = [0; 7];
|
||||
connection.read_exact(&mut request).await.expect("request");
|
||||
drop(connection);
|
||||
drop(listener);
|
||||
tokio::fs::remove_file(socket_path)
|
||||
.await
|
||||
.expect("remove exited updater socket");
|
||||
});
|
||||
// Without the marker, the ordinary startup path reports unsupported. A
|
||||
// retry that only waits for a successor would time out instead.
|
||||
std::fs::remove_file(home.path().join("packages/standalone/auto-update-version"))
|
||||
.expect("remove latest marker");
|
||||
let result = tokio::time::timeout(
|
||||
Duration::from_secs(5),
|
||||
super::manual_update::request(&daemon),
|
||||
)
|
||||
.await
|
||||
.expect("retry should return to normal startup")
|
||||
.expect("unsupported response");
|
||||
assert_eq!(result.status, UpdateStatus::Unsupported);
|
||||
server.await.expect("updater task");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn unsupported_request_preserves_updater_schedule() {
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let home = TempDir::new().expect("home");
|
||||
let (daemon, _) = manual_update_daemon(&home);
|
||||
let daemon = std::sync::Arc::new(daemon);
|
||||
let identity = executable_identity(&daemon.managed_codex_bin)
|
||||
.await
|
||||
.expect("updater identity");
|
||||
let socket_path = daemon.manual_update_socket_path();
|
||||
let http = FakeInstallerHttp::new(InstallerResponse::Success(Vec::new()));
|
||||
let updater_daemon = std::sync::Arc::clone(&daemon);
|
||||
let worker =
|
||||
tokio::spawn(async move { super::run_with_http(&http, &updater_daemon, &identity).await });
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
|
||||
while !socket_path.exists() {
|
||||
assert!(
|
||||
tokio::time::Instant::now() < deadline,
|
||||
"updater did not listen"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
std::fs::remove_file(home.path().join("packages/standalone/auto-update-version"))
|
||||
.expect("remove latest selection");
|
||||
let mut malformed = codex_uds::UnixStream::connect(&socket_path)
|
||||
.await
|
||||
.expect("connect malformed request");
|
||||
malformed
|
||||
.write_all(b"upd")
|
||||
.await
|
||||
.expect("send partial request");
|
||||
malformed.shutdown().await.expect("disconnect request");
|
||||
let mut discarded = Vec::new();
|
||||
malformed
|
||||
.read_to_end(&mut discarded)
|
||||
.await
|
||||
.expect("rejected request");
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
assert!(!worker.is_finished(), "malformed request stopped updater");
|
||||
assert_eq!(
|
||||
super::manual_update::request(&daemon)
|
||||
.await
|
||||
.expect("unsupported response")
|
||||
.status,
|
||||
UpdateStatus::Unsupported
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
assert!(!worker.is_finished(), "unsupported request stopped updater");
|
||||
worker.abort();
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn manual_update_restarts_managed_daemon_with_automatic_updates_disabled() {
|
||||
use futures::SinkExt;
|
||||
use futures::StreamExt;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let home = TempDir::new().expect("home");
|
||||
let (daemon, release) = manual_update_daemon(&home);
|
||||
let daemon = std::sync::Arc::new(daemon);
|
||||
std::fs::create_dir_all(daemon.settings_file.parent().expect("state directory"))
|
||||
.expect("state directory");
|
||||
std::fs::write(
|
||||
&daemon.settings_file,
|
||||
r#"{"updater":{"autoUpdateEnabled":false}}"#,
|
||||
)
|
||||
.expect("disabled updater");
|
||||
std::fs::create_dir_all(daemon.socket_path.parent().expect("socket parent"))
|
||||
.expect("socket directory");
|
||||
let mut listener = codex_uds::UnixListener::bind(&daemon.socket_path)
|
||||
.await
|
||||
.expect("control listener");
|
||||
let codex_home = home.path().to_path_buf();
|
||||
let server = tokio::spawn(async move {
|
||||
loop {
|
||||
let connection = listener.accept().await.expect("control connection");
|
||||
let mut websocket = tokio_tungstenite::accept_async(connection)
|
||||
.await
|
||||
.expect("websocket handshake");
|
||||
websocket
|
||||
.next()
|
||||
.await
|
||||
.expect("initialize request")
|
||||
.expect("frame");
|
||||
let version = if std::fs::read_to_string(
|
||||
codex_home.join("packages/standalone/auto-update-version"),
|
||||
)
|
||||
.expect("selected release")
|
||||
.starts_with("1.1.0")
|
||||
{
|
||||
"1.1.0"
|
||||
} else {
|
||||
"1.0.0"
|
||||
};
|
||||
websocket.send(tokio_tungstenite::tungstenite::Message::Text(
|
||||
serde_json::json!({"id": 1, "result": {
|
||||
"userAgent": format!("codex_app_server_daemon/{version}"),
|
||||
"codexHome": codex_home, "platformFamily": "unix", "platformOs": std::env::consts::OS,
|
||||
}}).to_string().into(),
|
||||
)).await.expect("initialize response");
|
||||
websocket
|
||||
.next()
|
||||
.await
|
||||
.expect("initialized notification")
|
||||
.expect("frame");
|
||||
}
|
||||
});
|
||||
let settings = crate::settings::DaemonSettings::default();
|
||||
let backend = crate::backend::pid_backend(daemon.backend_paths(&settings));
|
||||
backend.start().await.expect("start daemon");
|
||||
let current_pid = || {
|
||||
serde_json::from_slice::<serde_json::Value>(
|
||||
&std::fs::read(&daemon.pid_file).expect("daemon PID record"),
|
||||
)
|
||||
.expect("PID JSON")["pid"]
|
||||
.as_u64()
|
||||
.expect("PID")
|
||||
};
|
||||
let before = current_pid();
|
||||
let next = release.replacen("1.0.0", "1.1.0", 1);
|
||||
let standalone = home.path().join("packages/standalone");
|
||||
let ready = home.path().join("installer-ready");
|
||||
let proceed = home.path().join("installer-proceed");
|
||||
let script = format!(
|
||||
"#!/bin/sh\n# CODEX_INSTALL_IF_LATEST\ntest \"$CODEX_INSTALL_IF_LATEST\" = 1 || exit 4\nif [ \"$CODEX_UPDATE_FROM_RELEASE\" = '{next}' ]; then exit 0; fi\ntest \"$CODEX_UPDATE_FROM_RELEASE\" = '{release}' || exit 5\ntouch '{ready}'\nwhile [ ! -e '{proceed}' ]; do sleep .05; done\nmkdir -p '{root}/releases/{next}/bin'\nprintf '#!/bin/sh\\nif [ \"$1\" = --version ]; then echo codex 1.1.0; else exec sleep 30; fi\\n' > '{root}/releases/{next}/bin/codex'\nchmod +x '{root}/releases/{next}/bin/codex'\nln -sfn 'releases/{next}' '{root}/current'\nprintf '{next}' > '{root}/auto-update-version'\n",
|
||||
root = standalone.display(),
|
||||
ready = ready.display(),
|
||||
proceed = proceed.display(),
|
||||
);
|
||||
let http = FakeInstallerHttp::new(InstallerResponse::Success(script.into_bytes()));
|
||||
let updater_daemon = std::sync::Arc::clone(&daemon);
|
||||
let worker = tokio::spawn(async move {
|
||||
super::run_with_http(
|
||||
&http,
|
||||
&updater_daemon,
|
||||
&executable_identity_from_bytes(b"updater"),
|
||||
)
|
||||
.await
|
||||
});
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
|
||||
while !daemon.manual_update_socket_path().exists() {
|
||||
assert!(
|
||||
tokio::time::Instant::now() < deadline,
|
||||
"updater did not listen"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
let request_daemon = std::sync::Arc::clone(&daemon);
|
||||
let request = tokio::spawn(async move { super::manual_update::request(&request_daemon).await });
|
||||
while !ready.exists() {
|
||||
assert!(
|
||||
tokio::time::Instant::now() < deadline,
|
||||
"installer did not start"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
let mut queued = codex_uds::UnixStream::connect(&daemon.manual_update_socket_path())
|
||||
.await
|
||||
.expect("queue a second update");
|
||||
queued
|
||||
.write_all(b"update\n")
|
||||
.await
|
||||
.expect("send second update");
|
||||
std::fs::write(proceed, b"go").expect("release installer");
|
||||
let output = request
|
||||
.await
|
||||
.expect("first request task")
|
||||
.expect("manual update");
|
||||
assert_eq!(output.status, UpdateStatus::Updated);
|
||||
assert_eq!(output.installed_version.as_deref(), Some("1.1.0"));
|
||||
assert_eq!(output.running_version.as_deref(), Some("1.1.0"));
|
||||
assert_eq!(
|
||||
output.managed_codex_path,
|
||||
standalone.join("current/bin/codex")
|
||||
);
|
||||
let restarted = current_pid();
|
||||
assert_ne!(restarted, before);
|
||||
let mut response = Vec::new();
|
||||
queued
|
||||
.read_to_end(&mut response)
|
||||
.await
|
||||
.expect("second response");
|
||||
let second: Result<crate::UpdateOutput, String> =
|
||||
serde_json::from_slice(&response).expect("valid second response");
|
||||
assert_eq!(
|
||||
second.expect("second update").status,
|
||||
UpdateStatus::NoUpdate
|
||||
);
|
||||
assert_eq!(current_pid(), restarted);
|
||||
tokio::time::timeout(Duration::from_secs(5), worker)
|
||||
.await
|
||||
.expect("one-shot updater did not exit")
|
||||
.expect("updater task")
|
||||
.expect("updater loop");
|
||||
let no_op = FakeInstallerHttp::new(InstallerResponse::Success(
|
||||
b"#!/bin/sh\n# CODEX_INSTALL_IF_LATEST\nexit 0\n".to_vec(),
|
||||
));
|
||||
use std::io::Write;
|
||||
std::fs::OpenOptions::new()
|
||||
.append(true)
|
||||
.open(
|
||||
daemon
|
||||
.current_managed_codex_bin()
|
||||
.expect("current executable"),
|
||||
)
|
||||
.expect("managed executable")
|
||||
.write_all(b"\n# same-version replacement\n")
|
||||
.expect("replace binary bytes");
|
||||
let output = manual_update_once(
|
||||
&no_op,
|
||||
&daemon,
|
||||
&executable_identity_from_bytes(b"updater"),
|
||||
&mut test_terminate(),
|
||||
)
|
||||
.await
|
||||
.expect("retry with stale running binary");
|
||||
assert_eq!(output.status, UpdateStatus::NoUpdate);
|
||||
assert_ne!(current_pid(), restarted);
|
||||
backend.stop().await.expect("stop daemon");
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
#[tokio::test]
|
||||
async fn powershell_installer_is_noninteractive_and_reports_script_failure() {
|
||||
|
||||
Reference in New Issue
Block a user