runner, server: opt-in independent Runner that outlives its gateways (P1R-C01)

crates/codex-runtime/Cargo.toml

@@ -23,7 +23,7 @@ devguard = ["codespace-runner/devguard"]
 codespace-runner = { path = "../runner" }
 codex-process-hardening = { path = "../../third_party/codex/codex-rs/process-hardening" }
 codex-uds = { path = "../../third_party/codex/codex-rs/uds" }
-tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "io-util", "fs", "sync", "process", "time"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "io-util", "fs", "sync", "process", "time", "signal"] }
 
 [patch.crates-io]
 tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" }

crates/codex-runtime/tests/runtime_binary.rs

@@ -6,11 +6,11 @@ use std::process::Command;
 use std::sync::Arc;
 use std::time::Duration;
 
-use codespace_domain::{ProcessId, Profile, WorkspaceId};
+use codespace_domain::{ErrorCode, ProcessId, ProcessState, Profile, WorkspaceId};
 use codespace_policy::Workspace;
 use codespace_runner::{
-    read_frame, write_frame, Runner, RunnerExecRequest, RunnerOp, RunnerOpResult, UdsRunner,
-    WireEnvelope, WIRE_PROTOCOL,
+    read_frame, write_frame, Runner, RunnerError, RunnerExecRequest, RunnerLifecycle, RunnerOp,
+    RunnerOpResult, UdsRunner, WireEnvelope, WIRE_PROTOCOL,
 };
 use tokio::net::UnixStream;
 
@@ -144,7 +144,7 @@ async fn hello_on_live_socket() {
     match parsed.result {
         Some(RunnerOpResult::Hello { protocol, .. }) => {
             assert_eq!(protocol, WIRE_PROTOCOL);
-            assert_eq!(protocol, 7);
+            assert_eq!(protocol, 8);
         }
         other => panic!("unexpected hello {other:?}"),
     }
@@ -211,3 +211,220 @@ async fn worker_socket_drop_terminates_owned_child() {
         "owned child must stop updating heartbeat after worker disconnect (after={after:?} later={later:?})"
     );
 }
+
+/// An independent Runner (P1R-C01) at `socket`, whose directory it prepares itself.
+fn spawn_independent(socket: &Path) -> tokio::process::Child {
+    tokio::process::Command::new(BIN)
+        .arg(socket)
+        .args(["--lifecycle", "independent"])
+        .kill_on_drop(true)
+        .spawn()
+        .expect("spawn an independent Runner")
+}
+
+/// A gateway's connection to the Runner at `socket`, after its handshake.
+async fn gateway(socket: &Path) -> UdsRunner {
+    let runner = UdsRunner::from_stream(wait_connect(socket).await, Arc::new(|_| {}));
+    runner.handshake().await.expect("hello");
+    runner
+}
+
+fn mode(path: &Path) -> u32 {
+    std::fs::symlink_metadata(path)
+        .unwrap()
+        .permissions()
+        .mode()
+        & 0o777
+}
+
+/// A command that counts into `beat` ten times a second.
+fn beating(beat: &Path, process_id: &str, tty: bool) -> RunnerExecRequest {
+    let mut request = RunnerExecRequest::for_host(
+        vec![
+            "/bin/sh".into(),
+            "-c".into(),
+            format!(
+                "i=0; while :; do i=$((i+1)); echo $i > '{}'; sleep 0.1; done",
+                beat.display()
+            ),
+        ],
+        ProcessId(process_id.into()),
+        Profile::WorkspaceWrite,
+    );
+    request.timeout_ms = 60_000;
+    request.tty = tty;
+    request
+}
+
+/// Whether the count in `beat` still goes up.
+async fn still_beating(beat: &Path) -> bool {
+    let read = || std::fs::read_to_string(beat).unwrap_or_default();
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    while read().trim().is_empty() && tokio::time::Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    let before = read();
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    before != read()
+}
+
+async fn exited(runner: &mut tokio::process::Child) -> std::process::ExitStatus {
+    tokio::time::timeout(Duration::from_secs(15), runner.wait())
+        .await
+        .expect("the Runner exits")
+        .expect("wait for the Runner")
+}
+
+fn demo_workspace(root: &Path) -> Workspace {
+    Workspace::new(
+        WorkspaceId("demo".into()),
+        root.to_path_buf(),
+        Profile::WorkspaceWrite,
+    )
+}
+
+/// An independent Runner keeps its pipe and tty processes after its gateway detaches, the next
+/// gateway inherits them but starts nothing, and a stop ends them and the Runner, which removes
+/// its socket and the directory it created (P1R-C01).
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn an_independent_runner_outlives_its_gateways_until_one_stops_it() {
+    let base = tempfile::tempdir().unwrap();
+    std::fs::set_permissions(base.path(), std::fs::Permissions::from_mode(0o755)).unwrap();
+    let dir = base.path().join("runner");
+    let socket = runner_socket(&dir);
+    let mut runner = spawn_independent(&socket);
+    let first = gateway(&socket).await;
+    assert_eq!(mode(&dir), 0o700);
+    assert_eq!(mode(base.path()), 0o755);
+    let hello = first.hello();
+    assert_eq!(hello.lifecycle, RunnerLifecycle::Independent);
+    assert_eq!(
+        hello.identity.as_ref().map(|identity| identity.pid),
+        runner.id()
+    );
+    assert!(!hello.previous_controller);
+    // A second Runner neither takes nor removes the live socket.
+    let second = Command::new(BIN)
+        .arg(&socket)
+        .args(["--lifecycle", "independent"])
+        .output()
+        .expect("second Runner");
+    assert!(!second.status.success(), "{second:?}");
+    assert!(socket.exists());
+
+    let ws_dir = tempfile::tempdir().unwrap();
+    let ws = demo_workspace(ws_dir.path());
+    let pipe_beat = ws_dir.path().join("pipe-beat");
+    let tty_beat = ws_dir.path().join("tty-beat");
+    first
+        .exec(&ws, beating(&pipe_beat, "proc-pipe", false))
+        .await
+        .unwrap();
+    first
+        .exec(&ws, beating(&tty_beat, "proc-tty", true))
+        .await
+        .unwrap();
+    assert!(still_beating(&pipe_beat).await && still_beating(&tty_beat).await);
+    first.detach().await.unwrap();
+    // No gateway is connected, and both keep running.
+    assert!(still_beating(&pipe_beat).await && still_beating(&tty_beat).await);
+
+    let next = gateway(&socket).await;
+    let inherited = next.hello();
+    assert_eq!(inherited.identity, hello.identity);
+    assert!(inherited.previous_controller);
+    assert_eq!(inherited.inherited_processes, 2);
+    assert_eq!(
+        next.process_status(&ProcessId("proc-pipe".into()))
+            .await
+            .unwrap()
+            .state,
+        ProcessState::Running
+    );
+    match next
+        .exec(
+            &ws,
+            beating(&ws_dir.path().join("new-beat"), "proc-new", false),
+        )
+        .await
+    {
+        Err(RunnerError::Execution(body)) => assert_eq!(body.code, ErrorCode::RecoveryPending),
+        other => panic!("unexpected {other:?}"),
+    }
+    assert_eq!(next.stop_runner().await.unwrap(), 2);
+    let status = exited(&mut runner).await;
+    assert!(status.success(), "{status:?}");
+    assert!(!still_beating(&pipe_beat).await);
+    assert!(!still_beating(&tty_beat).await);
+    assert!(!socket.exists());
+    assert!(!dir.exists());
+}
+
+/// The Runner's own SIGTERM stops it as a gateway's `Stop` does, and the gateway connected to it
+/// refuses its later requests (P1R-C01).
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn sigterm_stops_an_independent_runner_and_its_processes() {
+    let base = tempfile::tempdir().unwrap();
+    let dir = base.path().join("runner");
+    let socket = runner_socket(&dir);
+    let mut runner = spawn_independent(&socket);
+    let connected = gateway(&socket).await;
+    let ws_dir = tempfile::tempdir().unwrap();
+    let ws = demo_workspace(ws_dir.path());
+    let beat = ws_dir.path().join("beat");
+    let process_id = ProcessId("proc-pipe".into());
+    connected
+        .exec(&ws, beating(&beat, &process_id.0, false))
+        .await
+        .unwrap();
+    assert!(still_beating(&beat).await);
+    let pid = runner.id().unwrap().to_string();
+    let sent = Command::new("kill").args(["-TERM", &pid]).status().unwrap();
+    assert!(sent.success());
+    let status = exited(&mut runner).await;
+    assert!(status.success(), "{status:?}");
+    assert!(!still_beating(&beat).await);
+    assert!(!socket.exists());
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        match connected.process_status(&process_id).await {
+            Err(RunnerError::Execution(body)) if body.code == ErrorCode::RunnerDisconnected => {
+                break
+            }
+            other => assert!(tokio::time::Instant::now() < deadline, "{other:?}"),
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
+
+/// An independent Runner does not start in a socket directory that is not private to its user,
+/// nor without a socket or with an unknown lifecycle, and changes nothing (P1R-C01).
+#[tokio::test]
+async fn an_independent_runner_refuses_a_shared_directory_and_unknown_settings() {
+    let base = tempfile::tempdir().unwrap();
+    let shared = base.path().join("shared");
+    std::fs::create_dir(&shared).unwrap();
+    std::fs::set_permissions(&shared, std::fs::Permissions::from_mode(0o755)).unwrap();
+    let socket = runner_socket(&shared);
+    let refused = Command::new(BIN)
+        .arg(&socket)
+        .args(["--lifecycle", "independent"])
+        .output()
+        .unwrap();
+    assert!(!refused.status.success(), "{refused:?}");
+    assert!(
+        String::from_utf8_lossy(&refused.stderr).contains("0700"),
+        "{refused:?}"
+    );
+    assert_eq!(mode(&shared), 0o755);
+    assert!(!socket.exists());
+    let socket = socket.display().to_string();
+    for args in [
+        &["--lifecycle", "independent"][..],
+        &[&socket, "--lifecycle", "later"],
+        &[&socket, "--runner-exit-intent", "stop"],
+    ] {
+        let refused = Command::new(BIN).args(args).output().unwrap();
+        assert_eq!(refused.status.code(), Some(2), "{args:?}: {refused:?}");
+    }
+}

crates/domain/src/error.rs

@@ -58,6 +58,16 @@ pub enum ErrorCode {
     /// so whether it reserved resources is not known. Nothing was started; do not assume the
     /// attempt does not exist.
     AdmissionUnknown,
+    /// An earlier gateway controlled this gateway's independent Runner, and recovering what it
+    /// left is not available yet (P1R-C01): its processes can be queried, read, written to,
+    /// resized and terminated, but no new execution, patch or approval resume starts, and an
+    /// approval being resumed is left as it was. Stopping the Runner and starting a new one ends
+    /// it. Not occupancy; do not retry as busy.
+    RecoveryPending,
+    /// This gateway lost its connection to its independent Runner and does not reconnect
+    /// (P1R-C01). The request never reached the Runner, whose processes keep running;
+    /// restarting the gateway reaches them again.
+    RunnerDisconnected,
     /// Scheduler or store invariant failed. Not occupancy; do not retry as busy.
     Internal,
 }
@@ -173,6 +183,10 @@ mod tests {
         assert_eq!(json, "\"RESOURCE_QUEUE_FULL\"");
         let json = serde_json::to_string(&ErrorCode::ResourcePolicyUnsupported).unwrap();
         assert_eq!(json, "\"RESOURCE_POLICY_UNSUPPORTED\"");
+        let json = serde_json::to_string(&ErrorCode::RecoveryPending).unwrap();
+        assert_eq!(json, "\"RECOVERY_PENDING\"");
+        let json = serde_json::to_string(&ErrorCode::RunnerDisconnected).unwrap();
+        assert_eq!(json, "\"RUNNER_DISCONNECTED\"");
         let json = serde_json::to_string(&ErrorCode::Internal).unwrap();
         assert_eq!(json, "\"INTERNAL\"");
     }

crates/runner/src/lib.rs

@@ -108,6 +108,7 @@ mod files;
 mod fork_handlers;
 #[cfg(feature = "devguard")]
 pub mod launch;
+mod lifecycle;
 mod linux_sandbox;
 mod patch_helper;
 mod patch_verify;
@@ -142,19 +143,21 @@ pub fn linux_sandbox_available() -> bool {
     linux_sandbox::probe()
 }
 
+pub use lifecycle::{RunnerIdentity, RunnerLifecycle};
 #[cfg(feature = "devguard")]
 pub use registration::OwnerRegistration;
 pub use socket::{
-    allocate_private_runner_dir, is_forbidden_runner_dir, reclaim_leftover_socket,
-    runner_socket_path, RUNNER_SOCKET_NAME,
+    allocate_private_runner_dir, is_forbidden_runner_dir, prepare_runner_socket_dir,
+    reclaim_leftover_socket, runner_socket_path, RUNNER_SOCKET_NAME,
 };
-pub use uds::{DisconnectHook, UdsRunner, RUNNER_CALL_DEADLINE};
+pub use uds::{runner_disconnected, DisconnectHook, RunnerHello, UdsRunner, RUNNER_CALL_DEADLINE};
 pub use watch::{FsWatchEvent, FsWatchKind, WatchClosed, WatchSubscription};
 #[cfg(feature = "devguard")]
 pub use wire::serve_runner_connection_with_registration;
 pub use wire::{
-    host_worker, read_frame, serve_runner_connection, write_frame, RunnerEvent, RunnerOp,
-    RunnerOpResult, WireEnvelope, WireKind, WIRE_PROTOCOL,
+    host_worker, read_frame, recovery_pending, serve_runner_connection, write_frame,
+    IndependentRunner, RunnerEvent, RunnerOp, RunnerOpResult, Served, StopHandle, WireEnvelope,
+    WireKind, STOP_DRAIN_BOUND, WIRE_PROTOCOL,
 };
 
 /// Execution-plane API. Control-plane fields (`work_id`, coordination,
@@ -318,6 +321,16 @@ impl RuntimeBackend {
     pub fn in_process(on_release: ShellRelease) -> Self {
         Self::InProcess(InProcessRunner::new(on_release))
     }
+
+    /// The backend is an independent Runner that an earlier gateway controlled, so this gateway
+    /// starts no new execution, patch or approval resume (P1R-C01): see
+    /// [`UdsRunner::refuses_new_work`].
+    pub fn refuses_new_work(&self) -> bool {
+        match self {
+            Self::InProcess(_) => false,
+            Self::Uds(runner) => runner.refuses_new_work(),
+        }
+    }
 }
 
 impl Runner for RuntimeBackend {