test: add isolated saved worker restore scenarios

Cargo.lock

@@ -6707,15 +6707,21 @@ name = "yoi-e2e"
 version = "0.0.0"
 dependencies = [
  "chrono",
+ "config-source",
  "libc",
+ "manifest",
+ "protocol",
  "reqwest",
  "serde",
  "serde_json",
  "server-api",
+ "session-store",
  "sha2 0.11.0",
  "tempfile",
+ "tungstenite 0.29.0",
  "webauthn-authenticator-rs",
  "webauthn-rs-proto",
+ "worker-runtime",
  "yoi-workspace-server",
 ]
 

tests/e2e/Cargo.toml

@@ -9,6 +9,7 @@ autotests = false
 [features]
 default = []
 e2e = []
+fixture-support = ["yoi-workspace-server/test-support", "worker-runtime/test-support"]
 
 [dependencies]
 libc.workspace = true
@@ -17,11 +18,17 @@ serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
 sha2.workspace = true
 tempfile.workspace = true
+protocol.workspace = true
+tungstenite = { version = "0.29", default-features = false, features = ["handshake"] }
 
 [dev-dependencies]
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 yoi-workspace-server.workspace = true
 server-api.workspace = true
+worker-runtime.workspace = true
+session-store.workspace = true
+config-source.workspace = true
+manifest.workspace = true
 webauthn-rs-proto = "0.5.5"
 webauthn-authenticator-rs = { version = "0.5.5", default-features = false, features = ["softpasskey"] }
 
@@ -44,3 +51,8 @@ required-features = ["e2e"]
 name = "worker_environment"
 path = "tests/worker_environment.rs"
 required-features = ["e2e"]
+
+[[test]]
+name = "worker_restore"
+path = "tests/worker_restore.rs"
+required-features = ["e2e", "fixture-support"]

tests/e2e/README.md

@@ -7,3 +7,41 @@
 - [`PanelHarness`]: PTY経由の入力、観測イベントによる同期、失敗時のartifact保存。
 - [`FixtureWorkspace`]: TUI用WorkspaceとHOME／XDG領域の隔離。
 - [`WorkerE2eEnvironment`]: Backend／Runtime用プロセスとfixtureの管理。
+- [`FixtureSource`]: 提供資産の全ファイルSHA-256記録、独立copy、原本不変の検証。
+- [`WorkerSessionSubscription`]: 公開Worker WebSocketの初期full Snapshot照合と停止確認。モデル呼出しは行わない。
+
+## 固定Workerの段階起動
+
+1. 提供fixtureを所有crateの型で検証し、`FixtureSource`で原本hashを記録する。各環境の一時領域へcopyする。fixtureが未提供・不適合なら失敗とし、代替Workerを生成しない。
+2. `initialize_runtime_identity`で固定Runtime IDに新しい鍵を発行する。Backendの型付きstore APIで新しいAccount認証・Workspace署名identity・Runtime binding・固定Worker登録を準備する。
+3. `start_seeded_backend`へ固定Workspace/Runtime IDとその環境の新しいbearerを渡す。Server起動後の公開APIからWorkspaceの公開identityを取得する。
+4. `start_runtime_with_seed`はtrust設定後、Runtime processの起動前にseed closureを呼ぶ。ここで実Server URLを使いRuntime所有crateの型付きinstallを呼ぶ。Runtime store rootは`runtime_data_dir.join("runtime")`。
+5. Worker一覧とdetailから停止状態・観測tokenを確認し、一つの固定request IDで`restore_worker`を一度だけ呼ぶ。結果不明・拒否は成功扱いせず、tokenを自動更新しない。
+6. `WorkerSessionSubscription::connect`で期待`protocol::SessionSnapshot`全体（entry ID、timestamp、順序、本文を含む）を照合する。二環境を同時に保持し、一方のstop/cleanup後も他方の購読・一覧・内容を再確認する。
+7. `stop_worker`、購読の`expect_stopped`、環境cleanup、原本`assert_unchanged`を確認する。
+
+所有crateの`test-support`依存はテスト側の非default `fixture-support` featureだけで有効にする。製品binary providerは別Cargo invocationと専用`target/e2e-products`へ`yoi`・`yoi-server`・`yoi-runtime`を追加featureなしでビルドする。support test自身をこのtargetへビルドしてはならない。`YOI_E2E_BIN`等の明示binary overrideを使用する場合も、呼出し側が同じ分離を保証する。既存TUI用`e2e-test` featureは別の`target/e2e-tui-products`へビルドし、restore製品binaryへ混入しない。
+
+固定資産の正常restore・並行隔離E2E:
+
+```sh
+cargo test -p yoi-e2e --features e2e,fixture-support --test worker_restore
+# 親から提供された候補の検証（読取専用、原本の自動採用や生成はしない）
+YOI_E2E_RESTORE_FIXTURE=target/fixture-candidate-v1 \
+  cargo test -p yoi-e2e --features e2e,fixture-support --test worker_restore
+```
+
+標準資産は`tests/e2e/fixtures/worker-codex-v1`の`manifest.json`・`worker-template.json`・`expected.json`・`workspace-config.dcdl`。manifestの各asset hash/size、schemaと固定ID、source revision、modelを検証する。Workspace configのimport参照を黙って最新へ置換せず、Gitでsource revisionに対する`resources`と`crates/builtin-source`の一致を確認する（Git履歴がない場合も失敗）。移植したSessionの正規公開projectionを供給された会話行と照合し、その全snapshotを実executionのWebSocketと比較する。欠落fixture・hash不一致・未実行は成功扱いしない。
+
+通常のharness回帰試験（製品processは起動しない）:
+
+```sh
+cargo test -p yoi-e2e --lib --no-default-features
+```
+
+既存の製品process E2Eは別結果として実行・報告する:
+
+```sh
+cargo test -p yoi-e2e --features e2e --test worker_environment
+cargo test -p yoi-e2e --features e2e --test panel --test rewind
+```

tests/e2e/src/fixture_source.rs

@@ -0,0 +1,173 @@
+//! Byte-preserving fixture transport, independent of the owning store's schema validation.
+use std::collections::BTreeMap;
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use sha2::{Digest, Sha256};
+
+use crate::{HarnessError, Result};
+
+#[derive(Debug, Clone)]
+pub struct FixtureSource {
+    root: PathBuf,
+    hashes: BTreeMap<PathBuf, String>,
+}
+
+impl FixtureSource {
+    /// Record every supplied file, including the manifest. No fixture is generated
+    /// or repaired here; the typed store importer must additionally validate it.
+    pub fn capture(root: &Path) -> Result<Self> {
+        let hashes = inventory(root)?;
+        if hashes.is_empty() {
+            return Err(HarnessError::Protocol("provided fixture is empty".into()));
+        }
+        Ok(Self {
+            root: root.to_owned(),
+            hashes,
+        })
+    }
+
+    pub fn hashes(&self) -> &BTreeMap<PathBuf, String> {
+        &self.hashes
+    }
+
+    pub fn assert_unchanged(&self) -> Result<()> {
+        if inventory(&self.root)? != self.hashes {
+            return Err(HarnessError::Protocol(
+                "provided fixture changed during execution".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    /// The destination must not exist. A copy is never a link to the source.
+    pub fn copy_to(&self, destination: &Path) -> Result<()> {
+        self.assert_unchanged()?;
+        let parent = destination
+            .parent()
+            .filter(|path| !path.as_os_str().is_empty())
+            .unwrap_or_else(|| Path::new("."));
+        if parent
+            .canonicalize()?
+            .starts_with(self.root.canonicalize()?)
+        {
+            return Err(HarnessError::Protocol(
+                "fixture copy destination is inside the original".into(),
+            ));
+        }
+        fs::create_dir(destination)?;
+        for (relative, expected) in &self.hashes {
+            let bytes = fs::read(self.root.join(relative))?;
+            if hash(&bytes) != *expected {
+                return Err(HarnessError::Protocol(
+                    "fixture changed while copying".into(),
+                ));
+            }
+            let target = destination.join(relative);
+            fs::create_dir_all(target.parent().unwrap())?;
+            use std::io::Write;
+            let mut file = fs::OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(target)?;
+            file.write_all(&bytes)?;
+        }
+        if inventory(destination)? != self.hashes {
+            return Err(HarnessError::Protocol(
+                "fixture copy failed hash verification".into(),
+            ));
+        }
+        self.assert_unchanged()
+    }
+}
+
+fn hash(bytes: &[u8]) -> String {
+    format!(
+        "sha256:{}",
+        Sha256::digest(bytes)
+            .iter()
+            .map(|byte| format!("{byte:02x}"))
+            .collect::<String>()
+    )
+}
+
+fn inventory(root: &Path) -> Result<BTreeMap<PathBuf, String>> {
+    fn visit(root: &Path, path: &Path, files: &mut BTreeMap<PathBuf, String>) -> Result<()> {
+        let metadata = fs::symlink_metadata(path)?;
+        if metadata.is_dir() {
+            for entry in fs::read_dir(path)? {
+                visit(root, &entry?.path(), files)?;
+            }
+        } else if metadata.is_file() {
+            files.insert(
+                path.strip_prefix(root).unwrap().to_owned(),
+                hash(&fs::read(path)?),
+            );
+        } else {
+            return Err(HarnessError::Protocol(
+                "fixture contains a symlink or special file".into(),
+            ));
+        }
+        Ok(())
+    }
+    if !fs::symlink_metadata(root)?.is_dir() {
+        return Err(HarnessError::Protocol(
+            "fixture source must be a real directory".into(),
+        ));
+    }
+    let mut files = BTreeMap::new();
+    visit(root, root, &mut files)?;
+    Ok(files)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // These are transport-only byte samples, never saved Worker substitutes.
+    #[test]
+    fn copies_are_independent_and_source_mutation_is_detected() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let source = temp.path().join("source");
+        fs::create_dir(&source)?;
+        fs::write(source.join("bytes"), b"original")?;
+        let fixture = FixtureSource::capture(&source)?;
+        assert!(fixture.copy_to(&source.join("nested-copy")).is_err());
+        assert!(!source.join("nested-copy").exists());
+        fixture.assert_unchanged()?;
+        let first = temp.path().join("first");
+        let second = temp.path().join("second");
+        fixture.copy_to(&first)?;
+        fixture.copy_to(&second)?;
+        fs::write(first.join("bytes"), b"changed")?;
+        fs::remove_dir_all(first)?;
+        assert_eq!(fs::read(second.join("bytes"))?, b"original");
+        fixture.assert_unchanged()?;
+        fs::write(source.join("bytes"), b"changed")?;
+        assert!(fixture.assert_unchanged().is_err());
+        Ok(())
+    }
+
+    #[test]
+    fn rejects_missing_empty_existing_destination_and_extra_file() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        assert!(FixtureSource::capture(&temp.path().join("missing")).is_err());
+        assert!(FixtureSource::capture(temp.path()).is_err());
+        fs::write(temp.path().join("bytes"), b"original")?;
+        let source = FixtureSource::capture(temp.path())?;
+        assert!(source.copy_to(temp.path()).is_err());
+        fs::write(temp.path().join("extra"), b"new")?;
+        assert!(source.assert_unchanged().is_err());
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn rejects_symbolic_links() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        fs::write(temp.path().join("bytes"), b"original")?;
+        std::os::unix::fs::symlink("bytes", temp.path().join("link"))?;
+        assert!(FixtureSource::capture(temp.path()).is_err());
+        Ok(())
+    }
+}

tests/e2e/src/lib.rs

@@ -1268,7 +1268,11 @@ impl Drop for PanelHarness {
     }
 }
 
+mod fixture_source;
+pub use fixture_source::FixtureSource;
 mod worker_environment;
+mod worker_session;
+pub use worker_session::{WorkerSessionSubscription, validate_full_snapshot};
 
 pub use worker_environment::{
     WorkerBinaryProviderInfo, WorkerE2eEnvironment, WorkerEnvironmentCleanupReport,
@@ -1627,6 +1631,8 @@ fn resolve_yoi_binary() -> Result<BinaryProviderInfo> {
         "e2e-test".to_owned(),
         "--bin".to_owned(),
         "yoi".to_owned(),
+        "--target-dir".to_owned(),
+        tui_product_target_dir()?.to_string_lossy().into_owned(),
     ];
     if test_profile() == "release" {
         args.push("--release".to_owned());
@@ -1648,7 +1654,9 @@ fn resolve_yoi_binary() -> Result<BinaryProviderInfo> {
         });
     }
 
-    let binary = current_target_profile_dir()?.join(binary_name());
+    let binary = tui_product_target_dir()?
+        .join(test_profile())
+        .join(binary_name());
     let info = BinaryProviderInfo {
         provider: "cargo-build".to_owned(),
         binary,
@@ -1675,6 +1683,26 @@ fn workspace_root() -> Result<PathBuf> {
         .ok_or_else(|| HarnessError::Protocol("could not resolve workspace root".to_owned()))
 }
 
+/// Product processes must never reuse the target used by test-support builds.
+fn product_target_dir() -> Result<PathBuf> {
+    dedicated_product_target("e2e-products")
+}
+
+// Existing PTY tests need their observer feature, never on restore binaries.
+fn tui_product_target_dir() -> Result<PathBuf> {
+    dedicated_product_target("e2e-tui-products")
+}
+
+fn dedicated_product_target(name: &str) -> Result<PathBuf> {
+    let target = workspace_root()?.join("target").join(name);
+    if current_target_profile_dir()?.parent() == Some(target.as_path()) {
+        return Err(HarnessError::Protocol(
+            "run support tests outside reserved product binary targets".to_owned(),
+        ));
+    }
+    Ok(target)
+}
+
 fn current_target_profile_dir() -> Result<PathBuf> {
     let mut path = std::env::current_exe()?;
     while let Some(name) = path.file_name().and_then(|name| name.to_str()) {

tests/e2e/src/worker_environment.rs

@@ -13,8 +13,8 @@ use serde_json::Value;
 use sha2::{Digest, Sha256};
 
 use super::{
-    FixtureAllocation, HarnessError, Result, allocate_fixture, command_display,
-    current_target_profile_dir, now_ms, test_profile, workspace_root, yoi_binary_info,
+    FixtureAllocation, HarnessError, Result, allocate_fixture, command_display, now_ms,
+    product_target_dir, test_profile, workspace_root,
 };
 
 const STARTUP_WAIT: Duration = Duration::from_secs(15);
@@ -61,22 +61,20 @@ pub fn worker_binary_provider_info() -> Result<WorkerBinaryProviderInfo> {
 fn resolve_worker_binaries() -> Result<WorkerBinaryProviderInfo> {
     let workspace_root = workspace_root()?;
     let profile = test_profile();
-    let yoi = yoi_binary_info()?;
-    let mut records = vec![binary_record(
-        "yoi",
-        yoi.binary,
-        yoi.provider,
-        yoi.build_command,
-    )?];
-
+    let mut records = Vec::new();
+    let yoi_override = resolve_override("YOI_E2E_BIN", &workspace_root)?;
     let server_override = resolve_override("YOI_E2E_SERVER_BIN", &workspace_root)?;
     let runtime_override = resolve_override("YOI_E2E_RUNTIME_BIN", &workspace_root)?;
     let mut build_command = None;
-    if server_override.is_none() || runtime_override.is_none() {
+    if yoi_override.is_none() || server_override.is_none() || runtime_override.is_none() {
         let cargo = PathBuf::from(std::env::var_os("CARGO").unwrap_or_else(|| "cargo".into()));
         let mut args = vec![
             "build".to_string(),
             "-p".to_string(),
+            "yoi".to_string(),
+            "--bin".to_string(),
+            "yoi".to_string(),
+            "-p".to_string(),
             "yoi-workspace-server".to_string(),
             "--bin".to_string(),
             "yoi-server".to_string(),
@@ -84,6 +82,8 @@ fn resolve_worker_binaries() -> Result<WorkerBinaryProviderInfo> {
             "worker-runtime".to_string(),
             "--bin".to_string(),
             "yoi-runtime".to_string(),
+            "--target-dir".to_string(),
+            product_target_dir()?.to_string_lossy().into_owned(),
         ];
         if profile == "release" {
             args.push("--release".to_string());
@@ -106,7 +106,16 @@ fn resolve_worker_binaries() -> Result<WorkerBinaryProviderInfo> {
         build_command = Some(rendered);
     }
 
-    let profile_dir = current_target_profile_dir()?;
+    let profile_dir = product_target_dir()?.join(&profile);
+    let (yoi_path, yoi_provider, yoi_build) = match yoi_override {
+        Some(path) => (path, "YOI_E2E_BIN".to_string(), None),
+        None => (
+            profile_dir.join(executable_name("yoi")),
+            "cargo-build".to_string(),
+            build_command.clone(),
+        ),
+    };
+    records.push(binary_record("yoi", yoi_path, yoi_provider, yoi_build)?);
     let (server_path, server_provider, server_build) = match server_override {
         Some(path) => (path, "YOI_E2E_SERVER_BIN".to_string(), None),
         None => (
@@ -231,6 +240,7 @@ pub struct WorkerE2eEnvironment {
     pub runtime_url: Option<String>,
     server: Option<ManagedProcess>,
     runtime: Option<ManagedProcess>,
+    runtime_public_identity: Option<Value>,
     access_token: Option<String>,
     client: Client,
     binaries: WorkerBinaryProviderInfo,
@@ -292,6 +302,7 @@ impl WorkerE2eEnvironment {
             runtime_url: None,
             server: None,
             runtime: None,
+            runtime_public_identity: None,
             access_token: None,
             client,
             binaries,
@@ -374,6 +385,47 @@ impl WorkerE2eEnvironment {
         Ok(())
     }
 
+    /// Start a store populated by the Backend's typed test-support API. The seed
+    /// supplies fixed catalog IDs, but the bearer must be freshly issued for this
+    /// environment. No database mutation or fixture credential import occurs here.
+    pub fn start_seeded_backend(
+        &mut self,
+        workspace_id: String,
+        runtime_id: String,
+        access_token: String,
+    ) -> Result<()> {
+        if self.server.is_some() || self.runtime.is_some() || self.workspace_id.is_some() {
+            return Err(HarnessError::Protocol(
+                "fixture seed requires an unstarted environment".into(),
+            ));
+        }
+        if workspace_id.is_empty() || runtime_id.is_empty() || access_token.is_empty() {
+            return Err(HarnessError::Protocol(
+                "fixture seed omitted an identity or fresh bearer".into(),
+            ));
+        }
+        if !self.server_data_dir.join("server.db").is_file() {
+            return Err(HarnessError::Protocol(
+                "typed Backend seed has not created server.db".into(),
+            ));
+        }
+        if self.runtime_public_identity.is_some() {
+            if self.runtime_id != runtime_id {
+                return Err(HarnessError::Protocol(
+                    "seeded Runtime ID differs from initialized identity".into(),
+                ));
+            }
+        } else {
+            assert_absent_or_empty(&self.runtime_data_dir, "Runtime data directory")?;
+        }
+        self.workspace_id = Some(workspace_id);
+        self.runtime_id = runtime_id;
+        self.access_token = Some(access_token);
+        self.start_server_process()?;
+        self.export_workspace_public_identity()?;
+        self.write_metadata("seeded-backend-running")
+    }
+
     pub fn start_backend(&mut self) -> Result<()> {
         self.start_empty_backend()?;
         let server_binary = self.binaries.binary("yoi-server")?;
@@ -417,7 +469,15 @@ impl WorkerE2eEnvironment {
             "create Workspace",
         )?;
         let workspace_id = json_string(&created, "/workspace/workspace_id")?;
-        self.workspace_id = Some(workspace_id.clone());
+        self.workspace_id = Some(workspace_id);
+        self.export_workspace_public_identity()?;
+        self.write_metadata("backend-running")?;
+        Ok(())
+    }
+
+    fn export_workspace_public_identity(&self) -> Result<()> {
+        let server_url = self.required_server_url()?;
+        let workspace_id = self.required_workspace_id()?;
         let identity = self.request_json(
             self.client.get(format!(
                 "{server_url}/api/w/{workspace_id}/settings/signing-identity"
@@ -429,17 +489,27 @@ impl WorkerE2eEnvironment {
         })?;
         let bundle_path = self.root.join("workspace-public-identity.json");
         fs::write(&bundle_path, serde_json::to_vec_pretty(&public_bundle)?)?;
-        self.write_metadata("backend-running")?;
         Ok(())
     }
 
     fn start_runtime(&mut self) -> Result<()> {
-        let bundle_path = self.root.join("workspace-public-identity.json");
-        let server_url = self.required_server_url()?.to_string();
-        let workspace_id = self.required_workspace_id()?.to_string();
-        let runtime_binary = self.binaries.binary("yoi-runtime")?;
+        self.start_runtime_with_seed(|_| Ok(()))?;
+        self.assert_connected_and_empty()
+    }
+
+    /// Generate fresh keys before offline Backend registration when fixed IDs are
+    /// required. Returns only the public bundle; no fixture keys are accepted.
+    pub fn initialize_runtime_identity(&mut self, runtime_id: String) -> Result<Value> {
+        if self.runtime_public_identity.is_some() || self.runtime.is_some() {
+            return Err(HarnessError::Protocol(
+                "Runtime identity is already initialized".into(),
+            ));
+        }
+        assert_absent_or_empty(&self.runtime_data_dir, "Runtime data directory")?;
+        self.runtime_id = runtime_id;
+        let binary = self.binaries.binary("yoi-runtime")?;
         self.run_command(
-            &runtime_binary,
+            &binary,
             &[
                 "identity",
                 "init",
@@ -450,8 +520,8 @@ impl WorkerE2eEnvironment {
             ],
             false,
         )?;
-        let runtime_identity = self.run_command(
-            &runtime_binary,
+        let identity = self.run_command(
+            &binary,
             &[
                 "identity",
                 "show",
@@ -461,7 +531,31 @@ impl WorkerE2eEnvironment {
             ],
             false,
         )?;
-        let runtime_identity: Value = serde_json::from_slice(&runtime_identity.stdout)?;
+        let identity: Value = serde_json::from_slice(&identity.stdout)?;
+        self.runtime_public_identity = Some(identity.clone());
+        Ok(identity)
+    }
+
+    /// Initialize fresh Runtime keys/trust, then invoke the typed Runtime seed
+    /// before starting its process. The actual Backend URL is now available.
+    pub fn start_runtime_with_seed(
+        &mut self,
+        seed: impl FnOnce(&Self) -> Result<()>,
+    ) -> Result<()> {
+        if self.runtime.is_some() {
+            return Err(HarnessError::Protocol("Runtime is already running".into()));
+        }
+        if self.runtime_public_identity.is_none() {
+            self.initialize_runtime_identity(self.runtime_id.clone())?;
+        }
+        let runtime_identity = self
+            .runtime_public_identity
+            .clone()
+            .expect("initialized identity");
+        let bundle_path = self.root.join("workspace-public-identity.json");
+        let server_url = self.required_server_url()?.to_string();
+        let workspace_id = self.required_workspace_id()?.to_string();
+        let runtime_binary = self.binaries.binary("yoi-runtime")?;
         self.run_command(
             &runtime_binary,
             &[
@@ -497,6 +591,8 @@ impl WorkerE2eEnvironment {
             ));
         }
 
+        seed(self)?;
+        self.write_metadata("runtime-seeded")?;
         let mut command = self.isolated_command(&runtime_binary);
         command.args([
             "--bind",
@@ -519,22 +615,38 @@ impl WorkerE2eEnvironment {
         self.runtime = Some(runtime);
         self.runtime_url = Some(runtime_url.clone());
 
+        let registered = self.request_json(
+            self.client
+                .get(format!("{server_url}/api/w/{workspace_id}/runtimes")),
+            "observe Runtime registration before endpoint update",
+        )?;
+        let expected_binding_id = registered
+            .get("items")
+            .and_then(Value::as_array)
+            .and_then(|items| {
+                items.iter().find(|item| {
+                    item.get("runtime_id").and_then(Value::as_str) == Some(self.runtime_id.as_str())
+                })
+            })
+            .and_then(|item| item.pointer("/management/binding/binding_id"))
+            .cloned();
         self.request_json(
             self.client
                 .post(format!("{server_url}/api/w/{workspace_id}/runtimes"))
                 .json(&serde_json::json!({
                     "public_bundle": runtime_identity,
                     "display_name": format!("E2E {}", self.runtime_id),
-                    "endpoint": runtime_url
+                    "endpoint": runtime_url,
+                    "expected_binding_id": expected_binding_id,
                 })),
             "register Runtime",
         )?;
-        self.assert_connected_and_empty()?;
+        self.assert_connected()?;
         self.write_metadata("running")?;
         Ok(())
     }
 
-    pub fn assert_connected_and_empty(&self) -> Result<()> {
+    pub fn assert_connected(&self) -> Result<()> {
         let server_url = self.required_server_url()?;
         let workspace_id = self.required_workspace_id()?;
         // connection testは実署名要求。保存済みの接続flagを認証成功の代わりにしない。
@@ -574,11 +686,22 @@ impl WorkerE2eEnvironment {
                 self.runtime_id
             )));
         }
-        let workers = self.request_json(
+        Ok(())
+    }
+
+    pub fn list_workers(&self) -> Result<Value> {
+        let server_url = self.required_server_url()?;
+        let workspace_id = self.required_workspace_id()?;
+        self.request_json(
             self.client
                 .get(format!("{server_url}/api/w/{workspace_id}/workers")),
             "list Workers",
-        )?;
+        )
+    }
+
+    pub fn assert_connected_and_empty(&self) -> Result<()> {
+        self.assert_connected()?;
+        let workers = self.list_workers()?;
         if workers
             .get("items")
             .and_then(Value::as_array)
@@ -591,6 +714,104 @@ impl WorkerE2eEnvironment {
         Ok(())
     }
 
+    fn worker_url(&self, worker_id: &str, suffix: &str) -> Result<String> {
+        let mut url = reqwest::Url::parse(self.required_server_url()?)
+            .map_err(|error| HarnessError::Protocol(format!("invalid Server URL: {error}")))?;
+        url.path_segments_mut()
+            .map_err(|_| HarnessError::Protocol("Server URL has no path segments".into()))?
+            .extend([
+                "api",
+                "w",
+                self.required_workspace_id()?,
+                "runtimes",
+                &self.runtime_id,
+                "workers",
+                worker_id,
+            ]);
+        if !suffix.is_empty() {
+            url.path_segments_mut().unwrap().extend(suffix.split('/'));
+        }
+        Ok(url.into())
+    }
+
+    pub fn worker_detail(&self, worker_id: &str) -> Result<Value> {
+        self.request_json(
+            self.client.get(self.worker_url(worker_id, "")?),
+            "read Worker detail",
+        )
+    }
+
+    /// Exactly one restore submission. No retries, replacement token or new intent.
+    pub fn restore_worker(
+        &self,
+        worker_id: &str,
+        request_id: &str,
+        expected_observation_token: &str,
+    ) -> Result<Value> {
+        self.request_json(
+            self.client
+                .post(self.worker_url(worker_id, "restore")?)
+                .timeout(STARTUP_WAIT)
+                .json(&serde_json::json!({
+                    "request_id": request_id,
+                    "expected_observation_token": expected_observation_token,
+                })),
+            "restore Worker",
+        )
+    }
+
+    pub fn stop_worker(&self, worker_id: &str) -> Result<Value> {
+        self.request_json(
+            self.client
+                .post(self.worker_url(worker_id, "stop")?)
+                .json(&serde_json::json!({})),
+            "stop Worker",
+        )
+    }
+
+    /// Authorization stays in memory; callers must not persist or debug this request.
+    pub fn worker_protocol_request(&self, worker_id: &str) -> Result<reqwest::blocking::Request> {
+        let token = self.access_token.as_deref().ok_or_else(|| {
+            HarnessError::Protocol("Backend access token is not initialized".into())
+        })?;
+        self.client
+            .get(self.worker_url(worker_id, "protocol/ws")?)
+            .bearer_auth(token)
+            .build()
+            .map_err(|_| HarnessError::Protocol("build Worker protocol request".into()))
+    }
+
+    /// The other environment must reject this environment's fresh bearer even
+    /// when both contain identical Workspace/Runtime/Worker IDs. Never logs tokens.
+    pub fn assert_authentication_isolated_from(&self, other: &Self) -> Result<()> {
+        let token = self
+            .access_token
+            .as_deref()
+            .ok_or_else(|| HarnessError::Protocol("missing source bearer".into()))?;
+        if other.access_token.as_deref() == Some(token) {
+            return Err(HarnessError::Protocol(
+                "environments share authentication material".into(),
+            ));
+        }
+        let (server, workspace) = other.backend_target()?;
+        let response = self
+            .client
+            .get(format!("{server}/api/w/{workspace}/workers"))
+            .bearer_auth(token)
+            .send()
+            .map_err(|_| {
+                HarnessError::Protocol("cross-environment authorization request failed".into())
+            })?;
+        let status = response.status().as_u16();
+        other.append_operation("reject other environment bearer", status)?;
+        if status != 401 {
+            return Err(HarnessError::Protocol(format!(
+                "cross-environment bearer was not rejected with 401: {status}"
+            )));
+        }
+        Ok(())
+    }
+
     pub fn install_backend_token(&self, config_home: &Path) -> Result<PathBuf> {
         let server_url = self.required_server_url()?;
         let access_token = self.access_token.as_deref().ok_or_else(|| {
@@ -636,7 +857,7 @@ impl WorkerE2eEnvironment {
                     "assignee": null,
                     "readiness": null,
                     "risk_flags": [],
-                    "workflow_state": "planning",
+                    "workflow_state": "open",
                     "queued_by": null,
                     "queued_at": null,
                     "repository_key": null,

tests/e2e/src/worker_session.rs

@@ -0,0 +1,246 @@
+//! Live public Worker protocol assertions. No Submit/Resume/model request is sent.
+use std::net::TcpStream;
+use std::time::{Duration, Instant};
+
+use protocol::{Event, SessionSnapshot, WorkerState};
+use tungstenite::{Message, WebSocket, client::IntoClientRequest};
+
+use crate::{HarnessError, Result, WorkerE2eEnvironment};
+
+const WS_WAIT: Duration = Duration::from_secs(10);
+
+pub struct WorkerSessionSubscription {
+    socket: WebSocket<TcpStream>,
+}
+
+impl WorkerSessionSubscription {
+    pub fn connect(
+        environment: &WorkerE2eEnvironment,
+        worker_id: &str,
+        expected: &SessionSnapshot,
+    ) -> Result<Self> {
+        let authorized = environment.worker_protocol_request(worker_id)?;
+        let mut url = authorized.url().clone();
+        url.set_scheme("ws")
+            .map_err(|_| error("invalid protocol URL"))?;
+        let mut request = url
+            .as_str()
+            .into_client_request()
+            .map_err(|_| error("build protocol handshake"))?;
+        request.headers_mut().extend(authorized.headers().clone());
+        // Only this environment's loopback endpoint is permitted, never a proxy.
+        let address = url
+            .socket_addrs(|| None)
+            .map_err(|_| error("resolve protocol address"))?
+            .into_iter()
+            .find(|address| address.ip().is_loopback())
+            .ok_or_else(|| error("protocol address is not loopback"))?;
+        let stream = TcpStream::connect_timeout(&address, WS_WAIT)?;
+        stream.set_read_timeout(Some(WS_WAIT))?;
+        stream.set_write_timeout(Some(WS_WAIT))?;
+        let (socket, _) =
+            tungstenite::client(request, stream).map_err(|_| error("protocol handshake failed"))?;
+        let mut subscription = Self { socket };
+        let deadline = Instant::now() + WS_WAIT;
+        loop {
+            match subscription.read(deadline)? {
+                Message::Text(text) => {
+                    let raw: serde_json::Value = serde_json::from_str(&text)?;
+                    validate_full_snapshot(&raw, expected)?;
+                    // Only validated public content is retained, never request headers.
+                    std::fs::write(
+                        environment.artifacts_dir.join("restored-session.json"),
+                        serde_json::to_vec_pretty(&raw)?,
+                    )?;
+                    return Ok(subscription);
+                }
+                Message::Ping(_) => subscription
+                    .socket
+                    .flush()
+                    .map_err(|_| error("flush protocol Pong"))?,
+                Message::Pong(_) => {}
+                _ => return Err(error("protocol ended before its initial full Snapshot")),
+            }
+        }
+    }
+
+    /// Prove an already-open subscription remains usable after the other
+    /// environment is stopped and cleaned up. This does not invoke the model.
+    pub fn assert_alive(&mut self) -> Result<()> {
+        let payload = b"restore-e2e-isolation".to_vec();
+        self.socket
+            .send(Message::Ping(payload.clone().into()))
+            .map_err(|_| error("send protocol Ping"))?;
+        let deadline = Instant::now() + WS_WAIT;
+        loop {
+            match self.read(deadline)? {
+                Message::Pong(bytes) if bytes.as_ref() == payload.as_slice() => return Ok(()),
+                Message::Ping(_) => self
+                    .socket
+                    .flush()
+                    .map_err(|_| error("flush protocol Pong"))?,
+                Message::Pong(_) => {}
+                _ => {
+                    return Err(error(
+                        "idle subscription changed or ended during isolation check",
+                    ));
+                }
+            }
+        }
+    }
+
+    /// A public stop must terminate the connected execution, not merely update
+    /// a retained catalog row. Require the shutdown event or websocket close.
+    pub fn expect_stopped(mut self) -> Result<()> {
+        let deadline = Instant::now() + WS_WAIT;
+        loop {
+            match self.read(deadline)? {
+                Message::Close(_) => return Ok(()),
+                Message::Text(text) => match serde_json::from_str::<Event>(&text)? {
+                    Event::Shutdown => return Ok(()),
+                    Event::WorkerState { .. } => {}
+                    _ => return Err(error("unexpected protocol event after Worker stop")),
+                },
+                Message::Ping(_) => self
+                    .socket
+                    .flush()
+                    .map_err(|_| error("flush protocol Pong"))?,
+                Message::Pong(_) => {}
+                _ => return Err(error("unexpected protocol frame after Worker stop")),
+            }
+        }
+    }
+
+    fn read(&mut self, deadline: Instant) -> Result<Message> {
+        let remaining = deadline
+            .checked_duration_since(Instant::now())
+            .filter(|duration| !duration.is_zero())
+            .ok_or_else(|| error("protocol deadline exceeded"))?;
+        self.socket.get_mut().set_read_timeout(Some(remaining))?;
+        self.socket
+            .read()
+            .map_err(|_| error("protocol read failed or timed out"))
+    }
+}
+
+fn error(message: &str) -> HarnessError {
+    HarnessError::Protocol(message.into())
+}
+
+pub fn validate_full_snapshot(raw: &serde_json::Value, expected: &SessionSnapshot) -> Result<()> {
+    let event: Event = serde_json::from_value(raw.clone())?;
+    let Event::Snapshot {
+        session,
+        state,
+        in_flight,
+        internal_workers,
+        ..
+    } = event
+    else {
+        return Err(error("first protocol event is not a full Snapshot"));
+    };
+    if state.state != WorkerState::Idle || !in_flight.is_empty() || !internal_workers.is_empty() {
+        return Err(error(
+            "restored Worker is not an idle, completed standalone Session",
+        ));
+    }
+    // Typed equality covers entry order/ID/timestamp/provenance/content; raw
+    // equality also refuses unknown fields that could leak private data.
+    if session != *expected
+        || raw.pointer("/data/session") != Some(&serde_json::to_value(expected)?)
+    {
+        return Err(error(
+            "full Session snapshot differs from supplied expected rows",
+        ));
+    }
+    if !session.pending_submissions.submissions.is_empty() {
+        return Err(error("fixture has pending submissions"));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use protocol::{
+        Greeting, InFlightSnapshot, Segment, SessionEntryProvenance, SessionSnapshotEntry,
+        SessionSnapshotEntryData, WorkerStateSnapshot, WorkerStatus,
+    };
+
+    // Protocol-validation samples only, not a substitute saved Worker fixture.
+    fn sample() -> (serde_json::Value, SessionSnapshot) {
+        let session = SessionSnapshot {
+            pending_submissions: Default::default(),
+            entries: vec![SessionSnapshotEntry {
+                entry_id: "E-unit-1".into(),
+                timestamp: 123,
+                provenance: SessionEntryProvenance::HumanInput,
+                derived_from: vec![],
+                use_policy: None,
+                data: SessionSnapshotEntryData::UserInput {
+                    segments: vec![Segment::text("unit input")],
+                },
+            }],
+        };
+        let event = Event::Snapshot {
+            session: session.clone(),
+            state: WorkerStateSnapshot::from(WorkerStatus::Idle),
+            in_flight: InFlightSnapshot::default(),
+            internal_workers: vec![],
+            greeting: Greeting {
+                worker_name: "unit".into(),
+                cwd: ".".into(),
+                provider: "unit".into(),
+                model: "unit".into(),
+                reasoning: None,
+                scope_summary: "none".into(),
+                tools: vec![],
+                context_window: 0,
+                context_tokens: 0,
+                context_usage: None,
+            },
+        };
+        (serde_json::to_value(event).unwrap(), session)
+    }
+
+    #[test]
+    fn full_snapshot_requires_exact_ids_timestamps_contents_and_no_unknown_fields() {
+        let (raw, expected) = sample();
+        validate_full_snapshot(&raw, &expected).unwrap();
+        for (key, value) in [
+            ("entry_id", serde_json::json!("different")),
+            ("timestamp", serde_json::json!(124)),
+            (
+                "hidden_reasoning",
+                serde_json::json!("must never be exposed"),
+            ),
+            ("segments", serde_json::json!([])),
+        ] {
+            let mut changed = raw.clone();
+            changed["data"]["session"]["entries"][0][key] = value;
+            assert!(
+                validate_full_snapshot(&changed, &expected).is_err(),
+                "{key}"
+            );
+        }
+        let mut missing = raw.clone();
+        missing["data"]["session"]["entries"] = serde_json::json!([]);
+        assert!(validate_full_snapshot(&missing, &expected).is_err());
+        let mut extra = raw;
+        let entry = extra["data"]["session"]["entries"][0].clone();
+        extra["data"]["session"]["entries"]
+            .as_array_mut()
+            .unwrap()
+            .push(entry);
+        assert!(validate_full_snapshot(&extra, &expected).is_err());
+    }
+
+    #[test]
+    fn non_snapshot_event_cannot_stand_in_for_live_snapshot() {
+        let (_, expected) = sample();
+        assert!(
+            validate_full_snapshot(&serde_json::to_value(Event::Shutdown).unwrap(), &expected)
+                .is_err()
+        );
+    }
+}

tests/e2e/tests/worker_restore.rs

@@ -0,0 +1,270 @@
+//! Fixed, externally supplied Worker restore. Never generates a fixture or sends model input.
+#[path = "worker_restore/fixture.rs"]
+mod fixture;
+#[path = "worker_restore/seed.rs"]
+mod seed;
+
+use std::fs;
+use std::time::{Duration, Instant};
+
+use fixture::{Fixture, failure};
+use seed::Installed;
+use server_api::{
+    RuntimeWorkerLifecycleResult, RuntimeWorkerShowResponse, WorkerListResponse,
+    WorkerOperationState, WorkerRestoreResponse, WorkerRestoreState, WorkerSummary,
+};
+use yoi_e2e::{FixtureSource, Result, WorkerE2eEnvironment, WorkerSessionSubscription};
+
+#[test]
+fn supplied_stopped_worker_restores_exact_live_session_and_cleans_up() -> Result<()> {
+    let fixture = Fixture::load()?;
+    let result = (|| {
+        let installed = seed::start(&fixture)?;
+        let subscription = restore(&installed, &fixture, "fixed-fixture-restore-single")?;
+        stop_and_cleanup(installed, subscription, &fixture)
+    })();
+    fixture.source.assert_unchanged()?;
+    result
+}
+
+#[test]
+fn concurrent_fixed_id_restores_keep_sessions_storage_and_cleanup_isolated() -> Result<()> {
+    let fixture = Fixture::load()?;
+    let result = (|| {
+        let (first, second) = std::thread::scope(|scope| {
+            let first = scope.spawn(|| seed::start(&fixture));
+            let second = seed::start(&fixture);
+            (
+                first
+                    .join()
+                    .map_err(|_| failure("first environment startup panicked"))
+                    .and_then(|result| result),
+                second,
+            )
+        });
+        let first = first?;
+        let second = second?;
+        assert_eq!(
+            first.environment.workspace_id,
+            second.environment.workspace_id
+        );
+        assert_eq!(first.environment.runtime_id, second.environment.runtime_id);
+        assert_ne!(first.environment.root, second.environment.root);
+        assert_ne!(
+            first.environment.server_data_dir,
+            second.environment.server_data_dir
+        );
+        assert_ne!(
+            first.environment.runtime_data_dir,
+            second.environment.runtime_data_dir
+        );
+        assert_ne!(
+            first.environment.artifacts_dir,
+            second.environment.artifacts_dir
+        );
+        assert_ne!(
+            first.environment.account_handle,
+            second.environment.account_handle
+        );
+        assert_ne!(first.runtime_public_key, second.runtime_public_key);
+        assert_ne!(first.workspace_public_key, second.workspace_public_key);
+        assert_ne!(
+            first.environment.server_port()?,
+            second.environment.server_port()?
+        );
+        assert_ne!(
+            first.environment.runtime_port()?,
+            second.environment.runtime_port()?
+        );
+        assert_ne!(
+            first.environment.server_pid(),
+            second.environment.server_pid()
+        );
+        assert_ne!(
+            first.environment.runtime_pid(),
+            second.environment.runtime_pid()
+        );
+        first
+            .environment
+            .assert_authentication_isolated_from(&second.environment)?;
+        second
+            .environment
+            .assert_authentication_isolated_from(&first.environment)?;
+        let (first_subscription, second_subscription) = std::thread::scope(|scope| {
+            // Deliberately reuse request identity in isolated stores, just like fixed IDs.
+            let a = scope.spawn(|| restore(&first, &fixture, "fixed-fixture-restore-concurrent"));
+            let b = restore(&second, &fixture, "fixed-fixture-restore-concurrent");
+            (
+                a.join()
+                    .map_err(|_| failure("first restore panicked"))
+                    .and_then(|result| result),
+                b,
+            )
+        });
+        let first_subscription = first_subscription?;
+        let mut second_subscription = second_subscription?;
+        let second_session_root = second
+            .environment
+            .runtime_data_dir
+            .join("runtime/workers")
+            .join(&fixture.manifest.worker_id)
+            .join("session");
+        let second_session = FixtureSource::capture(&second_session_root)?;
+        let first_root = first.environment.root.clone();
+        let first_artifacts = first.environment.artifacts_dir.clone();
+        stop_and_cleanup(first, first_subscription, &fixture)?;
+        assert!(!first_root.exists());
+        assert!(first_artifacts.join("cleanup.json").is_file());
+        assert!(second.environment.root.is_dir());
+        assert!(second.environment.server_data_dir.is_dir());
+        assert!(second.environment.runtime_data_dir.is_dir());
+        assert!(
+            second
+                .environment
+                .artifacts_dir
+                .join("restored-session.json")
+                .is_file()
+        );
+        second.environment.assert_connected()?;
+        assert_single_state(&second.environment, &fixture, "idle")?;
+        second_subscription.assert_alive()?;
+        // A second live connection independently verifies content after peer cleanup.
+        let mut reconnected = WorkerSessionSubscription::connect(
+            &second.environment,
+            &fixture.manifest.worker_id,
+            &second.snapshot,
+        )?;
+        reconnected.assert_alive()?;
+        second_session.assert_unchanged()?;
+        stop_and_cleanup(second, second_subscription, &fixture)?;
+        reconnected.expect_stopped()?;
+        Ok(())
+    })();
+    fixture.source.assert_unchanged()?;
+    result
+}
+
+fn restore(
+    installed: &Installed,
+    fixture: &Fixture,
+    request_id: &str,
+) -> Result<WorkerSessionSubscription> {
+    phase(&installed.environment, "observe-stopped")?;
+    let stopped = assert_single_state(&installed.environment, fixture, "stopped")?;
+    let token = stopped
+        .restore_observation_token
+        .as_deref()
+        .filter(|token| !token.is_empty())
+        .ok_or_else(|| failure("stopped Worker did not expose a restore observation token"))?;
+    phase(&installed.environment, "restore")?;
+    let result: WorkerRestoreResponse = serde_json::from_value(
+        installed
+            .environment
+            .restore_worker(&fixture.manifest.worker_id, request_id, token)?,
+    )?;
+    if result.workspace_id != fixture.manifest.workspace_id
+        || result.runtime_id != fixture.manifest.runtime_id
+        || result.worker_id != fixture.manifest.worker_id
+        || result.result.state != WorkerRestoreState::Accepted
+    {
+        return Err(failure(format!(
+            "restore did not commit accepted: {:?}",
+            result.result.state
+        )));
+    }
+    assert_single_state(&installed.environment, fixture, "idle")?;
+    let detail: RuntimeWorkerShowResponse = serde_json::from_value(
+        installed
+            .environment
+            .worker_detail(&fixture.manifest.worker_id)?,
+    )?;
+    if detail.worker.worker_id != fixture.manifest.worker_id
+        || detail.worker.runtime_id != fixture.manifest.runtime_id
+        || detail.worker.state != "idle"
+        || detail
+            .worker
+            .worker_state
+            .as_ref()
+            .is_none_or(|state| state.state != protocol::WorkerState::Idle)
+    {
+        return Err(failure(
+            "restored Worker detail is not the single live idle execution",
+        ));
+    }
+    phase(&installed.environment, "subscribe-live-session")?;
+    let mut subscription = WorkerSessionSubscription::connect(
+        &installed.environment,
+        &fixture.manifest.worker_id,
+        &installed.snapshot,
+    )?;
+    subscription.assert_alive()?;
+    phase(&installed.environment, "live-session-verified")?;
+    Ok(subscription)
+}
+
+fn assert_single_state(
+    environment: &WorkerE2eEnvironment,
+    fixture: &Fixture,
+    state: &str,
+) -> Result<WorkerSummary> {
+    let deadline = Instant::now() + Duration::from_secs(15);
+    loop {
+        let list: WorkerListResponse = serde_json::from_value(environment.list_workers()?)?;
+        if list.workspace_id != fixture.manifest.workspace_id || list.items.len() > 1 {
+            return Err(failure(
+                "Workspace Worker list has wrong identity or duplicate Workers",
+            ));
+        }
+        if let Some(worker) = list.items.into_iter().next() {
+            if worker.worker_id != fixture.manifest.worker_id
+                || worker.runtime_id != fixture.manifest.runtime_id
+            {
+                return Err(failure("Workspace listed an unexpected Worker identity"));
+            }
+            if worker.state == state {
+                return Ok(worker);
+            }
+        }
+        if Instant::now() >= deadline {
+            return Err(failure(format!(
+                "Worker list did not become exactly one {state} Worker"
+            )));
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+}
+
+fn stop_and_cleanup(
+    installed: Installed,
+    subscription: WorkerSessionSubscription,
+    fixture: &Fixture,
+) -> Result<()> {
+    phase(&installed.environment, "stop")?;
+    let result: RuntimeWorkerLifecycleResult = serde_json::from_value(
+        installed
+            .environment
+            .stop_worker(&fixture.manifest.worker_id)?,
+    )?;
+    if result.state != WorkerOperationState::Accepted {
+        return Err(failure("Worker stop was not accepted"));
+    }
+    subscription.expect_stopped()?;
+    assert_single_state(&installed.environment, fixture, "stopped")?;
+    phase(&installed.environment, "shutdown-cleanup")?;
+    let report = installed.environment.cleanup()?;
+    if !report.cleanup_success || report.fixture_root_exists_after {
+        return Err(failure("environment cleanup failed"));
+    }
+    if !report.report_path.is_file() {
+        return Err(failure("cleanup evidence is missing"));
+    }
+    Ok(())
+}
+
+fn phase(environment: &WorkerE2eEnvironment, phase: &str) -> Result<()> {
+    fs::write(
+        environment.artifacts_dir.join("restore-phase.json"),
+        serde_json::to_vec_pretty(&serde_json::json!({"phase":phase}))?,
+    )?;
+    Ok(())
+}

tests/e2e/tests/worker_restore/fixture.rs

@@ -0,0 +1,257 @@
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+use protocol::{SessionMessageRole, SessionSnapshot, SessionSnapshotEntryData};
+use serde::Deserialize;
+use serde_json::{Value, json};
+use worker_runtime::test_support::StoppedWorkerTemplate;
+use yoi_e2e::{FixtureSource, HarnessError, Result};
+
+pub const PROFILE: &str = "project:fixture-chat";
+const ASSETS: [&str; 3] = [
+    "expected.json",
+    "worker-template.json",
+    "workspace-config.dcdl",
+];
+const MAX_ASSET_BYTES: u64 = 8 * 1024 * 1024;
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Manifest {
+    pub format_version: u32,
+    pub source_revision: String,
+    pub runtime_schema_version: u32,
+    pub session_schema_version: u32,
+    pub workspace_id: String,
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub session_id: String,
+    pub segment_id: String,
+    pub model: String,
+    pub state: String,
+    pub acquisition: String,
+    pub source_file_sha256: BTreeMap<String, String>,
+    pub files: BTreeMap<String, Asset>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Asset {
+    pub bytes: u64,
+    pub sha256: String,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Expected {
+    source_revision: String,
+    model: String,
+    worker_id: String,
+    session_id: String,
+    rows: Vec<Value>,
+}
+
+pub struct Fixture {
+    pub root: PathBuf,
+    pub source: FixtureSource,
+    pub manifest: Manifest,
+    expected: Expected,
+}
+
+pub fn failure(message: impl Into<String>) -> HarnessError {
+    HarnessError::Protocol(message.into())
+}
+pub fn support<T, E: std::fmt::Display>(result: std::result::Result<T, E>) -> Result<T> {
+    result.map_err(|error| failure(error.to_string()))
+}
+
+impl Fixture {
+    pub fn load() -> Result<Self> {
+        // Candidate override is explicit and read-only. Missing assets are failures,
+        // never skips, and no test generates or repairs a fixture.
+        let root = std::env::var_os("YOI_E2E_RESTORE_FIXTURE")
+            .map(PathBuf::from)
+            .unwrap_or_else(|| {
+                PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("fixtures/worker-codex-v1")
+            });
+        let manifest: Manifest =
+            serde_json::from_slice(&bounded_read(&root.join("manifest.json"))?)?;
+        if manifest.format_version != 1
+            || manifest.state != "stopped"
+            || manifest.runtime_schema_version == 0
+            || manifest.session_schema_version == 0
+            || manifest.acquisition.is_empty()
+            || manifest.source_file_sha256.is_empty()
+            || manifest.source_revision.len() != 40
+            || !manifest
+                .source_revision
+                .bytes()
+                .all(|c| c.is_ascii_hexdigit())
+        {
+            return Err(failure("unsupported/incomplete fixture manifest"));
+        }
+        let names: BTreeSet<_> = manifest.files.keys().map(String::as_str).collect();
+        if names != BTreeSet::from(ASSETS) {
+            return Err(failure("unexpected fixture asset set"));
+        }
+        for name in ASSETS {
+            bounded_read(&root.join(name))?;
+        }
+        let source = FixtureSource::capture(&root)?;
+        if source.hashes().len() != 4 {
+            return Err(failure("fixture has unmanifested files"));
+        }
+        for (name, asset) in &manifest.files {
+            if asset.bytes != fs::metadata(root.join(name))?.len()
+                || source.hashes().get(Path::new(name)) != Some(&format!("sha256:{}", asset.sha256))
+            {
+                return Err(failure(format!("fixture asset hash/size mismatch: {name}")));
+            }
+        }
+        let expected: Expected =
+            serde_json::from_slice(&bounded_read(&root.join("expected.json"))?)?;
+        if expected.source_revision != manifest.source_revision
+            || expected.model != manifest.model
+            || expected.worker_id != manifest.worker_id
+            || expected.session_id != manifest.session_id
+            || expected.rows.is_empty()
+        {
+            return Err(failure(
+                "expected conversation identity differs from manifest",
+            ));
+        }
+        verify_builtin_revision(&manifest.source_revision)?;
+        let fixture = Self {
+            root,
+            source,
+            manifest,
+            expected,
+        };
+        fixture.template(&fixture.root)?;
+        Ok(fixture)
+    }
+
+    pub fn template(&self, copy: &Path) -> Result<StoppedWorkerTemplate> {
+        let bytes = bounded_read(&copy.join("worker-template.json"))?;
+        let template: StoppedWorkerTemplate = serde_json::from_slice(&bytes)?;
+        let shape: Value = serde_json::from_slice(&bytes)?;
+        let session_manifest_bytes: Vec<u8> =
+            serde_json::from_value(shape["session_manifest"].clone())?;
+        let segment_bytes: Vec<u8> = serde_json::from_value(shape["segment_jsonl"].clone())?;
+        for (name, bytes) in [
+            (
+                "session/session.json".to_owned(),
+                session_manifest_bytes.as_slice(),
+            ),
+            (
+                format!("session/segments/{}.jsonl", self.manifest.segment_id),
+                segment_bytes.as_slice(),
+            ),
+        ] {
+            use sha2::{Digest, Sha256};
+            let digest = Sha256::digest(bytes)
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect::<String>();
+            if self.manifest.source_file_sha256.get(&name) != Some(&digest) {
+                return Err(failure(
+                    "exported Session bytes differ from acquisition hash evidence",
+                ));
+            }
+        }
+        let session_manifest: Value = serde_json::from_slice(&session_manifest_bytes)?;
+        let m = &self.manifest;
+        if template.workspace_id() != m.workspace_id
+            || template.runtime_id() != m.runtime_id
+            || template.worker_id().to_string() != m.worker_id
+            || template.session_id().to_string() != m.session_id
+            || template.segment_id().to_string() != m.segment_id
+            || shape["runtime_schema_version"] != m.runtime_schema_version
+            || session_manifest["schema_version"] != m.session_schema_version
+        {
+            return Err(failure("template schema/identity differs from manifest"));
+        }
+        Ok(template)
+    }
+
+    pub fn verify_conversation(&self, snapshot: &SessionSnapshot) -> Result<()> {
+        let mut rows = Vec::new();
+        for entry in &snapshot.entries {
+            let (role, parts) = match &entry.data {
+                SessionSnapshotEntryData::UserInput { segments } => {
+                    ("user", serde_json::to_value(segments)?)
+                }
+                SessionSnapshotEntryData::Message {
+                    role: SessionMessageRole::Assistant,
+                    content,
+                } => ("assistant", serde_json::to_value(content)?),
+                // Display-only accounting is not a conversation row. Its exact
+                // IDs, timestamp and values are still checked in the full WS snapshot.
+                SessionSnapshotEntryData::RunStats { .. } => continue,
+                _ => {
+                    return Err(failure(
+                        "unexpected non-conversation entry in full Session snapshot",
+                    ));
+                }
+            };
+            rows.push(json!({"entry_id":entry.entry_id,"timestamp":entry.timestamp,"role":role,"parts":parts}));
+        }
+        if rows != self.expected.rows {
+            return Err(failure(
+                "full public conversation rows differ from expected IDs/time/roles/text/order",
+            ));
+        }
+        Ok(())
+    }
+}
+
+fn bounded_read(path: &Path) -> Result<Vec<u8>> {
+    let metadata = fs::symlink_metadata(path).map_err(|error| {
+        failure(format!(
+            "required fixture asset {}: {error}",
+            path.display()
+        ))
+    })?;
+    if !metadata.is_file() || metadata.len() > MAX_ASSET_BYTES {
+        return Err(failure("fixture asset is not a bounded regular file"));
+    }
+    Ok(fs::read(path)?)
+}
+
+fn verify_builtin_revision(revision: &str) -> Result<()> {
+    let workspace = Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap()
+        .parent()
+        .unwrap();
+    // Resolve the immutable commit before diff; bad revisions must not be treated
+    // as an empty diff. All resources are checked, including transitive imports.
+    let commit = Command::new("git")
+        .args(["cat-file", "-t", revision])
+        .current_dir(workspace)
+        .output()?;
+    if !commit.status.success() || commit.stdout != b"commit\n" {
+        return Err(failure(
+            "fixture source revision is unavailable for builtin verification",
+        ));
+    }
+    let status = Command::new("git")
+        .args([
+            "diff",
+            "--quiet",
+            revision,
+            "--",
+            "resources",
+            "crates/builtin-source",
+        ])
+        .current_dir(workspace)
+        .status()?;
+    if !status.success() {
+        return Err(failure(
+            "builtin/config resources differ from fixture source revision; provide a new version or frozen references",
+        ));
+    }
+    Ok(())
+}

tests/e2e/tests/worker_restore/seed.rs

@@ -0,0 +1,366 @@
+use std::fs;
+use std::net::TcpListener;
+use std::path::Path;
+use std::sync::Arc;
+
+use config_source::{ConfigTreeChange, VirtualPath};
+use protocol::SessionSnapshot;
+use serde_json::Value;
+use worker_runtime::FsRuntimeStoreOptions;
+use worker_runtime::catalog::{
+    ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource,
+    WorkspaceApiRef,
+};
+use yoi_e2e::{Result, WorkerE2eEnvironment};
+use yoi_workspace_server::auth::{mint_secret, new_id, now_rfc3339, token_hash};
+use yoi_workspace_server::config_source::{ConfigCommitRequest, WorkspaceConfigSchemaRegistry};
+use yoi_workspace_server::hosts::{
+    WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest,
+};
+use yoi_workspace_server::profile_settings::{
+    ProfileConfigSchemaProvider, build_virtual_profile_config_bundle,
+    project_profiles_from_workspace_config,
+};
+use yoi_workspace_server::store::{
+    AccountRecord, ApiTokenRecord, UserRecord, WorkspaceBootstrapRecord,
+    WorkspaceRuntimeAuthenticationMode, WorkspaceRuntimeBinding,
+};
+use yoi_workspace_server::test_support::{FixedWorkerSeed, seed_fixed_worker};
+use yoi_workspace_server::workspace_signing_identity::{
+    FsWorkspaceSigningMaterialStore, WorkspaceSigningIdentityService,
+    workspace_signing_material_root,
+};
+use yoi_workspace_server::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord};
+
+use super::fixture::{Fixture, PROFILE, failure, support};
+
+pub struct Installed {
+    pub environment: WorkerE2eEnvironment,
+    pub snapshot: SessionSnapshot,
+    pub runtime_public_key: String,
+    pub workspace_public_key: String,
+}
+
+pub fn start(fixture: &Fixture) -> Result<Installed> {
+    let mut environment = WorkerE2eEnvironment::new()?;
+    super::phase(&environment, "fixture-copy-verify")?;
+    let copy = environment.root.join("provided-fixture");
+    fixture.source.copy_to(&copy)?;
+    fs::write(
+        environment.artifacts_dir.join("fixture-manifest.json"),
+        fs::read(copy.join("manifest.json"))?,
+    )?;
+    let template = fixture.template(&copy)?;
+    let runtime_identity =
+        environment.initialize_runtime_identity(fixture.manifest.runtime_id.clone())?;
+    let runtime_public_key = runtime_identity["public_key"]
+        .as_str()
+        .ok_or_else(|| failure("Runtime public bundle omitted key"))?
+        .to_owned();
+
+    // Reserve an environment-local placeholder endpoint only while the Runtime is
+    // offline. Its real :0 endpoint replaces this binding through public API CAS.
+    let reserved_endpoint = TcpListener::bind("127.0.0.1:0")?;
+    super::phase(
+        &environment,
+        "offline-backend-bootstrap-profile-worker-seed",
+    )?;
+    let bootstrap = bootstrap(
+        &environment,
+        fixture,
+        &copy,
+        &runtime_identity,
+        &format!("http://{}", reserved_endpoint.local_addr()?),
+        &template.worker_name(),
+    )?;
+    super::phase(&environment, "start-seeded-backend")?;
+    environment.start_seeded_backend(
+        fixture.manifest.workspace_id.clone(),
+        fixture.manifest.runtime_id.clone(),
+        bootstrap.token,
+    )?;
+    let mut snapshot = None;
+    super::phase(&environment, "initialize-runtime-trust")?;
+    environment.start_runtime_with_seed(|environment| {
+        super::phase(environment, "install-runtime-fixture-and-validate-session")?;
+        let options = FsRuntimeStoreOptions::new(environment.runtime_data_dir.join("runtime"))
+            .with_runtime_id(&fixture.manifest.runtime_id);
+        let request = CreateWorkerRequest {
+            worker_id: template.worker_id(),
+            create_fingerprint: bootstrap.create_fingerprint.clone(),
+            profile: ProfileSelector::Named(PROFILE.into()),
+            display_name: Some("Restore E2E fixture".into()),
+            profile_source: ProfileSourceArchiveSource::WorkspaceConfig {
+                archive: bootstrap.archive.clone(),
+            },
+            config_bundle: Some(bootstrap.config_bundle.clone()),
+            workspace_api: Some(WorkspaceApiRef {
+                workspace_id: fixture.manifest.workspace_id.clone(),
+                base_url: environment.backend_target()?.0.into(),
+            }),
+            initial_input: None,
+            workdir_attachment_requests: vec![],
+            workdir_attachments: vec![],
+            worker_observation_enabled: false,
+            worker_observation_grants: vec![],
+            memory_settings: Some(bootstrap.memory_settings.clone()),
+            subjektiv_attached: false,
+            backend_job: None,
+        };
+        support(template.install(&options, request, &bootstrap.manifest))?;
+        let sessions = support(session_store::WorkerSessionStore::open_read_only(
+            options
+                .root
+                .join("workers")
+                .join(template.worker_id().to_string())
+                .join("session"),
+        ))?;
+        let (logs, _) = support(sessions.read_all_read_only_bounded(
+            template.session_id(),
+            template.segment_id(),
+            8 * 1024 * 1024,
+        ))?;
+        let projected = session_store::project_session_snapshot(template.session_id(), &logs);
+        fixture.verify_conversation(&projected)?;
+        snapshot = Some(projected);
+        Ok(())
+    })?;
+    drop(reserved_endpoint);
+    fixture.source.assert_unchanged()?;
+    Ok(Installed {
+        environment,
+        snapshot: snapshot.ok_or_else(|| failure("Runtime seed was not invoked"))?,
+        runtime_public_key,
+        workspace_public_key: bootstrap.workspace_public_key,
+    })
+}
+
+struct Bootstrap {
+    token: String,
+    workspace_public_key: String,
+    create_fingerprint: String,
+    archive: worker_runtime::profile_archive::ProfileSourceArchiveRef,
+    config_bundle: ConfigBundleRef,
+    manifest: manifest::WorkerManifest,
+    memory_settings: manifest::WorkspaceMemorySettingsSnapshot,
+}
+
+fn bootstrap(
+    environment: &WorkerE2eEnvironment,
+    fixture: &Fixture,
+    copy: &Path,
+    runtime_identity: &Value,
+    endpoint: &str,
+    worker_name: &str,
+) -> Result<Bootstrap> {
+    fs::create_dir_all(&environment.server_data_dir)?;
+    let database = environment.server_data_dir.join("server.db");
+    if database.exists() {
+        return Err(failure("Backend seed refuses an existing database"));
+    }
+    let store = support(SqliteWorkspaceStore::open(&database))?;
+    let now = now_rfc3339();
+    let account_id = new_id("acct-user");
+    let user_id = new_id("user");
+    support(store.upsert_account(&AccountRecord {
+        account_id: account_id.clone(),
+        kind: "user".into(),
+        handle: environment.account_handle.clone(),
+        display_name: "Restore E2E owner".into(),
+        created_at: now.clone(),
+        updated_at: now.clone(),
+    }))?;
+    support(store.upsert_user(&UserRecord {
+        user_id: user_id.clone(),
+        account_id: account_id.clone(),
+        handle: environment.account_handle.clone(),
+        display_name: "Restore E2E owner".into(),
+        created_at: now.clone(),
+        updated_at: now.clone(),
+    }))?;
+    let token = mint_secret("yoi_api");
+    support(store.create_api_token(&ApiTokenRecord {
+        token_id: new_id("api-token"),
+        token_hash: token_hash(&token),
+        user_id,
+        label: "restore-e2e".into(),
+        created_at: now.clone(),
+        expires_at: None,
+        revoked_at: None,
+        last_used_at: None,
+    }))?;
+    let identities = WorkspaceSigningIdentityService::new(
+        Arc::new(store.clone()),
+        Arc::new(FsWorkspaceSigningMaterialStore::new(
+            workspace_signing_material_root(&database),
+        )),
+    );
+    let (identity, operation) = support(identities.prepare_workspace_creation(
+        "restore-e2e-bootstrap",
+        "restore-e2e-bootstrap",
+        &fixture.manifest.workspace_id,
+        &account_id,
+    ))?;
+    let workspace_public_key = identity.public_key.clone();
+    support(store.create_workspace_bootstrap(
+        &WorkspaceBootstrapRecord {
+            operation_key: "restore-e2e-bootstrap".into(),
+            request_fingerprint: "restore-e2e-bootstrap".into(),
+            workspace: WorkspaceRecord {
+                workspace_id: fixture.manifest.workspace_id.clone(),
+                owner_account_id: account_id,
+                display_name: "Restore E2E".into(),
+                state: "active".into(),
+                created_at: now.clone(),
+                updated_at: now.clone(),
+            },
+            repository: None,
+        },
+        &identity,
+        &operation,
+    ))?;
+    support(
+        store.upsert_workspace_runtime_binding(
+            WorkspaceRuntimeBinding {
+                workspace_id: fixture.manifest.workspace_id.clone(),
+                runtime_id: fixture.manifest.runtime_id.clone(),
+                display_name: format!("E2E {}", fixture.manifest.runtime_id),
+                base_url: endpoint.into(),
+                public_key: runtime_identity["public_key"]
+                    .as_str()
+                    .ok_or_else(|| failure("missing Runtime public key"))?
+                    .into(),
+                public_key_fingerprint: String::new(),
+                binding_id: String::new(),
+                authentication_mode: WorkspaceRuntimeAuthenticationMode::WorkspaceIdentity,
+                created_at: now.clone(),
+                updated_at: now.clone(),
+                revoked_at: None,
+            },
+            false,
+        ),
+    )?;
+
+    let schema = support(
+        WorkspaceConfigSchemaRegistry::default()
+            .with_provider(Arc::new(ProfileConfigSchemaProvider))
+            .with_provider(Arc::new(
+                yoi_workspace_server::prompt_settings::PromptConfigSchemaProvider,
+            ))
+            .with_provider(Arc::new(
+                yoi_workspace_server::runtime_settings::RuntimeConfigSchemaProvider,
+            ))
+            .with_provider(Arc::new(
+                yoi_workspace_server::repository_access::RepositoryAccessConfigSchemaProvider,
+            ))
+            .with_provider(Arc::new(
+                yoi_workspace_server::skills::SkillConfigSchemaProvider,
+            ))
+            .compose(),
+    )?;
+    let state = support(store.ensure_workspace_config_materialized_with_schema(
+        &fixture.manifest.workspace_id,
+        &now,
+        schema.clone(),
+    ))?;
+    let main = support(VirtualPath::parse("main.dcdl"))?;
+    let current = state
+        .snapshot
+        .get(&main)
+        .ok_or_else(|| failure("Workspace config omitted main.dcdl"))?;
+    let candidate = support(store.evaluate_workspace_config_candidate_with_schema(
+        &fixture.manifest.workspace_id,
+        &ConfigCommitRequest {
+            base_digest: state.snapshot.digest.clone(),
+            changes: vec![ConfigTreeChange::Update {
+                path: main.clone(),
+                expected_digest: current.content_digest.clone(),
+                content: fs::read_to_string(copy.join("workspace-config.dcdl"))?,
+            }],
+            entrypoints: vec![main],
+        },
+        schema,
+    ))?;
+    let state = support(
+        store.commit_evaluated_workspace_config(&fixture.manifest.workspace_id, &candidate),
+    )?;
+    let projection = support(project_profiles_from_workspace_config(
+        &fixture.manifest.workspace_id,
+        &state,
+    ))?;
+    let bundle = support(build_virtual_profile_config_bundle(
+        &projection,
+        &state,
+        &fixture.manifest.workspace_id,
+        &now,
+        PROFILE,
+    ))?
+    .ok_or_else(|| failure("fixture Profile is not registered"))?;
+    let archive = bundle
+        .profile_source_archive
+        .as_ref()
+        .ok_or_else(|| failure("fixture Profile archive is absent"))?;
+    let mut resolved = support(
+        support(archive.verify())?.resolve_profile(
+            PROFILE,
+            &environment
+                .runtime_data_dir
+                .join("runtime/workers")
+                .join(&fixture.manifest.worker_id),
+            worker_name,
+        ),
+    )?;
+    if resolved.model.ref_.as_deref() != Some(fixture.manifest.model.as_str()) {
+        return Err(failure("resolved Profile changed the supplied model"));
+    }
+    let request = WorkerSpawnRequest {
+        intent: WorkerSpawnIntent::WorkspaceCoding,
+        requested_worker_name: Some("Restore E2E fixture".into()),
+        singleton_key: None,
+        acceptance: WorkerSpawnAcceptanceRequirement::SocketReady,
+        profile: ProfileSelector::Named(PROFILE.into()),
+        ticket_assignment: None,
+        initial_submit: vec![],
+        workdir_attachment_requests: vec![],
+        resolved_workdir_attachment_requests: vec![],
+        resolved_workdir_attachments: vec![],
+        resolved_config_bundle: None,
+        resolved_workspace_api: None,
+        resolved_memory_settings: None,
+        resolved_subjektiv_attached: false,
+        resolved_worker_observation_enabled: false,
+        resolved_worker_observation_grants: vec![],
+        resolved_control_operation: None,
+        resolved_execution_grant: None,
+    };
+    let reservation = support(seed_fixed_worker(
+        &store,
+        FixedWorkerSeed {
+            workspace_id: &fixture.manifest.workspace_id,
+            runtime_id: &fixture.manifest.runtime_id,
+            worker_id: support(fixture.manifest.worker_id.parse())?,
+            allocation_key: "restore-e2e-worker",
+            request: &request,
+        },
+    ))?;
+    if resolved.feature.memory.profile.enabled {
+        support(
+            resolved
+                .feature
+                .memory
+                .bind_workspace_settings(reservation.memory_settings.clone()),
+        )?;
+    }
+    Ok(Bootstrap {
+        manifest: resolved,
+        memory_settings: reservation.memory_settings,
+        token,
+        workspace_public_key,
+        create_fingerprint: reservation.create_fingerprint,
+        archive: archive.reference.clone(),
+        config_bundle: ConfigBundleRef {
+            id: bundle.metadata.id,
+            digest: bundle.metadata.digest,
+        },
+    })
+}