diff --git a/Cargo.lock b/Cargo.lock index f2901cb1f..c88fc1d23 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/tests/e2e/Cargo.toml b/tests/e2e/Cargo.toml index b308eeb49..53b60d5e3 100644 --- a/tests/e2e/Cargo.toml +++ b/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"] diff --git a/tests/e2e/README.md b/tests/e2e/README.md index 678300006..f2a9414ae 100644 --- a/tests/e2e/README.md +++ b/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 +``` diff --git a/tests/e2e/src/fixture_source.rs b/tests/e2e/src/fixture_source.rs new file mode 100644 index 000000000..6b2e80e2b --- /dev/null +++ b/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, +} + +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 { + 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 { + &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::() + ) +} + +fn inventory(root: &Path) -> Result> { + fn visit(root: &Path, path: &Path, files: &mut BTreeMap) -> 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(()) + } +} diff --git a/tests/e2e/src/lib.rs b/tests/e2e/src/lib.rs index 176c7c2ff..989308dd1 100644 --- a/tests/e2e/src/lib.rs +++ b/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 { "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 { }); } - 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 { .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 { + dedicated_product_target("e2e-products") +} + +// Existing PTY tests need their observer feature, never on restore binaries. +fn tui_product_target_dir() -> Result { + dedicated_product_target("e2e-tui-products") +} + +fn dedicated_product_target(name: &str) -> Result { + 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 { let mut path = std::env::current_exe()?; while let Some(name) = path.file_name().and_then(|name| name.to_str()) { diff --git a/tests/e2e/src/worker_environment.rs b/tests/e2e/src/worker_environment.rs index 2183560cd..679ce38e9 100644 --- a/tests/e2e/src/worker_environment.rs +++ b/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 { fn resolve_worker_binaries() -> Result { 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 { "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 { 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, server: Option, runtime: Option, + runtime_public_identity: Option, access_token: Option, 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { 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, diff --git a/tests/e2e/src/worker_session.rs b/tests/e2e/src/worker_session.rs new file mode 100644 index 000000000..a00e8c814 --- /dev/null +++ b/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, +} + +impl WorkerSessionSubscription { + pub fn connect( + environment: &WorkerE2eEnvironment, + worker_id: &str, + expected: &SessionSnapshot, + ) -> Result { + 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::(&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 { + 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() + ); + } +} diff --git a/tests/e2e/tests/worker_restore.rs b/tests/e2e/tests/worker_restore.rs new file mode 100644 index 000000000..d79ce570c --- /dev/null +++ b/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 { + 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 { + 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(()) +} diff --git a/tests/e2e/tests/worker_restore/fixture.rs b/tests/e2e/tests/worker_restore/fixture.rs new file mode 100644 index 000000000..f4d5dadc7 --- /dev/null +++ b/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, + pub files: BTreeMap, +} + +#[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, +} + +pub struct Fixture { + pub root: PathBuf, + pub source: FixtureSource, + pub manifest: Manifest, + expected: Expected, +} + +pub fn failure(message: impl Into) -> HarnessError { + HarnessError::Protocol(message.into()) +} +pub fn support(result: std::result::Result) -> Result { + result.map_err(|error| failure(error.to_string())) +} + +impl Fixture { + pub fn load() -> Result { + // 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 { + let bytes = bounded_read(©.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 = + serde_json::from_value(shape["session_manifest"].clone())?; + let segment_bytes: Vec = 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::(); + 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> { + 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(()) +} diff --git a/tests/e2e/tests/worker_restore/seed.rs b/tests/e2e/tests/worker_restore/seed.rs new file mode 100644 index 000000000..8df363352 --- /dev/null +++ b/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 { + let mut environment = WorkerE2eEnvironment::new()?; + super::phase(&environment, "fixture-copy-verify")?; + let copy = environment.root.join("provided-fixture"); + fixture.source.copy_to(©)?; + fs::write( + environment.artifacts_dir.join("fixture-manifest.json"), + fs::read(copy.join("manifest.json"))?, + )?; + let template = fixture.template(©)?; + 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, + ©, + &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 { + 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, + }, + }) +}