fix(native): keep interrupted restarts recoverable under concurrent ps

docs/guides/native-candidate.md

@@ -3080,6 +3080,24 @@ separately after restart.
 
 ### Explicit cleanup after a dead foreground owner
 
+`hack up` and `hack restart` refuse a run whose foreground publication is still
+present and name the next step: a running owner must be stopped with Ctrl-C or
+`hack down`; an owner that exited while its services were running is stopped by
+`hack down --recover`. That command selects the run's exact receipt with the
+read-only `graph foreground-owner --run-id RUN --json` and then uses the
+same-boot `graph recover-live-owner` described below, retaining data and
+retiring the stale publication. If the frontend that owned the run also exited
+before confirming its cleanup, `hack down --recover` prints the exact
+`hack restart --recover-frontend --expect-finalization-attempt ATTEMPT` step.
+An owner that exited before its restore was admitted is reported, with data
+retained, but no command retires that attempt yet.
+
+Read-only `hack ps` and `hack logs` hold the provider lease only briefly. Native
+startup, retained-start eligibility, restore admission, inspections inside
+mutating commands and owned foreground cleanup wait out such a holder within a
+bound instead of failing, so polling status during `up` or `restart` does not
+make them fail or strand a foreground publication.
+
 For a fully ready graph whose foreground owner died while its pool remains on the
 same boot, `graph recover-live-owner --run-id RUN --expect-receipt SHA256` selects
 only that graph. It requires matching dead foreground and relay-owner identities,

docs/reference/cli.md

@@ -194,7 +194,7 @@ hack down [options]
 | `--remove-data` | Native backend: delete a stopped project's owned data volumes so the next up starts fresh (asks first) |
 | `--yes` | Confirm --prune-caches or --remove-data without prompting |
 | `--json` | Output JSON (machine-readable) |
-| `--recover` | Explicitly reconcile an interrupted native Compose stop while retaining data |
+| `--recover` | Explicitly reconcile an interrupted native stop, or stop the services of a native run whose foreground owner exited, while retaining data |
 | `--no-interactive` | Never prompt: apply documented defaults or fail with E_INTERACTIVE_REQUIRED (also via HACK_NO_INTERACTIVE=1) |
 | `--help, -h` | Show help |
 | `--version, -v` | Show version |

packages/runtime-core/src/graph_cli.rs

@@ -20,7 +20,7 @@ use std::{collections::BTreeMap, path::Path, time::Duration};
 fn invalid() -> CandidateError {
     CandidateError::new(
         "graph_arguments",
-        "Use graph run|restart|restore with --project, --file, --expect-plan, --run-id and --ready service=started|healthy|completed; planning actions accept optional --branch <canonical-lowercase-dns-label>. Source mounts may select --source-revision. Fresh run/serve may explicitly select --release-initializer-cache service for quiescent guest page/dentry cache release (package contents retained); explicit --live-source binds directory mounts to that initial acknowledged workspace revision; inspect/reconcile/cleanup/archive/export/reconcile-export/prune require --run-id. Cleanup alone may use --remove-data. Dead-owner recover-cleanup and same-boot recover-live-owner require --expect-receipt; retire-recovered-publisher requires --expect-owner. A missing publisher lock requires inspect-missing-publication-lock --run-id and explicitly selected retire-missing-publication-lock --run-id --expect-selection --accept-quiesced-candidate-home. Foreground serve may opt into --auto-dependency-slots for pool-wide transport allocation; dependency-reservations inspects claims and recover-dependency-reservation --run-id --expect-reservation releases a verified dead pre-admission claim. Foreground serve additionally requires --dependencies and --expect-dependencies and accepts explicit --environment-stdin and --route-slot service=index|auto for reviewed local routes; dependency-plan requires --dependencies; dependency-discover requires --host-port and exactly one of --executable or --ancestor-pid; owner-status requires --run-id. Fresh foreground owner-restore requires exactly --run-id, --expect-plan, --expect-generation and --environment-stdin, plus optional --json. Bridge reservation requires --run-id, --service, --slot and --expect-generation; start/release require --run-id, --slot and --expect-reservation. bridges/reconcile-bridges require --run-id. Foreground publish-bridge requires --run-id, --slot, --expect-reservation and exactly one of --port or --unix (no --json); --unix accepts up to eight --hostname claims; unpublish-bridge requires --run-id and --expect-reservation.",
+        "Use graph run|restart|restore with --project, --file, --expect-plan, --run-id and --ready service=started|healthy|completed; planning actions accept optional --branch <canonical-lowercase-dns-label>. Source mounts may select --source-revision. Fresh run/serve may explicitly select --release-initializer-cache service for quiescent guest page/dentry cache release (package contents retained); explicit --live-source binds directory mounts to that initial acknowledged workspace revision; inspect/reconcile/cleanup/archive/export/reconcile-export/prune require --run-id. Cleanup alone may use --remove-data. foreground-owner --run-id selects a run's receipt digest and publication state read-only. Dead-owner recover-cleanup and same-boot recover-live-owner require --expect-receipt; retire-recovered-publisher requires --expect-owner. A missing publisher lock requires inspect-missing-publication-lock --run-id and explicitly selected retire-missing-publication-lock --run-id --expect-selection --accept-quiesced-candidate-home. Foreground serve may opt into --auto-dependency-slots for pool-wide transport allocation; dependency-reservations inspects claims and recover-dependency-reservation --run-id --expect-reservation releases a verified dead pre-admission claim. Foreground serve additionally requires --dependencies and --expect-dependencies and accepts explicit --environment-stdin and --route-slot service=index|auto for reviewed local routes; dependency-plan requires --dependencies; dependency-discover requires --host-port and exactly one of --executable or --ancestor-pid; owner-status requires --run-id. Fresh foreground owner-restore requires exactly --run-id, --expect-plan, --expect-generation and --environment-stdin, plus optional --json. Bridge reservation requires --run-id, --service, --slot and --expect-generation; start/release require --run-id, --slot and --expect-reservation. bridges/reconcile-bridges require --run-id. Foreground publish-bridge requires --run-id, --slot, --expect-reservation and exactly one of --port or --unix (no --json); --unix accepts up to eight --hostname claims; unpublish-bridge requires --run-id and --expect-reservation.",
     )
 }
 pub fn command(candidate: &Candidate, args: &[&str]) -> Result<Value, CandidateError> {
@@ -123,6 +123,21 @@ pub fn command(candidate: &Candidate, args: &[&str]) -> Result<Value, CandidateE
             return Err(invalid());
         }
     }
+    if *action == "foreground-owner" {
+        let run = match *args {
+            ["--run-id", run] | ["--run-id", run, "--json"] => run,
+            _ => return Err(invalid()),
+        };
+        #[cfg(target_os = "macos")]
+        {
+            return graph::foreground_owner(candidate, run);
+        }
+        #[cfg(not(target_os = "macos"))]
+        {
+            let _ = run;
+            return Err(invalid());
+        }
+    }
     if *action == "retire-acknowledged-publisher" {
         let (run, owner, receipt, publisher) = match *args {
             [

packages/runtime-core/src/provider/graph/dependency_slots.rs

@@ -41,6 +41,16 @@ fn refused() -> CandidateError {
 fn root(candidate: &Candidate) -> PathBuf {
     candidate.state_root.join("run/dependency-assignments")
 }
+/// Whether a dependency reservation record exists for this run. Observation only.
+pub(super) fn reservation_recorded(
+    candidate: &Candidate,
+    run: &str,
+) -> Result<bool, CandidateError> {
+    if !hex(run, 32) {
+        return Err(refused());
+    }
+    Ok(!absent(&root(candidate).join(format!("{run}.json")))?)
+}
 fn socket(home: &Path, slot: u8) -> PathBuf {
     home.join(format!("dependency-{slot:02}.sock"))
 }

packages/runtime-core/src/provider/graph/foreground.rs

@@ -285,7 +285,10 @@ fn serve_input<'a>(
             // containers. It travels only in this reply and is never persisted.
             let error = error.with_startup_logs(super::startup_logs::capture(candidate, &run_id));
             if fresh_generation.is_some() && !runtime.admission_started() {
-                let engine = Engine::connect_cleanup(candidate)?;
+                // Wait out a concurrent read-only holder of the lease: refusing here
+                // would exit with this attempt's publication still bound, which
+                // blocks every later start of the retained run.
+                let engine = Engine::connect_cleanup_wait(candidate)?;
                 let (retained, root) = super::load(candidate, &engine, &run_id)?;
                 if retained.phase == "stopped-data-retained"
                     && std::fs::symlink_metadata(root.join("state.pending"))

packages/runtime-core/src/provider/graph/foreground/transport.rs

@@ -843,6 +843,53 @@ pub(in crate::provider::graph) struct LegacyPublisher {
     pub process: ProcessIdentity,
 }
 
+/// What a remaining foreground publication shows about its recorded process.
+/// Presentation only: it names the next step for a refused start and grants no
+/// retirement, cleanup or adoption authority.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(in crate::provider::graph) enum PublishedOwner {
+    /// Neither the owner record nor its control socket exists.
+    Absent,
+    /// The recorded owner process is alive with its exact identity.
+    Running,
+    /// The recorded owner process is gone and nothing listens on its socket.
+    Exited,
+    /// The publication is partial, foreign or otherwise unclassified.
+    Uncertain,
+}
+impl PublishedOwner {
+    pub(in crate::provider::graph) fn word(self) -> &'static str {
+        match self {
+            Self::Absent => "absent",
+            Self::Running => "running",
+            Self::Exited => "exited",
+            Self::Uncertain => "uncertain",
+        }
+    }
+}
+pub(in crate::provider::graph) fn published_owner(
+    candidate: &Candidate,
+    run: &str,
+) -> Result<PublishedOwner, CandidateError> {
+    let root = root(candidate, run)?;
+    let present = |name: &str| match fs::symlink_metadata(root.join(name)) {
+        Ok(_) => Ok(true),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+        Err(_) => Err(refused()),
+    };
+    if !present("owner.json")? && !present("control.sock")? {
+        return Ok(PublishedOwner::Absent);
+    }
+    let Ok(pin) = Pin::read(candidate, run) else {
+        return Ok(PublishedOwner::Uncertain);
+    };
+    Ok(match identity::alive(pin.record.process.pid) {
+        Ok(true) if pin.verify().is_ok() => PublishedOwner::Running,
+        Ok(false) if no_listener(&pin.root.join("control.sock")).is_ok() => PublishedOwner::Exited,
+        _ => PublishedOwner::Uncertain,
+    })
+}
+
 /// Select an exact dead current-device publisher without requiring its missing
 /// operation lock. This observation grants no file or guest effect.
 pub(in crate::provider::graph) fn dead_publisher_without_lock(
@@ -879,6 +926,29 @@ pub(in crate::provider::graph) fn fixture_dead_publisher_without_lock(
     .unwrap();
     fs::remove_file(root.join("operation.lock")).unwrap();
 }
+/// Bind a retired run's next publication in this test process. The returned value
+/// keeps it alive (its owner is this process) until dropped.
+#[cfg(test)]
+pub(in crate::provider::graph) fn fixture_live_publication(
+    candidate: &Candidate,
+    run: &str,
+) -> Box<dyn std::any::Any> {
+    Box::new(Publication::bind_retired(candidate, run).unwrap())
+}
+/// Leave a retired run's next publication behind with an exited owner process:
+/// the listener is closed and the record names a process that does not exist.
+#[cfg(test)]
+pub(in crate::provider::graph) fn fixture_exited_publication(candidate: &Candidate, run: &str) {
+    drop(Publication::bind_retired(candidate, run).unwrap());
+    let root = root(candidate, run).unwrap();
+    let mut record: Record = state::read(&root.join("owner.json")).unwrap();
+    record.process.pid = 2_000_000;
+    fs::write(
+        root.join("owner.json"),
+        serde_json::to_vec(&record).unwrap(),
+    )
+    .unwrap();
+}
 #[cfg(test)]
 pub(in crate::provider::graph) fn fixture_next_publication(candidate: &Candidate, run: &str) {
     let mut publication = Publication::bind(candidate, run).unwrap();

packages/runtime-core/src/provider/graph/live_owner_cleanup.rs

@@ -34,6 +34,36 @@ pub(super) fn current_completion(
     Ok(intent.complete_sha256.as_deref() == Some(complete.as_str()))
 }
 
+/// The original receipt digest of a same-boot recovery whose durable intent exists but
+/// whose recovery has not finished, so an interrupted `hack down --recover` resumes it
+/// by its own selection. Unfinished means cleanup is still pending, or cleanup completed
+/// for this exact receipt while a retirement effect of `finish_retirement` remains: the
+/// stale publication, or this run's dependency reservation (released last).
+/// Selection only: `recover_live_owner` revalidates the intent against current evidence.
+pub(super) fn resumable_selection(
+    candidate: &Candidate,
+    root: &Path,
+    receipt: &Receipt,
+    publication_present: bool,
+) -> Result<Option<String>, CandidateError> {
+    if !exists(&root.join(FILE))? {
+        return Ok(None);
+    }
+    let intent: Intent = state::read(&root.join(FILE))?;
+    let current = digest(receipt)?;
+    let remaining =
+        publication_present || dependency_slots::reservation_recorded(candidate, &receipt.run)?;
+    Ok(
+        unfinished(intent.complete_sha256.as_deref(), &current, remaining)
+            .then_some(intent.original_sha256),
+    )
+}
+/// Cleanup not yet complete, or complete for this exact receipt with a retirement effect
+/// remaining. A completed recovery of an older receipt is history.
+fn unfinished(complete: Option<&str>, current: &str, retirement_remaining: bool) -> bool {
+    complete.is_none_or(|complete| retirement_remaining && complete == current)
+}
+
 /// Select an independent data-removal request from the exact completed same-boot
 /// generation. A retry pins the entire committed proof, not just receipt phases.
 pub(super) fn removal_selection(
@@ -649,6 +679,20 @@ pub(super) fn retire(
     ))
 }
 
+#[cfg(test)]
+mod resumable_tests {
+    #[test]
+    fn only_an_unfinished_recovery_is_resumed() {
+        // Interrupted after the cleanup intent: resumed whatever the receipt shows.
+        assert!(super::unfinished(None, "a", true));
+        assert!(super::unfinished(None, "a", false));
+        // Cleanup completed for this receipt but a retirement effect remains.
+        assert!(super::unfinished(Some("a"), "a", true));
+        // Fully retired, or a completed recovery of an older generation: history.
+        assert!(!super::unfinished(Some("a"), "a", false));
+        assert!(!super::unfinished(Some("a"), "b", true));
+    }
+}
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -811,6 +855,43 @@ mod tests {
         );
         value
     }
+    #[test]
+    fn an_interrupted_recovery_is_selected_by_its_original_receipt_until_retired() {
+        let fixture = super::super::tests::Fixture::new();
+        let candidate = Candidate::discover(&fixture.0).unwrap();
+        let select = |receipt: &Receipt, publication: bool| {
+            resumable_selection(&candidate, &fixture.0, receipt, publication).unwrap()
+        };
+        let mut current = receipt();
+        // No recovery intent: nothing to resume (an unadmitted exit stays refused).
+        assert_eq!(select(&current, true), None);
+        let mut intent = completed(receipt());
+        intent.original_sha256 = "1".repeat(64);
+        intent.complete_sha256 = None;
+        state::write(&fixture.0.join(FILE), &intent).unwrap();
+        // Interrupted after the cleanup intent.
+        current.phase = "cleanup-intent".into();
+        assert_eq!(select(&current, true), Some("1".repeat(64)));
+        current.phase = "stopped-data-retained".into();
+        intent.complete_sha256 = Some(digest(&current).unwrap());
+        fs::remove_file(fixture.0.join(FILE)).unwrap();
+        state::write(&fixture.0.join(FILE), &intent).unwrap();
+        // Interrupted after cleanup, before publisher retirement.
+        assert_eq!(select(&current, true), Some("1".repeat(64)));
+        // Interrupted after both publisher paths were retired but before the
+        // dependency reservation was released: still resumed by its receipt.
+        let reservations = candidate.state_root.join("run/dependency-assignments");
+        state::private_directory(&reservations).unwrap();
+        let record = reservations.join(format!("{}.json", current.run));
+        fs::write(&record, b"{}").unwrap();
+        assert_eq!(select(&current, false), Some("1".repeat(64)));
+        // Every retirement effect done: history, not a resumable recovery.
+        fs::remove_file(&record).unwrap();
+        assert_eq!(select(&current, false), None);
+        current.owner = "9".repeat(32);
+        assert_eq!(select(&current, true), None);
+    }
+
     fn completed(original: Receipt) -> Intent {
         let mut stopped = original.clone();
         stopped.phase = "stopped-data-retained".into();

packages/runtime-core/src/provider/graph/mod.rs

@@ -1668,6 +1668,39 @@ fn endpoint_network<'a>(
     Ok(Some(network))
 }
 
+/// Read-only selection for an explicit dead-owner recovery such as `hack down --recover`:
+/// the run's receipt phase, the SHA-256 of its exact receipt bytes, what its
+/// foreground publication shows (`absent`, `running`, `exited` or `uncertain`) and,
+/// for an interrupted same-boot dead-owner recovery, the original receipt digest that
+/// resumes it (`live_owner_recovery`).
+///
+/// It takes no lease, contacts no guest and grants no authority. The digest only lets
+/// a caller select the receipt it observed; each recovery command revalidates its own
+/// evidence (owner death, endpoints, inventories) against that selection.
+#[cfg(target_os = "macos")]
+pub fn foreground_owner(candidate: &Candidate, run: &str) -> Result<Value, CandidateError> {
+    use sha2::{Digest, Sha256};
+    let owner = state::Owner::load(candidate)?;
+    let (receipt, root) = load_at(directory(candidate, run)?, run, &owner.token)?;
+    let bytes = host_pin_recovery::read_raw(&root.join("state.json"), 1024 * 1024)?;
+    let publication = foreground::transport::published_owner(candidate, run)?;
+    let resumable = live_owner_cleanup::resumable_selection(
+        candidate,
+        &root,
+        &receipt,
+        publication != foreground::transport::PublishedOwner::Absent,
+    )?;
+    Ok(json!({
+        "version": 1,
+        "run": receipt.run,
+        "owner": receipt.owner,
+        "phase": receipt.phase,
+        "receipt_sha256": format!("{:x}", Sha256::digest(&bytes)),
+        "foreground": publication.word(),
+        "live_owner_recovery": resumable,
+    }))
+}
+
 /// Read-only recovery inspection; historical ready phase is not a current health assertion.
 /// Read-only inspection of one owned graph attempt.
 ///

packages/runtime-core/src/provider/graph/restore.rs

@@ -251,6 +251,23 @@ fn restore_inputs(
     if options.timeout.is_zero() || options.timeout > Duration::from_secs(600) {
         return Err(error("graph_budget", "Invalid restore timeout."));
     }
+    // Nothing above has effects, and acquisition admits nothing: wait out a concurrent
+    // read-only holder of the lease (the inspection behind `hack ps`) within the restore
+    // budget instead of failing the restore, as graph startup does. Cancellation stops
+    // the wait before admission.
+    #[cfg(target_os = "macos")]
+    let engine = {
+        let budget = std::time::Instant::now() + options.timeout;
+        let deadline = fresh
+            .as_ref()
+            .map_or(budget, |fresh| fresh.deadline.min(budget));
+        Engine::connect_until(candidate, deadline, || {
+            startup
+                .as_ref()
+                .is_some_and(|driver| driver.check_cancelled().is_err())
+        })?
+    };
+    #[cfg(not(target_os = "macos"))]
     let engine = Engine::connect(candidate)?;
     if engine.guest().profile() != super::super::Profile::Development {
         return Err(error(