runner, server, store: Claim and controller epoch for the independent Runner (P1R-C03)

crates/codex-runtime/src/bin/codespace-codex-runtime.rs

@@ -15,11 +15,12 @@
 //!
 //! With `--lifecycle independent` (P1R-C01) an operator or a harness starts the worker, not a
 //! gateway: it is an independent Runner. It prepares its socket's directory itself, created
-//! with mode 0700 or accepted only as a private directory of its user, and serves one gateway
-//! connection at a time. Its processes, their output and their timeouts outlive each connection
-//! until a gateway's `Stop`, or the Runner's own SIGTERM or SIGINT, ends them and the Runner.
-//! With DevGuard it reads the consumer secret from `--devguard-credential-file`, as no gateway
-//! hands it one.
+//! with mode 0700 or accepted only as a private directory of its user, and serves every
+//! connection it accepts, of which one at a time, after its `Claim`, controls it (P1R-C03).
+//! It logs each Claim with the process ID of its peer. Its processes, their output and their
+//! timeouts outlive each connection until its controller's `Stop`, or the Runner's own SIGTERM
+//! or SIGINT, ends them and the Runner. With DevGuard it reads the consumer secret from
+//! `--devguard-credential-file`, as no gateway hands it one.
 
 use std::ffi::OsString;
 use std::path::{Path, PathBuf};
@@ -30,7 +31,7 @@ use codespace_runner::serve_runner_connection;
 use codespace_runner::serve_runner_connection_with_registration;
 use codespace_runner::{
     allocate_private_runner_dir, host_worker, prepare_runner_socket_dir, reclaim_leftover_socket,
-    runner_socket_path, IndependentRunner, RunnerLifecycle, Served, STOP_DRAIN_BOUND,
+    runner_socket_path, ClaimNote, IndependentRunner, RunnerLifecycle, Served, STOP_DRAIN_BOUND,
 };
 
 fn main() {
@@ -138,17 +139,48 @@ async fn serve() {
                 refuse(2, "--lifecycle independent needs the socket's path first")
             });
             let (runner, events) = host_worker();
-            let independent = IndependentRunner::new(runner, events);
+            let independent = IndependentRunner::new(runner, events)
+                .with_claim_observer(std::sync::Arc::new(log_claim));
             #[cfg(feature = "devguard")]
             let independent = independent.with_registration(registration, launcher);
             std::process::exit(serve_independent(&socket, independent).await);
         }
     }
 }
 
-/// An independent Runner's life (P1R-C01): its socket, the connections it serves one at a
-/// time, and its stop, after which its socket, and the directory it created for it, are
-/// removed. The exit code.
+/// What an independent Runner logs of each Claim (P1R-C03): any process of its user that can
+/// reach its socket may claim it while no connection controls it.
+fn log_claim(note: ClaimNote) {
+    let peer = |pid: Option<i32>| pid.map_or_else(|| "unknown".into(), |pid| pid.to_string());
+    match note {
+        ClaimNote::Claimed {
+            epoch,
+            previous_controller,
+            peer_pid,
+        } => eprintln!(
+            "codespace-codex-runtime: process {} claimed the Runner under epoch {epoch}{}",
+            peer(peer_pid),
+            if previous_controller {
+                "; an earlier controller started work, so it starts nothing"
+            } else {
+                ""
+            }
+        ),
+        ClaimNote::Refused {
+            refusal,
+            message,
+            peer_pid,
+        } => eprintln!(
+            "codespace-codex-runtime: refused the Claim of process {} ({}): {message}",
+            peer(peer_pid),
+            refusal.as_str()
+        ),
+    }
+}
+
+/// An independent Runner's life (P1R-C01): its socket, the connections it serves, and its
+/// stop, after which its socket, and the directory it created for it, are removed. The exit
+/// code.
 async fn serve_independent(socket: &Path, independent: IndependentRunner) -> i32 {
     let dir = match socket.parent() {
         Some(dir) if !dir.as_os_str().is_empty() => dir,
@@ -163,7 +195,9 @@ async fn serve_independent(socket: &Path, independent: IndependentRunner) -> i32
         rendezvous.remove_dir();
         refuse(1, err);
     }
-    let listener = match codex_uds::UnixListener::bind(socket).await {
+    // Tokio's listener, as codex-uds binds on Unix: a Claim needs its peer's credentials from
+    // the accepted stream (P1R-C03).
+    let listener = match tokio::net::UnixListener::bind(socket) {
         Ok(listener) => listener,
         Err(err) => {
             rendezvous.remove_dir();
@@ -196,20 +230,20 @@ async fn serve_independent(socket: &Path, independent: IndependentRunner) -> i32
         std::process::id(),
         socket.display()
     );
-    let listener = std::sync::Arc::new(tokio::sync::Mutex::new(listener));
+    let listener = std::sync::Arc::new(listener);
     let terminated = independent
         .run(
             move || {
                 let listener = listener.clone();
-                async move { listener.lock().await.accept().await }
+                async move { listener.accept().await.map(|(stream, _)| stream) }
             },
             |ending| match ending {
-                Ok(Served::Detached) => {
-                    eprintln!("codespace-codex-runtime: a gateway detached; its processes go on")
-                }
+                Ok(Served::Detached) => eprintln!(
+                    "codespace-codex-runtime: the controlling gateway detached; its processes go on"
+                ),
                 Ok(Served::Lost) => eprintln!(
-                    "codespace-codex-runtime: a gateway's connection ended without a detach or a \
-                     stop; its processes go on"
+                    "codespace-codex-runtime: the controlling gateway's connection ended without a \
+                     detach or a stop; its processes go on"
                 ),
                 Ok(Served::Stopped { .. }) => {}
                 Err(err) => eprintln!("codespace-codex-runtime: accept: {err}"),

crates/codex-runtime/tests/runtime_binary.rs

@@ -9,10 +9,12 @@ use std::time::Duration;
 use codespace_domain::{ErrorCode, ProcessId, ProcessState, Profile, WorkspaceId};
 use codespace_policy::Workspace;
 use codespace_runner::{
-    read_frame, write_frame, Runner, RunnerError, RunnerExecRequest, RunnerLifecycle, RunnerOp,
-    RunnerOpResult, UdsRunner, WireEnvelope, WIRE_PROTOCOL,
+    read_frame, write_frame, Runner, RunnerClaim, RunnerError, RunnerExecRequest, RunnerLifecycle,
+    RunnerOp, RunnerOpResult, UdsRunner, WireEnvelope, WIRE_PROTOCOL,
 };
+use tokio::io::AsyncBufReadExt;
 use tokio::net::UnixStream;
+use tokio::sync::mpsc;
 
 const BIN: &str = env!("CARGO_BIN_EXE_codespace-codex-runtime");
 
@@ -144,7 +146,7 @@ async fn hello_on_live_socket() {
     match parsed.result {
         Some(RunnerOpResult::Hello { protocol, .. }) => {
             assert_eq!(protocol, WIRE_PROTOCOL);
-            assert_eq!(protocol, 9);
+            assert_eq!(protocol, 10);
         }
         other => panic!("unexpected hello {other:?}"),
     }
@@ -212,21 +214,47 @@ async fn worker_socket_drop_terminates_owned_child() {
     );
 }
 
-/// 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)
+/// An independent Runner (P1R-C01) at `socket`, whose directory it prepares itself, and the
+/// lines it logs.
+fn spawn_independent(socket: &Path) -> (tokio::process::Child, mpsc::UnboundedReceiver<String>) {
+    let mut runner = tokio::process::Command::new(BIN)
         .arg(socket)
         .args(["--lifecycle", "independent"])
+        .stderr(std::process::Stdio::piped())
         .kill_on_drop(true)
         .spawn()
-        .expect("spawn an independent Runner")
+        .expect("spawn an independent Runner");
+    let (lines, logged) = mpsc::unbounded_channel();
+    let mut read = tokio::io::BufReader::new(runner.stderr.take().unwrap()).lines();
+    tokio::spawn(async move {
+        while let Ok(Some(line)) = read.next_line().await {
+            let _ = lines.send(line);
+        }
+    });
+    (runner, logged)
+}
+
+/// The first line logged that contains `text`.
+async fn logged(lines: &mut mpsc::UnboundedReceiver<String>, text: &str) -> String {
+    tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            let line = lines.recv().await.expect("the Runner logs it");
+            if line.contains(text) {
+                return line;
+            }
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("the Runner does not log {text:?}"))
 }
 
-/// A gateway's connection to the Runner at `socket`, after its handshake.
-async fn gateway(socket: &Path) -> UdsRunner {
+/// A gateway's connection to the Runner at `socket`, after its handshake and its Claim
+/// (P1R-C03).
+async fn gateway(socket: &Path) -> (UdsRunner, RunnerClaim) {
     let runner = UdsRunner::from_stream(wait_connect(socket).await, Arc::new(|_| {}));
     runner.handshake().await.expect("hello");
-    runner
+    let claim = runner.claim("store-test").await.expect("claim");
+    (runner, claim)
 }
 
 fn mode(path: &Path) -> u32 {
@@ -285,15 +313,16 @@ fn demo_workspace(root: &Path) -> Workspace {
 
 /// 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).
+/// its socket and the directory it created (P1R-C01). The Runner logs each Claim with the
+/// process that made it (P1R-C03).
 #[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;
+    let (mut runner, mut lines) = spawn_independent(&socket);
+    let (first, claim) = gateway(&socket).await;
     assert_eq!(mode(&dir), 0o700);
     assert_eq!(mode(base.path()), 0o755);
     let hello = first.hello();
@@ -302,7 +331,15 @@ async fn an_independent_runner_outlives_its_gateways_until_one_stops_it() {
         hello.identity.as_ref().map(|identity| identity.pid),
         runner.id()
     );
-    assert!(!hello.previous_controller);
+    assert_eq!((claim.epoch, claim.previous_controller), (1, false));
+    let line = logged(&mut lines, "claimed the Runner").await;
+    assert!(
+        line.contains(&format!(
+            "process {} claimed the Runner under epoch 1",
+            std::process::id()
+        )),
+        "{line}"
+    );
     // A second Runner neither takes nor removes the live socket.
     let second = Command::new(BIN)
         .arg(&socket)
@@ -329,11 +366,27 @@ async fn an_independent_runner_outlives_its_gateways_until_one_stops_it() {
     // 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);
+    let (next, inherited) = gateway(&socket).await;
+    assert_eq!(next.hello().identity, hello.identity);
+    assert_eq!((inherited.epoch, inherited.previous_controller), (2, true));
+    let mut processes: Vec<_> = inherited
+        .processes
+        .iter()
+        .map(|entry| (entry.status.process_id.0.as_str(), entry.status.state))
+        .collect();
+    processes.sort_by_key(|(process_id, _)| *process_id);
+    assert_eq!(
+        processes,
+        [
+            ("proc-pipe", ProcessState::Running),
+            ("proc-tty", ProcessState::Running)
+        ]
+    );
+    let line = logged(&mut lines, "under epoch 2").await;
+    assert!(
+        line.contains("an earlier controller started work"),
+        "{line}"
+    );
     assert_eq!(
         next.process_status(&ProcessId("proc-pipe".into()))
             .await
@@ -367,8 +420,8 @@ 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 (mut runner, _lines) = 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");

crates/domain/src/error.rs

@@ -64,10 +64,20 @@ pub enum ErrorCode {
     /// 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.
+    /// This gateway lost its connection to its independent Runner and cannot claim it again
+    /// (P1R-C01, P1R-C03): the Runner stopped, another Runner answers at its socket, or it
+    /// refused this gateway. The request never reached a Runner. Restart the gateway once the
+    /// Runner can be reached; a Runner that did not stop keeps its processes running.
     RunnerDisconnected,
+    /// This gateway lost its connection to its independent Runner and is claiming it again
+    /// (P1R-C03). The request never reached the Runner, whose processes keep running; retry
+    /// once the gateway has reconnected. Not occupancy.
+    RunnerReconnecting,
+    /// The independent Runner refused the request because it does not come from its current
+    /// controller (P1R-C03): another connection claimed the Runner after the request was
+    /// prepared, or the request's connection never claimed it. Nothing was started or changed.
+    /// Not occupancy; do not retry as busy.
+    StaleController,
     /// Scheduler or store invariant failed. Not occupancy; do not retry as busy.
     Internal,
 }
@@ -187,6 +197,10 @@ mod tests {
         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::RunnerReconnecting).unwrap();
+        assert_eq!(json, "\"RUNNER_RECONNECTING\"");
+        let json = serde_json::to_string(&ErrorCode::StaleController).unwrap();
+        assert_eq!(json, "\"STALE_CONTROLLER\"");
         let json = serde_json::to_string(&ErrorCode::Internal).unwrap();
         assert_eq!(json, "\"INTERNAL\"");
     }

crates/runner/src/lib.rs

@@ -150,14 +150,18 @@ pub use socket::{
     allocate_private_runner_dir, is_forbidden_runner_dir, prepare_runner_socket_dir,
     reclaim_leftover_socket, runner_socket_path, RUNNER_SOCKET_NAME,
 };
-pub use uds::{runner_disconnected, DisconnectHook, RunnerHello, UdsRunner, RUNNER_CALL_DEADLINE};
+pub use uds::{
+    runner_disconnected, runner_reconnecting, ClaimError, DisconnectHook, LinkEvent, LinkObserver,
+    RunnerHello, UdsRunner, CLAIM_START_BOUND, RECLAIM_BACKOFF_MAX, 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, recovery_pending, serve_runner_connection, write_frame,
-    IndependentRunner, RunnerEvent, RunnerOp, RunnerOpResult, Served, StopHandle, WireEnvelope,
-    WireKind, STOP_DRAIN_BOUND, WIRE_PROTOCOL,
+    host_worker, read_frame, recovery_pending, serve_runner_connection, write_frame, ClaimNote,
+    ClaimObserver, ClaimRefusal, IndependentRunner, RunnerClaim, RunnerEvent, RunnerEvents,
+    RunnerInventoryEntry, RunnerOp, RunnerOpResult, Served, StopHandle, WireEnvelope, WireKind,
+    CLAIM_WAIT_BOUND, STOP_DRAIN_BOUND, WIRE_PROTOCOL,
 };
 
 /// Execution-plane API. Control-plane fields (`work_id`, coordination,
@@ -322,13 +326,28 @@ impl RuntimeBackend {
         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 {
+    /// The controller epoch under which this gateway sends a new execution, patch or approval
+    /// resume, none for a runner it does not claim, or why it starts none (P1R-C01, P1R-C03):
+    /// see [`UdsRunner::dispatch_epoch`].
+    pub fn dispatch_epoch(&self) -> Result<Option<u64>, ErrorBody> {
         match self {
-            Self::InProcess(_) => false,
-            Self::Uds(runner) => runner.refuses_new_work(),
+            Self::InProcess(_) => Ok(None),
+            Self::Uds(runner) => runner.dispatch_epoch(),
+        }
+    }
+
+    /// Why this gateway starts no new execution, patch or approval resume, if it does not
+    /// (P1R-C01, P1R-C03): see [`UdsRunner::dispatch_epoch`].
+    pub fn new_work_refusal(&self) -> Option<ErrorBody> {
+        self.dispatch_epoch().err()
+    }
+
+    /// This backend, sending its requests under the controller epoch `epoch` when there is one
+    /// (P1R-C03): see [`UdsRunner::at_epoch`].
+    pub fn at_epoch(&self, epoch: Option<u64>) -> Self {
+        match (self, epoch) {
+            (Self::Uds(runner), Some(epoch)) => Self::Uds(runner.at_epoch(epoch)),
+            _ => self.clone(),
         }
     }
 }