runner, server, store: reconcile the independent Runner per workspace (P1R-C04)

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

@@ -148,8 +148,8 @@ async fn serve() {
     }
 }
 
-/// 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.
+/// What an independent Runner logs of each Claim (P1R-C03, P1R-C04): 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 {
@@ -161,7 +161,8 @@ fn log_claim(note: ClaimNote) {
             "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"
+                "; an earlier controller started work, so it starts nothing where that \
+                 controller did until it reconciled it"
             } else {
                 ""
             }

crates/codex-runtime/tests/runtime_binary.rs

@@ -6,7 +6,9 @@ use std::process::Command;
 use std::sync::Arc;
 use std::time::Duration;
 
-use codespace_domain::{ErrorCode, ProcessId, ProcessState, Profile, WorkspaceId};
+use codespace_domain::{
+    ErrorCode, ProcessId, ProcessState, ProcessTermination, Profile, WorkspaceId,
+};
 use codespace_policy::Workspace;
 use codespace_runner::{
     read_frame, write_frame, Runner, RunnerClaim, RunnerError, RunnerExecRequest, RunnerLifecycle,
@@ -146,7 +148,7 @@ async fn hello_on_live_socket() {
     match parsed.result {
         Some(RunnerOpResult::Hello { protocol, .. }) => {
             assert_eq!(protocol, WIRE_PROTOCOL);
-            assert_eq!(protocol, 10);
+            assert_eq!(protocol, 11);
         }
         other => panic!("unexpected hello {other:?}"),
     }
@@ -369,6 +371,7 @@ async fn an_independent_runner_outlives_its_gateways_until_one_stops_it() {
     let (next, inherited) = gateway(&socket).await;
     assert_eq!(next.hello().identity, hello.identity);
     assert_eq!((inherited.epoch, inherited.previous_controller), (2, true));
+    assert_eq!(inherited.barrier, ["demo"]);
     let mut processes: Vec<_> = inherited
         .processes
         .iter()
@@ -404,7 +407,35 @@ async fn an_independent_runner_outlives_its_gateways_until_one_stops_it() {
         Err(RunnerError::Execution(body)) => assert_eq!(body.code, ErrorCode::RecoveryPending),
         other => panic!("unexpected {other:?}"),
     }
-    assert_eq!(next.stop_runner().await.unwrap(), 2);
+    let stopped = next.stop_runner().await.unwrap();
+    assert_eq!(stopped.terminated, 2);
+    let mut ended: Vec<_> = stopped
+        .processes
+        .iter()
+        .map(|entry| {
+            (
+                entry.status.process_id.0.as_str(),
+                entry.status.state,
+                entry.status.termination,
+            )
+        })
+        .collect();
+    ended.sort_by_key(|(process_id, ..)| *process_id);
+    assert_eq!(
+        ended,
+        [
+            (
+                "proc-pipe",
+                ProcessState::Exited,
+                Some(ProcessTermination::Terminated)
+            ),
+            (
+                "proc-tty",
+                ProcessState::Exited,
+                Some(ProcessTermination::Terminated)
+            )
+        ]
+    );
     let status = exited(&mut runner).await;
     assert!(status.success(), "{status:?}");
     assert!(!still_beating(&pipe_beat).await);

crates/runner/src/admission.rs

@@ -29,7 +29,7 @@
 //! Cancelling is the one settlement used, and only where its precondition holds: this owner
 //! never asked DevGuard to launch the attempt, so a cancellation finds it prepared or settled.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::ffi::OsString;
 use std::path::Path;
 use std::sync::Arc;
@@ -564,6 +564,95 @@ fn wire_name(value: &impl Serialize) -> String {
         .unwrap_or_default()
 }
 
+/// The admissions an independent Runner asked the authority for, by attempt ID, each with the
+/// process it was for (P1R-C04): what it needs to look its attempts up again, since the
+/// authority checks a lookup against the attempt's meaning. Kept for the Runner's lifetime.
+#[derive(Clone, Default)]
+pub struct Admissions(Arc<std::sync::Mutex<HashMap<String, (ProcessId, Admission)>>>);
+
+impl Admissions {
+    fn remember(&self, process_id: &ProcessId, admission: &Admission) {
+        self.0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(
+                admission.attempt_id.clone(),
+                (process_id.clone(), admission.clone()),
+            );
+    }
+
+    fn admitted(&self, attempt_id: &str, process_id: &ProcessId) -> Option<Admission> {
+        self.0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(attempt_id)
+            .filter(|(admitted_for, _)| admitted_for == process_id)
+            .map(|(_, admission)| admission.clone())
+    }
+}
+
+/// A governed execution's attempt, which a gateway asks its independent Runner to look up
+/// (P1R-C04).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AttemptRef {
+    pub attempt_id: String,
+    pub process_id: ProcessId,
+}
+
+/// What the authority states of an attempt that an independent Runner looked up (P1R-C04).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AttemptLookup {
+    pub attempt_id: String,
+    pub process_id: ProcessId,
+    pub found: LookedUp,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "lookup", content = "attempt", rename_all = "snake_case")]
+pub enum LookedUp {
+    /// This Runner never asked the authority to admit this attempt for this process.
+    NotAdmitted,
+    /// The authority has no record of the attempt.
+    NotFound,
+    /// The authority's record of the attempt.
+    Found(AttemptState),
+    /// The authority could not be asked, or gave no usable answer: ask again later.
+    Failed,
+}
+
+/// Look up each of `attempts` with `authority`, as this owner admitted it (P1R-C04). An attempt
+/// that this owner never admitted for its process is not asked about.
+pub async fn look_up(
+    authority: Option<Arc<dyn AttemptAuthority>>,
+    admissions: &Admissions,
+    attempts: Vec<AttemptRef>,
+) -> Vec<AttemptLookup> {
+    let mut found = Vec::with_capacity(attempts.len());
+    for AttemptRef {
+        attempt_id,
+        process_id,
+    } in attempts
+    {
+        let looked_up = match (admissions.admitted(&attempt_id, &process_id), &authority) {
+            (None, _) => LookedUp::NotAdmitted,
+            (Some(_), None) => LookedUp::Failed,
+            (Some(admission), Some(authority)) => {
+                match call(authority, &admission, Call::Lookup).await {
+                    Answer::Attempt(attempt) => LookedUp::Found(attempt_state(&attempt)),
+                    Answer::Refused(devguard::ErrorCode::NotFound) => LookedUp::NotFound,
+                    Answer::Refused(_) | Answer::NotSent(_) | Answer::Unknown => LookedUp::Failed,
+                }
+            }
+        };
+        found.push(AttemptLookup {
+            attempt_id,
+            process_id,
+            found: looked_up,
+        });
+    }
+    found
+}
+
 /// Prepares and launches governed executions for the owner that holds both the process slots
 /// and the registered identity.
 #[derive(Clone)]
@@ -572,6 +661,7 @@ pub struct Preparer {
     runner: InProcessRunner,
     ttl_bound: Duration,
     launcher: Option<Launcher>,
+    admissions: Option<Admissions>,
 }
 
 impl Preparer {
@@ -581,9 +671,16 @@ impl Preparer {
             runner,
             ttl_bound: Duration::from_millis(devguard_prepared_ttl()),
             launcher: None,
+            admissions: None,
         }
     }
 
+    /// Remember each admission asked for in `admissions`, to look it up again (P1R-C04).
+    pub fn with_admissions(mut self, admissions: Option<Admissions>) -> Self {
+        self.admissions = admissions;
+        self
+    }
+
     /// Hold prepared attempts for at most `bound` here, whatever the authority grants.
     pub fn with_ttl_bound(mut self, bound: Duration) -> Self {
         self.ttl_bound = bound;
@@ -662,6 +759,9 @@ impl Preparer {
             reserved: None,
             applied: Applied::NotLaunched,
         };
+        if let Some(admissions) = &self.admissions {
+            admissions.remember(&req.process_id, &admission);
+        }
         let started = Instant::now();
         match call(&self.authority, &admission, Call::Admit).await {
             Answer::Attempt(attempt) => {
@@ -1874,4 +1974,71 @@ pub(crate) mod tests {
             assert_ne!(meaning(ws, req).digest().unwrap(), digest);
         }
     }
+
+    /// An independent Runner looks up only the attempts it admitted, each for the process it
+    /// admitted it for and as it admitted it. An authority that cannot answer, or a Runner
+    /// without its registration, leaves the attempt to be asked again (P1R-C04).
+    #[tokio::test]
+    async fn an_independent_runner_looks_up_only_the_attempts_it_admitted() {
+        let dir = tempfile::tempdir().unwrap();
+        let runner = runner();
+        let admissions = Admissions::default();
+        let fake = Fake::admitting([prepared()])
+            .then_lookup(Answer::Attempt(record(dg::Phase::LaunchCommitted, None)))
+            .then_lookup(Answer::Attempt(record(
+                dg::Phase::Released,
+                Some(dg::Release::NoHelperCreated),
+            )))
+            .then_lookup(Answer::Refused(devguard::ErrorCode::NotFound))
+            .then_lookup(Answer::Unknown)
+            .then_lookup(Answer::NotSent(unreachable_registration()));
+        let report = Preparer::new(fake.clone(), runner.clone())
+            .with_admissions(Some(admissions.clone()))
+            .prepare_and_dispose(
+                governed(dir.path()),
+                touch(dir.path(), "proc-l"),
+                "cs-l".into(),
+            )
+            .await;
+        assert_eq!(report.attempt, AttemptState::Cancelled);
+        let authority: Arc<dyn AttemptAuthority> = fake.clone();
+        let attempt = |attempt_id: &str, process_id: &str| AttemptRef {
+            attempt_id: attempt_id.into(),
+            process_id: ProcessId(process_id.into()),
+        };
+        let found = |lookups: Vec<AttemptLookup>| -> Vec<LookedUp> {
+            lookups.into_iter().map(|lookup| lookup.found).collect()
+        };
+
+        let elsewhere = look_up(
+            Some(authority.clone()),
+            &admissions,
+            vec![attempt("cs-other", "proc-l"), attempt("cs-l", "proc-other")],
+        )
+        .await;
+        assert_eq!(found(elsewhere), [LookedUp::NotAdmitted; 2]);
+        assert_eq!(fake.count(LOOKUP), 0);
+        for expected in [
+            LookedUp::Found(AttemptState::Launched),
+            LookedUp::Found(AttemptState::Released),
+            LookedUp::NotFound,
+            LookedUp::Failed,
+            LookedUp::Failed,
+        ] {
+            let lookups = look_up(
+                Some(authority.clone()),
+                &admissions,
+                vec![attempt("cs-l", "proc-l")],
+            )
+            .await;
+            assert_eq!(found(lookups), [expected]);
+        }
+        let unregistered = look_up(None, &admissions, vec![attempt("cs-l", "proc-l")]).await;
+        assert_eq!(found(unregistered), [LookedUp::Failed]);
+        assert_eq!(fake.count(LOOKUP), 5);
+        // Each lookup asked about the admission itself, under its digest.
+        let seen = fake.seen.lock().unwrap();
+        assert!(seen.iter().all(|admission| *admission == seen[0]));
+        assert_nothing_started(dir.path(), &runner);
+    }
 }

crates/runner/src/api.rs

@@ -153,6 +153,25 @@ pub struct RunnerProcessStatus {
     pub eof: bool,
 }
 
+/// What an independent Runner knows of a process that it was asked to start (P1R-C04). It
+/// remembers, for its whole lifetime, whether it created a process for each `process_id`, and
+/// nothing else of a process it no longer holds.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "recalled", rename_all = "snake_case")]
+pub enum Recalled {
+    /// No request to start it reached this Runner.
+    NotReceived,
+    /// A request to start it reached this Runner, which created no process for it.
+    Refused,
+    /// This Runner created it and holds it, running or ended.
+    Held {
+        workspace_id: String,
+        status: RunnerProcessStatus,
+    },
+    /// This Runner created it, and no longer holds it: it ended, and how is not known any more.
+    Dropped,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RunnerResizeResult {
     pub rows: u16,

crates/runner/src/lib.rs

@@ -122,7 +122,7 @@ mod watch;
 mod wire;
 
 pub use api::{
-    default_exec_timeout_ms, runner_local_exec_env, RunnerApplyPatchRequest,
+    default_exec_timeout_ms, runner_local_exec_env, Recalled, RunnerApplyPatchRequest,
     RunnerApplyPatchResult, RunnerCwd, RunnerError, RunnerExecEnv, RunnerExecPolicy,
     RunnerExecRequest, RunnerExecResult, RunnerProcessRoot, RunnerProcessStatus, RunnerReadProcess,
     RunnerReadResult, RunnerResizeResult, RunnerWriteStdin, DEFAULT_TIMEOUT_MS, MAX_OUTPUT_BYTES,
@@ -134,7 +134,7 @@ pub use fork_handlers::prepare_fork_spawns;
 pub use patch_helper::ensure_helper_for_tests;
 pub use process::{
     InProcessRunner, RetentionPolicy, ShellRelease, SlotReservation, DEFAULT_COMPLETED_TTL,
-    DEFAULT_MAX_COMPLETED, DEFAULT_MAX_PROCESSES, DEFAULT_TIMEOUT,
+    DEFAULT_MAX_COMPLETED, DEFAULT_MAX_PROCESSES, DEFAULT_MAX_UNACKNOWLEDGED, DEFAULT_TIMEOUT,
 };
 
 /// True when this process probed a working Linux sandbox helper.
@@ -152,16 +152,17 @@ pub use socket::{
 };
 pub use uds::{
     runner_disconnected, runner_reconnecting, ClaimError, DisconnectHook, LinkEvent, LinkObserver,
-    RunnerHello, UdsRunner, CLAIM_START_BOUND, RECLAIM_BACKOFF_MAX, RUNNER_CALL_DEADLINE,
+    RunnerHello, RunnerStopped, 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, ClaimNote,
-    ClaimObserver, ClaimRefusal, IndependentRunner, RunnerClaim, RunnerEvent, RunnerEvents,
-    RunnerInventoryEntry, RunnerOp, RunnerOpResult, Served, StopHandle, WireEnvelope, WireKind,
-    CLAIM_WAIT_BOUND, STOP_DRAIN_BOUND, WIRE_PROTOCOL,
+    ClaimObserver, ClaimRefusal, IndependentRunner, RecalledProcess, 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,
@@ -342,6 +343,18 @@ impl RuntimeBackend {
         self.dispatch_epoch().err()
     }
 
+    /// Tell an independent Runner that this gateway recorded how these processes ended
+    /// (P1R-C04): see [`UdsRunner::acknowledge`]. Any other runner keeps nothing for a
+    /// controller, so it is told nothing.
+    pub async fn acknowledge(&self, process_ids: &[ProcessId]) -> Result<(), RunnerError> {
+        match self {
+            Self::Uds(runner) if runner.hello().lifecycle == RunnerLifecycle::Independent => {
+                runner.acknowledge(process_ids).await
+            }
+            _ => Ok(()),
+        }
+    }
+
     /// 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 {