Clean up older remote run dirs and unused source tars (OR-575)

agent-skills/orx-compute/references/configuration.md

@@ -12,6 +12,7 @@ orx compute configure ssh --host lab --container research
 orx compute configure ssh --host lab --clear container
 orx compute configure ssh --host lab --remote-root /data/me/orx  # runs + source cache; default ~/.orx
 orx compute configure ssh --host lab --clear remote-root
+orx compute configure ssh --host lab --keep-remote-runs true   # keep every finished run dir on the host
 orx compute test ssh --host lab                     # tests saved container too
 orx compute test ssh --host lab --no-container      # explicitly tests host
 orx compute connect ssh --host lab                 # interactive login/MFA
@@ -20,6 +21,7 @@ orx compute configure slurm --cpus-per-task 8 --mem 64G   # defaults; exp run --
 orx compute configure slurm --clear cpus-per-task --clear mem
 orx compute configure slurm --clear time-limit     # cluster chooses
 orx compute configure slurm --remote-root /scratch/me/orx   # shared FS; --clear remote-root for ~/.orx
+orx compute configure slurm --keep-remote-runs true          # keep every finished run dir on the cluster
 orx compute connect slurm --host cluster
 orx compute configure k8s --context research --namespace experiments
 orx compute configure ray --address http://127.0.0.1:8265

agent-skills/orx-compute/references/slurm.md

@@ -27,5 +27,14 @@ orx exp run <expId> --backend slurm
   absolute directory, such as scratch space when home quota is small. It must
   be on a filesystem the compute nodes share. Each run keeps the root it was
   submitted with; `--clear remote-root` restores the default for new runs.
+- Each project's three newest SSH and Slurm runs, counted together across
+  hosts and including unfinished ones, stay whole. Once an older finished
+  run's log is collected, the running OpenResearch app removes its remote run
+  dir except `tensorboard/`. It also removes orx's cached source archives that
+  no active run uses and that have been idle for an hour, then checks again an
+  hour later. Anything the job wrote into `repo/` goes with it. To keep every run
+  on the cluster, set `orx compute configure slurm --keep-remote-runs true`
+  (it covers runs not yet cleaned up), or write outputs outside the run dir.
+  Deleting a project removes its runs' remote dirs entirely regardless.
 - A detached `orx supervise` process records scheduler status and logs; do not
   kill it.

agent-skills/orx-compute/references/ssh.md

@@ -43,6 +43,18 @@ 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.
 
+Each project's three newest SSH and Slurm runs, counted together across hosts
+and including unfinished ones, stay whole. Once an older finished run's log is
+collected, the running OpenResearch app removes its host dir except
+`tensorboard/`, and the container's copy entirely (retried while the container
+is stopped), so anything written into `repo/` goes with it. While cleaning a host it also removes orx's cached source
+archives that no active run uses and that have been idle for an hour, and
+checks again an hour later.
+Unreachable hosts are retried later.
+`orx compute configure ssh --host lab --keep-remote-runs true` keeps all of
+that host's runs not yet cleaned up. Deleting a project removes its runs'
+remote dirs entirely, regardless of that setting.
+
 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
@@ -58,8 +70,9 @@ 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 `<root>/runs/<runId>/` (`~/.orx` unless a remote root is saved), 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.
+container removal does not erase them before they are collected. 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

src/commands/compute.rs

@@ -238,6 +238,10 @@ pub struct ConfigureArgs {
     /// Absolute remote dir for runs and the source cache (SSH per host, or Slurm); default ~/.orx.
     #[arg(long)]
     remote_root: Option<String>,
+    /// true keeps every finished run's remote dir; false (the default) keeps each project's 3 newest
+    /// SSH/Slurm runs and removes older ones except tensorboard/.
+    #[arg(long)]
+    keep_remote_runs: Option<bool>,
     #[arg(long)]
     default_host: Option<String>,
     #[arg(long)]
@@ -324,7 +328,13 @@ impl ConfigureArgs {
     fn body(self) -> Result<Value> {
         let backend = self.backend.name();
         let allowed: &[&str] = match backend {
-            "ssh" => &["host", "container", "remoteRoot", "defaultHost"],
+            "ssh" => &[
+                "host",
+                "container",
+                "remoteRoot",
+                "keepRemoteRuns",
+                "defaultHost",
+            ],
             "slurm" => &[
                 "host",
                 "partition",
@@ -333,6 +343,7 @@ impl ConfigureArgs {
                 "cpusPerTask",
                 "mem",
                 "remoteRoot",
+                "keepRemoteRuns",
             ],
             "k8s" => &["context", "namespace"],
             "ray" => &["address"],
@@ -359,6 +370,9 @@ impl ConfigureArgs {
                 body.insert(key.into(), json!(value));
             }
         }
+        if let Some(keep) = self.keep_remote_runs {
+            body.insert("keepRemoteRuns".into(), json!(keep));
+        }
         if let Some(cpus) = self.cpus_per_task {
             body.insert("cpusPerTask".into(), json!(cpus));
         }
@@ -395,6 +409,7 @@ impl ConfigureArgs {
                 "cpus-per-task" => "cpusPerTask",
                 "default-host" => "defaultHost",
                 "remote-root" => "remoteRoot",
+                "keep-remote-runs" => "keepRemoteRuns",
                 other => other,
             };
             if !allowed.contains(&key)
@@ -412,6 +427,8 @@ impl ConfigureArgs {
                     Value::Null
                 } else if key == "cpusPerTask" {
                     json!(0)
+                } else if key == "keepRemoteRuns" {
+                    json!(false)
                 } else {
                     json!("")
                 },
@@ -427,13 +444,15 @@ impl ConfigureArgs {
                 "Supply settings to change; use compute show {backend} to inspect."
             ));
         }
-        let host_setting = body.contains_key("container") || body.contains_key("remoteRoot");
+        let host_setting = ["container", "remoteRoot", "keepRemoteRuns"]
+            .iter()
+            .any(|key| body.contains_key(*key));
         if backend == "ssh" && body.contains_key("host") && !host_setting {
-            return Err(anyhow!("--host needs --container, --remote-root, or a matching --clear; use --default-host to pick a default."));
+            return Err(anyhow!("--host needs --container, --remote-root, --keep-remote-runs, or a matching --clear; use --default-host to pick a default."));
         }
         if backend == "ssh" && host_setting && !body.contains_key("host") {
             return Err(anyhow!(
-                "A container or remote root change requires --host."
+                "A container, remote root, or keep-remote-runs change requires --host."
             ));
         }
         Ok(Value::Object(body))
@@ -783,6 +802,10 @@ mod tests {
             body(&["slurm", "--clear", "remote-root"]).unwrap(),
             json!({"remoteRoot":""})
         );
+        assert_eq!(
+            body(&["ssh", "--host", "lab", "--keep-remote-runs", "true"]).unwrap(),
+            json!({"host":"lab", "keepRemoteRuns":true})
+        );
         assert!(body(&["ssh", "--remote-root", "/scratch/me/orx"]).is_err());
         assert!(crate::Cli::try_parse_from(["orx", "compute", "status", "--json"]).is_ok());
     }

src/commands/supervise.rs

@@ -849,6 +849,9 @@ async fn watch_ssh_job(
                 )
                 .await;
             }
+            if collected {
+                queue_remote_cleanup(store, run_id);
+            }
             eprintln!("supervise {run_id}: finished ({status})");
             return Ok(status);
         }
@@ -1436,10 +1439,11 @@ async fn run_slurm(
 
         if is_terminal_stage(stage) {
             let _ = done_tx.send(true);
-            if tokio::time::timeout(Duration::from_secs(20), &mut log_task)
-                .await
-                .is_err()
-            {
+            let collected = matches!(
+                tokio::time::timeout(Duration::from_secs(20), &mut log_task).await,
+                Ok(Ok(true))
+            );
+            if !collected {
                 log_task.abort();
             }
             let applied = store.update_status(&run_id, status, Some(now_ms()), None)?;
@@ -1452,6 +1456,9 @@ async fn run_slurm(
                     }
                 }
             }
+            if collected {
+                queue_remote_cleanup(&store, &run_id);
+            }
             eprintln!("supervise {run_id}: finished ({status})");
             return Ok(());
         }
@@ -1465,6 +1472,14 @@ async fn run_slurm(
     }
 }
 
+/// Queues the remote run dir once logs are local; `orx up` removes it when it falls past its
+/// project's newest few, unless the target keeps results.
+fn queue_remote_cleanup(store: &Store, run_id: &str) {
+    if let Err(err) = store.queue_finished_remote_cleanup(run_id) {
+        eprintln!("supervise {run_id}: could not queue remote cleanup: {err}");
+    }
+}
+
 async fn cancel_slurm(host: &str, job_id: &str, run_id: &str, cancel_sent: &mut bool) {
     eprintln!("supervise {run_id}: cancel requested — scancel {job_id}");
     match slurm::cancel_job(host, job_id).await {

src/commands/up.rs

@@ -179,6 +179,10 @@ pub async fn run(args: UpArgs) -> Result<()> {
         state.data_dir_move_in_progress.clone(),
         state.data_dir_gate.clone(),
     ));
+    tokio::spawn(reclaim_remote_run_dirs(
+        state.data_dir_move_in_progress.clone(),
+        state.data_dir_gate.clone(),
+    ));
     spawn_claude_auth_monitor(
         state.chat.clone(),
         claude.clone(),
@@ -4616,6 +4620,7 @@ impl axum::body::HttpBody for ActiveBody {
 }
 
 const ORPHANED_RUN_SCAN_INTERVAL: Duration = Duration::from_secs(60);
+const REMOTE_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
 /// SSH container launches record their handle before the job starts; leave them this long to finish.
 const ORPHANED_RUN_SETTLE_MS: i64 = 5 * 60 * 1000;
 
@@ -4705,6 +4710,73 @@ async fn reclaim_orphaned_run_artifacts(
     }
 }
 
+/// Runs `f` on a fresh store under the data-dir gate; `None` while the data dir is moving.
+async fn with_gated_store<T: Send + 'static>(
+    moving: &AtomicBool,
+    gate: &tokio::sync::Mutex<()>,
+    f: impl FnOnce(&Store) -> Result<T> + Send + 'static,
+) -> Option<Result<T>> {
+    if moving.load(Ordering::SeqCst) {
+        return None;
+    }
+    let _gate = gate.lock().await;
+    if moving.load(Ordering::SeqCst) {
+        return None;
+    }
+    Some(
+        tokio::task::spawn_blocking(move || f(&Store::open()?))
+            .await
+            .unwrap_or_else(|error| Err(anyhow!("store task failed: {error}"))),
+    )
+}
+
+/// Removes queued remote run dirs; the gate is held only around store access, never over SSH.
+async fn reclaim_remote_run_dirs(moving: Arc<AtomicBool>, gate: Arc<tokio::sync::Mutex<()>>) {
+    // Startup's own SSH traffic (supervisor adoption, settings checks) goes first.
+    tokio::time::sleep(Duration::from_secs(30)).await;
+    let mut reported = HashSet::new();
+    let mut plan_error = None;
+    let mut report = |failures: Vec<(String, String)>| {
+        for (run_id, error) in failures {
+            if reported.insert(run_id.clone()) {
+                eprintln!("orx up: remote cleanup for run {run_id}: {error}");
+            }
+        }
+    };
+    loop {
+        match with_gated_store(&moving, &gate, local::remote_cleanup::plan).await {
+            Some(Ok((batches, rejected))) => {
+                plan_error = None;
+                report(rejected);
+                if !batches.is_empty() {
+                    let results = local::remote_cleanup::clean(batches).await;
+                    match with_gated_store(&moving, &gate, move |store| {
+                        local::remote_cleanup::record(store, results)
+                    })
+                    .await
+                    {
+                        Some(Ok(failures)) => report(failures),
+                        Some(Err(error)) => {
+                            eprintln!("orx up: could not record remote cleanup: {error}")
+                        }
+                        // Unrecorded entries stay due and removal is idempotent.
+                        None => {}
+                    }
+                }
+            }
+            Some(Err(error)) => {
+                let error = error.to_string();
+                if plan_error.as_ref() != Some(&error) {
+                    eprintln!("orx up: could not plan remote cleanup: {error}");
+                    plan_error = Some(error);
+                }
+            }
+            None => {}
+        }
+        tokio::time::sleep(REMOTE_CLEANUP_INTERVAL).await;
+    }
+}
+
 /// Spawns the missing supervisors and returns the runs whose spawn failed.
 fn adopt_orphaned_runs_once() -> Result<Vec<(String, crate::error::Error)>> {
     let store = Store::open()?;

src/commands/up/compute_settings.rs

@@ -588,6 +588,7 @@ pub(super) async fn ssh_settings() -> ApiResult {
                 .unwrap_or_default();
             host["container"] = json!(options.container);
             host["remoteRoot"] = json!(options.remote_root);
+            host["keepRemoteRuns"] = json!(options.keep_remote_runs);
         }
         // Best-effort, like the preflight write: a store hiccup shouldn't take
         // out the host listing — hosts just render as never tested.
@@ -625,22 +626,26 @@ pub(super) struct SaveSshSettingsReq {
     container: Option<String>,
     /// Unlike `container`, `None` keeps the saved root (the UI never sends it); `Some("")` clears it.
     remote_root: Option<String>,
+    /// `None` keeps the saved value.
+    keep_remote_runs: Option<bool>,
 }
 
 pub(super) async fn save_ssh_settings(Json(req): Json<SaveSshSettingsReq>) -> ApiResult {
     let host = req.host.trim().to_string();
     require_configured_ssh_host(&host).await?;
     blocking_api(move || {
-        let remote_root = match req.remote_root {
-            Some(root) => Some(root.trim().to_string()).filter(|root| !root.is_empty()),
-            None => crate::config::ssh_settings()?
-                .hosts
-                .get(&host)
-                .and_then(|saved| saved.remote_root.clone()),
-        };
+        let saved = crate::config::ssh_settings()?
+            .hosts
+            .get(&host)
+            .cloned()
+            .unwrap_or_default();
         let options = crate::config::SshHostSettings {
             container: req.container,
-            remote_root,
+            remote_root: match req.remote_root {
+                Some(root) => Some(root.trim().to_string()).filter(|root| !root.is_empty()),
+                None => saved.remote_root,
+            },
+            keep_remote_runs: req.keep_remote_runs.unwrap_or(saved.keep_remote_runs),
         };
         crate::jobs::ssh::validate_host_options(&options).map_err(bad_request)?;
         crate::config::set_ssh_host(host, options)?;
@@ -860,6 +865,7 @@ pub(super) fn slurm_settings_json() -> Value {
         "cpusPerTask": settings.cpus_per_task,
         "mem": settings.mem,
         "remoteRoot": settings.remote_root,
+        "keepRemoteRuns": settings.keep_remote_runs,
         "hosts": list_ssh_hosts(),
     })
 }
@@ -882,6 +888,7 @@ pub(super) struct SetSlurmSettingsReq {
     cpus_per_task: Option<u32>,
     mem: Option<String>,
     remote_root: Option<String>,
+    keep_remote_runs: Option<bool>,
 }
 
 pub(super) async fn set_slurm_settings(Json(req): Json<SetSlurmSettingsReq>) -> ApiResult {
@@ -925,6 +932,9 @@ pub(super) async fn set_slurm_settings(Json(req): Json<SetSlurmSettingsReq>) ->
             }
             settings.remote_root = root;
         }
+        if let Some(keep) = req.keep_remote_runs {
+            settings.keep_remote_runs = keep;
+        }
         slurm::save_settings(&settings)?;
         Ok(Json(slurm_settings_json()))
     })
@@ -1475,8 +1485,11 @@ pub(crate) async fn configure(backend: &str, body: Value) -> Result<Value> {
                     .await
                     .map_err(|ApiError(_, message)| anyhow!(message))?;
             }
-            if body.get("container").is_some() || body.get("remoteRoot").is_some() {
-                // A root-only change keeps the host's saved container.
+            if ["container", "remoteRoot", "keepRemoteRuns"]
+                .iter()
+                .any(|key| body.get(*key).is_some())
+            {
+                // A change without --container keeps the host's saved container.
                 let container = match body.get("container") {
                     Some(container) => container.clone(),
                     None => {
@@ -1488,8 +1501,10 @@ pub(crate) async fn configure(backend: &str, body: Value) -> Result<Value> {
                     }
                 };
                 let mut req = json!({"host": body["host"], "container": container});
-                if let Some(root) = body.get("remoteRoot") {
-                    req["remoteRoot"] = root.clone();
+                for key in ["remoteRoot", "keepRemoteRuns"] {
+                    if let Some(value) = body.get(key) {
+                        req[key] = value.clone();
+                    }
                 }
                 let req: SaveSshSettingsReq = serde_json::from_value(req)?;
                 cli_result(save_ssh_settings(Json(req)).await)?;