test: synchronize owned runtime crashes at restore decision checkpoints

crates/worker-runtime/Cargo.toml

@@ -21,6 +21,8 @@ default = ["ws-server", "fs-store"]
 fs-store = ["dep:fs4"]
 # Offline fixture tooling only; never enabled by the product defaults.
 test-support = ["fs-store", "ws-server"]
+# Explicit process-test synchronization; independent of offline fixture tooling.
+restore-checkpoints = ["fs-store"]
 http-server = ["dep:axum", "dep:tower", "dep:reqwest"]
 ws-server = ["http-server", "axum/ws", "tokio/sync"]
 

crates/worker-runtime/README.md

@@ -39,3 +39,30 @@ not stripped or rewritten. Portability/secret-marker checks do not replace
 review of the supplied artificial conversation or its independently verified hashes.
 The installer refuses any destination content other than initialized `auth.toml`;
 interrupted publication is fail-closed, not an overwrite/recovery facility.
+
+## Owned-process Restore checkpoints
+
+The independent, nondefault `restore-checkpoints` feature exists only for explicit
+process tests in the shared `tests/e2e` harness. Offline `test-support` does not enable
+it. Normal binaries reject `--test-restore-checkpoint`; no environment variable or
+HTTP endpoint activates this facility. Do not deploy a checkpoint-enabled binary
+as a replacement for the default product.
+
+A test-enabled Runtime accepts `--test-restore-checkpoint <config.json>` with a
+bounded regular JSON file selecting a loopback socket, Worker ID, one stage and a
+diagnostic nonce. The nonce is only a handshake label, never execution identity,
+request authority or routing state. The one-shot stages are real prepared candidate
+with `NotCommitted`, and durable `Committed` before activation/terminal receipt.
+Before acknowledging, the existing bounded canonical reader validates the exact
+Worker/operation/decision, syncs its file/directories and requires no saved receipt.
+No pending JSON is fabricated and no lifecycle work is skipped.
+
+The Runtime sends bounded newline-delimited JSON evidence, then waits at most eight
+seconds for byte `R`. EOF, invalid reply, unavailable peer or deadline releases the
+wait into the original operation. The Runtime-state mutex is not held across that
+wait; existing per-Worker ownership still covers the actual candidate and cleanup.
+The harness verifies the ack, kills/reaps only its owned child and inspects the
+canonical aggregate offline before restarting the same roots/identity without the
+checkpoint flag or another fixture install. These tests are a separate cfg/binary
+lane from featureless product regression; they do not certify arbitrary OS failure,
+external power loss or a real model call.

crates/worker-runtime/src/lib.rs

@@ -16,6 +16,8 @@ pub mod management;
 pub mod observation;
 pub mod profile_archive;
 pub mod resource;
+#[cfg(feature = "restore-checkpoints")]
+pub mod restore_checkpoints;
 #[cfg(test)]
 #[path = "../../../tests/support/restore_environment.rs"]
 mod restore_test_environment;

crates/worker-runtime/src/main.rs

@@ -117,6 +117,12 @@ fn run() -> Result<(), ProcessError> {
         let listener = tokio::net::TcpListener::bind(config.http.bind_addr).await?;
         let local_addr = listener.local_addr()?;
         let worker_runtime = build_runtime(&config)?;
+        #[cfg(feature = "restore-checkpoints")]
+        if let Some(checkpoint) = &config.test_restore_checkpoint {
+            worker_runtime
+                .install_test_restore_checkpoint(checkpoint.clone())
+                .map_err(ProcessError::Runtime)?;
+        }
         eprintln!(
             "yoi-runtime listening on {local_addr}; intended client is a trusted backend/proxy, not a browser"
         );
@@ -391,6 +397,21 @@ where
         let (flag, inline_value) = split_flag_value(arg)?;
         match flag.as_str() {
             "--help" | "-h" => return Ok(None),
+            #[cfg(feature = "restore-checkpoints")]
+            "--test-restore-checkpoint" => {
+                let path = take_value(&flag, inline_value, &mut args)?;
+                if config.test_restore_checkpoint.is_some() {
+                    return Err(ProcessError::usage(
+                        "--test-restore-checkpoint may only be specified once".into(),
+                    ));
+                }
+                config.test_restore_checkpoint = Some(
+                    worker_runtime::restore_checkpoints::RestoreCheckpointConfig::from_path(
+                        Path::new(&path),
+                    )
+                    .map_err(ProcessError::Runtime)?,
+                );
+            }
             "--bind" => {
                 let value = take_value(&flag, inline_value, &mut args)?;
                 config.http.bind_addr = value.parse::<SocketAddr>().map_err(|error| {
@@ -465,6 +486,12 @@ where
     }
 
     apply_store_selection(&mut config);
+    #[cfg(feature = "restore-checkpoints")]
+    if config.test_restore_checkpoint.is_some() && config.no_store {
+        return Err(ProcessError::usage(
+            "--test-restore-checkpoint requires an fs Runtime store".into(),
+        ));
+    }
     Ok(Some(config))
 }
 
@@ -521,6 +548,8 @@ struct ProcessConfig {
     no_store: bool,
     backend_resource_endpoint: Option<String>,
     backend_resource_token: Option<String>,
+    #[cfg(feature = "restore-checkpoints")]
+    test_restore_checkpoint: Option<worker_runtime::restore_checkpoints::RestoreCheckpointConfig>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -542,6 +571,8 @@ impl ProcessConfig {
             no_store: false,
             backend_resource_endpoint: None,
             backend_resource_token: None,
+            #[cfg(feature = "restore-checkpoints")]
+            test_restore_checkpoint: None,
         })
     }
 
@@ -1227,6 +1258,47 @@ Auth commands:
 mod tests {
     use super::*;
 
+    #[test]
+    fn checkpoint_cli_is_absent_by_default() {
+        let config = parse_args([] as [&str; 0]).unwrap().unwrap();
+        #[cfg(feature = "restore-checkpoints")]
+        assert!(config.test_restore_checkpoint.is_none());
+        #[cfg(not(feature = "restore-checkpoints"))]
+        {
+            let _ = config;
+            let error = parse_args(["--test-restore-checkpoint", "never-opened.json"]).unwrap_err();
+            assert!(error.to_string().contains("unknown argument"));
+        }
+    }
+
+    #[cfg(feature = "restore-checkpoints")]
+    #[test]
+    fn checkpoint_cli_requires_valid_explicit_file_and_fs_store() {
+        use worker_runtime::restore_checkpoints::{
+            RestoreCheckpointConfig, RestoreCheckpointStage,
+        };
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("checkpoint.json");
+        let checkpoint = RestoreCheckpointConfig {
+            endpoint: "127.0.0.1:1234".parse().unwrap(),
+            worker_id: protocol::WorkerId::now_v7(),
+            stage: RestoreCheckpointStage::CommittedPreterminal,
+            nonce: "cli-test".into(),
+        };
+        std::fs::write(&path, serde_json::to_vec(&checkpoint).unwrap()).unwrap();
+        let args = ["--test-restore-checkpoint", path.to_str().unwrap()];
+        let config = parse_args(args).unwrap().unwrap();
+        assert_eq!(config.test_restore_checkpoint, Some(checkpoint));
+        assert!(parse_args([args[0], args[1], "--no-store"]).is_err());
+        assert!(parse_args([args[0], args[1], args[0], args[1]]).is_err());
+        assert!(parse_args([args[0]]).is_err());
+        std::fs::write(&path, b"private-invalid-body").unwrap();
+        let error = parse_args(args).unwrap_err().to_string();
+        assert!(!error.contains("private-invalid-body"));
+        std::fs::remove_file(&path).unwrap();
+        assert!(parse_args(args).is_err());
+    }
+
     #[test]
     fn help_does_not_start_server() {
         assert!(parse_args(["--help"]).unwrap().is_none());

crates/worker-runtime/src/restore_checkpoints.rs

@@ -0,0 +1,143 @@
+//! Explicit, ephemeral synchronization for owned Runtime process tests only.
+//!
+//! The nonce authenticates no Runtime action and is never a routing identity.
+//! Socket failure, EOF and the fixed deadline only release the gate; the lifecycle
+//! owner continues its original operation and remains responsible for cleanup.
+
+use std::fs::OpenOptions;
+use std::io::{Read, Write};
+use std::net::{SocketAddr, TcpStream};
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+use protocol::WorkerId;
+use serde::{Deserialize, Serialize};
+
+use crate::error::RuntimeError;
+use crate::execution::WorkerLifecycleOperationId;
+
+pub const RESTORE_CHECKPOINT_TIMEOUT: Duration = Duration::from_secs(8);
+pub const MAX_RESTORE_CHECKPOINT_CONFIG_BYTES: usize = 4096;
+pub const MAX_RESTORE_CHECKPOINT_ACK_BYTES: usize = 4096;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RestoreCheckpointStage {
+    PreparedNotCommitted,
+    CommittedPreterminal,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RestoreCheckpointCommit {
+    NotCommitted,
+    Committed,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct RestoreCheckpointConfig {
+    pub endpoint: SocketAddr,
+    pub worker_id: WorkerId,
+    pub stage: RestoreCheckpointStage,
+    pub nonce: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct RestoreCheckpointAck {
+    pub nonce: String,
+    pub stage: RestoreCheckpointStage,
+    pub runtime_id: String,
+    pub worker_id: WorkerId,
+    pub operation_id: WorkerLifecycleOperationId,
+    pub commit: RestoreCheckpointCommit,
+    pub receipt_absent: bool,
+    pub pid: u32,
+}
+
+fn invalid_config() -> RuntimeError {
+    // Do not include serde errors, paths or input contents in process diagnostics.
+    RuntimeError::InvalidRequest("invalid test restore checkpoint configuration".into())
+}
+
+impl RestoreCheckpointConfig {
+    pub fn validate(&self) -> Result<(), RuntimeError> {
+        if !self.endpoint.ip().is_loopback()
+            || self.endpoint.port() == 0
+            || self.nonce.is_empty()
+            || self.nonce.len() > 256
+            || self.nonce.chars().any(char::is_control)
+        {
+            return Err(invalid_config());
+        }
+        Ok(())
+    }
+
+    pub fn from_path(path: &Path) -> Result<Self, RuntimeError> {
+        let mut options = OpenOptions::new();
+        options.read(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            // A FIFO must not stall startup before its regular-file check.
+            options.custom_flags(libc::O_NONBLOCK | libc::O_NOFOLLOW);
+        }
+        let file = options.open(path).map_err(|_| invalid_config())?;
+        let metadata = file.metadata().map_err(|_| invalid_config())?;
+        if !metadata.is_file() || metadata.len() > MAX_RESTORE_CHECKPOINT_CONFIG_BYTES as u64 {
+            return Err(invalid_config());
+        }
+        let mut bytes = Vec::new();
+        file.take(MAX_RESTORE_CHECKPOINT_CONFIG_BYTES as u64 + 1)
+            .read_to_end(&mut bytes)
+            .map_err(|_| invalid_config())?;
+        if bytes.len() > MAX_RESTORE_CHECKPOINT_CONFIG_BYTES {
+            return Err(invalid_config());
+        }
+        let config: Self = serde_json::from_slice(&bytes).map_err(|_| invalid_config())?;
+        config.validate()?;
+        Ok(config)
+    }
+}
+
+pub(crate) fn wait_for_release(config: &RestoreCheckpointConfig, ack: &RestoreCheckpointAck) {
+    let _ = exchange(config, ack);
+}
+
+fn remaining(deadline: Instant) -> std::io::Result<Duration> {
+    deadline
+        .checked_duration_since(Instant::now())
+        .filter(|remaining| !remaining.is_zero())
+        .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::TimedOut))
+}
+
+fn exchange(config: &RestoreCheckpointConfig, ack: &RestoreCheckpointAck) -> std::io::Result<()> {
+    let deadline = Instant::now() + RESTORE_CHECKPOINT_TIMEOUT;
+    let mut bytes = serde_json::to_vec(ack)?;
+    bytes.push(b'\n');
+    if bytes.len() > MAX_RESTORE_CHECKPOINT_ACK_BYTES {
+        return Err(std::io::Error::from(std::io::ErrorKind::InvalidData));
+    }
+    let mut stream = TcpStream::connect_timeout(&config.endpoint, remaining(deadline)?)?;
+    // Each syscall shares one budget; partial writes cannot reset the deadline.
+    let mut pending = bytes.as_slice();
+    while !pending.is_empty() {
+        stream.set_write_timeout(Some(remaining(deadline)?))?;
+        let written = stream.write(pending)?;
+        if written == 0 {
+            return Err(std::io::Error::from(std::io::ErrorKind::WriteZero));
+        }
+        pending = &pending[written..];
+    }
+    stream.set_read_timeout(Some(remaining(deadline)?))?;
+    let mut release = [0];
+    stream.read_exact(&mut release)?;
+    if release != [b'R'] {
+        return Err(std::io::Error::from(std::io::ErrorKind::InvalidData));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests;

crates/worker-runtime/src/restore_checkpoints/tests.rs

@@ -0,0 +1,139 @@
+use super::*;
+use std::io::{BufRead, BufReader};
+use std::net::TcpListener;
+use std::thread;
+
+fn config(endpoint: SocketAddr) -> RestoreCheckpointConfig {
+    RestoreCheckpointConfig {
+        endpoint,
+        worker_id: WorkerId::now_v7(),
+        stage: RestoreCheckpointStage::PreparedNotCommitted,
+        nonce: "checkpoint-test-handshake".into(),
+    }
+}
+
+fn ack(config: &RestoreCheckpointConfig) -> RestoreCheckpointAck {
+    RestoreCheckpointAck {
+        nonce: config.nonce.clone(),
+        stage: config.stage,
+        runtime_id: "checkpoint-runtime".into(),
+        worker_id: config.worker_id,
+        operation_id: WorkerLifecycleOperationId::new(),
+        commit: RestoreCheckpointCommit::NotCommitted,
+        receipt_absent: true,
+        pid: std::process::id(),
+    }
+}
+
+#[test]
+fn config_rejects_nonliteral_nonloopback_bad_stage_unknown_fields_and_invalid_worker() {
+    let valid = serde_json::to_value(config("127.0.0.1:1234".parse().unwrap())).unwrap();
+    for (field, value) in [
+        ("endpoint", "localhost:1234"),
+        ("endpoint", "192.0.2.1:1234"),
+        ("endpoint", "0.0.0.0:1234"),
+        ("endpoint", "127.0.0.1:0"),
+        ("endpoint", "[::ffff:127.0.0.1]:1234"),
+        ("worker_id", "not-a-worker-id"),
+        ("stage", "after_terminal"),
+        ("nonce", ""),
+        ("nonce", "secret\nbody"),
+        ("timeout", "unbounded"),
+    ] {
+        let mut invalid = valid.clone();
+        invalid[field] = value.into();
+        let parsed = serde_json::from_value::<RestoreCheckpointConfig>(invalid);
+        assert!(
+            parsed.is_err() || parsed.unwrap().validate().is_err(),
+            "{field}"
+        );
+    }
+    for endpoint in ["127.0.0.1:1234", "[::1]:1234"] {
+        config(endpoint.parse().unwrap()).validate().unwrap();
+    }
+}
+
+#[test]
+fn config_file_bounds_and_errors_do_not_disclose_contents() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("config.json");
+    assert!(RestoreCheckpointConfig::from_path(&path).is_err());
+    std::fs::write(&path, vec![b'x'; MAX_RESTORE_CHECKPOINT_CONFIG_BYTES + 1]).unwrap();
+    assert!(RestoreCheckpointConfig::from_path(&path).is_err());
+    std::fs::write(
+        &path,
+        br#"{"nonce":"private-config-body","stage":"secret"}"#,
+    )
+    .unwrap();
+    let error = RestoreCheckpointConfig::from_path(&path)
+        .unwrap_err()
+        .to_string();
+    assert!(!error.contains("private-config-body"));
+    assert!(!error.contains("secret"));
+    let mut valid = config("127.0.0.1:1234".parse().unwrap());
+    valid.nonce = "x".repeat(257);
+    assert!(valid.validate().is_err());
+    valid.nonce = "valid".into();
+    std::fs::write(&path, serde_json::to_vec(&valid).unwrap()).unwrap();
+    assert_eq!(RestoreCheckpointConfig::from_path(&path).unwrap(), valid);
+    assert!(RestoreCheckpointConfig::from_path(dir.path()).is_err());
+}
+
+#[test]
+fn explicit_release_eof_and_invalid_release_close_the_owned_stream() {
+    for release in [Some(b'R'), None, Some(b'X')] {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let config = config(listener.local_addr().unwrap());
+        let expected = ack(&config);
+        let sent = expected.clone();
+        let task = thread::spawn(move || exchange(&config, &sent));
+        let (mut stream, _) = listener.accept().unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(2)))
+            .unwrap();
+        let mut line = String::new();
+        BufReader::new(&mut stream).read_line(&mut line).unwrap();
+        assert!(line.len() <= MAX_RESTORE_CHECKPOINT_ACK_BYTES);
+        assert_eq!(
+            serde_json::from_str::<RestoreCheckpointAck>(&line).unwrap(),
+            expected
+        );
+        if let Some(byte) = release {
+            stream.write_all(&[byte]).unwrap();
+        } else {
+            stream.shutdown(std::net::Shutdown::Write).unwrap();
+        }
+        assert_eq!(task.join().unwrap().is_ok(), release == Some(b'R'));
+        let mut byte = [0];
+        assert_eq!(stream.read(&mut byte).unwrap(), 0);
+    }
+}
+
+#[test]
+fn absent_listener_resumes_without_waiting_or_retrying() {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let config = config(listener.local_addr().unwrap());
+    drop(listener);
+    let start = Instant::now();
+    wait_for_release(&config, &ack(&config));
+    assert!(start.elapsed() < Duration::from_secs(2));
+}
+
+#[test]
+fn silent_peer_hits_fixed_deadline_and_stream_is_dropped() {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let config = config(listener.local_addr().unwrap());
+    let start = Instant::now();
+    let task = thread::spawn(move || exchange(&config, &ack(&config)));
+    let (mut stream, _) = listener.accept().unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_secs(12)))
+        .unwrap();
+    let mut reader = BufReader::new(&mut stream);
+    let mut line = String::new();
+    reader.read_line(&mut line).unwrap();
+    assert!(task.join().unwrap().is_err());
+    assert!(start.elapsed() >= RESTORE_CHECKPOINT_TIMEOUT - Duration::from_millis(100));
+    assert!(start.elapsed() < RESTORE_CHECKPOINT_TIMEOUT + Duration::from_secs(3));
+    assert_eq!(reader.read(&mut [0]).unwrap(), 0);
+}

crates/worker-runtime/src/runtime.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "restore-checkpoints")]
+mod restore_checkpoint;
 mod restore_continuation;
 use crate::catalog::{
     ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector,
@@ -2776,6 +2778,12 @@ impl Runtime {
                 worker_state,
                 workdir_attachments,
             } => {
+                #[cfg(feature = "restore-checkpoints")]
+                self.test_restore_checkpoint(
+                    worker_ref,
+                    operation_id,
+                    crate::restore_checkpoints::RestoreCheckpointStage::PreparedNotCommitted,
+                );
                 let commit = self.commit_restored_worker_execution(
                     worker_ref,
                     worker_state.clone(),
@@ -2784,6 +2792,12 @@ impl Runtime {
                 );
                 match commit {
                     Ok(_) => {
+                        #[cfg(feature = "restore-checkpoints")]
+                        self.test_restore_checkpoint(
+                            worker_ref,
+                            operation_id,
+                            crate::restore_checkpoints::RestoreCheckpointStage::CommittedPreterminal,
+                        );
                         if let Err(error) =
                             backend.activate_restored_worker(operation_id, worker_ref)
                         {
@@ -5419,6 +5433,8 @@ impl std::fmt::Debug for BackendResourceClientRef {
 
 #[derive(Debug)]
 struct RuntimeState {
+    #[cfg(feature = "restore-checkpoints")]
+    restore_checkpoint: Option<crate::restore_checkpoints::RestoreCheckpointConfig>,
     display_name: Option<String>,
     backend: RuntimeBackendKind,
     /// Backend-bound stable identity used for cross-boundary retention evidence.
@@ -5466,6 +5482,8 @@ impl RuntimeState {
             display_name,
             backend: RuntimeBackendKind::Memory,
             runtime_identity: None,
+            #[cfg(feature = "restore-checkpoints")]
+            restore_checkpoint: None,
             persistence: RuntimePersistence::Memory,
             status: RuntimeStatus::Running,
             execution_backend: None,
@@ -5500,6 +5518,8 @@ impl RuntimeState {
             display_name,
             backend: RuntimeBackendKind::FsStore,
             runtime_identity: None,
+            #[cfg(feature = "restore-checkpoints")]
+            restore_checkpoint: None,
             persistence: RuntimePersistence::Fs(store),
             status: RuntimeStatus::Running,
             execution_backend: None,
@@ -5550,6 +5570,8 @@ impl RuntimeState {
             display_name: persisted.display_name,
             backend: RuntimeBackendKind::FsStore,
             runtime_identity: None,
+            #[cfg(feature = "restore-checkpoints")]
+            restore_checkpoint: None,
             persistence: RuntimePersistence::Fs(store),
             status: persisted.status,
             execution_backend: None,
@@ -7553,6 +7575,10 @@ fn subscription_worker_state(status: WorkerStatus) -> SubscriptionWorkerState {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(feature = "restore-checkpoints")]
+    mod restore_checkpoint_tests {
+        include!("runtime/restore_checkpoint_tests.rs");
+    }
     use super::*;
     use crate::catalog::{
         ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext,

crates/worker-runtime/src/runtime/restore_checkpoint.rs

@@ -0,0 +1,111 @@
+use super::*;
+use crate::restore_checkpoints::{
+    RestoreCheckpointAck, RestoreCheckpointCommit, RestoreCheckpointConfig, RestoreCheckpointStage,
+    wait_for_release,
+};
+
+impl Runtime {
+    /// Install an explicitly selected, one-shot process-test gate after normal startup.
+    /// It is not persisted or consulted by admission, routing or recovery authority.
+    pub fn install_test_restore_checkpoint(
+        &self,
+        config: RestoreCheckpointConfig,
+    ) -> Result<(), RuntimeError> {
+        config.validate()?;
+        let mut state = self.lock()?;
+        if !matches!(state.persistence, RuntimePersistence::Fs(_)) {
+            return Err(RuntimeError::InvalidRequest(
+                "test restore checkpoint requires an fs Runtime store".into(),
+            ));
+        }
+        state.restore_checkpoint = Some(config);
+        Ok(())
+    }
+
+    // The lifecycle caller holds the per-Worker operation lock. No Runtime-state
+    // mutex is retained while waiting; the real backend candidate remains owned by
+    // the same operation throughout this synchronous call.
+    pub(super) fn test_restore_checkpoint(
+        &self,
+        worker_ref: &WorkerRef,
+        operation_id: WorkerLifecycleOperationId,
+        stage: RestoreCheckpointStage,
+    ) {
+        let config = {
+            let Ok(mut state) = self.lock() else {
+                return;
+            };
+            let slot = &mut state.restore_checkpoint;
+            if !slot.as_ref().is_some_and(|config| {
+                config.worker_id == worker_ref.worker_id && config.stage == stage
+            }) {
+                return;
+            }
+            // Consume even if canonical verification or the handshake fails.
+            slot.take().expect("matched checkpoint")
+        };
+        let Ok(ack) = self.restore_checkpoint_ack(&config, worker_ref, operation_id) else {
+            return;
+        };
+        wait_for_release(&config, &ack);
+    }
+
+    pub(super) fn restore_checkpoint_ack(
+        &self,
+        config: &RestoreCheckpointConfig,
+        worker_ref: &WorkerRef,
+        operation_id: WorkerLifecycleOperationId,
+    ) -> Result<RestoreCheckpointAck, RuntimeError> {
+        let invalid = || {
+            RuntimeError::InvalidRequest("test restore checkpoint authority did not match".into())
+        };
+        let state = self.lock()?;
+        let RuntimePersistence::Fs(store) = &state.persistence else {
+            return Err(invalid());
+        };
+        // The existing helper validates the exact bounded canonical aggregate and
+        // syncs that same file plus its owning directories. Never manufacture a
+        // journal marker or infer commit from an in-memory projection.
+        let record = store.read_durable_worker_record(worker_ref)?;
+        let PersistedWorkerExecutionState::ReconciliationRequired(
+            PersistedWorkerLifecycleOperation::Restore(pending),
+        ) = &record.execution_state
+        else {
+            return Err(invalid());
+        };
+        let expected = match config.stage {
+            RestoreCheckpointStage::PreparedNotCommitted => RestoreCommitFact::NotCommitted,
+            RestoreCheckpointStage::CommittedPreterminal => RestoreCommitFact::Committed,
+        };
+        if config.worker_id != worker_ref.worker_id
+            || pending.operation_id != operation_id
+            || pending.commit != expected
+        {
+            return Err(invalid());
+        }
+        let guard = &record.restore_guard;
+        let id = guard.active_request_id.as_ref().ok_or_else(invalid)?;
+        let owner = guard.owners.get(id).ok_or_else(invalid)?;
+        if owner.operation_id != operation_id
+            || owner.request.request_id != *id
+            || owner.receipt.is_some()
+        {
+            return Err(invalid());
+        }
+        let runtime_id = state.runtime_identity.clone().ok_or_else(invalid)?;
+        Ok(RestoreCheckpointAck {
+            nonce: config.nonce.clone(),
+            stage: config.stage,
+            runtime_id,
+            worker_id: record.worker_id,
+            operation_id,
+            commit: match expected {
+                RestoreCommitFact::NotCommitted => RestoreCheckpointCommit::NotCommitted,
+                RestoreCommitFact::Committed => RestoreCheckpointCommit::Committed,
+                RestoreCommitFact::LegacyUnknown => unreachable!(),
+            },
+            receipt_absent: true,
+            pid: std::process::id(),
+        })
+    }
+}

crates/worker-runtime/src/runtime/restore_checkpoint_tests.rs

@@ -0,0 +1,296 @@
+use super::*;
+use crate::restore_checkpoints::{
+    RestoreCheckpointAck, RestoreCheckpointCommit, RestoreCheckpointConfig, RestoreCheckpointStage,
+};
+use std::io::{BufRead, BufReader, Write};
+use std::net::TcpListener;
+use std::time::Duration;
+
+#[test]
+fn checkpoint_stages_ack_canonical_decision_and_eof_resumes_same_operation_once() {
+    // Component scope: durable Runtime with a substituted execution backend.
+    // Real prepared Controllers and process death are owned by the E2E suite.
+    for (stage, release) in [
+        (RestoreCheckpointStage::PreparedNotCommitted, "release"),
+        (RestoreCheckpointStage::CommittedPreterminal, "eof"),
+        (RestoreCheckpointStage::CommittedPreterminal, "timeout"),
+    ] {
+        let dir = tempfile::tempdir().unwrap();
+        let backend = Arc::new(TestExecutionBackend::default());
+        let runtime = Runtime::with_fs_store_and_execution_backend(
+            FsRuntimeStoreOptions::new(dir.path().join("runtime"))
+                .with_runtime_id("checkpoint-runtime"),
+            backend.clone(),
+        )
+        .unwrap();
+        runtime.bind_runtime_identity("checkpoint-runtime").unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime
+            .create_worker(task_request("checkpoint original operation"))
+            .unwrap();
+        runtime.stop_worker(&worker.worker_ref, None).unwrap();
+        let request = runtime.test_restore_request(&worker.worker_ref);
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        runtime
+            .install_test_restore_checkpoint(RestoreCheckpointConfig {
+                endpoint: listener.local_addr().unwrap(),
+                worker_id: worker.worker_id,
+                stage,
+                nonce: "runtime-stage-test".into(),
+            })
+            .unwrap();
+        let cloned = runtime.clone();
+        let worker_ref = worker.worker_ref.clone();
+        let sent_request = request.clone();
+        let (sent, received) = std::sync::mpsc::channel();
+        let task = std::thread::spawn(move || {
+            sent.send(cloned.restore_worker_operation(&worker_ref, sent_request))
+                .unwrap();
+        });
+        listener.set_nonblocking(true).unwrap();
+        let stream = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                let listener = tokio::net::TcpListener::from_std(listener).unwrap();
+                let (stream, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept())
+                    .await
+                    .expect("checkpoint did not connect")
+                    .unwrap();
+                stream.into_std().unwrap()
+            });
+        stream.set_nonblocking(false).unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(3)))
+            .unwrap();
+        let mut reader = BufReader::new(stream);
+        let mut line = String::new();
+        reader.read_line(&mut line).unwrap();
+        let ack: RestoreCheckpointAck = serde_json::from_str(&line).unwrap();
+        assert_eq!(ack.stage, stage);
+        assert_eq!(ack.nonce, "runtime-stage-test");
+        assert_eq!(ack.runtime_id, "checkpoint-runtime");
+        assert_eq!(ack.worker_id, worker.worker_id);
+        assert_eq!(ack.pid, std::process::id());
+        assert!(ack.receipt_absent);
+        assert_eq!(
+            ack.commit,
+            match stage {
+                RestoreCheckpointStage::PreparedNotCommitted =>
+                    RestoreCheckpointCommit::NotCommitted,
+                RestoreCheckpointStage::CommittedPreterminal => RestoreCheckpointCommit::Committed,
+            }
+        );
+        assert_eq!(
+            backend.restore_operation_ids.lock().unwrap().as_slice(),
+            &[ack.operation_id]
+        );
+        assert!(backend.activation_operation_ids.lock().unwrap().is_empty());
+        assert!(received.try_recv().is_err());
+        let canonical = runtime_store(&runtime)
+            .read_durable_worker_record(&worker.worker_ref)
+            .unwrap();
+        let id = canonical.restore_guard.active_request_id.as_ref().unwrap();
+        assert_eq!(id, &request.request_id);
+        assert!(canonical.restore_guard.owners[id].receipt.is_none());
+        // Release, EOF and timeout all resume the original lifecycle owner.
+        let start = std::time::Instant::now();
+        if release == "release" {
+            reader.get_mut().write_all(b"R").unwrap();
+        } else if release == "eof" {
+            reader
+                .get_mut()
+                .shutdown(std::net::Shutdown::Write)
+                .unwrap();
+        }
+        let result = received
+            .recv_timeout(Duration::from_secs(12))
+            .unwrap()
+            .unwrap();
+        assert!(start.elapsed() < Duration::from_secs(11));
+        task.join().unwrap();
+        assert_eq!(result.state, WorkerRestoreState::Accepted);
+        assert_eq!(
+            backend.activation_operation_ids.lock().unwrap().as_slice(),
+            &[ack.operation_id]
+        );
+        assert_eq!(
+            runtime
+                .restore_worker_operation(&worker.worker_ref, request)
+                .unwrap(),
+            result
+        );
+        assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+        assert!(runtime.lock().unwrap().restore_checkpoint.is_none());
+        runtime.stop_worker(&worker.worker_ref, None).unwrap();
+        let next = runtime.test_restore_request(&worker.worker_ref);
+        assert_eq!(
+            runtime
+                .restore_worker_operation(&worker.worker_ref, next)
+                .unwrap()
+                .state,
+            WorkerRestoreState::Accepted
+        );
+        assert_eq!(*backend.restore_count.lock().unwrap(), 2);
+    }
+}
+
+#[test]
+fn checkpoint_disabled_mismatched_or_unreachable_does_not_change_restore_result() {
+    for mode in ["disabled", "mismatch", "unreachable"] {
+        let dir = tempfile::tempdir().unwrap();
+        let backend = Arc::new(TestExecutionBackend::default());
+        let runtime = Runtime::with_fs_store_and_execution_backend(
+            FsRuntimeStoreOptions::new(dir.path().join("runtime"))
+                .with_runtime_id("checkpoint-runtime"),
+            backend.clone(),
+        )
+        .unwrap();
+        runtime.bind_runtime_identity("checkpoint-runtime").unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime
+            .create_worker(task_request("checkpoint absent"))
+            .unwrap();
+        runtime.stop_worker(&worker.worker_ref, None).unwrap();
+        let request = runtime.test_restore_request(&worker.worker_ref);
+        if mode != "disabled" {
+            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+            let endpoint = listener.local_addr().unwrap();
+            drop(listener);
+            runtime
+                .install_test_restore_checkpoint(RestoreCheckpointConfig {
+                    endpoint,
+                    worker_id: if mode == "mismatch" {
+                        WorkerId::now_v7()
+                    } else {
+                        worker.worker_id
+                    },
+                    stage: RestoreCheckpointStage::PreparedNotCommitted,
+                    nonce: "absent-test".into(),
+                })
+                .unwrap();
+        }
+        let result = runtime
+            .restore_worker_operation(&worker.worker_ref, request.clone())
+            .unwrap();
+        assert_eq!(result.state, WorkerRestoreState::Accepted, "{mode}");
+        assert_eq!(
+            runtime
+                .restore_worker_operation(&worker.worker_ref, request)
+                .unwrap(),
+            result
+        );
+        assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+        assert_eq!(
+            runtime.lock().unwrap().restore_checkpoint.is_some(),
+            mode == "mismatch"
+        );
+    }
+}
+
+#[test]
+fn checkpoint_verification_requires_canonical_worker_operation_commit_and_absent_receipt() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().join("runtime");
+    let backend = Arc::new(TestExecutionBackend::default());
+    let runtime = Runtime::with_fs_store_and_execution_backend(
+        FsRuntimeStoreOptions::new(&root).with_runtime_id("checkpoint-runtime"),
+        backend.clone(),
+    )
+    .unwrap();
+    runtime.bind_runtime_identity("checkpoint-runtime").unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    let worker = runtime
+        .create_worker(task_request("checkpoint validation"))
+        .unwrap();
+    runtime.stop_worker(&worker.worker_ref, None).unwrap();
+    let request = runtime.test_restore_request(&worker.worker_ref);
+    *backend.restore_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::Errored(
+        WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, "not yet connected"),
+    ));
+    assert_eq!(
+        runtime
+            .restore_worker_operation(&worker.worker_ref, request.clone())
+            .unwrap()
+            .state,
+        WorkerRestoreState::ReconciliationRequired
+    );
+    let operation = backend.restore_operation_ids.lock().unwrap()[0];
+    let mut config = RestoreCheckpointConfig {
+        endpoint: "127.0.0.1:1234".parse().unwrap(),
+        worker_id: worker.worker_id,
+        stage: RestoreCheckpointStage::PreparedNotCommitted,
+        nonce: "verification-test".into(),
+    };
+    let path = root
+        .join("workers")
+        .join(worker.worker_id.to_string())
+        .join("worker.json");
+    let before = std::fs::read(&path).unwrap();
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_ok()
+    );
+    assert!(
+        runtime
+            .restore_checkpoint_ack(
+                &config,
+                &worker.worker_ref,
+                WorkerLifecycleOperationId::new()
+            )
+            .is_err()
+    );
+    config.worker_id = WorkerId::now_v7();
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_err()
+    );
+    config.worker_id = worker.worker_id;
+    config.stage = RestoreCheckpointStage::CommittedPreterminal;
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_err()
+    );
+    config.stage = RestoreCheckpointStage::PreparedNotCommitted;
+    runtime_store(&runtime).fail_next_worker_reconcile_sync();
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_err()
+    );
+    assert_eq!(
+        std::fs::read(&path).unwrap(),
+        before,
+        "verification must not write authority"
+    );
+    // In-memory equality must not hide an unreadable canonical aggregate.
+    std::fs::write(&path, b"unreadable canonical aggregate").unwrap();
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_err()
+    );
+    assert_eq!(
+        std::fs::read(&path).unwrap(),
+        b"unreadable canonical aggregate"
+    );
+    std::fs::write(&path, before).unwrap();
+    backend.restore_result.lock().unwrap().take();
+    assert_eq!(
+        runtime
+            .restore_worker_operation(&worker.worker_ref, request)
+            .unwrap()
+            .state,
+        WorkerRestoreState::Accepted
+    );
+    assert!(
+        runtime
+            .restore_checkpoint_ack(&config, &worker.worker_ref, operation)
+            .is_err(),
+        "terminal operation cannot ack a preterminal checkpoint"
+    );
+}

tests/e2e/Cargo.toml

@@ -10,6 +10,7 @@ autotests = false
 default = []
 e2e = []
 fixture-support = ["yoi-workspace-server/test-support", "worker-runtime/test-support"]
+restore-checkpoints = ["fixture-support", "worker-runtime/restore-checkpoints"]
 
 [dependencies]
 libc.workspace = true

tests/e2e/README.md

@@ -39,6 +39,24 @@ process終了ケースはそのAccepted receipt確認後に環境が所有する
 
 これは**commit済みかつterminal receipt保存後**のprocess終了であり、未commit候補やcommit直後・receipt前の終了証明ではない。固定Codex資産へSubmit/Notifyを送らないため、通常入力・Paused admissionの製品E2Eの証明でもない。検証jobはこの明示targetを選択する必要があり、通常suiteだけでは代替しない。
 
+## Opt-in Runtime checkpoint process試験
+
+上のfeatureless製品4case（target内の既存seed projection試験1件とは別）に加え、`restore-checkpoints`を明示した実Runtimeで2つのlive windowを試験する。このfeatureはoffline fixture-supportと独立しており、通常RuntimeではCLI入口も環境変数による入口も無い。
+
+```sh
+# 同期を有効にしたRuntimeだけを専用targetへ。通常製品binaryを上書きしない。
+cargo build --locked -p worker-runtime --bin yoi-runtime \
+  --features restore-checkpoints --target-dir target/e2e-checkpoint-products
+cargo test --locked -p yoi-e2e --features e2e,restore-checkpoints \
+  --test worker_restore checkpoint:: -- --test-threads=1
+```
+
+`WorkerE2eEnvironment::use_checkpoint_runtime`はその環境のRuntime binary/hashと、隔離root内の明示configだけを選択する。global provider・環境変数・通常起動への暗黙設定は変更しない。configは一度だけCLIへ渡し、再起動時は渡さない。test側のlink featureをfeatureless製品binaryへ混入させない。
+
+共有harnessはloopback ackのstage/Worker/operation/PID/nonceを確認し、直ちに所有childをkill/reapする。**offline正本を確認してから**同roots/identity/trustを再起動するため、ack後のdeadlineとの競合を成功扱いしない。NotCommittedの実私有候補は同operationのcleanup-only→RolledBack/Stopped、Committed/preterminalは同じdecision→実protocol/full Session snapshot→Acceptedへ回収する。元のrequest/token、owner数1、同operation、stale fresh intent拒否、保存source hash、restore run path数、最終cleanupを検証する。JSONでpending状態を作らず、seedは初回のみ。
+
+明示解除・peer EOF・observer scopeの意図的panicも試験する。gateの8秒deadlineとscoped HTTP callの15秒deadlineが待機を制限し、通常operationの継続後に環境のRAIIがprocess/rootを回収する。これはtest scope失敗の回収証明であり、OSによるharness強制終了や電源断の一般的な保証ではない。原fixtureへのSubmit/Notifyや実モデル呼出しは行わない。checkpoint ack/at-death/recovery、kill/reap、binary/hashとcleanup artifactはfeaturelessログと分けて報告する。
+
 通常のharness回帰試験（製品processは起動しない）:
 
 ```sh

tests/e2e/src/worker_environment.rs

@@ -266,6 +266,8 @@ pub struct WorkerE2eEnvironment {
     access_token: Option<String>,
     client: Client,
     binaries: WorkerBinaryProviderInfo,
+    #[cfg(feature = "restore-checkpoints")]
+    restore_checkpoint_config: Option<PathBuf>,
 }
 
 impl WorkerE2eEnvironment {
@@ -331,11 +333,49 @@ impl WorkerE2eEnvironment {
             access_token: None,
             client,
             binaries,
+            #[cfg(feature = "restore-checkpoints")]
+            restore_checkpoint_config: None,
         };
         environment.write_metadata("created")?;
         Ok(environment)
     }
 
+    /// Select the separately built test-enabled Runtime for this environment
+    /// only. The checkpoint file must already belong to its isolated root.
+    #[cfg(feature = "restore-checkpoints")]
+    pub fn use_checkpoint_runtime(&mut self, binary: PathBuf, config: &Path) -> Result<()> {
+        if self.runtime.is_some() || self.server.is_some() {
+            return Err(HarnessError::Protocol(
+                "checkpoint selection requires an unstarted environment".into(),
+            ));
+        }
+        let config = config.canonicalize()?;
+        if !config.starts_with(self.root.canonicalize()?) || !config.is_file() {
+            return Err(HarnessError::Protocol(
+                "checkpoint config is outside the owned environment".into(),
+            ));
+        }
+        let record = binary_record(
+            "yoi-runtime",
+            binary.canonicalize()?,
+            "restore-checkpoints-opt-in".into(),
+            None,
+        )?;
+        let runtime = self
+            .binaries
+            .binaries
+            .iter_mut()
+            .find(|record| record.name == "yoi-runtime")
+            .ok_or_else(|| HarnessError::Protocol("missing Runtime binary record".into()))?;
+        *runtime = record;
+        self.restore_checkpoint_config = Some(config);
+        fs::write(
+            self.artifacts_dir.join("binary-provider.json"),
+            serde_json::to_vec_pretty(&self.binaries)?,
+        )?;
+        Ok(())
+    }
+
     pub fn start(&mut self) -> Result<()> {
         self.start_backend()?;
         self.start_runtime()
@@ -639,6 +679,10 @@ impl WorkerE2eEnvironment {
             "--fs-root",
             self.runtime_data_dir.to_string_lossy().as_ref(),
         ]);
+        #[cfg(feature = "restore-checkpoints")]
+        if let Some(config) = self.restore_checkpoint_config.take() {
+            command.arg("--test-restore-checkpoint").arg(config);
+        }
         let runtime = ManagedProcess::spawn(
             "runtime",
             command,
@@ -684,6 +728,13 @@ impl WorkerE2eEnvironment {
     /// Kill/reap only the Runtime child owned by this environment, retaining roots
     /// and identities. Restart neither seeds storage nor rotates restore requests.
     pub fn crash_and_restart_runtime(&mut self) -> Result<()> {
+        self.crash_runtime()?;
+        self.start_runtime_process()
+    }
+
+    /// Leave the owned Runtime reaped and offline for read-only canonical crash
+    /// inspection before restart. No root, identity or request is recreated.
+    pub fn crash_runtime(&mut self) -> Result<()> {
         let runtime = self
             .runtime
             .as_mut()
@@ -708,6 +759,17 @@ impl WorkerE2eEnvironment {
         }
         self.runtime_url = None;
         self.runtime_shutdown = WorkerProcessShutdownOutcome::NotStarted;
+        Ok(())
+    }
+
+    /// Restart the previously initialized identity without seed or checkpoint flags.
+    #[cfg(feature = "restore-checkpoints")]
+    pub fn restart_runtime_after_crash(&mut self) -> Result<()> {
+        if self.runtime.is_some() || self.runtime_public_identity.is_none() {
+            return Err(HarnessError::Protocol(
+                "Runtime is not an offline initialized child".into(),
+            ));
+        }
         self.start_runtime_process()
     }
 
@@ -794,6 +856,37 @@ impl WorkerE2eEnvironment {
         )
     }
 
+    /// Build one authenticated call that may run in a scoped test thread while
+    /// the owner observes a checkpoint. The HTTP request deadline bounds joining
+    /// that thread even on test failure; credentials never enter artifacts.
+    #[cfg(feature = "restore-checkpoints")]
+    pub fn pending_restore_call(
+        &self,
+        worker_id: &str,
+        request_id: &str,
+        expected_observation_token: &str,
+    ) -> Result<impl FnOnce() -> Result<Value> + Send + 'static + use<>> {
+        let token = self.access_token.as_deref().ok_or_else(|| {
+            HarnessError::Protocol("Backend access token is not initialized".into())
+        })?;
+        let request = self
+            .client
+            .post(self.worker_url(worker_id, "restore")?)
+            .bearer_auth(token)
+            .timeout(STARTUP_WAIT)
+            .json(&serde_json::json!({"request_id": request_id,
+                "expected_observation_token": expected_observation_token}))
+            .build()
+            .map_err(|_| HarnessError::Protocol("build pending restore request".into()))?;
+        let client = self.client.clone();
+        Ok(move || {
+            let response = client
+                .execute(request)
+                .map_err(|_| HarnessError::Protocol("pending restore transport failed".into()))?;
+            decode_api_response(response, "pending restore")
+        })
+    }
+
     /// Forward one unchanged public restore request through an owned loopback
     /// relay, consume its complete upstream response, and close without delivering
     /// any response bytes. The returned JSON is relay evidence, not caller success.

tests/e2e/tests/worker_restore.rs

@@ -1,4 +1,7 @@
 //! Fixed, externally supplied Worker restore. Never generates a fixture or sends model input.
+#[cfg(feature = "restore-checkpoints")]
+#[path = "worker_restore/checkpoint.rs"]
+mod checkpoint;
 #[path = "worker_restore/fixture.rs"]
 mod fixture;
 #[path = "worker_restore/seed.rs"]

tests/e2e/tests/worker_restore/checkpoint.rs

@@ -0,0 +1,496 @@
+//! Opt-in real-process checkpoint cases; never substitute the featureless lane.
+use super::*;
+use std::collections::BTreeMap;
+use std::io::{Read, Write};
+use std::net::{TcpListener, TcpStream};
+use std::path::{Path, PathBuf};
+use worker_runtime::restore_checkpoints::{
+    MAX_RESTORE_CHECKPOINT_ACK_BYTES, RestoreCheckpointAck, RestoreCheckpointCommit,
+    RestoreCheckpointConfig, RestoreCheckpointStage,
+};
+
+struct Peer {
+    listener: TcpListener,
+    config: RestoreCheckpointConfig,
+    connection: Option<TcpStream>,
+}
+
+impl Peer {
+    fn configure(
+        environment: &mut WorkerE2eEnvironment,
+        fixture: &Fixture,
+        stage: RestoreCheckpointStage,
+    ) -> Result<Self> {
+        let listener = TcpListener::bind("127.0.0.1:0")?;
+        listener.set_nonblocking(true)?;
+        let config = RestoreCheckpointConfig {
+            endpoint: listener.local_addr()?,
+            worker_id: fixture::support(fixture.manifest.worker_id.parse())?,
+            stage,
+            nonce: format!(
+                "owned-checkpoint-{}-{}",
+                std::process::id(),
+                listener.local_addr()?.port()
+            ),
+        };
+        let path = environment.root.join("restore-checkpoint.json");
+        fs::write(&path, serde_json::to_vec(&config)?)?;
+        let provider = yoi_e2e::worker_binary_provider_info()?;
+        let binary = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("../../target/e2e-checkpoint-products")
+            .join(provider.profile)
+            .join("yoi-runtime");
+        environment.use_checkpoint_runtime(binary, &path)?;
+        Ok(Self {
+            listener,
+            config,
+            connection: None,
+        })
+    }
+
+    fn acknowledge(&mut self, environment: &WorkerE2eEnvironment) -> Result<RestoreCheckpointAck> {
+        let deadline = Instant::now() + Duration::from_secs(15);
+        let (mut stream, address) = loop {
+            match self.listener.accept() {
+                Ok(value) => break value,
+                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                    if Instant::now() >= deadline {
+                        return Err(failure("checkpoint ack deadline exceeded"));
+                    }
+                    std::thread::sleep(Duration::from_millis(10));
+                }
+                Err(error) => return Err(error.into()),
+            }
+        };
+        if !address.ip().is_loopback() {
+            return Err(failure("non-loopback checkpoint peer"));
+        }
+        let mut bytes = Vec::new();
+        loop {
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                return Err(failure("checkpoint ack read deadline exceeded"));
+            }
+            stream.set_read_timeout(Some(remaining))?;
+            let mut byte = [0];
+            stream.read_exact(&mut byte)?;
+            if byte == [b'\n'] {
+                break;
+            }
+            bytes.push(byte[0]);
+            if bytes.len() > MAX_RESTORE_CHECKPOINT_ACK_BYTES {
+                return Err(failure("checkpoint ack oversized"));
+            }
+        }
+        let ack: RestoreCheckpointAck = serde_json::from_slice(&bytes)?;
+        if ack.nonce != self.config.nonce
+            || ack.stage != self.config.stage
+            || ack.worker_id != self.config.worker_id
+            || ack.runtime_id != environment.runtime_id
+            || Some(ack.pid) != environment.runtime_pid()
+            || !ack.receipt_absent
+        {
+            return Err(failure(
+                "checkpoint ack does not match owned process and stage",
+            ));
+        }
+        fs::write(
+            environment.artifacts_dir.join("checkpoint-ack.json"),
+            serde_json::to_vec_pretty(&ack)?,
+        )?;
+        self.connection = Some(stream);
+        Ok(ack)
+    }
+
+    fn release(&mut self) -> Result<()> {
+        let mut stream = self
+            .connection
+            .take()
+            .ok_or_else(|| failure("checkpoint not acknowledged"))?;
+        stream.set_write_timeout(Some(Duration::from_secs(1)))?;
+        stream.write_all(b"R")?;
+        Ok(())
+    }
+}
+
+fn aggregate_path(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> PathBuf {
+    environment
+        .runtime_data_dir
+        .join("runtime/workers")
+        .join(&fixture.manifest.worker_id)
+}
+
+fn canonical(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> Result<serde_json::Value> {
+    let mut bytes = Vec::new();
+    fs::File::open(aggregate_path(environment, fixture).join("worker.json"))?
+        .take(8 * 1024 * 1024 + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() > 8 * 1024 * 1024 {
+        return Err(failure("canonical aggregate oversized"));
+    }
+    Ok(serde_json::from_slice(&bytes)?)
+}
+
+fn saved_source_hashes(
+    environment: &WorkerE2eEnvironment,
+    fixture: &Fixture,
+) -> Result<BTreeMap<String, String>> {
+    fn visit(root: &Path, path: &Path, files: &mut BTreeMap<String, String>) -> Result<()> {
+        if files.len() > 128 {
+            return Err(failure("saved source file budget exceeded"));
+        }
+        let metadata = fs::symlink_metadata(path)?;
+        if metadata.file_type().is_symlink() {
+            return Err(failure("unexpected saved source symlink"));
+        }
+        if metadata.is_dir() {
+            for entry in fs::read_dir(path)? {
+                visit(root, &entry?.path(), files)?;
+            }
+        } else if metadata.is_file() {
+            if metadata.len() > 8 * 1024 * 1024 {
+                return Err(failure("saved source byte budget exceeded"));
+            }
+            use sha2::{Digest, Sha256};
+            let digest: String = Sha256::digest(fs::read(path)?)
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect();
+            files.insert(
+                path.strip_prefix(root)
+                    .unwrap()
+                    .to_string_lossy()
+                    .into_owned(),
+                digest,
+            );
+        } else {
+            return Err(failure("unexpected saved source entry type"));
+        }
+        Ok(())
+    }
+    let root = aggregate_path(environment, fixture);
+    let mut files = BTreeMap::new();
+    for name in ["metadata.json", "session"] {
+        visit(&root, &root.join(name), &mut files)?;
+    }
+    if files.len() < 2 {
+        return Err(failure("saved source inventory is incomplete"));
+    }
+    Ok(files)
+}
+
+fn restore_runs(environment: &WorkerE2eEnvironment, fixture: &Fixture) -> Result<Vec<String>> {
+    let runs = aggregate_path(environment, fixture).join("runs");
+    if !runs.exists() {
+        return Ok(Vec::new());
+    }
+    let mut names = Vec::new();
+    for entry in fs::read_dir(runs)? {
+        let entry = entry?;
+        if entry.file_name().to_string_lossy().starts_with("restore-") {
+            if !entry.file_type()?.is_dir() {
+                return Err(failure("restore run is not a directory"));
+            }
+            names.push(entry.file_name().to_string_lossy().into_owned());
+        }
+    }
+    names.sort();
+    Ok(names)
+}
+
+#[test]
+fn acknowledged_not_committed_candidate_process_death_recovers_same_request_as_rollback()
+-> Result<()> {
+    crash_window(RestoreCheckpointStage::PreparedNotCommitted)
+}
+
+#[test]
+fn acknowledged_committed_preterminal_process_death_reconstructs_same_decision() -> Result<()> {
+    crash_window(RestoreCheckpointStage::CommittedPreterminal)
+}
+
+fn crash_window(stage: RestoreCheckpointStage) -> Result<()> {
+    let fixture = Fixture::load()?;
+    let result = (|| {
+        let mut environment = WorkerE2eEnvironment::new()?;
+        let mut peer = Peer::configure(&mut environment, &fixture, stage)?;
+        let mut installed = seed::start_in(&fixture, environment)?;
+        let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?;
+        let token = stopped
+            .restore_observation_token
+            .ok_or_else(|| failure("missing original observation"))?;
+        let original_sources = saved_source_hashes(&installed.environment, &fixture)?;
+        let request_id = "owned-process-checkpoint-request";
+        let call = installed.environment.pending_restore_call(
+            &fixture.manifest.worker_id,
+            request_id,
+            &token,
+        )?;
+        let ack = std::thread::scope(|scope| -> Result<RestoreCheckpointAck> {
+            let request = scope.spawn(call);
+            let ack = peer.acknowledge(&installed.environment)?;
+            installed.environment.crash_runtime()?;
+            // With the actual child reaped there can be no deadline race while
+            // inspecting the exact decision and nonterminal owner at process death.
+            let saved = canonical(&installed.environment, &fixture)?;
+            let op = saved
+                .pointer("/execution_state/execution/intent")
+                .ok_or_else(|| failure("checkpoint lacks canonical pending operation"))?;
+            let owner = &saved["restore_guard"]["owners"][request_id];
+            assert_eq!(op["operation_id"], serde_json::to_value(ack.operation_id)?);
+            assert_eq!(owner["operation_id"], op["operation_id"]);
+            assert!(owner["receipt"].is_null());
+            assert_eq!(owner["request"]["expected_observation_token"], token);
+            assert_eq!(op["source_repair_possible"], false);
+            let committed = stage == RestoreCheckpointStage::CommittedPreterminal;
+            assert_eq!(
+                op["commit"],
+                if committed {
+                    "committed"
+                } else {
+                    "not_committed"
+                }
+            );
+            assert_eq!(
+                ack.commit,
+                if committed {
+                    RestoreCheckpointCommit::Committed
+                } else {
+                    RestoreCheckpointCommit::NotCommitted
+                }
+            );
+            assert_eq!(
+                saved_source_hashes(&installed.environment, &fixture)?,
+                original_sources
+            );
+            assert_eq!(
+                restore_runs(&installed.environment, &fixture)?,
+                vec![format!("restore-{}", ack.operation_id)]
+            );
+            fs::write(
+                installed
+                    .environment
+                    .artifacts_dir
+                    .join("checkpoint-at-death.json"),
+                serde_json::to_vec_pretty(&serde_json::json!({
+                    "runtime_reaped_before_inspection": installed.environment.runtime_pid().is_none(),
+                    "operation_id": op["operation_id"], "commit": op["commit"],
+                    "receipt_absent": owner["receipt"].is_null(),
+                    "source_repair_possible": op["source_repair_possible"],
+                    "same_original_request_and_token": true,
+                    "restore_runs": restore_runs(&installed.environment, &fixture)?,
+                    "saved_source_hashes": saved_source_hashes(&installed.environment, &fixture)?,
+                }))?,
+            )?;
+            let outcome = request
+                .join()
+                .map_err(|_| failure("public restore caller panicked"))?;
+            if let Ok(value) = outcome {
+                let receipt: WorkerRestoreResponse = serde_json::from_value(value)?;
+                assert_ne!(
+                    receipt.result.state,
+                    WorkerRestoreState::Accepted,
+                    "no caller Accepted may precede either checkpoint kill"
+                );
+            }
+            installed.environment.restart_runtime_after_crash()?;
+            assert_ne!(installed.environment.runtime_pid(), Some(ack.pid));
+            Ok(ack)
+        })?;
+        drop(peer);
+        let recovered = installed.environment.restore_worker(
+            &fixture.manifest.worker_id,
+            request_id,
+            &token,
+        )?;
+        let receipt: WorkerRestoreResponse = serde_json::from_value(recovered.clone())?;
+        let committed = stage == RestoreCheckpointStage::CommittedPreterminal;
+        assert_eq!(
+            receipt.result.state,
+            if committed {
+                WorkerRestoreState::Accepted
+            } else {
+                WorkerRestoreState::RolledBack
+            }
+        );
+        assert_eq!(
+            installed.environment.restore_worker(
+                &fixture.manifest.worker_id,
+                request_id,
+                &token
+            )?,
+            recovered
+        );
+        let final_saved = canonical(&installed.environment, &fixture)?;
+        let owners = final_saved["restore_guard"]["owners"]
+            .as_object()
+            .ok_or_else(|| failure("missing canonical owners"))?;
+        assert_eq!(owners.len(), 1);
+        assert_eq!(
+            owners[request_id]["operation_id"],
+            serde_json::to_value(ack.operation_id)?
+        );
+        assert_eq!(
+            owners[request_id]["request"]["expected_observation_token"],
+            token
+        );
+        assert_eq!(
+            owners[request_id]["receipt"]["state"],
+            if committed { "accepted" } else { "rolled_back" }
+        );
+        let stale = installed.environment.restore_worker(
+            &fixture.manifest.worker_id,
+            "new-stale-checkpoint-intent",
+            &token,
+        );
+        if !matches!(stale, Err(error) if error.to_string().contains("HTTP 409")) {
+            return Err(failure("fresh stale intent was not rejected"));
+        }
+        fs::write(
+            installed
+                .environment
+                .artifacts_dir
+                .join("checkpoint-recovery.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "ack": ack, "recovered_state": receipt.result.state,
+                "canonical_operation_id": owners[request_id]["operation_id"],
+                "owner_count": owners.len(), "same_original_request_and_token": true,
+                "saved_source_hashes_before": original_sources,
+                "saved_source_hashes_after": saved_source_hashes(&installed.environment, &fixture)?,
+                "restore_runs_after": restore_runs(&installed.environment, &fixture)?,
+            }))?,
+        )?;
+        if committed {
+            assert_single_state(&installed.environment, &fixture, "idle")?;
+            assert_eq!(
+                restore_runs(&installed.environment, &fixture)?,
+                vec![format!("restore-{}", ack.operation_id)]
+            );
+            let mut subscription = WorkerSessionSubscription::connect(
+                &installed.environment,
+                &fixture.manifest.worker_id,
+                &installed.snapshot,
+            )?;
+            subscription.assert_alive()?;
+            stop_and_cleanup(installed, vec![subscription], &fixture)
+        } else {
+            assert_single_state(&installed.environment, &fixture, "stopped")?;
+            assert!(restore_runs(&installed.environment, &fixture)?.is_empty());
+            assert_eq!(
+                saved_source_hashes(&installed.environment, &fixture)?,
+                original_sources
+            );
+            let report = installed.environment.cleanup()?;
+            assert!(report.cleanup_success && !report.fixture_root_exists_after);
+            Ok(())
+        }
+    })();
+    fixture.source.assert_unchanged()?;
+    result
+}
+
+#[test]
+fn acknowledged_checkpoint_peer_disconnect_resumes_original_operation_and_tears_down() -> Result<()>
+{
+    observer_completion(false)
+}
+
+#[test]
+fn acknowledged_checkpoint_explicit_release_resumes_original_operation_and_tears_down() -> Result<()>
+{
+    observer_completion(true)
+}
+
+fn observer_completion(release: bool) -> Result<()> {
+    let fixture = Fixture::load()?;
+    let result = (|| {
+        let mut environment = WorkerE2eEnvironment::new()?;
+        let mut peer = Peer::configure(
+            &mut environment,
+            &fixture,
+            RestoreCheckpointStage::PreparedNotCommitted,
+        )?;
+        let installed = seed::start_in(&fixture, environment)?;
+        let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?;
+        let token = stopped
+            .restore_observation_token
+            .ok_or_else(|| failure("missing observation"))?;
+        let call = installed.environment.pending_restore_call(
+            &fixture.manifest.worker_id,
+            "peer-disconnect-request",
+            &token,
+        )?;
+        std::thread::scope(|scope| -> Result<()> {
+            let request = scope.spawn(call);
+            peer.acknowledge(&installed.environment)?;
+            // EOF models an abandoned observer, not a successful restore result.
+            if release {
+                peer.release()?;
+            } else {
+                drop(peer);
+            }
+            let value = request
+                .join()
+                .map_err(|_| failure("public caller panicked"))??;
+            let response: WorkerRestoreResponse = serde_json::from_value(value)?;
+            assert_eq!(response.result.state, WorkerRestoreState::Accepted);
+            Ok(())
+        })?;
+        let mut subscription = WorkerSessionSubscription::connect(
+            &installed.environment,
+            &fixture.manifest.worker_id,
+            &installed.snapshot,
+        )?;
+        subscription.assert_alive()?;
+        stop_and_cleanup(installed, vec![subscription], &fixture)
+    })();
+    fixture.source.assert_unchanged()?;
+    result
+}
+
+#[test]
+fn acknowledged_checkpoint_failed_observer_scope_has_bounded_owned_cleanup() -> Result<()> {
+    let fixture = Fixture::load()?;
+    let environment = WorkerE2eEnvironment::new()?;
+    let root = environment.root.clone();
+    let artifacts = environment.artifacts_dir.clone();
+    let started = Instant::now();
+    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| -> Result<()> {
+        let mut environment = environment;
+        let mut peer = Peer::configure(
+            &mut environment,
+            &fixture,
+            RestoreCheckpointStage::PreparedNotCommitted,
+        )?;
+        let installed = seed::start_in(&fixture, environment)?;
+        let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?;
+        let token = stopped
+            .restore_observation_token
+            .ok_or_else(|| failure("missing observation"))?;
+        let call = installed.environment.pending_restore_call(
+            &fixture.manifest.worker_id,
+            "failed-observer-request",
+            &token,
+        )?;
+        std::thread::scope(|scope| -> Result<()> {
+            scope.spawn(call);
+            peer.acknowledge(&installed.environment)?;
+            // Leave the peer connected while unwinding the scoped HTTP caller.
+            // The Runtime's own deadline must unblock it, then RAII must reap
+            // both owned processes and remove the environment, not wait forever.
+            panic!("intentional checkpoint observer failure");
+        })
+    }));
+    assert!(outcome.is_err(), "the observer failure was not exercised");
+    assert!(
+        started.elapsed() < Duration::from_secs(45),
+        "owned cleanup exceeded scenario budget"
+    );
+    assert!(!root.exists(), "failed observer left its isolated root");
+    let cleanup: serde_json::Value =
+        serde_json::from_slice(&fs::read(artifacts.join("cleanup.json"))?)?;
+    assert_eq!(cleanup["cleanup_success"], true);
+    assert_eq!(cleanup["fixture_root_exists_after"], false);
+    fixture.source.assert_unchanged()?;
+    Ok(())
+}

tests/e2e/tests/worker_restore/seed.rs

@@ -42,7 +42,10 @@ pub struct Installed {
 }
 
 pub fn start(fixture: &Fixture) -> Result<Installed> {
-    let mut environment = WorkerE2eEnvironment::new()?;
+    start_in(fixture, WorkerE2eEnvironment::new()?)
+}
+
+pub fn start_in(fixture: &Fixture, mut environment: WorkerE2eEnvironment) -> Result<Installed> {
     super::phase(&environment, "fixture-copy-verify")?;
     let copy = environment.root.join("provided-fixture");
     fixture.source.copy_to(&copy)?;