mirror of
https://github.com/alphaXiv/OpenResearch.git
synced 2026-10-02 01:34:34 +08:00
Run SSH experiments in existing Docker containers (#407)
* feat: run SSH experiments in existing Docker containers * Simplify SSH container settings and harden launch recovery
This commit is contained in:
@@ -1,20 +1,47 @@
|
||||
# SSH (`--backend ssh`)
|
||||
|
||||
Use this backend only when the user explicitly asks to run on their own machine
|
||||
or server, or when it is the configured default. It runs a detached process on
|
||||
a host from `~/.ssh/config` using that host's environment.
|
||||
Use this backend when the user asks to run on their own server, or when SSH is
|
||||
the configured compute default. Authentication uses SSH config, keys and the
|
||||
agent; `orx` never reads private keys.
|
||||
|
||||
```sh
|
||||
orx exp run <expId> --backend ssh --host my-gpu-box
|
||||
orx exp run <expId> --backend ssh --host lab
|
||||
orx exp run <expId> --backend ssh --host lab --container research
|
||||
orx exp run <expId> --backend ssh --host lab --no-container
|
||||
```
|
||||
|
||||
- `--host` is required on every launch and must be an SSH config alias. There is
|
||||
no flavor.
|
||||
- Authentication uses the user's SSH keys or agent. `orx` invokes SSH but never
|
||||
reads a private key. The host needs `bash` and `tar`.
|
||||
- The committed snapshot is streamed to the host and extracted before the fixed
|
||||
run command starts.
|
||||
- There is no image or timeout flag; the host environment is used as-is.
|
||||
- Runs live under `~/.orx/runs/<runId>/` on the host. Cancellation terminates
|
||||
the remote process group.
|
||||
- A detached `orx supervise` process polls the host; do not kill it.
|
||||
Machine-local SSH settings can save a container name or ID for each host.
|
||||
A saved default SSH host lets launches omit `--host`.
|
||||
These defaults apply only to SSH experiments; dashboards, terminals and coding
|
||||
agents still connect directly to the host.
|
||||
|
||||
- Omitted container selection uses the host's saved target. `--container`
|
||||
overrides it; `--no-container` explicitly runs on the host. The flags conflict.
|
||||
|
||||
Keep environment preparation in the project's committed setup/run scripts and
|
||||
invoke them through the fixed run command, just as with other backends. For
|
||||
example, a committed `run.sh` can source Conda, activate the environment, and
|
||||
then run the experiment. Those paths must exist in the selected execution
|
||||
environment (host or container). There is no separate SSH setup command.
|
||||
|
||||
The SSH host needs Bash and tar. Container execution also requires host Docker
|
||||
access and an already running, unpaused Linux container with Bash, tar, working
|
||||
`setsid --wait`, and an absolute writable `$HOME`. Execution uses the container's
|
||||
configured user. There is no `sudo`, user override, Podman support, container
|
||||
creation or lifecycle management. `--image` and `--flavor` are unsupported.
|
||||
|
||||
The committed source archive is cached on the host, then streamed into the
|
||||
container's `$HOME/.orx/runs/<runId>/repo`. Host logs and exit status stay in the
|
||||
host's `~/.orx/runs/<runId>/`, so container removal does not erase them. A detached
|
||||
supervisor reattaches using the saved container ID and start time; later settings
|
||||
changes cannot redirect an existing run. Do not kill the supervisor.
|
||||
|
||||
Cancellation targets only the experiment process group, waits five seconds for
|
||||
TERM, then uses KILL if needed. It leaves the container and unrelated processes
|
||||
running. Paused containers must be unpaused by the user before cancellation can
|
||||
finish. Stopping, removing or restarting a container fails its experiment.
|
||||
Docker/SSH outages are retried rather than reported as container removal.
|
||||
|
||||
Host authentication and container readiness are separate checks. Preflight must
|
||||
check the explicitly selected container; host readiness does not imply container
|
||||
readiness.
|
||||
|
||||
Executable
+44
@@ -0,0 +1,44 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
# Only the uniquely named test containers are changed or removed.
|
||||
fixture=$(mktemp -d)
|
||||
prefix="orx-ssh-test-$$"
|
||||
cleanup() {
|
||||
docker rm -f "$prefix-host" "$prefix-inner" >/dev/null 2>&1 || true
|
||||
docker image rm "$prefix" >/dev/null 2>&1 || true
|
||||
rm -rf "$fixture"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
ssh-keygen -q -t ed25519 -N '' -f "$fixture/key"
|
||||
cat > "$fixture/Dockerfile" <<'DOCKER'
|
||||
FROM docker:28-cli
|
||||
RUN apk add --no-cache openssh bash util-linux coreutils procps && ssh-keygen -A && mkdir -p /root/.ssh && chmod 700 /root/.ssh
|
||||
COPY key.pub /root/.ssh/authorized_keys
|
||||
COPY ssh-dispatch /usr/local/bin/ssh-dispatch
|
||||
RUN chmod +x /usr/local/bin/ssh-dispatch && echo "ForceCommand /usr/local/bin/ssh-dispatch" >> /etc/ssh/sshd_config
|
||||
RUN chmod 600 /root/.ssh/authorized_keys
|
||||
ENTRYPOINT ["/usr/sbin/sshd", "-D", "-e"]
|
||||
DOCKER
|
||||
cat > "$fixture/ssh-dispatch" <<'DISPATCH'
|
||||
#!/bin/sh
|
||||
case "$SSH_ORIGINAL_COMMAND" in
|
||||
*'date +%s > launch_time'*)
|
||||
if [ -f /tmp/drop-launch-ack ]; then
|
||||
rm /tmp/drop-launch-ack
|
||||
sh -c "$SSH_ORIGINAL_COMMAND"
|
||||
exit 255
|
||||
fi ;;
|
||||
esac
|
||||
exec sh -c "$SSH_ORIGINAL_COMMAND"
|
||||
DISPATCH
|
||||
docker build -q -t "$prefix" "$fixture"
|
||||
docker run -d --name "$prefix-host" -p 127.0.0.1::22 -v /var/run/docker.sock:/var/run/docker.sock "$prefix"
|
||||
docker run -d --name "$prefix-inner" --user 1000:1000 --env CONDA_ENVS_PATH=/tmp/conda-envs --env "HOME=/tmp/orx home'quoted" --entrypoint bash condaforge/miniforge3 -c 'mkdir -p "$HOME"; exec sleep infinity'
|
||||
port=$(docker port "$prefix-host" 22/tcp)
|
||||
export ORX_SSH_TEST_PORT="${port##*:}" ORX_SSH_TEST_KEY="$fixture/key" ORX_SSH_TEST_CONTAINER="$prefix-inner"
|
||||
for attempt in {1..30}; do
|
||||
if ssh -i "$fixture/key" -p "$ORX_SSH_TEST_PORT" -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes root@127.0.0.1 true 2>/dev/null; then break; fi
|
||||
sleep 1
|
||||
done
|
||||
cargo build --locked
|
||||
cargo test --locked ssh_container_lifecycle -- --ignored --nocapture
|
||||
@@ -584,7 +584,15 @@ async fn run_ssh(
|
||||
eprintln!("supervise {run_id}: watching ssh job {host}:{dir}");
|
||||
let target = ssh::SshTarget::alias(host);
|
||||
let dir = dir.to_string();
|
||||
watch_ssh_job(&store, status_of(&stored)?, target, dir, &run_id).await?;
|
||||
watch_ssh_job(
|
||||
&store,
|
||||
status_of(&stored)?,
|
||||
target,
|
||||
dir,
|
||||
descriptor.ssh_container,
|
||||
&run_id,
|
||||
)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -596,6 +604,7 @@ async fn watch_ssh_job(
|
||||
initial_status: RunStatus,
|
||||
target: ssh::SshTarget,
|
||||
dir: String,
|
||||
container: Option<ssh::ContainerRun>,
|
||||
run_id: &str,
|
||||
) -> Result<RunStatus> {
|
||||
let path = log_path(run_id);
|
||||
@@ -610,9 +619,10 @@ async fn watch_ssh_job(
|
||||
|
||||
let mut last_status = initial_status;
|
||||
let mut cancel_sent = false;
|
||||
let mut last_message = None;
|
||||
|
||||
loop {
|
||||
let job = match ssh::inspect_job(&target, &dir).await {
|
||||
let job = match ssh::inspect_job(&target, &dir, container.as_ref()).await {
|
||||
Ok(j) => j,
|
||||
Err(err) => {
|
||||
eprintln!("supervise {run_id}: inspect failed (will retry): {err}");
|
||||
@@ -645,15 +655,32 @@ async fn watch_ssh_job(
|
||||
return Ok(status);
|
||||
}
|
||||
|
||||
if container.is_some() && job.message != last_message {
|
||||
let message = format!(
|
||||
"[orx] {}",
|
||||
job.message.as_deref().unwrap_or("Container running.")
|
||||
);
|
||||
let command = format!(
|
||||
"printf '%s\\n' {} >> \"$HOME/{dir}/log\"",
|
||||
ssh::sh_quote(&message)
|
||||
);
|
||||
match ssh::ssh_run(&target, &command, None).await {
|
||||
Ok(_) => last_message = job.message,
|
||||
Err(error) => {
|
||||
eprintln!("supervise {run_id}: could not log container status: {error}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if status != last_status && store.update_status(run_id, status, None, None)? {
|
||||
let cancel_requested = local_cancel_requested(store, run_id);
|
||||
eprintln!("supervise {run_id}: {last_status} -> {status} (stage {stage})");
|
||||
last_status = status;
|
||||
if cancel_requested && !cancel_sent {
|
||||
cancel_ssh(&target, &dir, run_id, &mut cancel_sent).await;
|
||||
cancel_ssh(&target, &dir, container.as_ref(), run_id, &mut cancel_sent).await;
|
||||
}
|
||||
} else if !cancel_sent && local_cancel_requested(store, run_id) {
|
||||
cancel_ssh(&target, &dir, run_id, &mut cancel_sent).await;
|
||||
cancel_ssh(&target, &dir, container.as_ref(), run_id, &mut cancel_sent).await;
|
||||
}
|
||||
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
@@ -701,9 +728,15 @@ async fn tail_logs_ssh(
|
||||
}
|
||||
}
|
||||
|
||||
async fn cancel_ssh(target: &ssh::SshTarget, dir: &str, run_id: &str, cancel_sent: &mut bool) {
|
||||
async fn cancel_ssh(
|
||||
target: &ssh::SshTarget,
|
||||
dir: &str,
|
||||
container: Option<&ssh::ContainerRun>,
|
||||
run_id: &str,
|
||||
cancel_sent: &mut bool,
|
||||
) {
|
||||
eprintln!("supervise {run_id}: cancel requested — killing remote process group");
|
||||
match ssh::cancel_job(target, dir).await {
|
||||
match ssh::cancel_job(target, dir, container).await {
|
||||
Ok(()) => *cancel_sent = true,
|
||||
Err(err) => eprintln!("supervise {run_id}: ssh cancel failed (will retry): {err}"),
|
||||
}
|
||||
@@ -839,13 +872,15 @@ async fn run_openresearch(
|
||||
teardown_box(&store, &lifecycle, &sandbox_id, &run_id).await;
|
||||
return Ok(());
|
||||
}
|
||||
let staged = ssh::stage_source(&target, &run_id, &source.path, &source.digest).await;
|
||||
let staged =
|
||||
ssh::stage_source(&target, &run_id, &source.path, &source.digest, None).await;
|
||||
if let Err(err) = staged {
|
||||
eprintln!("supervise {run_id}: source staging failed (will retry): {err}");
|
||||
launch_err = Some(err);
|
||||
continue;
|
||||
}
|
||||
match ssh::run_job(&ssh::SshJobSpec {
|
||||
container: None,
|
||||
target: target.clone(),
|
||||
run_id: run_id.clone(),
|
||||
script: script.clone(),
|
||||
@@ -885,7 +920,7 @@ async fn run_openresearch(
|
||||
// The shared ssh loop owns status and logs; the box is deleted after
|
||||
// it returns (logs are drained from the box BEFORE teardown), and even
|
||||
// when it errors.
|
||||
let watch = watch_ssh_job(&store, status_of(&stored)?, target, dir, &run_id).await;
|
||||
let watch = watch_ssh_job(&store, status_of(&stored)?, target, dir, None, &run_id).await;
|
||||
teardown_box(&store, &lifecycle, &sandbox_id, &run_id).await;
|
||||
watch?;
|
||||
Ok(())
|
||||
|
||||
+88
-3
@@ -592,7 +592,11 @@ fn router(state: AppState, remote_auth: Option<RemoteAuth>) -> Router {
|
||||
"/api/projects/{id}/ui-state",
|
||||
get(project_ui_state).post(set_project_ui_state),
|
||||
)
|
||||
.route("/api/settings/ssh", get(ssh_settings))
|
||||
.route(
|
||||
"/api/settings/ssh",
|
||||
get(ssh_settings).post(save_ssh_settings),
|
||||
)
|
||||
.route("/api/settings/ssh/default", post(save_ssh_default))
|
||||
.route("/api/settings/ssh/master", get(ssh_master_status))
|
||||
.route("/api/settings/ssh/preflight", post(ssh_preflight))
|
||||
.route("/api/settings/ssh/connect", get(ssh_connect))
|
||||
@@ -1927,6 +1931,9 @@ struct CreateRunReq {
|
||||
backend: Option<String>,
|
||||
flavor: Option<String>,
|
||||
host: Option<String>,
|
||||
container: Option<String>,
|
||||
#[serde(default)]
|
||||
no_container: bool,
|
||||
manifest: Option<String>,
|
||||
image: Option<String>,
|
||||
timeout: Option<String>,
|
||||
@@ -2004,6 +2011,8 @@ pub(crate) async fn submit_run_via_up(
|
||||
backend: args.backend.clone(),
|
||||
flavor: args.flavor.clone(),
|
||||
host: args.host.clone(),
|
||||
container: args.container.clone(),
|
||||
no_container: args.no_container,
|
||||
manifest: args.manifest.clone(),
|
||||
image: args.image.clone(),
|
||||
timeout: args.timeout.clone(),
|
||||
@@ -2068,6 +2077,8 @@ async fn create_run(State(state): State<AppState>, Json(req): Json<CreateRunReq>
|
||||
flavor,
|
||||
org: req.org,
|
||||
host: req.host,
|
||||
container: req.container,
|
||||
no_container: req.no_container,
|
||||
manifest: req.manifest,
|
||||
image: req.image,
|
||||
timeout: req.timeout,
|
||||
@@ -6072,6 +6083,16 @@ fn list_ssh_hosts() -> Vec<Value> {
|
||||
async fn ssh_settings() -> ApiResult {
|
||||
tokio::task::spawn_blocking(|| {
|
||||
let mut hosts = list_ssh_hosts();
|
||||
let settings = crate::config::ssh_settings()?;
|
||||
for host in &mut hosts {
|
||||
let options = host
|
||||
.get("host")
|
||||
.and_then(Value::as_str)
|
||||
.and_then(|host| settings.hosts.get(host))
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
host["container"] = json!(options.container);
|
||||
}
|
||||
// Best-effort, like the preflight write: a store hiccup shouldn't take
|
||||
// out the host listing — hosts just render as never tested.
|
||||
let tests: HashMap<String, SshHostTest> = Store::open()
|
||||
@@ -6093,12 +6114,52 @@ async fn ssh_settings() -> ApiResult {
|
||||
};
|
||||
h["lastTest"] = json!(t);
|
||||
}
|
||||
Ok(Json(json!({ "hosts": hosts })))
|
||||
Ok(Json(
|
||||
json!({ "hosts": hosts, "defaultHost": settings.default_host }),
|
||||
))
|
||||
})
|
||||
.await
|
||||
.map_err(|e| ApiError::from(anyhow!("ssh task failed: {e}")))?
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct SaveSshSettingsReq {
|
||||
host: String,
|
||||
container: Option<String>,
|
||||
}
|
||||
|
||||
async fn save_ssh_settings(Json(req): Json<SaveSshSettingsReq>) -> ApiResult {
|
||||
let host = req.host.trim().to_string();
|
||||
require_configured_ssh_host(&host)?;
|
||||
let options = crate::config::SshHostSettings {
|
||||
container: req.container,
|
||||
};
|
||||
crate::jobs::ssh::validate_host_options(&options).map_err(bad_request)?;
|
||||
blocking_api(move || {
|
||||
crate::config::set_ssh_host(host, options)?;
|
||||
Ok(Json(json!({"ok": true})))
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SaveSshDefaultReq {
|
||||
host: Option<String>,
|
||||
}
|
||||
|
||||
async fn save_ssh_default(Json(req): Json<SaveSshDefaultReq>) -> ApiResult {
|
||||
let host = req.host.map(|host| host.trim().to_string());
|
||||
if let Some(host) = &host {
|
||||
require_configured_ssh_host(host)?;
|
||||
}
|
||||
blocking_api(move || {
|
||||
crate::config::set_ssh_default(host)?;
|
||||
Ok(Json(json!({"ok": true})))
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
fn ssh_config_path() -> Result<std::path::PathBuf> {
|
||||
Ok(dirs::home_dir()
|
||||
.ok_or_else(|| anyhow!("no home directory"))?
|
||||
@@ -6177,6 +6238,7 @@ async fn save_ssh_config(Json(req): Json<SaveSshConfigReq>) -> ApiResult {
|
||||
#[derive(Deserialize)]
|
||||
struct SshPreflightReq {
|
||||
host: String,
|
||||
container: Option<String>,
|
||||
}
|
||||
|
||||
async fn ssh_master_status(Query(req): Query<SshPreflightReq>) -> ApiResult {
|
||||
@@ -6201,7 +6263,27 @@ async fn ssh_preflight(Json(req): Json<SshPreflightReq>) -> ApiResult {
|
||||
if host.is_empty() {
|
||||
return Err(bad_request("host is required"));
|
||||
}
|
||||
Ok(Json(json!(run_ssh_host_preflight(host).await)))
|
||||
require_configured_ssh_host(&host)?;
|
||||
if let Some(reference) = &req.container {
|
||||
crate::jobs::ssh::validate_container_reference(reference).map_err(bad_request)?;
|
||||
}
|
||||
let test = run_ssh_host_preflight(host.clone()).await;
|
||||
let container = if let Some(reference) = req.container {
|
||||
let result = crate::jobs::ssh::resolve_container(
|
||||
&crate::jobs::ssh::SshTarget::alias(&host),
|
||||
&reference,
|
||||
)
|
||||
.await;
|
||||
Some(
|
||||
json!({"reference": reference, "ready": result.is_ok(), "error": result.err().map(|error| error.to_string())}),
|
||||
)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let mut result =
|
||||
serde_json::to_value(test).map_err(|error| ApiError::from(anyhow!("{error}")))?;
|
||||
result["container"] = json!(container);
|
||||
Ok(Json(result))
|
||||
}
|
||||
|
||||
fn require_configured_ssh_host(host: &str) -> Result<(), ApiError> {
|
||||
@@ -8184,6 +8266,7 @@ mod tests {
|
||||
async fn windows_ssh_master_status_is_not_a_disconnection() {
|
||||
let response = ssh_master_status(Query(SshPreflightReq {
|
||||
host: "unused-host".into(),
|
||||
container: None,
|
||||
}))
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("{}", error.1));
|
||||
@@ -8438,6 +8521,8 @@ mod tests {
|
||||
backend: Some("local".into()),
|
||||
flavor: None,
|
||||
host: None,
|
||||
container: None,
|
||||
no_container: false,
|
||||
manifest: None,
|
||||
image: None,
|
||||
timeout: None,
|
||||
|
||||
+127
-28
@@ -523,35 +523,52 @@ backend_adapter!(
|
||||
run_id | crate::local::k8s::submit_local_k8s_with_source(args, source, run_id).await
|
||||
);
|
||||
|
||||
backend_adapter!(
|
||||
SshCompute,
|
||||
"ssh",
|
||||
"SSH",
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
"SSH tar stream",
|
||||
false,
|
||||
preflight | args | {
|
||||
let host = args
|
||||
.host
|
||||
.as_deref()
|
||||
.ok_or_else(|| anyhow!("SSH requires --host <alias>."))?;
|
||||
let check = crate::jobs::ssh::preflight(&crate::jobs::ssh::SshTarget::alias(host)).await;
|
||||
if !check.reachable || !check.tools_found {
|
||||
return Ok(not_ready(
|
||||
check
|
||||
.error
|
||||
.as_deref()
|
||||
.unwrap_or("The SSH host needs bash and tar."),
|
||||
));
|
||||
#[derive(Default)]
|
||||
pub struct SshCompute {
|
||||
launch: tokio::sync::OnceCell<crate::jobs::ssh::ResolvedLaunch>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ComputeBackend for SshCompute {
|
||||
fn capabilities(&self) -> Capabilities {
|
||||
Capabilities {
|
||||
id: "ssh",
|
||||
label: "SSH",
|
||||
remote: true,
|
||||
flavors: false,
|
||||
requires_flavor: false,
|
||||
source_transport: "SSH tar stream",
|
||||
}
|
||||
}
|
||||
|
||||
async fn preflight(&self, args: &crate::ExpRunArgs) -> Result<Preflight> {
|
||||
self.launch
|
||||
.get_or_try_init(|| crate::jobs::ssh::resolve_launch(args))
|
||||
.await?;
|
||||
ready()
|
||||
},
|
||||
submit | args,
|
||||
source,
|
||||
run_id | crate::local::ssh::submit_local_ssh_with_source(args, source, run_id).await
|
||||
);
|
||||
}
|
||||
|
||||
async fn stage_source(
|
||||
&self,
|
||||
project: &LocalProject,
|
||||
experiment: &LocalExperiment,
|
||||
) -> Result<StagedSource> {
|
||||
stage_snapshot(project, experiment, false).await
|
||||
}
|
||||
|
||||
async fn submit(
|
||||
&self,
|
||||
args: &crate::ExpRunArgs,
|
||||
source: StagedSource,
|
||||
run_id: String,
|
||||
) -> Result<StoredRun> {
|
||||
let launch = self
|
||||
.launch
|
||||
.get()
|
||||
.ok_or_else(|| anyhow!("SSH launch was not prepared."))?;
|
||||
crate::local::ssh::submit_local_ssh_with_source(args, source.0, run_id, launch).await
|
||||
}
|
||||
}
|
||||
|
||||
backend_adapter!(
|
||||
SlurmCompute,
|
||||
@@ -633,7 +650,7 @@ pub fn backend(id: &str) -> Result<Box<dyn ComputeBackend>> {
|
||||
"hf" => Ok(Box::new(HuggingFaceCompute)),
|
||||
"modal" => Ok(Box::new(ModalCompute)),
|
||||
"k8s" => Ok(Box::new(KubernetesCompute)),
|
||||
"ssh" => Ok(Box::new(SshCompute)),
|
||||
"ssh" => Ok(Box::new(SshCompute::default())),
|
||||
"slurm" => Ok(Box::new(SlurmCompute)),
|
||||
"ray" => Ok(Box::new(RayCompute)),
|
||||
"openresearch" => Ok(Box::new(OpenResearchCompute)),
|
||||
@@ -650,6 +667,20 @@ pub fn capabilities() -> Vec<Capabilities> {
|
||||
}
|
||||
|
||||
pub fn validate_run_args(args: &crate::ExpRunArgs) -> Result<()> {
|
||||
if (args.container.is_some() || args.no_container) && args.backend.as_deref() != Some("ssh") {
|
||||
return Err(anyhow!(
|
||||
"--container and --no-container only apply with --backend ssh."
|
||||
));
|
||||
}
|
||||
if args.container.is_some() && args.no_container {
|
||||
return Err(anyhow!("--container conflicts with --no-container."));
|
||||
}
|
||||
if let Some(reference) = &args.container {
|
||||
crate::jobs::ssh::validate_container_reference(reference)?;
|
||||
}
|
||||
if args.backend.as_deref() == Some("ssh") && (args.image.is_some() || args.flavor.is_some()) {
|
||||
return Err(anyhow!("SSH does not support --flavor or --image."));
|
||||
}
|
||||
if args.manifest.is_some() && args.backend.as_deref() != Some("k8s") {
|
||||
return Err(anyhow!("--manifest only applies with --backend k8s."));
|
||||
}
|
||||
@@ -724,6 +755,7 @@ pub async fn submit(args: &crate::ExpRunArgs) -> Result<StoredRun> {
|
||||
})
|
||||
.unwrap_or_default();
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: format!("{}_job", backend_id),
|
||||
namespace: None,
|
||||
job_id: None,
|
||||
@@ -891,6 +923,8 @@ mod tests {
|
||||
flavor: None,
|
||||
org: None,
|
||||
host: None,
|
||||
container: None,
|
||||
no_container: false,
|
||||
manifest: None,
|
||||
image: None,
|
||||
timeout: None,
|
||||
@@ -899,6 +933,71 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ssh_selectors_and_saved_target_precedence() {
|
||||
use crate::config::{SshHostSettings, SshSettings};
|
||||
let settings = SshSettings {
|
||||
default_host: Some("lab".into()),
|
||||
hosts: std::collections::BTreeMap::from([
|
||||
(
|
||||
"lab".into(),
|
||||
SshHostSettings {
|
||||
container: Some("research".into()),
|
||||
},
|
||||
),
|
||||
("direct".into(), SshHostSettings { container: None }),
|
||||
]),
|
||||
};
|
||||
let mut args = tinker_args();
|
||||
args.backend = Some("ssh".into());
|
||||
for (host, container, no_container, expected_host, expected_container) in [
|
||||
(None, None, false, "lab", Some("research")),
|
||||
(
|
||||
Some("lab"),
|
||||
Some("research"),
|
||||
false,
|
||||
"lab",
|
||||
Some("research"),
|
||||
),
|
||||
(None, Some("other"), false, "lab", Some("other")),
|
||||
(None, Some("none"), false, "lab", Some("none")),
|
||||
(None, None, true, "lab", None),
|
||||
(Some("direct"), None, true, "direct", None),
|
||||
(Some("unknown"), None, false, "unknown", None),
|
||||
] {
|
||||
args.host = host.map(str::to_string);
|
||||
args.container = container.map(str::to_string);
|
||||
args.no_container = no_container;
|
||||
validate_run_args(&args).unwrap();
|
||||
let (host, options) = crate::jobs::ssh::resolve_options(&args, &settings).unwrap();
|
||||
assert_eq!(host, expected_host);
|
||||
assert_eq!(options.container.as_deref(), expected_container);
|
||||
}
|
||||
args.host = None;
|
||||
assert!(crate::jobs::ssh::resolve_options(&args, &SshSettings::default()).is_err());
|
||||
args.container = Some(String::new());
|
||||
assert!(validate_run_args(&args).is_err());
|
||||
args.container = Some("research".into());
|
||||
args.no_container = true;
|
||||
assert!(validate_run_args(&args).is_err());
|
||||
args.no_container = false;
|
||||
for backend in ["local", "slurm", "openresearch"] {
|
||||
args.backend = Some(backend.into());
|
||||
assert!(validate_run_args(&args).is_err());
|
||||
}
|
||||
use clap::Parser;
|
||||
assert!(crate::Cli::try_parse_from([
|
||||
"orx",
|
||||
"exp",
|
||||
"run",
|
||||
"exp",
|
||||
"--container",
|
||||
"research",
|
||||
"--no-container"
|
||||
])
|
||||
.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tinker_is_registered_as_a_local_controller_without_flavors() {
|
||||
let capabilities = backend("tinker").unwrap().capabilities();
|
||||
|
||||
@@ -431,3 +431,32 @@ pub fn write_synced_env_vars(values: &[(&str, &str)]) -> Result<()> {
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct SshHostSettings {
|
||||
pub container: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct SshSettings {
|
||||
pub default_host: Option<String>,
|
||||
#[serde(default)]
|
||||
pub hosts: std::collections::BTreeMap<String, SshHostSettings>,
|
||||
}
|
||||
|
||||
pub fn ssh_settings() -> Result<SshSettings> {
|
||||
crate::telemetry::ssh_settings()
|
||||
}
|
||||
|
||||
pub fn set_ssh_host(host: String, options: SshHostSettings) -> Result<()> {
|
||||
crate::jobs::ssh::validate_host_options(&options)?;
|
||||
crate::telemetry::set_ssh_host(host, options)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn set_ssh_default(host: Option<String>) -> Result<()> {
|
||||
crate::telemetry::set_ssh_default(host)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -50,6 +50,8 @@ pub fn default_python_env(env: &HashMap<String, String>) -> HashMap<String, Stri
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackendDescriptor {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub ssh_container: Option<ssh::ContainerRun>,
|
||||
pub kind: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub namespace: Option<String>,
|
||||
@@ -249,6 +251,7 @@ mod tests {
|
||||
|
||||
fn openresearch_descriptor() -> BackendDescriptor {
|
||||
BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "openresearch_job".to_string(),
|
||||
namespace: Some("org_1".to_string()),
|
||||
job_id: Some("sb_1".to_string()),
|
||||
@@ -301,6 +304,20 @@ mod tests {
|
||||
assert_eq!(d.ssh_ref().unwrap(), ("mybox", ".orx/runs/r1"));
|
||||
assert_eq!(d.ssh_host, None);
|
||||
assert_eq!(d.timeout_secs, None);
|
||||
assert_eq!(d.ssh_container, None);
|
||||
let mut d = d;
|
||||
d.ssh_container = Some(ssh::ContainerRun {
|
||||
reference: "research".into(),
|
||||
id: "immutable-id".into(),
|
||||
started_at: "2026-09-21T00:00:00Z".into(),
|
||||
run_dir: "/home/user/.orx/runs/r1".into(),
|
||||
});
|
||||
let json = d.to_json();
|
||||
assert!(json.contains("sshContainer"));
|
||||
assert_eq!(
|
||||
BackendDescriptor::parse(&json).unwrap().ssh_container,
|
||||
d.ssh_container
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
+140
-34
@@ -14,6 +14,11 @@
|
||||
//! exit_code written when the payload finishes
|
||||
//! A restarted `orx supervise` reattaches purely from that directory.
|
||||
|
||||
mod container;
|
||||
pub use container::{
|
||||
resolve as resolve_container, validate_reference as validate_container_reference, ContainerRun,
|
||||
};
|
||||
|
||||
use std::collections::HashMap;
|
||||
#[cfg(unix)]
|
||||
use std::path::PathBuf;
|
||||
@@ -65,6 +70,63 @@ fn prepare_control_dir() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ResolvedLaunch {
|
||||
pub target: SshTarget,
|
||||
pub container: Option<ContainerRun>,
|
||||
}
|
||||
|
||||
pub fn validate_host_options(options: &crate::config::SshHostSettings) -> Result<()> {
|
||||
if let Some(reference) = &options.container {
|
||||
validate_container_reference(reference)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn resolve_options(
|
||||
args: &crate::ExpRunArgs,
|
||||
settings: &crate::config::SshSettings,
|
||||
) -> Result<(String, crate::config::SshHostSettings)> {
|
||||
let host = args
|
||||
.host
|
||||
.as_ref()
|
||||
.or(settings.default_host.as_ref())
|
||||
.filter(|host| !host.trim().is_empty())
|
||||
.ok_or_else(|| {
|
||||
anyhow!("SSH requires --host <alias> or a default host saved in SSH compute settings.")
|
||||
})?
|
||||
.clone();
|
||||
let saved = settings.hosts.get(&host).cloned().unwrap_or_default();
|
||||
let container = if args.no_container {
|
||||
None
|
||||
} else {
|
||||
args.container.clone().or_else(|| saved.container.clone())
|
||||
};
|
||||
let options = crate::config::SshHostSettings { container };
|
||||
validate_host_options(&options)?;
|
||||
Ok((host, options))
|
||||
}
|
||||
|
||||
pub async fn resolve_launch(args: &crate::ExpRunArgs) -> Result<ResolvedLaunch> {
|
||||
let settings = crate::config::ssh_settings()?;
|
||||
let (host, options) = resolve_options(args, &settings)?;
|
||||
let target = SshTarget::alias(&host);
|
||||
let host_check = preflight(&target).await;
|
||||
if !host_check.reachable || !host_check.tools_found {
|
||||
return Err(anyhow!(
|
||||
"{}",
|
||||
host_check
|
||||
.error
|
||||
.unwrap_or_else(|| "SSH host needs bash and tar.".into())
|
||||
));
|
||||
}
|
||||
let container = match options.container {
|
||||
Some(reference) => Some(container::resolve(&target, &reference).await?),
|
||||
None => None,
|
||||
};
|
||||
Ok(ResolvedLaunch { target, container })
|
||||
}
|
||||
|
||||
/// An ssh endpoint. The classic ssh backend connects by `~/.ssh/config` alias
|
||||
/// (`SshTarget::alias`); backends that learn an endpoint at runtime (an
|
||||
/// OpenResearch box on a provider-assigned host:port) pass an explicit
|
||||
@@ -366,6 +428,7 @@ pub async fn stage_source(
|
||||
run_id: &str,
|
||||
archive: &std::path::Path,
|
||||
digest: &str,
|
||||
container: Option<&ContainerRun>,
|
||||
) -> Result<String> {
|
||||
let dir = format!(".orx/runs/{run_id}");
|
||||
let cache = format!(".orx/source/{digest}.tar");
|
||||
@@ -382,16 +445,24 @@ pub async fn stage_source(
|
||||
);
|
||||
ssh_run_file(target, &upload, archive).await?;
|
||||
}
|
||||
ssh_run(
|
||||
target,
|
||||
&format!(
|
||||
"umask 077; mkdir -p \"$HOME/.orx/runs\" \"$HOME/{dir}/repo\"; \
|
||||
if let Some(container) = container {
|
||||
container.require_running(target).await?;
|
||||
let path = sh_quote(&container.run_dir);
|
||||
let extract = container.exec(&format!("set -e; umask 077; mkdir -p {path}/repo; chmod 700 {path} {path}/repo; tar -xf - -C {path}/repo"));
|
||||
// Start the bound here too: submission may fail before detached launch.
|
||||
ssh_run(target, &format!("umask 077; mkdir -p \"$HOME/{dir}\" && chmod 700 \"$HOME/{dir}\" && {extract} < \"$HOME/{cache}\" && date +%s > \"$HOME/{dir}/launch_time.tmp\" && mv \"$HOME/{dir}/launch_time.tmp\" \"$HOME/{dir}/launch_time\""), None).await?;
|
||||
} else {
|
||||
ssh_run(
|
||||
target,
|
||||
&format!(
|
||||
"umask 077; mkdir -p \"$HOME/.orx/runs\" \"$HOME/{dir}/repo\"; \
|
||||
chmod 700 \"$HOME/.orx/runs\" \"$HOME/{dir}\" \"$HOME/{dir}/repo\"; \
|
||||
tar -xf \"$HOME/{cache}\" -C \"$HOME/{dir}/repo\""
|
||||
),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Ok(dir)
|
||||
}
|
||||
|
||||
@@ -409,8 +480,20 @@ pub struct SshJobSpec {
|
||||
pub script: String,
|
||||
/// Exported inside run.sh on the remote (tokens, synced env).
|
||||
pub env: HashMap<String, String>,
|
||||
pub container: Option<ContainerRun>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct LaunchUncertain;
|
||||
|
||||
impl std::fmt::Display for LaunchUncertain {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str("SSH launch acknowledgement failed")
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LaunchUncertain {}
|
||||
|
||||
/// Submit the job: write run.sh, launch it detached, record its pid. Returns
|
||||
/// the remote run dir (relative to `$HOME`) — the reattach handle.
|
||||
pub async fn run_job(spec: &SshJobSpec) -> Result<String> {
|
||||
@@ -421,31 +504,32 @@ pub async fn run_job(spec: &SshJobSpec) -> Result<String> {
|
||||
.map(|(k, v)| format!("export {}={}", k, sh_quote(v)))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
// run.sh: set up env, run the payload capturing all output to `log`, then
|
||||
// record the exit status. The payload runs in a SUBSHELL `( … )` — not a
|
||||
// `{ … }` group — so an `exit`/`set -e` failure inside it ends the subshell,
|
||||
// not run.sh, and we still reach `echo $? > exit_code`.
|
||||
let run_sh = format!(
|
||||
"#!/usr/bin/env bash\n{exports}\ncd \"$HOME/{dir}\" || exit 97\n(\n{script}\n) > log 2>&1\necho $? > exit_code\n",
|
||||
script = spec.script,
|
||||
);
|
||||
let run_sh = if let Some(container) = &spec.container {
|
||||
container.require_running(&spec.target).await?;
|
||||
let inner = container::inner_script(container, &exports, &spec.script);
|
||||
container::upload_script(&spec.target, container, &inner).await?;
|
||||
container::host_script(&dir, container)
|
||||
} else {
|
||||
let script = &spec.script;
|
||||
// A payload exit must leave the outer shell alive to record its status.
|
||||
format!("#!/usr/bin/env bash\ncd \"$HOME/{dir}\" || exit 97\n(\n{exports}\n{script}\n) > log 2>&1\necho $? > exit_code\n")
|
||||
};
|
||||
|
||||
// Create the dir (owner-only) and write run.sh from stdin.
|
||||
let setup = format!(
|
||||
"mkdir -p \"$HOME/{dir}\" && chmod 700 \"$HOME/{dir}\" && cat > \"$HOME/{dir}/run.sh\"",
|
||||
"umask 077; mkdir -p \"$HOME/{dir}\" && chmod 700 \"$HOME/{dir}\" && cat > \"$HOME/{dir}/run.sh\"",
|
||||
);
|
||||
ssh_run(&spec.target, &setup, Some(&run_sh)).await?;
|
||||
|
||||
// Launch detached so it survives the ssh channel closing. Prefer `setsid`
|
||||
// (new session → pid == pgid, so cancel can TERM the whole group); fall back
|
||||
// to `nohup` where setsid is absent (e.g. a macOS host). Record the pid.
|
||||
let launch = format!(
|
||||
"cd \"$HOME/{dir}\" && \
|
||||
if command -v setsid >/dev/null 2>&1; then setsid bash run.sh </dev/null >/dev/null 2>&1 & \
|
||||
else nohup bash run.sh </dev/null >/dev/null 2>&1 & fi; \
|
||||
echo $! > pid",
|
||||
);
|
||||
ssh_run(&spec.target, &launch, None).await?;
|
||||
// The container host wrapper records its own PID; direct runs retain the existing launcher.
|
||||
let launch = if spec.container.is_some() {
|
||||
format!("cd \"$HOME/{dir}\" && date +%s > launch_time.tmp && mv launch_time.tmp launch_time && {{ if command -v setsid >/dev/null 2>&1; then setsid bash run.sh </dev/null >/dev/null 2>&1 & else nohup bash run.sh </dev/null >/dev/null 2>&1 & fi; }}")
|
||||
} else {
|
||||
format!("cd \"$HOME/{dir}\" && if command -v setsid >/dev/null 2>&1; then setsid bash run.sh </dev/null >/dev/null 2>&1 & else nohup bash run.sh </dev/null >/dev/null 2>&1 & fi; echo $! > pid")
|
||||
};
|
||||
ssh_run(&spec.target, &launch, None)
|
||||
.await
|
||||
.map_err(|error| error.context(LaunchUncertain))?;
|
||||
Ok(dir)
|
||||
}
|
||||
|
||||
@@ -456,7 +540,18 @@ pub struct JobState {
|
||||
pub message: Option<String>,
|
||||
}
|
||||
|
||||
pub async fn inspect_job(target: &SshTarget, dir: &str) -> Result<JobState> {
|
||||
pub async fn inspect_job(
|
||||
target: &SshTarget,
|
||||
dir: &str,
|
||||
container: Option<&ContainerRun>,
|
||||
) -> Result<JobState> {
|
||||
match container {
|
||||
Some(container) => container::inspect(target, dir, container).await,
|
||||
None => inspect_host_job(target, dir).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn inspect_host_job(target: &SshTarget, dir: &str) -> Result<JobState> {
|
||||
// exit_code present -> finished; pid alive -> running; pid dead & no
|
||||
// exit_code -> killed/crashed; no pid yet -> just starting.
|
||||
let cmd = format!(
|
||||
@@ -466,10 +561,13 @@ pub async fn inspect_job(target: &SshTarget, dir: &str) -> Result<JobState> {
|
||||
elif [ -f \"$d/pid\" ]; then echo DEAD; else echo PENDING; fi",
|
||||
);
|
||||
let out = ssh_run(target, &cmd, None).await?;
|
||||
let out = out.trim();
|
||||
Ok(parse_job_state(out.trim()))
|
||||
}
|
||||
|
||||
fn parse_job_state(out: &str) -> JobState {
|
||||
if let Some(code) = out.strip_prefix("EXIT ") {
|
||||
let code: i32 = code.trim().parse().unwrap_or(-1);
|
||||
return Ok(if code == 0 {
|
||||
return if code == 0 {
|
||||
JobState {
|
||||
stage: "COMPLETED".into(),
|
||||
message: None,
|
||||
@@ -479,9 +577,9 @@ pub async fn inspect_job(target: &SshTarget, dir: &str) -> Result<JobState> {
|
||||
stage: "ERROR".into(),
|
||||
message: Some(format!("exited with code {code}")),
|
||||
}
|
||||
});
|
||||
};
|
||||
}
|
||||
Ok(match out {
|
||||
match out {
|
||||
"RUNNING" | "PENDING" => JobState {
|
||||
stage: "RUNNING".into(),
|
||||
message: None,
|
||||
@@ -494,7 +592,7 @@ pub async fn inspect_job(target: &SshTarget, dir: &str) -> Result<JobState> {
|
||||
stage: "RUNNING".into(),
|
||||
message: Some(format!("unexpected inspect output: {other}")),
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// One poll of the remote log past `skip` lines. Unlike the streaming backends
|
||||
@@ -524,7 +622,15 @@ pub async fn stream_logs(
|
||||
|
||||
/// Cancel = TERM the process group if we have one (setsid case), else the pid
|
||||
/// (nohup fallback). The negative-pid form targets the whole group.
|
||||
pub async fn cancel_job(target: &SshTarget, dir: &str) -> Result<()> {
|
||||
pub async fn cancel_job(
|
||||
target: &SshTarget,
|
||||
dir: &str,
|
||||
container: Option<&ContainerRun>,
|
||||
) -> Result<()> {
|
||||
if let Some(container) = container {
|
||||
container::cancel(target, container).await?;
|
||||
}
|
||||
|
||||
let cmd = format!(
|
||||
"p=$(cat \"$HOME/{dir}/pid\" 2>/dev/null); \
|
||||
[ -n \"$p\" ] && {{ kill -TERM -\"$p\" 2>/dev/null || kill -TERM \"$p\" 2>/dev/null; }}; true",
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::{sh_quote, ssh_run, JobState, SshTarget};
|
||||
use crate::error::{anyhow, Result};
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ContainerRun {
|
||||
pub reference: String,
|
||||
pub id: String,
|
||||
pub started_at: String,
|
||||
pub run_dir: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct Inspection {
|
||||
id: String,
|
||||
status: String,
|
||||
started_at: String,
|
||||
}
|
||||
|
||||
pub fn validate_reference(reference: &str) -> Result<()> {
|
||||
if reference.is_empty()
|
||||
|| !reference.starts_with(|c: char| c.is_ascii_alphanumeric())
|
||||
|| !reference
|
||||
.chars()
|
||||
.all(|c| c.is_ascii_alphanumeric() || "_.-".contains(c))
|
||||
{
|
||||
return Err(anyhow!(
|
||||
"Container must be a Docker name or ID (letters, numbers, '.', '_' or '-')."
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn inspection(target: &SshTarget, reference: &str) -> Result<Option<Inspection>> {
|
||||
let template = r#"{"id":{{json .Id}},"status":{{json .State.Status}},"started_at":{{json .State.StartedAt}}}"#;
|
||||
match ssh_run(
|
||||
target,
|
||||
&format!(
|
||||
"docker container inspect --format {} -- {}",
|
||||
sh_quote(template),
|
||||
sh_quote(reference)
|
||||
),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(output) => Ok(Some(serde_json::from_str(output.trim())?)),
|
||||
Err(error) => {
|
||||
// A successful listing distinguishes absence from an unreachable Docker daemon.
|
||||
let list = ssh_run(
|
||||
target,
|
||||
"docker container ls --all --no-trunc --format '{{.ID}} {{.Names}}'",
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
if list.lines().any(|line| {
|
||||
line.split_whitespace().any(|value| {
|
||||
value == reference || (value.len() == 64 && value.starts_with(reference))
|
||||
})
|
||||
}) {
|
||||
Err(error)
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn resolve(target: &SshTarget, reference: &str) -> Result<ContainerRun> {
|
||||
validate_reference(reference)?;
|
||||
let found = inspection(target, reference)
|
||||
.await?
|
||||
.ok_or_else(|| anyhow!("Container '{reference}' does not exist on {}.", target.dest))?;
|
||||
if found.status != "running" {
|
||||
return Err(anyhow!(
|
||||
"Container '{reference}' is {}. Start or unpause it before running an experiment.",
|
||||
found.status
|
||||
));
|
||||
}
|
||||
let probe = r#"set -e
|
||||
[ "$(uname -s)" = Linux ] || { echo 'Container must run Linux' >&2; exit 1; }
|
||||
command -v tar >/dev/null
|
||||
if setsid --wait bash -c 'exit 42'; then rc=0; else rc=$?; fi
|
||||
[ "$rc" = 42 ] || { echo 'Container needs working setsid --wait' >&2; exit 1; }
|
||||
case "$HOME" in /*) ;; *) echo 'Container needs an absolute HOME' >&2; exit 1;; esac
|
||||
[ -d "$HOME" ] && [ -w "$HOME" ] || { echo 'Container HOME is not writable' >&2; exit 1; }
|
||||
printf '%s' "$HOME"
|
||||
"#;
|
||||
let home = ssh_run(
|
||||
target,
|
||||
&format!(
|
||||
"docker exec {} bash -c {}",
|
||||
sh_quote(&found.id),
|
||||
sh_quote(probe)
|
||||
),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
if !home.starts_with('/') || home.contains(['\n', '\r', '\0']) {
|
||||
return Err(anyhow!("Container returned an invalid home directory."));
|
||||
}
|
||||
Ok(ContainerRun {
|
||||
reference: reference.into(),
|
||||
id: found.id,
|
||||
started_at: found.started_at,
|
||||
run_dir: format!("{}/.orx/runs", home.trim_end_matches('/')),
|
||||
})
|
||||
}
|
||||
|
||||
impl ContainerRun {
|
||||
pub fn for_run(&self, run_id: &str) -> Self {
|
||||
Self {
|
||||
run_dir: format!("{}/{run_id}", self.run_dir),
|
||||
..self.clone()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn exec(&self, script: &str) -> String {
|
||||
format!(
|
||||
"docker exec -i {} bash -c {}",
|
||||
sh_quote(&self.id),
|
||||
sh_quote(script)
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn require_running(&self, target: &SshTarget) -> Result<()> {
|
||||
if let Some(reason) = unavailable(target, self).await? {
|
||||
return Err(anyhow!("{reason}"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn unavailable(target: &SshTarget, container: &ContainerRun) -> Result<Option<String>> {
|
||||
let Some(found) = inspection(target, &container.id).await? else {
|
||||
return Ok(Some(format!(
|
||||
"Container '{}' was removed; host logs are preserved.",
|
||||
container.reference
|
||||
)));
|
||||
};
|
||||
if found.id != container.id || found.started_at != container.started_at {
|
||||
return Ok(Some(format!(
|
||||
"Container '{}' restarted; the experiment cannot resume.",
|
||||
container.reference
|
||||
)));
|
||||
}
|
||||
Ok((found.status != "running")
|
||||
.then(|| format!("Container '{}' is {}.", container.reference, found.status)))
|
||||
}
|
||||
|
||||
// Linux process start time prevents a recycled PID from identifying another experiment.
|
||||
// Minimal containers need no procps; ignore zombies when checking the process group.
|
||||
pub(super) const PROCESS_HELPERS: &str = r#"
|
||||
start_time() {
|
||||
local stat
|
||||
IFS= read -r stat < "/proc/$1/stat" 2>/dev/null || return 1
|
||||
stat=${stat##*) }
|
||||
local fields
|
||||
read -ra fields <<< "$stat"
|
||||
printf '%s' "${fields[19]}"
|
||||
}
|
||||
group_alive() {
|
||||
local file stat fields
|
||||
for file in /proc/[0-9]*/stat; do
|
||||
IFS= read -r stat < "$file" 2>/dev/null || continue
|
||||
stat=${stat##*) }
|
||||
read -ra fields <<< "$stat"
|
||||
if [ "${fields[2]}" = "$p" ] && [ "${fields[0]}" != Z ] && [ "${fields[0]}" != X ]; then return 0; fi
|
||||
done
|
||||
return 1
|
||||
}
|
||||
"#;
|
||||
|
||||
pub(super) fn inner_script(container: &ContainerRun, exports: &str, script: &str) -> String {
|
||||
let dir = sh_quote(&container.run_dir);
|
||||
format!(
|
||||
r#"#!/usr/bin/env bash
|
||||
{PROCESS_HELPERS}
|
||||
umask 077
|
||||
cd {dir} || exit 97
|
||||
printf '%s %s\n' "$$" "$(start_time $$)" > identity.tmp
|
||||
mv identity.tmp identity
|
||||
[ ! -f cancel ] || exit 143
|
||||
(
|
||||
set -eo pipefail
|
||||
{exports}
|
||||
{script}
|
||||
)
|
||||
exit "$?"
|
||||
"#
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) async fn upload_script(
|
||||
target: &SshTarget,
|
||||
container: &ContainerRun,
|
||||
script: &str,
|
||||
) -> Result<()> {
|
||||
let dir = sh_quote(&container.run_dir);
|
||||
ssh_run(
|
||||
target,
|
||||
&container.exec(&format!(
|
||||
"umask 077; cat > {dir}/run.sh && chmod 600 {dir}/run.sh"
|
||||
)),
|
||||
Some(script),
|
||||
)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn host_script(dir: &str, container: &ContainerRun) -> String {
|
||||
let inner = sh_quote(&format!("{}/run.sh", container.run_dir));
|
||||
let check = sh_quote("{{.State.StartedAt}} {{.State.Status}}");
|
||||
let expected = sh_quote(&format!("{} running", container.started_at));
|
||||
let id = sh_quote(&container.id);
|
||||
format!("#!/usr/bin/env bash\ncd \"$HOME/{dir}\" || exit 97\nprintf '%s\\n' \"$$\" > pid\n(\nactual=$(docker container inspect --format {check} -- {id}) || exit $?\n[ \"$actual\" = {expected} ] || {{ echo 'Container generation or readiness changed before launch' >&2; exit 97; }}\ndocker exec {id} setsid --wait bash {inner} </dev/null\n) >> log 2>&1\nprintf '%s\\n' \"$?\" > exit_code\n")
|
||||
}
|
||||
|
||||
fn failed(message: String) -> JobState {
|
||||
JobState {
|
||||
stage: "ERROR".into(),
|
||||
message: Some(message),
|
||||
}
|
||||
}
|
||||
|
||||
async fn host_state(target: &SshTarget, dir: &str) -> Result<String> {
|
||||
ssh_run(target, &format!(r#"d="$HOME/{dir}"; if [ -f "$d/exit_code" ]; then echo "EXIT $(cat "$d/exit_code")"; elif [ -f "$d/pid" ] && kill -0 "$(cat "$d/pid")" 2>/dev/null; then echo RUNNING; else echo DEAD; fi; if [ -f "$d/launch_time" ]; then echo "$(( $(date +%s) - $(cat "$d/launch_time") ))"; else echo 0; fi"#), None).await
|
||||
}
|
||||
|
||||
pub(super) async fn inspect(
|
||||
target: &SshTarget,
|
||||
dir: &str,
|
||||
container: &ContainerRun,
|
||||
) -> Result<JobState> {
|
||||
let host = host_state(target, dir).await?;
|
||||
let mut lines = host.lines();
|
||||
let state = lines.next().unwrap_or_default();
|
||||
if state.starts_with("EXIT ") {
|
||||
return Ok(super::parse_job_state(state));
|
||||
}
|
||||
let age: i64 = lines.next().unwrap_or("0").parse()?;
|
||||
if let Some(reason) = unavailable(target, container).await? {
|
||||
if reason.ends_with(" is paused.") {
|
||||
return Ok(JobState {
|
||||
stage: "RUNNING".into(),
|
||||
message: Some(format!(
|
||||
"{reason} Unpause it to continue or finish cancellation."
|
||||
)),
|
||||
});
|
||||
}
|
||||
return Ok(failed(reason));
|
||||
}
|
||||
let directory = sh_quote(&container.run_dir);
|
||||
let probe = format!(
|
||||
r#"{PROCESS_HELPERS}
|
||||
cd {directory} || exit 1
|
||||
if [ ! -f identity ]; then echo PENDING; exit 0; fi
|
||||
read -r p started < identity
|
||||
current=$(start_time "$p")
|
||||
if [ -n "$current" ] && [ "$current" != "$started" ]; then echo REUSED
|
||||
elif group_alive; then echo RUNNING
|
||||
else echo DEAD; fi
|
||||
"#
|
||||
);
|
||||
let inner = ssh_run(target, &container.exec(&probe), None).await?;
|
||||
match inner.trim() {
|
||||
"RUNNING" if state == "RUNNING" => Ok(JobState {
|
||||
stage: "RUNNING".into(),
|
||||
message: None,
|
||||
}),
|
||||
"PENDING" if age < 30 => Ok(JobState {
|
||||
stage: "RUNNING".into(),
|
||||
message: Some("Waiting for the container launcher.".into()),
|
||||
}),
|
||||
"RUNNING" | "PENDING" => {
|
||||
if inner.trim() == "RUNNING" {
|
||||
// The wrapper can start between the first host read and the inner probe.
|
||||
let current = host_state(target, dir).await?;
|
||||
let state = current.lines().next().unwrap_or_default();
|
||||
if state == "RUNNING" || state.starts_with("EXIT ") {
|
||||
return Ok(super::parse_job_state(state));
|
||||
}
|
||||
}
|
||||
cancel(target, container).await?;
|
||||
super::cancel_job(target, dir, None).await?;
|
||||
Ok(failed("SSH launcher disappeared or did not start within 30 seconds; the container experiment was stopped.".into()))
|
||||
}
|
||||
"REUSED" => Ok(failed(
|
||||
"The container experiment's process identity changed.".into(),
|
||||
)),
|
||||
"DEAD" => {
|
||||
// The host wrapper may still be writing the final exit code.
|
||||
let final_state = super::inspect_host_job(target, dir).await?;
|
||||
if final_state.stage != "RUNNING" {
|
||||
return Ok(final_state);
|
||||
}
|
||||
let waited = ssh_run(target, &format!(r#"d="$HOME/{dir}"; umask 077; if [ ! -f "$d/completion_wait" ]; then date +%s > "$d/completion_wait.tmp" && mv "$d/completion_wait.tmp" "$d/completion_wait"; fi; echo "$(( $(date +%s) - $(cat "$d/completion_wait") ))""#), None).await?;
|
||||
if waited.trim().parse::<i64>()? >= 30 {
|
||||
let final_state = super::inspect_host_job(target, dir).await?;
|
||||
if final_state.stage != "RUNNING" {
|
||||
return Ok(final_state);
|
||||
}
|
||||
super::cancel_job(target, dir, None).await?;
|
||||
return Ok(failed("The container experiment ended, but the SSH launcher did not record its exit status within 30 seconds.".into()));
|
||||
}
|
||||
Ok(JobState {
|
||||
stage: "RUNNING".into(),
|
||||
message: Some("Waiting for the launcher to record the exit status.".into()),
|
||||
})
|
||||
}
|
||||
other => Err(anyhow!("Unexpected container process state: {other}")),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn cancel(target: &SshTarget, container: &ContainerRun) -> Result<()> {
|
||||
if let Some(reason) = unavailable(target, container).await? {
|
||||
if reason.ends_with(" is paused.") {
|
||||
return Err(anyhow!("{reason} Unpause it to complete cancellation."));
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
let directory = sh_quote(&container.run_dir);
|
||||
let script = format!(
|
||||
r#"{PROCESS_HELPERS}
|
||||
cd {directory} || exit 1
|
||||
umask 077
|
||||
: > cancel
|
||||
[ -f identity ] || exit 0
|
||||
read -r p started < identity
|
||||
case "$p" in ''|*[!0-9]*) exit 1;; esac
|
||||
[ "$p" -gt 1 ] || exit 1
|
||||
current=$(start_time "$p")
|
||||
if [ -n "$current" ] && [ "$current" != "$started" ]; then echo 'Process identity changed; refusing to signal it' >&2; exit 1; fi
|
||||
if ! group_alive; then exit 0; fi
|
||||
kill -TERM -- -"$p" 2>/dev/null || true
|
||||
for ((i=0; i<50; i++)); do group_alive || exit 0; sleep 0.1; done
|
||||
kill -KILL -- -"$p" 2>/dev/null || true
|
||||
for ((i=0; i<10; i++)); do group_alive || exit 0; sleep 0.1; done
|
||||
echo 'Experiment process group is still alive' >&2
|
||||
exit 1
|
||||
"#
|
||||
);
|
||||
ssh_run(target, &container.exec(&script), None).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,563 @@
|
||||
use super::*;
|
||||
use crate::jobs::ssh::{self, HostKeyPolicy, SshJobSpec};
|
||||
use std::{collections::HashMap, time::Duration};
|
||||
|
||||
async fn prepare(
|
||||
target: &SshTarget,
|
||||
reference: Option<&str>,
|
||||
command: &str,
|
||||
) -> (String, SshJobSpec) {
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let container = match reference {
|
||||
Some(reference) => Some(resolve(target, reference).await.unwrap().for_run(&id)),
|
||||
None => None,
|
||||
};
|
||||
let temp = crate::local::git::TemporaryDirectory::new("orx-container-test").unwrap();
|
||||
std::fs::write(temp.path().join("source.txt"), "snapshot").unwrap();
|
||||
std::fs::write(
|
||||
temp.path().join("run.sh"),
|
||||
format!("set -eo pipefail\n{command}\n"),
|
||||
)
|
||||
.unwrap();
|
||||
let archive = temp.path().join("source.tar");
|
||||
assert!(std::process::Command::new("tar")
|
||||
.arg("-cf")
|
||||
.arg(&archive)
|
||||
.arg("-C")
|
||||
.arg(temp.path())
|
||||
.arg("source.txt")
|
||||
.arg("run.sh")
|
||||
.status()
|
||||
.unwrap()
|
||||
.success());
|
||||
let dir = ssh::stage_source(target, &id, &archive, &id, container.as_ref())
|
||||
.await
|
||||
.unwrap();
|
||||
let spec = SshJobSpec {
|
||||
target: target.clone(),
|
||||
run_id: id,
|
||||
script: crate::compute::staged_script("bash run.sh"),
|
||||
env: HashMap::from([(
|
||||
"ORX_TEST_SECRET".into(),
|
||||
"quote' and $literal\nsecond line".into(),
|
||||
)]),
|
||||
container,
|
||||
};
|
||||
(dir, spec)
|
||||
}
|
||||
|
||||
async fn launch(
|
||||
target: &SshTarget,
|
||||
reference: Option<&str>,
|
||||
command: &str,
|
||||
) -> (String, Option<ContainerRun>) {
|
||||
let (dir, spec) = prepare(target, reference, command).await;
|
||||
ssh::run_job(&spec).await.unwrap();
|
||||
(dir, spec.container)
|
||||
}
|
||||
|
||||
async fn terminal(target: &SshTarget, dir: &str, container: Option<&ContainerRun>) -> JobState {
|
||||
tokio::time::timeout(Duration::from_secs(45), async {
|
||||
loop {
|
||||
let state = ssh::inspect_job(target, dir, container).await.unwrap();
|
||||
if crate::jobs::is_terminal_stage(&state.stage) {
|
||||
return state;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("run never became terminal")
|
||||
}
|
||||
|
||||
async fn logs(target: &SshTarget, dir: &str) -> String {
|
||||
let mut output = String::new();
|
||||
ssh::stream_logs(target, dir, 0, Duration::ZERO, &mut |line| {
|
||||
output.push_str(line);
|
||||
output.push('\n');
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
output
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn references_are_names_not_shell_or_docker_options() {
|
||||
for value in ["", " ", "--help", "a/b", "a;id", "a\nb"] {
|
||||
assert!(validate_reference(value).is_err());
|
||||
}
|
||||
for value in ["none", "research-v1.2", "a_b", "012345abcdef"] {
|
||||
validate_reference(value).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires disposable fixtures; run scripts/test-ssh-container.sh"]
|
||||
async fn ssh_container_lifecycle() {
|
||||
let port = std::env::var("ORX_SSH_TEST_PORT")
|
||||
.expect("fixture port")
|
||||
.parse()
|
||||
.unwrap();
|
||||
let reference = std::env::var("ORX_SSH_TEST_CONTAINER").expect("fixture container");
|
||||
let mut target = SshTarget::host_port("root@127.0.0.1".into(), port, HostKeyPolicy::Ephemeral);
|
||||
target.extra_opts.extend([
|
||||
"-i".into(),
|
||||
std::env::var("ORX_SSH_TEST_KEY").expect("fixture key"),
|
||||
]);
|
||||
let base = resolve(&target, &reference).await.unwrap();
|
||||
ssh_run(
|
||||
&target,
|
||||
&base.exec("/opt/conda/bin/conda create -y -n research --offline"),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let setup = "source /opt/conda/etc/profile.d/conda.sh\nconda activate research\nprintf 'setup:%s\\n' \"$CONDA_DEFAULT_ENV\"";
|
||||
let (dir, container) = launch(&target, Some(&reference), &format!("{setup}\ncat source.txt; printf '%s\\n' \"$ORX_TEST_SECRET\" \"$PYTHONUNBUFFERED\" \"$CONDA_DEFAULT_ENV\"; echo early; sleep 2; echo late")).await;
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
let early = logs(&target, &dir).await;
|
||||
assert!(
|
||||
early.contains("early") && !early.contains("late"),
|
||||
"{early}"
|
||||
);
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, container.as_ref()).await.stage,
|
||||
"COMPLETED"
|
||||
);
|
||||
let output = logs(&target, &dir).await;
|
||||
assert!(
|
||||
output.contains("snapshot")
|
||||
&& output.contains("quote' and $literal\nsecond line")
|
||||
&& output.contains("\n1\n/tmp/conda-envs/research\n"),
|
||||
"{output}"
|
||||
);
|
||||
assert_eq!(output.matches("setup:/tmp/conda-envs/research").count(), 1);
|
||||
ssh_run(&target, &format!("test ! -d \"$HOME/{dir}/repo\""), None)
|
||||
.await
|
||||
.unwrap();
|
||||
for code in [7, 125, 126, 127] {
|
||||
let (dir, container) = launch(&target, Some(&reference), &format!("exit {code}")).await;
|
||||
let state = terminal(&target, &dir, container.as_ref()).await;
|
||||
assert_eq!(
|
||||
state.message.as_deref(),
|
||||
Some(format!("exited with code {code}").as_str())
|
||||
);
|
||||
}
|
||||
let (dir, container) = launch(
|
||||
&target,
|
||||
Some(&reference),
|
||||
"echo SETUP_FAILED; false; echo SHOULD_NOT_RUN",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, container.as_ref()).await.stage,
|
||||
"ERROR"
|
||||
);
|
||||
assert_eq!(logs(&target, &dir).await, "SETUP_FAILED\n");
|
||||
let (dir, _) = launch(
|
||||
&target,
|
||||
None,
|
||||
"cat source.txt; export HOME=/tmp; cd /tmp; echo DIRECT",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(terminal(&target, &dir, None).await.stage, "COMPLETED");
|
||||
assert!(logs(&target, &dir).await.contains("DIRECT"));
|
||||
|
||||
let (dir, container) = launch(
|
||||
&target,
|
||||
Some(&reference),
|
||||
"trap '' TERM; (trap '' TERM; sleep 300) & wait",
|
||||
)
|
||||
.await;
|
||||
let container = container.unwrap();
|
||||
tokio::time::sleep(Duration::from_millis(300)).await;
|
||||
let restored: ContainerRun =
|
||||
serde_json::from_str(&serde_json::to_string(&container).unwrap()).unwrap();
|
||||
assert_eq!(
|
||||
ssh::inspect_job(&target, &dir, Some(&restored))
|
||||
.await
|
||||
.unwrap()
|
||||
.stage,
|
||||
"RUNNING"
|
||||
);
|
||||
let processes = ssh_run(&target, &container.exec("ps auxww"), None)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
!processes.contains("quote' and $literal"),
|
||||
"secret leaked to argv"
|
||||
);
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("docker exec -d {} sleep 301", sh_quote(&base.id)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("docker pause {}", sh_quote(&base.id)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let paused = ssh::inspect_job(&target, &dir, Some(&container))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(paused.stage, "RUNNING");
|
||||
assert!(paused.message.unwrap().contains("paused"));
|
||||
assert!(ssh::cancel_job(&target, &dir, Some(&container))
|
||||
.await
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("Unpause"));
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("docker unpause {}", sh_quote(&base.id)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
ssh::cancel_job(&target, &dir, Some(&container))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, Some(&container)).await.stage,
|
||||
"ERROR"
|
||||
);
|
||||
let processes = ssh_run(&target, &base.exec("ps auxww"), None)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
!processes.contains("sleep 300"),
|
||||
"child escaped cancellation: {processes}"
|
||||
);
|
||||
assert!(
|
||||
processes.contains("sleep 301"),
|
||||
"unrelated process was killed"
|
||||
);
|
||||
base.require_running(&target).await.unwrap();
|
||||
|
||||
let (dir, spec) = prepare(&target, Some(&reference), "echo ONCE; sleep 1").await;
|
||||
ssh_run(&target, "touch /tmp/drop-launch-ack", None)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
ssh::run_job(&spec)
|
||||
.await
|
||||
.unwrap_err()
|
||||
.is::<ssh::LaunchUncertain>(),
|
||||
"fixture must drop the launch acknowledgement"
|
||||
);
|
||||
let container = spec.container;
|
||||
let saved: ContainerRun =
|
||||
serde_json::from_str(&serde_json::to_string(&container.unwrap()).unwrap()).unwrap();
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, Some(&saved)).await.stage,
|
||||
"COMPLETED"
|
||||
);
|
||||
assert_eq!(logs(&target, &dir).await.matches("ONCE").count(), 1);
|
||||
let (dir, spec) = prepare(&target, Some(&reference), "echo NEVER_LAUNCHED").await;
|
||||
ssh_run(
|
||||
&target,
|
||||
&spec.container.as_ref().unwrap().exec(&format!(
|
||||
"mkdir {}",
|
||||
sh_quote(&format!(
|
||||
"{}/run.sh",
|
||||
spec.container.as_ref().unwrap().run_dir
|
||||
))
|
||||
)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let error = ssh::run_job(&spec).await.unwrap_err();
|
||||
assert!(
|
||||
!error.is::<ssh::LaunchUncertain>(),
|
||||
"script upload failure is not a lost acknowledgement"
|
||||
);
|
||||
assert!(!logs(&target, &dir).await.contains("NEVER_LAUNCHED"));
|
||||
|
||||
let (dir, container) = launch(&target, Some(&reference), "true").await;
|
||||
let container = container.unwrap();
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, Some(&container)).await.stage,
|
||||
"COMPLETED"
|
||||
);
|
||||
ssh_run(&target, &format!("cd \"$HOME/{dir}\"; rm exit_code; setsid sleep 300 </dev/null >/dev/null 2>&1 & echo $! > pid"), None).await.unwrap();
|
||||
assert_eq!(
|
||||
inspect(&target, &dir, &container).await.unwrap().stage,
|
||||
"RUNNING"
|
||||
);
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("echo 0 > \"$HOME/{dir}/completion_wait\""),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let state = inspect(&target, &dir, &container).await.unwrap();
|
||||
assert_eq!(state.stage, "ERROR");
|
||||
assert!(state
|
||||
.message
|
||||
.unwrap()
|
||||
.contains("did not record its exit status"));
|
||||
|
||||
let (dir, container) = launch(&target, Some(&reference), "sleep 300").await;
|
||||
let container = container.unwrap();
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("kill -KILL $(cat \"$HOME/{dir}/pid\")"),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, Some(&container)).await.stage,
|
||||
"ERROR"
|
||||
);
|
||||
assert!(!ssh_run(&target, &base.exec("ps auxww"), None)
|
||||
.await
|
||||
.unwrap()
|
||||
.contains("sleep 300"));
|
||||
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let pending = base.for_run(&id);
|
||||
let pending_dir = format!(".orx/runs/{id}");
|
||||
ssh_run(
|
||||
&target,
|
||||
&pending.exec(&format!("mkdir -p {}", sh_quote(&pending.run_dir))),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!(
|
||||
"mkdir -p \"$HOME/{pending_dir}\"; date +%s > \"$HOME/{pending_dir}/launch_time\""
|
||||
),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
inspect(&target, &pending_dir, &pending)
|
||||
.await
|
||||
.unwrap()
|
||||
.stage,
|
||||
"RUNNING"
|
||||
);
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("echo 0 > \"$HOME/{pending_dir}/launch_time\""),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
inspect(&target, &pending_dir, &pending)
|
||||
.await
|
||||
.unwrap()
|
||||
.stage,
|
||||
"ERROR"
|
||||
);
|
||||
|
||||
let (dir, container) = launch(&target, Some(&reference), "sleep 300").await;
|
||||
let container = container.unwrap();
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let identity = sh_quote(&format!("{}/identity", container.run_dir));
|
||||
ssh_run(&target, &container.exec(&format!("cp {identity} {identity}.saved; read -r p started < {identity}; echo \"$p 0\" > {identity}")), None).await.unwrap();
|
||||
assert!(cancel(&target, &container)
|
||||
.await
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("identity changed"));
|
||||
ssh_run(
|
||||
&target,
|
||||
&container.exec(&format!("mv {identity}.saved {identity}")),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
cancel(&target, &container).await.unwrap();
|
||||
terminal(&target, &dir, Some(&container)).await;
|
||||
|
||||
ssh_run(&target, "mv /usr/local/bin/docker /usr/local/bin/docker-real; printf '#!/bin/sh\\nDOCKER_HOST=unix:///missing-docker.sock exec /usr/local/bin/docker-real \"$@\"\\n' > /usr/local/bin/docker; chmod +x /usr/local/bin/docker", None).await.unwrap();
|
||||
assert!(inspection(&target, &base.id).await.is_err());
|
||||
ssh_run(
|
||||
&target,
|
||||
"mv /usr/local/bin/docker-real /usr/local/bin/docker",
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
supervisor_restart(&target, &reference, port).await;
|
||||
for action in ["stop", "restart", "rm -f"] {
|
||||
let (dir, container) = launch(&target, Some(&reference), "echo SURVIVES; sleep 300").await;
|
||||
tokio::time::sleep(Duration::from_millis(300)).await;
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("docker {action} {}", sh_quote(&reference)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
terminal(&target, &dir, container.as_ref()).await.stage,
|
||||
"ERROR"
|
||||
);
|
||||
assert!(logs(&target, &dir).await.contains("SURVIVES"));
|
||||
if action == "stop" {
|
||||
ssh_run(
|
||||
&target,
|
||||
&format!("docker start {}", sh_quote(&reference)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
if action == "rm -f" {
|
||||
ssh_run(&target, &format!("docker run -d --name {} --entrypoint bash condaforge/miniforge3 -c 'exec sleep infinity'", sh_quote(&reference)), None).await.unwrap();
|
||||
assert!(unavailable(&target, container.as_ref().unwrap())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.contains("removed"));
|
||||
cancel(&target, container.as_ref().unwrap()).await.unwrap();
|
||||
resolve(&target, &reference).await.unwrap();
|
||||
}
|
||||
}
|
||||
eprintln!("SSH container lifecycle: passed");
|
||||
}
|
||||
|
||||
async fn supervisor_restart(target: &SshTarget, reference: &str, port: u16) {
|
||||
use crate::local::model::{LocalExperiment, LocalProject};
|
||||
use crate::store::{Store, StoredRun};
|
||||
let temp = crate::local::git::TemporaryDirectory::new("orx-supervisor-restart").unwrap();
|
||||
let root = temp.path();
|
||||
let ssh_dir = root.join(".ssh");
|
||||
std::fs::create_dir(&ssh_dir).unwrap();
|
||||
std::fs::write(ssh_dir.join("config"), format!("Host fixture\n HostName 127.0.0.1\n User root\n Port {port}\n IdentityFile {}\n StrictHostKeyChecking no\n UserKnownHostsFile /dev/null\n", std::env::var("ORX_SSH_TEST_KEY").unwrap())).unwrap();
|
||||
let store = Store::open_at(root.join("data")).unwrap();
|
||||
store
|
||||
.create_local_project(&LocalProject {
|
||||
id: "project".into(),
|
||||
name: "test".into(),
|
||||
slug: "test".into(),
|
||||
github_owner: String::new(),
|
||||
github_repo: String::new(),
|
||||
github_sync_enabled: false,
|
||||
baseline_branch: "main".into(),
|
||||
repo_path: root.to_string_lossy().into_owned(),
|
||||
run_command: None,
|
||||
paper_id: None,
|
||||
created_at: 1,
|
||||
updated_at: 1,
|
||||
})
|
||||
.unwrap();
|
||||
store
|
||||
.create_local_experiment(&LocalExperiment {
|
||||
id: "experiment".into(),
|
||||
project_id: "project".into(),
|
||||
parent_experiment_id: None,
|
||||
slug: "test".into(),
|
||||
branch_name: "main".into(),
|
||||
title: None,
|
||||
description: None,
|
||||
run_command: "echo ONCE; sleep 8; echo AFTER".into(),
|
||||
agent_status: "idle".into(),
|
||||
created_at: 1,
|
||||
updated_at: 1,
|
||||
chat_session_id: None,
|
||||
})
|
||||
.unwrap();
|
||||
let (dir, container) = launch(target, Some(reference), "echo ONCE; sleep 8; echo AFTER").await;
|
||||
let id = dir.rsplit('/').next().unwrap();
|
||||
let descriptor = serde_json::json!({"kind":"ssh_job","namespace":"fixture","jobId":dir,"sshContainer":container});
|
||||
store
|
||||
.upsert_run(&StoredRun {
|
||||
id: id.into(),
|
||||
experiment_id: "experiment".into(),
|
||||
project_id: "project".into(),
|
||||
status: "starting".into(),
|
||||
backend_json: descriptor.to_string(),
|
||||
command: "test".into(),
|
||||
created_at: 1,
|
||||
updated_at: 1,
|
||||
ended_at: None,
|
||||
exit_code: None,
|
||||
commit_sha: None,
|
||||
result_markdown: None,
|
||||
cancel_requested: false,
|
||||
chat_session_id: None,
|
||||
})
|
||||
.unwrap();
|
||||
// OpenSSH ignores HOME for its config lookup, so confine this override to the supervisor child.
|
||||
let bin = root.join("bin");
|
||||
std::fs::create_dir(&bin).unwrap();
|
||||
let ssh = bin.join("ssh");
|
||||
std::fs::write(
|
||||
&ssh,
|
||||
format!(
|
||||
"#!/bin/sh\nexec /usr/bin/ssh -F {} \"$@\"\n",
|
||||
sh_quote(&ssh_dir.join("config").to_string_lossy())
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&ssh, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||
}
|
||||
let executable = std::env::current_exe()
|
||||
.unwrap()
|
||||
.parent()
|
||||
.unwrap()
|
||||
.parent()
|
||||
.unwrap()
|
||||
.join(format!("orx{}", std::env::consts::EXE_SUFFIX));
|
||||
let spawn = || {
|
||||
let mut command = tokio::process::Command::new(&executable);
|
||||
command
|
||||
.args(["--no-telemetry", "supervise", id])
|
||||
.env("HOME", root)
|
||||
.env("ORX_DATA_DIR", root.join("data"))
|
||||
.env("ORX_CACHE_DIR", root.join("cache"))
|
||||
.env("XDG_CONFIG_HOME", root.join("config"))
|
||||
.env("ORX_NO_UPDATE_CHECK", "1")
|
||||
.env(
|
||||
"PATH",
|
||||
format!("{}:{}", bin.display(), std::env::var("PATH").unwrap()),
|
||||
)
|
||||
.stdout(std::process::Stdio::null())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.kill_on_drop(true);
|
||||
command.spawn().unwrap()
|
||||
};
|
||||
let mut first = spawn();
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
assert!(
|
||||
first.try_wait().unwrap().is_none(),
|
||||
"supervisor exited before restart"
|
||||
);
|
||||
first.kill().await.unwrap();
|
||||
first.wait().await.unwrap();
|
||||
let config = root.join("config/openresearch");
|
||||
std::fs::create_dir_all(&config).unwrap();
|
||||
std::fs::write(config.join("settings.json"), r#"{"ssh":{"defaultHost":"wrong-host","hosts":{"fixture":{"container":"wrong-container"}}}}"#).unwrap();
|
||||
let second = spawn();
|
||||
let result = tokio::time::timeout(Duration::from_secs(25), second.wait_with_output())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
result.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&result.stderr)
|
||||
);
|
||||
assert_eq!(store.get_run(id).unwrap().unwrap().status, "done");
|
||||
let output = logs(target, &dir).await;
|
||||
assert_eq!(output.matches("ONCE").count(), 1);
|
||||
assert!(output.contains("AFTER"));
|
||||
}
|
||||
@@ -106,6 +106,7 @@ pub async fn submit_local_hf_with_source(
|
||||
.await?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "hf_job".to_string(),
|
||||
namespace: Some(namespace.clone()),
|
||||
job_id: Some(job.id.clone()),
|
||||
|
||||
@@ -150,6 +150,7 @@ pub async fn submit_local_k8s_with_source(
|
||||
.await?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "k8s_job".to_string(),
|
||||
namespace: Some(namespace.clone()),
|
||||
job_id: Some(submitted.job_name.clone()),
|
||||
|
||||
@@ -118,6 +118,7 @@ async fn submit_controller_run(
|
||||
})?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: kind.to_string(),
|
||||
namespace: None,
|
||||
job_id: Some(dir.to_string_lossy().into_owned()),
|
||||
|
||||
@@ -110,6 +110,7 @@ pub async fn submit_local_modal_with_source(
|
||||
.await?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "modal_job".to_string(),
|
||||
namespace: Some(MODAL_APP.to_string()),
|
||||
job_id: Some(sandbox_id.clone()),
|
||||
|
||||
@@ -163,6 +163,7 @@ pub async fn submit_local_openresearch_with_source(
|
||||
.sandbox;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "openresearch_job".to_string(),
|
||||
namespace: Some(org_id),
|
||||
job_id: Some(sandbox.id.clone()),
|
||||
|
||||
@@ -123,6 +123,7 @@ pub async fn submit_local_ray_with_source(
|
||||
let watch = ray::job_url(&address, &submission_id);
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "ray_job".to_string(),
|
||||
namespace: Some(address.clone()),
|
||||
job_id: Some(submission_id.clone()),
|
||||
|
||||
@@ -106,6 +106,7 @@ pub async fn submit_local_slurm_with_source(
|
||||
&run_id,
|
||||
&source.path,
|
||||
&source.digest,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
let job_id = slurm::run_job(&slurm::SlurmJobSpec {
|
||||
@@ -122,6 +123,7 @@ pub async fn submit_local_slurm_with_source(
|
||||
.await?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: None,
|
||||
kind: "slurm_job".to_string(),
|
||||
namespace: Some(host.clone()),
|
||||
job_id: Some(job_id.clone()),
|
||||
|
||||
+51
-35
@@ -1,5 +1,5 @@
|
||||
//! Local SSH launch — the SSH twin of `local/k8s.rs`: run the experiment as a
|
||||
//! detached process on one of your own boxes over ssh. `--flavor` names an
|
||||
//! detached process on one of your own boxes over ssh. `--host` names an
|
||||
//! `~/.ssh/config` host alias (there's no hardware scheduler on a plain
|
||||
//! server). The run row lives in the local store only; a detached
|
||||
//! `orx supervise` watches the remote process.
|
||||
@@ -16,7 +16,7 @@ use crate::store::{now_ms, Store, StoredRun};
|
||||
pub async fn launch_local_ssh(args: &crate::ExpRunArgs) -> Result<()> {
|
||||
let run = submit_local_ssh(args).await?;
|
||||
let backend = BackendDescriptor::parse(&run.backend_json)?;
|
||||
println!("\u{2713} SSH job started.");
|
||||
println!("\u{2713} SSH run submitted.");
|
||||
println!(
|
||||
" host {} ({})",
|
||||
backend.namespace.as_deref().unwrap_or(""),
|
||||
@@ -31,7 +31,7 @@ pub async fn launch_local_ssh(args: &crate::ExpRunArgs) -> Result<()> {
|
||||
}
|
||||
|
||||
/// Submit the local experiment's run as a detached process on an ssh host and
|
||||
/// detach a supervisor. Requires `--backend ssh` and `--flavor <host>` where
|
||||
/// detach a supervisor. Requires `--backend ssh` and `--host <host>` where
|
||||
/// the host is an `~/.ssh/config` alias.
|
||||
pub async fn submit_local_ssh(args: &crate::ExpRunArgs) -> Result<StoredRun> {
|
||||
crate::compute::submit(args).await
|
||||
@@ -41,25 +41,8 @@ pub async fn submit_local_ssh_with_source(
|
||||
args: &crate::ExpRunArgs,
|
||||
source: SourceSnapshot,
|
||||
run_id: String,
|
||||
launch: &ssh::ResolvedLaunch,
|
||||
) -> Result<StoredRun> {
|
||||
if args.flavor.is_some() {
|
||||
return Err(anyhow!(
|
||||
"--backend ssh has no flavors — a machine is an address, not a shape. \
|
||||
Pass --host <alias> (an ~/.ssh/config alias)."
|
||||
));
|
||||
}
|
||||
if args.image.is_some() {
|
||||
return Err(anyhow!(
|
||||
"--image doesn't apply to --backend ssh — the run uses the host's own environment."
|
||||
));
|
||||
}
|
||||
let host = args.host.clone().ok_or_else(|| {
|
||||
anyhow!(
|
||||
"--backend ssh requires --host <alias> from the user's ~/.ssh/config. \
|
||||
The host needs git and bash."
|
||||
)
|
||||
})?;
|
||||
|
||||
let store = Store::open()?;
|
||||
let exp = store
|
||||
.get_local_experiment(&args.exp_id)?
|
||||
@@ -75,8 +58,19 @@ pub async fn submit_local_ssh_with_source(
|
||||
.or_else(|| project.run_command.clone().filter(|c| !c.trim().is_empty()))
|
||||
.ok_or_else(|| anyhow!("{}", crate::invocation::no_run_command(&project.id)))?;
|
||||
|
||||
let target = ssh::SshTarget::alias(&host);
|
||||
ssh::stage_source(&target, &run_id, &source.path, &source.digest).await?;
|
||||
let target = launch.target.clone();
|
||||
let container = launch
|
||||
.container
|
||||
.as_ref()
|
||||
.map(|container| container.for_run(&run_id));
|
||||
let remote_dir = ssh::stage_source(
|
||||
&target,
|
||||
&run_id,
|
||||
&source.path,
|
||||
&source.digest,
|
||||
container.as_ref(),
|
||||
)
|
||||
.await?;
|
||||
let script = crate::compute::staged_script(&run_command);
|
||||
|
||||
// The remote env: everything the user synced (API keys), plus the tokens
|
||||
@@ -86,17 +80,10 @@ pub async fn submit_local_ssh_with_source(
|
||||
env.entry("HF_TOKEN".to_string()).or_insert(hf_token);
|
||||
}
|
||||
|
||||
let remote_dir = ssh::run_job(&ssh::SshJobSpec {
|
||||
target: target.clone(),
|
||||
run_id: run_id.clone(),
|
||||
script,
|
||||
env,
|
||||
})
|
||||
.await?;
|
||||
|
||||
let mut descriptor = BackendDescriptor {
|
||||
ssh_container: container.clone(),
|
||||
kind: "ssh_job".to_string(),
|
||||
namespace: Some(host),
|
||||
namespace: Some(target.dest.clone()),
|
||||
job_id: Some(remote_dir.clone()),
|
||||
flavor: None,
|
||||
image: None,
|
||||
@@ -113,9 +100,38 @@ pub async fn submit_local_ssh_with_source(
|
||||
source_size: None,
|
||||
};
|
||||
source.apply_to_descriptor(&mut descriptor);
|
||||
if let Err(error) = crate::compute::record_submission_handle(&run_id, &descriptor) {
|
||||
let _ = ssh::cancel_job(&target, &remote_dir).await;
|
||||
return Err(error);
|
||||
if container.is_some() {
|
||||
crate::compute::record_submission_handle(&run_id, &descriptor)?;
|
||||
}
|
||||
let submission = ssh::run_job(&ssh::SshJobSpec {
|
||||
target: target.clone(),
|
||||
run_id: run_id.clone(),
|
||||
script,
|
||||
env,
|
||||
container: container.clone(),
|
||||
})
|
||||
.await;
|
||||
match submission {
|
||||
Err(error) if container.is_some() && error.is::<ssh::LaunchUncertain>() => {
|
||||
eprintln!("SSH launch response failed for run {run_id}: {error:#}. Reconciling the saved container handle.");
|
||||
}
|
||||
Err(error) => {
|
||||
store.update_status(
|
||||
&run_id,
|
||||
crate::store::RunStatus::Failed,
|
||||
Some(now_ms()),
|
||||
None,
|
||||
)?;
|
||||
store.set_result_markdown(&run_id, &format!("Compute submission failed: {error:#}"))?;
|
||||
return Err(error);
|
||||
}
|
||||
Ok(_) if container.is_none() => {
|
||||
if let Err(error) = crate::compute::record_submission_handle(&run_id, &descriptor) {
|
||||
let _ = ssh::cancel_job(&target, &remote_dir, None).await;
|
||||
return Err(error);
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
}
|
||||
let run = StoredRun {
|
||||
id: run_id.clone(),
|
||||
|
||||
+7
-1
@@ -492,11 +492,17 @@ pub struct ExpRunArgs {
|
||||
/// you belong to exactly one org.
|
||||
#[arg(long)]
|
||||
pub org: Option<String>,
|
||||
/// The ~/.ssh/config host alias to run on (with `--backend ssh`), or the
|
||||
/// The ~/.ssh/config host alias (SSH defaults to its saved default host), or the
|
||||
/// cluster login node (with `--backend slurm`; defaults to the slurm
|
||||
/// settings' host).
|
||||
#[arg(long)]
|
||||
pub host: Option<String>,
|
||||
/// Existing running Docker container on the SSH host (name or ID).
|
||||
#[arg(long, conflicts_with = "no_container")]
|
||||
pub container: Option<String>,
|
||||
/// Run directly on the SSH host, overriding its saved container.
|
||||
#[arg(long)]
|
||||
pub no_container: bool,
|
||||
/// Repo-relative path to the k8s manifest on the experiment branch (with
|
||||
/// `--backend k8s`; default .orx/k8s.yaml). The manifest declares the run's
|
||||
/// resources — image, GPUs, topology — and orx injects the run script, env
|
||||
|
||||
@@ -107,6 +107,8 @@ fn flush_window() -> Duration {
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub(crate) struct Settings {
|
||||
#[serde(default)]
|
||||
pub ssh: crate::config::SshSettings,
|
||||
/// Random anonymous id (uuid v4), generated once on first enabled run.
|
||||
#[serde(default)]
|
||||
pub install_id: Option<String>,
|
||||
@@ -319,6 +321,33 @@ fn settings_path() -> PathBuf {
|
||||
crate::config::config_dir().join("settings.json")
|
||||
}
|
||||
|
||||
pub(crate) fn ssh_settings() -> crate::error::Result<crate::config::SshSettings> {
|
||||
let raw = match std::fs::read_to_string(settings_path()) {
|
||||
Ok(raw) => raw,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Default::default()),
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
let settings: Settings = serde_json::from_str(&raw)
|
||||
.map_err(|error| crate::error::anyhow!("Cannot read SSH settings: {error}"))?;
|
||||
for options in settings.ssh.hosts.values() {
|
||||
crate::jobs::ssh::validate_host_options(options)?;
|
||||
}
|
||||
Ok(settings.ssh)
|
||||
}
|
||||
|
||||
pub(crate) fn set_ssh_host(
|
||||
host: String,
|
||||
options: crate::config::SshHostSettings,
|
||||
) -> std::io::Result<()> {
|
||||
mutate_settings(|settings| {
|
||||
settings.ssh.hosts.insert(host, options);
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn set_ssh_default(host: Option<String>) -> std::io::Result<()> {
|
||||
mutate_settings(|settings| settings.ssh.default_host = host)
|
||||
}
|
||||
|
||||
fn outbox_dir() -> PathBuf {
|
||||
crate::config::config_dir().join("telemetry-outbox")
|
||||
}
|
||||
@@ -1437,6 +1466,42 @@ mod tests {
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ssh_settings_preserve_siblings_and_reject_corrupt_config() {
|
||||
use crate::config::SshHostSettings;
|
||||
let _g = EnvGuard::new(OPT_VARS);
|
||||
let dir = std::env::temp_dir().join(format!("orx-tel-ssh-{}", uuid::Uuid::new_v4()));
|
||||
std::env::set_var("XDG_CONFIG_HOME", &dir);
|
||||
assert!(ssh_settings().unwrap().hosts.is_empty());
|
||||
set_persisted_disabled(true).unwrap();
|
||||
let options = SshHostSettings {
|
||||
container: Some("research".into()),
|
||||
};
|
||||
set_ssh_host("lab".into(), options.clone()).unwrap();
|
||||
set_ssh_default(Some("lab".into())).unwrap();
|
||||
set_compute_default(Some("ssh".into()), None).unwrap();
|
||||
let settings = ssh_settings().unwrap();
|
||||
assert_eq!(settings.default_host.as_deref(), Some("lab"));
|
||||
assert_eq!(settings.hosts["lab"], options);
|
||||
assert_eq!(load_settings().unwrap().telemetry_disabled, Some(true));
|
||||
set_ssh_host("lab".into(), SshHostSettings::default()).unwrap();
|
||||
set_ssh_default(None).unwrap();
|
||||
assert_eq!(
|
||||
ssh_settings().unwrap().hosts["lab"],
|
||||
SshHostSettings::default()
|
||||
);
|
||||
for raw in [
|
||||
r#"{"ssh":null}"#,
|
||||
r#"{"ssh":{"hosts":{"lab":{"container":""}}}}"#,
|
||||
r#"{"ssh":{"hosts":[]}}"#,
|
||||
"{",
|
||||
] {
|
||||
std::fs::write(settings_path(), raw).unwrap();
|
||||
assert!(ssh_settings().is_err(), "{raw}");
|
||||
}
|
||||
std::fs::remove_dir_all(dir).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn compute_default_roundtrip_preserves_siblings() {
|
||||
// Same single-writer contract as data_dir: the Compute settings persist
|
||||
|
||||
+210
-210
File diff suppressed because one or more lines are too long
Vendored
+1
-1
@@ -49,7 +49,7 @@
|
||||
html { background: #ffffff; }
|
||||
html[data-theme="dark"] { background: #0e0c0c; }
|
||||
</style>
|
||||
<script type="module" crossorigin src="/assets/index-Ph8YqAap.js"></script>
|
||||
<script type="module" crossorigin src="/assets/index-C3cRg0nD.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-DCbOCW9V.css">
|
||||
</head>
|
||||
<body>
|
||||
|
||||
+6
-1
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "اختيار وكيل برمجة آخر",
|
||||
"image_zoom_out": "تصغير",
|
||||
"image_zoom_in": "تكبير",
|
||||
"image_zoom_reset": "إعادة ضبط التكبير للملاءمة"
|
||||
"image_zoom_reset": "إعادة ضبط التكبير للملاءمة",
|
||||
"ssh_default_host": "مضيف SSH الافتراضي",
|
||||
"ssh_run_in": "التشغيل في",
|
||||
"ssh_direct_host": "المضيف مباشرة",
|
||||
"ssh_existing_container": "حاوية موجودة",
|
||||
"ssh_container_reference": "اسم الحاوية أو معرّفها"
|
||||
}
|
||||
|
||||
+6
-1
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "Choose another coding agent",
|
||||
"image_zoom_out": "Zoom out",
|
||||
"image_zoom_in": "Zoom in",
|
||||
"image_zoom_reset": "Reset zoom to fit"
|
||||
"image_zoom_reset": "Reset zoom to fit",
|
||||
"ssh_default_host": "Default SSH host",
|
||||
"ssh_run_in": "Run in",
|
||||
"ssh_direct_host": "Direct host",
|
||||
"ssh_existing_container": "Existing container",
|
||||
"ssh_container_reference": "Container name or ID"
|
||||
}
|
||||
|
||||
+6
-1
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "Elegir otro agente de programación",
|
||||
"image_zoom_out": "Alejar",
|
||||
"image_zoom_in": "Acercar",
|
||||
"image_zoom_reset": "Restablecer zoom para ajustar"
|
||||
"image_zoom_reset": "Restablecer zoom para ajustar",
|
||||
"ssh_default_host": "Servidor SSH predeterminado",
|
||||
"ssh_run_in": "Ejecutar en",
|
||||
"ssh_direct_host": "Servidor directamente",
|
||||
"ssh_existing_container": "Contenedor existente",
|
||||
"ssh_container_reference": "Nombre o ID del contenedor"
|
||||
}
|
||||
|
||||
+6
-1
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "انتخاب یک عامل برنامهنویسی دیگر",
|
||||
"image_zoom_out": "کوچکنمایی",
|
||||
"image_zoom_in": "بزرگنمایی",
|
||||
"image_zoom_reset": "بازنشانی بزرگنمایی برای جا شدن"
|
||||
"image_zoom_reset": "بازنشانی بزرگنمایی برای جا شدن",
|
||||
"ssh_default_host": "میزبان پیشفرض SSH",
|
||||
"ssh_run_in": "اجرا در",
|
||||
"ssh_direct_host": "مستقیماً روی میزبان",
|
||||
"ssh_existing_container": "کانتینر موجود",
|
||||
"ssh_container_reference": "نام یا شناسهٔ کانتینر"
|
||||
}
|
||||
|
||||
+6
-1
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "दूसरा कोडिंग एजेंट चुनें",
|
||||
"image_zoom_out": "ज़ूम आउट",
|
||||
"image_zoom_in": "ज़ूम इन",
|
||||
"image_zoom_reset": "फ़िट करने के लिए ज़ूम रीसेट करें"
|
||||
"image_zoom_reset": "फ़िट करने के लिए ज़ूम रीसेट करें",
|
||||
"ssh_default_host": "डिफ़ॉल्ट SSH होस्ट",
|
||||
"ssh_run_in": "यहाँ चलाएँ",
|
||||
"ssh_direct_host": "सीधे होस्ट पर",
|
||||
"ssh_existing_container": "मौजूदा कंटेनर",
|
||||
"ssh_container_reference": "कंटेनर का नाम या ID"
|
||||
}
|
||||
|
||||
@@ -1187,5 +1187,10 @@
|
||||
"onboarding_choose_another_agent": "选择其他编程智能体",
|
||||
"image_zoom_out": "缩小",
|
||||
"image_zoom_in": "放大",
|
||||
"image_zoom_reset": "重置为适合窗口"
|
||||
"image_zoom_reset": "重置为适合窗口",
|
||||
"ssh_default_host": "默认 SSH 主机",
|
||||
"ssh_run_in": "运行位置",
|
||||
"ssh_direct_host": "直接在主机上",
|
||||
"ssh_existing_container": "现有容器",
|
||||
"ssh_container_reference": "容器名称或 ID"
|
||||
}
|
||||
|
||||
+18
-3
@@ -953,6 +953,7 @@ export const moveDataDir = (path: string) =>
|
||||
post<{ started: boolean }>("/api/settings/data-dir/move", { path });
|
||||
|
||||
export interface SshHost {
|
||||
container?: string | null;
|
||||
host: string;
|
||||
hostname?: string;
|
||||
user?: string;
|
||||
@@ -962,8 +963,16 @@ export interface SshHost {
|
||||
lastTest?: SshPreflight;
|
||||
}
|
||||
|
||||
export const getSshHosts = (signal?: AbortSignal) =>
|
||||
get<{ hosts: SshHost[] }>("/api/settings/ssh", signal).then((r) => r.hosts);
|
||||
export interface SshSettings { hosts: SshHost[]; defaultHost: string | null }
|
||||
export const getSshSettings = (signal?: AbortSignal) => get<SshSettings>("/api/settings/ssh", signal);
|
||||
export const saveSshHost = (body: { host: string; container: string | null }) =>
|
||||
post<{ ok: boolean }>("/api/settings/ssh", body);
|
||||
export const saveSshDefault = (host: string | null) => post<{ ok: boolean }>("/api/settings/ssh/default", { host });
|
||||
export interface SshExecutionPreflight extends SshPreflight {
|
||||
container: { reference: string; ready: boolean; error: string | null } | null;
|
||||
}
|
||||
export const testSshExecution = (host: string, container: string | null) =>
|
||||
post<SshExecutionPreflight>("/api/settings/ssh/preflight", { host, container });
|
||||
|
||||
export interface SshConfigFile {
|
||||
path: string;
|
||||
@@ -2123,7 +2132,13 @@ export function backendDetail(backend: Run["backend"]): string {
|
||||
if (typeof backend.manifest === "string" && backend.manifest) return backend.manifest;
|
||||
// Ray's namespace is the whole Jobs URL — too long for a badge.
|
||||
if (backendKind(backend) === "ray_job") return "";
|
||||
if (typeof backend.namespace === "string" && backend.namespace) return backend.namespace;
|
||||
if (typeof backend.namespace === "string" && backend.namespace) {
|
||||
const container = backend.sshContainer;
|
||||
if (container && typeof container === "object" && "reference" in container && typeof container.reference === "string") {
|
||||
return `${backend.namespace} / ${container.reference}`;
|
||||
}
|
||||
return backend.namespace;
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ import { createPortal } from "react-dom";
|
||||
import { createRemoteSession } from "../api";
|
||||
import { m } from "../paraglide/messages.js";
|
||||
import { getLocale } from "../paraglide/runtime.js";
|
||||
import { getSshHostsQuery, listRemoteSessionsQuery } from "../queries/settings";
|
||||
import { getSshSettingsQuery, listRemoteSessionsQuery } from "../queries/settings";
|
||||
import { getThemePreference } from "../theme";
|
||||
import { useDialogFocus } from "./useDialogFocus";
|
||||
import { Button, IconButton, Input, showAlert, Spinner } from "./ui";
|
||||
@@ -19,9 +19,9 @@ export function RemoteHostDialog({
|
||||
}) {
|
||||
const createRemoteSessionMutation = useMutation({ mutationFn: (args: Parameters<typeof createRemoteSession>) => createRemoteSession(...args) });
|
||||
|
||||
const hostsQuery = useQuery(getSshHostsQuery());
|
||||
const hostsQuery = useQuery(getSshSettingsQuery());
|
||||
const sessionsQuery = useQuery(listRemoteSessionsQuery());
|
||||
const hosts = hostsQuery.data ?? null;
|
||||
const hosts = hostsQuery.data?.hosts ?? null;
|
||||
const sessions = sessionsQuery.data ?? [];
|
||||
const [query, setQuery] = useState("");
|
||||
const loadError = !hosts ? hostsQuery.error?.message ?? sessionsQuery.error?.message ?? null : null;
|
||||
|
||||
@@ -16,7 +16,7 @@ import {
|
||||
getK8sSettingsQuery,
|
||||
getModalSettingsQuery,
|
||||
getSshMasterStatusQuery,
|
||||
getSshHostsQuery,
|
||||
getSshSettingsQuery,
|
||||
getSlurmSettingsQuery,
|
||||
getRaySettingsQuery,
|
||||
getOpenResearchSettingsQuery,
|
||||
@@ -103,6 +103,8 @@ import {
|
||||
type SlurmPreflight,
|
||||
type SlurmSettings,
|
||||
type SshPreflight,
|
||||
type SshExecutionPreflight,
|
||||
testSshExecution,
|
||||
applyUpdate,
|
||||
harnessModelLabel,
|
||||
installCli,
|
||||
@@ -126,6 +128,7 @@ import { HarnessLogo } from "./HarnessLogo";
|
||||
import { LocalModelSetup } from "./LocalModelSetup";
|
||||
import { StatusBadge } from "./StatusBadge";
|
||||
import { OpenResearchSetupTerminal, SettingsCommandTerminal, SshConnectTerminal, SshTerminalTranscript } from "./SshConnectTerminal";
|
||||
import { SshExecutionSettings, SshDefaultHost } from "./SshExecutionSettings";
|
||||
import { SshConfigDialog } from "./SshConfigDialog";
|
||||
import {
|
||||
Badge,
|
||||
@@ -790,7 +793,7 @@ function useSshMasterStatuses(hosts: string[]) {
|
||||
return [statuses, markRunning] as const;
|
||||
}
|
||||
|
||||
function HostTestCell({ test, connecting, masterRunning }: { test: SshPreflight | undefined; connecting: boolean; masterRunning: boolean | null | undefined }) {
|
||||
function HostTestCell({ test, connecting, masterRunning, containerFailed = false }: { test: SshPreflight | undefined; connecting: boolean; masterRunning: boolean | null | undefined; containerFailed?: boolean }) {
|
||||
if (connecting)
|
||||
return (
|
||||
<span role="status">
|
||||
@@ -800,7 +803,7 @@ function HostTestCell({ test, connecting, masterRunning }: { test: SshPreflight
|
||||
if (test === undefined) return <Badge className={CONNECTION_BADGE_IDLE_CLASS}>{m.settings_page_not_checked()}</Badge>;
|
||||
const missingTools = test.missingTools ?? [];
|
||||
const disconnected = test.reachable && test.toolsFound && masterRunning === false;
|
||||
const badge = !test.reachable ? (
|
||||
const badge = !test.reachable || containerFailed ? (
|
||||
<Badge className="rounded-sm" variant="error">{m.settings_page_failed()}</Badge>
|
||||
) : !test.toolsFound ? (
|
||||
<Badge className="rounded-sm" variant="error">
|
||||
@@ -822,11 +825,14 @@ function HostTestCell({ test, connecting, masterRunning }: { test: SshPreflight
|
||||
}
|
||||
|
||||
function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
const hostsOptions = getSshHostsQuery();
|
||||
const hostsOptions = getSshSettingsQuery();
|
||||
const hostsQuery = useQuery(hostsOptions);
|
||||
const hosts = hostsQuery.data ?? (hostsQuery.isError ? [] : null);
|
||||
const hosts = hostsQuery.data?.hosts ?? (hostsQuery.isError ? [] : null);
|
||||
const [configOpen, setConfigOpen] = useState(false);
|
||||
const [tests, setTests] = useState<Record<string, SshPreflight>>({});
|
||||
const [tests, setTests] = useState<Record<string, SshExecutionPreflight>>({});
|
||||
const [drafts, setDrafts] = useState<Record<string, string | null>>({});
|
||||
const activeAttempt = useRef(0);
|
||||
const [probing, setProbing] = useState(false);
|
||||
const [expandedHosts, setExpandedHosts] = useState<Record<string, boolean>>({});
|
||||
const [connectingHost, setConnectingHost] = useState<string | null>(null);
|
||||
const [connectionFailed, setConnectionFailed] = useState(false);
|
||||
@@ -840,6 +846,8 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
const [masterRunning, markMasterRunning] = useSshMasterStatuses(checkedHosts);
|
||||
|
||||
function connect(host: string) {
|
||||
activeAttempt.current += 1;
|
||||
setProbing(false);
|
||||
setConnectionFailed(false);
|
||||
setConnectionAttempt((attempt) => attempt + 1);
|
||||
setConnectingHost(host);
|
||||
@@ -847,6 +855,8 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
}
|
||||
|
||||
function cancelConnect() {
|
||||
activeAttempt.current += 1;
|
||||
setProbing(false);
|
||||
setConnectionFailed(false);
|
||||
setConnectingHost(null);
|
||||
}
|
||||
@@ -857,8 +867,10 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
|
||||
return (
|
||||
<>
|
||||
<div className="mb-3 flex justify-end">
|
||||
<Button variant="ghost" onClick={() => setConfigOpen(true)}>
|
||||
{hostsQuery.error && <p className="text-sm text-accent-red">{hostsQuery.error.message}</p>}
|
||||
<div className="mb-3 flex flex-wrap items-center justify-between gap-3">
|
||||
{!remote && hostsQuery.data && <SshDefaultHost settings={hostsQuery.data} />}
|
||||
<Button variant="ghost" className="ms-auto" onClick={() => setConfigOpen(true)}>
|
||||
<Settings size={14} /> {m.ssh_configure_hosts()}
|
||||
</Button>
|
||||
</div>
|
||||
@@ -872,45 +884,51 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
<div className="border-y border-border-variant divide-y divide-border-variant">
|
||||
{hosts.map((h) => {
|
||||
// Session-local result wins; the persisted one covers restarts.
|
||||
const hostTest = tests[h.host] ?? h.lastTest;
|
||||
const reference = drafts[h.host] === undefined ? h.container ?? null : drafts[h.host];
|
||||
const executionTest = tests[h.host];
|
||||
const containerTest = reference !== null && executionTest?.container?.reference === reference.trim()
|
||||
? executionTest : undefined;
|
||||
const hostTest = reference !== null ? containerTest
|
||||
: executionTest?.container ? h.lastTest : executionTest ?? h.lastTest;
|
||||
const containerError = containerTest?.container?.error;
|
||||
const containerFailed = containerTest?.container?.ready === false;
|
||||
const connectionError = hostTest?.error || containerError;
|
||||
const connecting = connectingHost === h.host;
|
||||
const open = expandedHosts[h.host] ?? false;
|
||||
const hasTerminal = !remote && (connecting || hostTest?.reachable === false);
|
||||
const hasTerminal = !remote && (connecting || Boolean(connectionError));
|
||||
const address =
|
||||
`${h.user ? `${h.user}@` : ""}${h.hostname ?? h.host}${h.port ? `:${h.port}` : ""}`;
|
||||
return (
|
||||
<div key={h.host}>
|
||||
<div
|
||||
className="flex items-center gap-3 py-3 px-2"
|
||||
className="flex items-center gap-3 py-3"
|
||||
>
|
||||
<div className="flex min-w-0 flex-1 items-center gap-2.5">
|
||||
{hasTerminal ? (
|
||||
<button
|
||||
type="button"
|
||||
className="flex-none inline-flex items-center p-0.5 rounded-sm [&:hover]:bg-panel"
|
||||
aria-expanded={open}
|
||||
aria-label={open ? m.a11y_collapse_item({ name: ltr(h.host) }) : m.a11y_expand_item({ name: ltr(h.host) })}
|
||||
onClick={(event) => {
|
||||
event.stopPropagation();
|
||||
toggle(h.host, open);
|
||||
}}
|
||||
>
|
||||
<ChevronDown
|
||||
size={15}
|
||||
className={`text-muted transition-transform duration-120 ease-standard${open ? " rotate-180" : ""}`}
|
||||
/>
|
||||
</button>
|
||||
) : (
|
||||
<span className="w-5 flex-none" aria-hidden="true" />
|
||||
)}
|
||||
<div className="min-w-0">
|
||||
<div className="truncate text-base font-medium text-text" title={h.host}>{h.host}</div>
|
||||
<div className="mt-1 truncate text-sm text-subtext" title={address}>{address}</div>
|
||||
</div>
|
||||
</div>
|
||||
{!remote && <div className="grid flex-none grid-cols-[8.5rem_5rem] items-center gap-x-12">
|
||||
<div className="text-start">
|
||||
<HostTestCell test={hostTest} connecting={connecting && !connectionFailed} masterRunning={masterRunning[h.host]} />
|
||||
<div className="flex items-center gap-2 text-start">
|
||||
<HostTestCell test={hostTest} connecting={connecting && !connectionFailed} masterRunning={masterRunning[h.host]} containerFailed={containerFailed} />
|
||||
{hasTerminal && (
|
||||
<button
|
||||
type="button"
|
||||
className="flex-none inline-flex items-center p-0.5 rounded-sm [&:hover]:bg-panel"
|
||||
aria-expanded={open}
|
||||
aria-label={open ? m.a11y_collapse_item({ name: ltr(h.host) }) : m.a11y_expand_item({ name: ltr(h.host) })}
|
||||
onClick={(event) => {
|
||||
event.stopPropagation();
|
||||
toggle(h.host, open);
|
||||
}}
|
||||
>
|
||||
<ChevronDown
|
||||
size={18}
|
||||
className={`text-muted transition-transform duration-120 ease-standard${open ? " rotate-180" : ""}`}
|
||||
/>
|
||||
</button>
|
||||
)}
|
||||
</div>
|
||||
<Button size="small"
|
||||
type="button"
|
||||
@@ -920,13 +938,13 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
if (connecting && !connectionFailed) cancelConnect();
|
||||
else connect(h.host);
|
||||
}}
|
||||
disabled={!connecting && connectingHost !== null && !connectionFailed}
|
||||
disabled={!connecting && ((connectingHost !== null && !connectionFailed) || (reference !== null && !reference.trim()))}
|
||||
>
|
||||
{connecting
|
||||
? connectionFailed
|
||||
? m.app_retry()
|
||||
: m.settings_page_cancel()
|
||||
: hostTest?.reachable === false
|
||||
: hostTest?.reachable === false || containerFailed
|
||||
? m.app_retry()
|
||||
: hostTest
|
||||
? m.settings_reconnect()
|
||||
@@ -934,12 +952,22 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
</Button>
|
||||
</div>}
|
||||
</div>
|
||||
{!remote && <SshExecutionSettings host={h} connecting={connecting && !connectionFailed} reference={reference}
|
||||
onChange={(value) => {
|
||||
if (connecting) cancelConnect();
|
||||
setDrafts((drafts) => ({ ...drafts, [h.host]: value }));
|
||||
setTests((tests) => {
|
||||
const next = { ...tests };
|
||||
delete next[h.host];
|
||||
return next;
|
||||
});
|
||||
}} />}
|
||||
{hasTerminal && (open || connecting) && (
|
||||
<div className={`border-t border-t-border-variant py-3 pe-2 ps-10${open ? "" : " hidden"}`}>
|
||||
{!connecting && hostTest?.error && (
|
||||
<SshTerminalTranscript host={h.host} transcript={hostTest.error} />
|
||||
<div className={`border-t border-t-border-variant py-3${open ? "" : " hidden"}`}>
|
||||
{!connecting && connectionError && (
|
||||
<SshTerminalTranscript host={h.host} transcript={connectionError} />
|
||||
)}
|
||||
{connecting && (
|
||||
{connecting && !probing && (
|
||||
<SshConnectTerminal
|
||||
key={connectionAttempt}
|
||||
host={h.host}
|
||||
@@ -947,16 +975,40 @@ function SshSection({ remote = false }: { remote?: boolean }) {
|
||||
active={open}
|
||||
onComplete={(complete) => {
|
||||
if (complete.backend !== "ssh") return;
|
||||
setTests((tests) => ({ ...tests, [h.host]: complete.result }));
|
||||
markMasterRunning(h.host);
|
||||
setConnectionFailed(false);
|
||||
setConnectingHost(null);
|
||||
const attempt = activeAttempt.current;
|
||||
if (reference === null) {
|
||||
setTests((tests) => ({ ...tests, [h.host]: { ...complete.result, container: null } }));
|
||||
setConnectionFailed(false);
|
||||
setConnectingHost(null);
|
||||
return;
|
||||
}
|
||||
setProbing(true);
|
||||
void testSshExecution(h.host, reference.trim()).then((result) => {
|
||||
if (activeAttempt.current !== attempt) return;
|
||||
setTests((tests) => ({ ...tests, [h.host]: {
|
||||
...result,
|
||||
container: result.container ?? { reference: reference.trim(), ready: false, error: result.error ?? null },
|
||||
} }));
|
||||
}).catch((error: unknown) => {
|
||||
if (activeAttempt.current !== attempt) return;
|
||||
setTests((tests) => ({ ...tests, [h.host]: {
|
||||
...complete.result,
|
||||
container: { reference: reference.trim(), ready: false, error: error instanceof Error ? error.message : String(error) },
|
||||
} }));
|
||||
}).finally(() => {
|
||||
if (activeAttempt.current !== attempt) return;
|
||||
setProbing(false);
|
||||
setConnectionFailed(false);
|
||||
setConnectingHost(null);
|
||||
});
|
||||
}}
|
||||
onError={(error) => {
|
||||
setConnectionFailed(true);
|
||||
setTests((tests) => ({
|
||||
...tests,
|
||||
[h.host]: {
|
||||
container: reference === null ? null : { reference: reference.trim(), ready: false, error },
|
||||
reachable: false,
|
||||
toolsFound: false,
|
||||
missingTools: [],
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { useId, useState } from "react";
|
||||
import { useMutation } from "@tanstack/react-query";
|
||||
import { saveSshDefault, saveSshHost, type SshHost, type SshSettings } from "../api";
|
||||
import { m } from "../paraglide/messages.js";
|
||||
import { OptionPicker } from "./ModelPicker";
|
||||
import { Input, showAlert } from "./ui";
|
||||
|
||||
export function SshDefaultHost({ settings }: { settings: SshSettings }) {
|
||||
const [saving, setSaving] = useState(false);
|
||||
return <label className="inline-flex max-w-full flex-wrap items-center gap-2 text-sm text-subtext">
|
||||
{m.ssh_default_host()}
|
||||
<span className="inline-block w-max max-w-full">
|
||||
<OptionPicker variant="field" dropDown value={settings.defaultHost ?? ""} disabled={saving}
|
||||
choices={[
|
||||
{ id: "", label: m.settings_page_not_set_pass_host_per_launch() },
|
||||
...(settings.defaultHost && !settings.hosts.some((host) => host.host === settings.defaultHost)
|
||||
? [{ id: settings.defaultHost, label: settings.defaultHost }] : []),
|
||||
...settings.hosts.map((host) => ({ id: host.host, label: host.host })),
|
||||
]}
|
||||
onSelect={(host) => {
|
||||
setSaving(true);
|
||||
void saveSshDefault(host || null).catch((error: unknown) => {
|
||||
showAlert(error instanceof Error ? error.message : String(error), "error");
|
||||
}).finally(() => setSaving(false));
|
||||
}} />
|
||||
</span>
|
||||
</label>;
|
||||
}
|
||||
|
||||
export function SshExecutionSettings({ host, connecting, reference, onChange }: {
|
||||
host: SshHost;
|
||||
connecting: boolean;
|
||||
reference: string | null;
|
||||
onChange: (reference: string | null) => void;
|
||||
}) {
|
||||
const id = useId();
|
||||
const save = useMutation({
|
||||
mutationFn: saveSshHost,
|
||||
scope: { id: `ssh-settings-${host.host}` },
|
||||
});
|
||||
function change(value: string | null) {
|
||||
onChange(value);
|
||||
save.reset();
|
||||
if (value !== null && !value.trim()) return;
|
||||
save.mutate({ host: host.host, container: value?.trim() ?? null });
|
||||
}
|
||||
|
||||
return <details className="pb-3">
|
||||
<summary className="w-fit cursor-pointer rounded-sm text-sm text-subtext focus-visible:outline-2 focus-visible:outline-text">{m.new_project_advanced()}</summary>
|
||||
<div className="mt-3 grid max-w-xl gap-3">
|
||||
<label className="text-sm text-subtext">
|
||||
{m.ssh_run_in()}
|
||||
<OptionPicker variant="field" dropDown value={reference !== null ? "container" : "host"} disabled={connecting}
|
||||
choices={[{ id: "host", label: m.ssh_direct_host() }, { id: "container", label: m.ssh_existing_container() }]}
|
||||
onSelect={(value) => change(value === "container" ? "" : null)} />
|
||||
</label>
|
||||
{reference !== null && <label className="text-sm text-subtext" htmlFor={`${id}-container`}>
|
||||
{m.ssh_container_reference()}
|
||||
<Input id={`${id}-container`} value={reference} disabled={connecting} required
|
||||
onChange={(event) => change(event.target.value)} />
|
||||
</label>}
|
||||
{save.error && <p className="m-0 text-sm text-accent-red" role="status">{save.error.message}</p>}
|
||||
</div>
|
||||
</details>;
|
||||
}
|
||||
@@ -67,7 +67,7 @@ const settingsFamilies: Record<string, readonly string[]> = {
|
||||
ray: ["getRaySettings", "getComputeSettings"],
|
||||
env: ["getEnvVars", "getHfSettings", "getTinkerSettings", "getModalSettings", "getSlurmSettings", "getRaySettings", "getK8sSettings", "getOpenResearchSettings", "getComputeSettings", "getHarnesses"],
|
||||
"data-dir": ["getDataDir"],
|
||||
ssh: ["getSshHosts", "getSshConfig", "getSshMasterStatus", "getComputeSettings"],
|
||||
ssh: ["getSshSettings", "getSshConfig", "getSshMasterStatus", "getComputeSettings"],
|
||||
compute: ["getComputeSettings"],
|
||||
git: ["githubAccount", "repoAccess", "getProjectGitStatus"],
|
||||
profile: ["getProfile"],
|
||||
|
||||
@@ -69,10 +69,10 @@ export const getDataDirQuery = () => queryOptions({
|
||||
staleTime: 300_000,
|
||||
});
|
||||
|
||||
export const getSshHostsQuery = () => queryOptions({
|
||||
queryKey: workspaceKey("getSshHosts"),
|
||||
queryFn: ({ signal }) => api.getSshHosts(signal),
|
||||
staleTime: 300_000,
|
||||
export const getSshSettingsQuery = () => queryOptions({
|
||||
queryKey: workspaceKey("getSshSettings"),
|
||||
queryFn: ({ signal }) => api.getSshSettings(signal),
|
||||
staleTime: 30_000,
|
||||
});
|
||||
|
||||
export const getSshConfigQuery = () => queryOptions({
|
||||
|
||||
@@ -119,7 +119,7 @@ for (const outcome of ["success", "failure", "popup-blocked"]) {
|
||||
react: state.react,
|
||||
"react-dom": { createPortal: (tree) => tree },
|
||||
"@tanstack/react-query": {
|
||||
useQuery: ({ kind }) => ({ data: kind === "hosts" ? [{ host: "research" }] : [] }),
|
||||
useQuery: ({ kind }) => ({ data: kind === "hosts" ? { hosts: [{ host: "research" }], defaultHost: null } : [] }),
|
||||
useMutation: () => ({ mutateAsync: async (args) => {
|
||||
calls.push(["connect", ...args]);
|
||||
if (outcome === "failure") throw new Error("connection failed");
|
||||
@@ -129,7 +129,7 @@ for (const outcome of ["success", "failure", "popup-blocked"]) {
|
||||
"lucide-react": { SlidersHorizontal: "Icon", X: "Icon" },
|
||||
"../api": {}, "../paraglide/messages.js": messages,
|
||||
"../paraglide/runtime.js": { getLocale: () => "en" },
|
||||
"../queries/settings": { getSshHostsQuery: () => ({ kind: "hosts" }), listRemoteSessionsQuery: () => ({ kind: "sessions" }) },
|
||||
"../queries/settings": { getSshSettingsQuery: () => ({ kind: "hosts" }), listRemoteSessionsQuery: () => ({ kind: "sessions" }) },
|
||||
"../theme": { getThemePreference: () => "dark" },
|
||||
"./useDialogFocus": { useDialogFocus: () => {} },
|
||||
"./ui": {
|
||||
|
||||
Reference in New Issue
Block a user