diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 0eba6a589..32f6ea537 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -19,6 +19,8 @@ required-features = ["ws-server", "fs-store"] [features] default = ["ws-server", "fs-store"] fs-store = ["dep:fs4"] +# Offline fixture tooling only; never enabled by the product defaults. +test-support = ["fs-store", "ws-server"] http-server = ["dep:axum", "dep:tower", "dep:reqwest"] ws-server = ["http-server", "axum/ws", "dep:futures", "tokio/sync"] diff --git a/crates/worker-runtime/README.md b/crates/worker-runtime/README.md index 4984c2e18..5df99279d 100644 --- a/crates/worker-runtime/README.md +++ b/crates/worker-runtime/README.md @@ -8,3 +8,30 @@ Manages Worker creation, stopping, and Restore, execution connections within the - [`execution`] — execution backends and operation-specific results. - [`working_directory`] — repository materialization, session resources, and safe deletion. - [`workspace_issuer`] — Workspace signature verification and replay prevention. + +## Offline stopped-Worker fixtures + +The nondefault `test-support` feature exposes `test_support::StoppedWorkerTemplate`. +It is intentionally limited to current-schema, stopped, attachment-free artificial +conversations with one Segment, text history, and no pending execution/activation. +It does not add a product endpoint or change create/restore authorization. + +1. With the source Runtime offline, call `StoppedWorkerTemplate::extract(worker_dir)`. + The source must have its existing Runtime owner lock and identity. Extraction is + read-only; the returned serde template excludes source auth, saved Manifest, + request/config references, run directories, and Runtime operation history. +2. Initialize **new** Runtime identity and Workspace trust in a fresh destination + using normal setup. Preserve the template's Runtime and Workspace IDs. +3. Call `template.install(&options, request, &resolved_manifest)` before starting + the Runtime. Supply a fresh typed `CreateWorkerRequest` (including config/profile + references and trusted memory settings) and `manifest::WorkerManifest` resolved + for that environment. Its Worker name must equal `template.worker_name()`. +4. Start normally and use the ordinary explicit restore operation. Installation + itself creates no execution and leaves the Worker stopped with a fresh guard. + +The template preserves the original Session manifest/JSONL bytes, including stable +entry IDs, timestamps, and settled submission receipts. It rejects resumable queues +and Feature invocation history. Portability/secret-marker checks do not replace +review of the supplied artificial conversation or its independently verified hashes. +The installer refuses any destination content other than initialized `auth.toml`; +interrupted publication is fail-closed, not an overwrite/recovery facility. diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index de799b323..a4a7d679c 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -39,6 +39,10 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; +#[cfg(feature = "test-support")] +#[path = "test_support.rs"] +pub mod test_support; + const SCHEMA_VERSION: u32 = 12; // Frozen source schemas: do not move their meaning when adding a migration. const SOURCE_SCHEMA_VERSION: u32 = 11; diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index 1e160a6f5..251416449 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -34,6 +34,8 @@ pub mod workspace_request; #[cfg(feature = "fs-store")] pub use fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; +#[cfg(feature = "test-support")] +pub use fs_store::test_support; pub use management::RuntimeOptions; pub use runtime::{ Runtime, RuntimeEventSelectorSubscription, RuntimeSubscriptionRecvError, diff --git a/crates/worker-runtime/src/test_support.rs b/crates/worker-runtime/src/test_support.rs new file mode 100644 index 000000000..8edfcf441 --- /dev/null +++ b/crates/worker-runtime/src/test_support.rs @@ -0,0 +1,770 @@ +//! Offline support for the approved, attachment-free stopped Worker fixture. +//! +//! Enabled only by `test-support`. This is not an import API: extraction accepts a +//! current-schema, settled, single-Segment aggregate; installation accepts only a +//! fresh Runtime directory containing its newly initialized `auth.toml`. No +//! Runtime or execution backend is constructed. Original Session bytes are never +//! rewritten. The caller must separately review/attest the artificial conversation; +//! structural checks and secret-marker checks cannot prove arbitrary prose safe. + +use super::*; +use crate::catalog::ProfileSourceArchiveSource; +use crate::workspace_issuer::{ + WorkspaceIssuerTrustRecord, WorkspaceIssuerTrustState, validate_workspace_issuer_trust_records, +}; +use session_store::{ + SegmentId, SessionId, WorkerActiveSegmentRef, WorkerMetadata, WorkerSessionStore, +}; + +#[cfg(test)] +#[path = "test_support_tests.rs"] +mod tests; + +const TEMPLATE_VERSION: u32 = 1; +const MAX_SESSION_BYTES: usize = 8 * 1024 * 1024; + +/// Portable fixture, intentionally without Debug (Session material is private). +/// Fields are private and every deserialized template is validated on install. +/// Serialize this value to retain the fixture; do not serialize its source aggregate. +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StoppedWorkerTemplate { + template_version: u32, + runtime_schema_version: u32, + runtime_id: String, + workspace_id: String, + worker_id: WorkerId, + created_at_ms: Option, + session_id: SessionId, + segment_id: SegmentId, + session_manifest: Vec, + segment_jsonl: Vec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct SessionManifest { + schema_version: u32, + session_id: SessionId, +} + +// Read-only projection: never retain source key material in a template. +#[derive(Deserialize)] +struct AuthIdentity { + identity_id: String, +} + +#[derive(Deserialize)] +struct SourceAuth { + identity: AuthIdentity, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct InitializedAuth { + identity: crate::auth::RuntimeIdentityMaterial, + workspace_issuers: Vec, + #[serde(default)] + archived_legacy_issuers: Vec, +} + +#[derive(Deserialize)] +struct GuardState { + active_request_id: Option, + owners: BTreeMap, +} + +#[derive(Deserialize)] +struct GuardOwner { + receipt: Option, +} + +fn reject(message: &'static str) -> RuntimeError { + RuntimeError::InvalidRequest(format!("stopped Worker fixture: {message}")) +} + +fn io(result: std::io::Result) -> Result { + result.map_err(|_| reject("filesystem operation failed")) +} + +fn decode(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(|_| reject("invalid typed document")) +} + +fn no_symlink_components(path: &Path) -> Result<(), RuntimeError> { + let mut current = PathBuf::new(); + for part in path.components() { + if matches!(part, std::path::Component::ParentDir) { + return Err(reject("parent traversal is unsupported")); + } + current.push(part); + let metadata = io(fs::symlink_metadata(¤t))?; + if metadata.file_type().is_symlink() { + return Err(reject("symlinks are unsupported")); + } + } + Ok(()) +} + +fn check_tree(path: &Path, budget: &mut usize) -> Result<(), RuntimeError> { + if *budget == 0 { + return Err(reject("file count exceeds fixture limit")); + } + *budget -= 1; + let metadata = io(fs::symlink_metadata(path))?; + if metadata.is_dir() { + for entry in io(fs::read_dir(path))? { + check_tree(&io(entry)?.path(), budget)?; + } + } else if !metadata.is_file() { + return Err(reject("symlinks and special files are unsupported")); + } + Ok(()) +} + +fn regular_file(path: &Path) -> Result<(), RuntimeError> { + no_symlink_components(path)?; + if !io(fs::symlink_metadata(path))?.is_file() { + return Err(reject("expected a regular file")); + } + Ok(()) +} + +fn bytes(path: &Path) -> Result, RuntimeError> { + regular_file(path)?; + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(target_os = "linux")] + { + use std::os::unix::fs::OpenOptionsExt; + options.custom_flags(libc::O_NOATIME | libc::O_NOFOLLOW); + } + let file = io(options.open(path))?; + let mut result = Vec::new(); + io(file + .take(MAX_SESSION_BYTES as u64 + 1) + .read_to_end(&mut result))?; + if result.len() > MAX_SESSION_BYTES { + return Err(reject("file exceeds fixture limit")); + } + Ok(result) +} + +fn text(path: &Path) -> Result { + String::from_utf8(bytes(path)?).map_err(|_| reject("invalid UTF-8 document")) +} + +fn portable_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= 256 + && id + .bytes() + .all(|c| c.is_ascii_alphanumeric() || b"-_".contains(&c)) +} + +impl StoppedWorkerTemplate { + pub fn runtime_id(&self) -> &str { + &self.runtime_id + } + pub fn workspace_id(&self) -> &str { + &self.workspace_id + } + pub fn worker_id(&self) -> WorkerId { + self.worker_id + } + pub fn worker_name(&self) -> String { + runtime_worker_name(self.worker_id) + } + pub fn session_id(&self) -> SessionId { + self.session_id + } + pub fn segment_id(&self) -> SegmentId { + self.segment_id + } + + /// Extract without migration, rewriting metadata, or copying credentials, + /// resolved manifests, execution-operation receipts, or run directories. + /// The source Runtime must be offline; its existing owner lock is held for + /// the entire read. The path must be `/workers/`. + pub fn extract(worker_dir: &Path) -> Result { + no_symlink_components(worker_dir)?; + let workers = worker_dir + .parent() + .ok_or_else(|| reject("invalid aggregate path"))?; + if workers.file_name().and_then(|s| s.to_str()) != Some(WORKERS_DIR) { + return Err(reject("invalid aggregate path")); + } + let root = workers + .parent() + .ok_or_else(|| reject("invalid Runtime path"))?; + // Require the existing lock, rather than creating a file in the source. + let lock_path = root.with_file_name(format!( + ".{}.runtime-owner.lock", + root.file_name() + .and_then(|s| s.to_str()) + .ok_or_else(|| reject("invalid Runtime path"))? + )); + regular_file(&lock_path)?; + let lock = io(OpenOptions::new().read(true).open(lock_path))?; + if !io(FileExt::try_lock_shared(&lock))? { + return Err(reject("source Runtime is not offline")); + } + let _lock = RuntimeStoreOwnerLock { file: lock }; + check_tree(worker_dir, &mut 256)?; + let snapshot: RuntimeSnapshot = decode(&bytes(&root.join(RUNTIME_FILE))?)?; + snapshot + .validate(&root.join(RUNTIME_FILE)) + .map_err(|_| reject("unsupported Runtime schema"))?; + if !snapshot.subject_create_recovery_fences.is_empty() + || !snapshot.subject_releases.is_empty() + { + return Err(reject("source has lifecycle operations")); + } + let record: WorkerAggregateRecord = decode(&bytes(&worker_dir.join(WORKER_FILE))?)?; + let record = record + .validate(&worker_dir.join(WORKER_FILE)) + .map_err(|_| reject("invalid Worker schema or identity"))?; + let guard: GuardState = decode( + &serde_json::to_vec(&record.restore_guard) + .map_err(|_| reject("invalid restore guard"))?, + )?; + if record.status != WorkerStatus::Stopped + || guard.active_request_id.is_some() + || guard.owners.values().any(|owner| owner.receipt.is_none()) + { + return Err(reject("Worker is not settled and stopped")); + } + let PersistedWorkerExecutionState::Available(execution) = &record.execution_state else { + return Err(reject("pending or unavailable execution metadata")); + }; + // A stopped aggregate may retain the empty historical binding marker. + // It is deliberately not copied to the installed Worker. + if execution.restore_intent != WorkerRestoreIntent::Explicit { + return Err(reject("Worker requires explicit restore")); + } + if !record.workdir_attachments.is_empty() || !record.logical_workdir_attachments.is_empty() + { + return Err(reject("attachments are unsupported")); + } + check_request(&execution.request, false)?; + let workspace_id = record + .workspace_id + .ok_or_else(|| reject("missing Workspace identity"))?; + let metadata: WorkerMetadata = decode(&bytes(&worker_dir.join(WORKER_METADATA_FILE))?)?; + crate::worker_backend::validate_fixture_restore_metadata(&metadata, &execution.request) + .map_err(|_| reject("invalid source restore Manifest or settings"))?; + if metadata.worker_name != runtime_worker_name(record.worker_id) + || metadata.workspace_id.as_deref() != Some(&workspace_id) + || !metadata.spawned_children.is_empty() + || !metadata.reclaimed_children.is_empty() + || !metadata.peers.is_empty() + || metadata.subjektiv_session_attribution.is_some() + { + return Err(reject("unsupported or inconsistent Worker metadata")); + } + let active = metadata + .active + .ok_or_else(|| reject("missing active Session"))?; + let segment_id = active + .segment_id + .ok_or_else(|| reject("unmaterialized Session"))?; + let source_auth: SourceAuth = toml::from_str(&text(&root.join("auth.toml"))?) + .map_err(|_| reject("missing Runtime identity"))?; + let session = worker_dir.join("session"); + let fixture = Self { + template_version: TEMPLATE_VERSION, + runtime_schema_version: SCHEMA_VERSION, + runtime_id: source_auth.identity.identity_id, + workspace_id, + worker_id: record.worker_id, + created_at_ms: record.created_at_ms, + session_id: active.session_id, + segment_id, + session_manifest: bytes(&session.join("session.json"))?, + segment_jsonl: bytes(&session.join("segments").join(format!("{segment_id}.jsonl")))?, + }; + fixture.validate_shape()?; + fixture.validate_session(&session)?; + Ok(fixture) + } + + /// Install after normal identity/trust initialization, before Runtime start. + /// `request` and `manifest` supply NEW backend/config/profile references and + /// freshly resolved policy, not a launch. The saved source Manifest is never + /// reused. `manifest.worker.name` must equal `worker_name()`. + /// Only `auth.toml` may already exist. The owner lock fences live Runtimes; + /// staged files are synced and published without overwriting any Worker. + /// An interrupted publication fails closed as an incomplete Runtime layout. + pub fn install( + &self, + options: &FsRuntimeStoreOptions, + request: CreateWorkerRequest, + manifest: &manifest::WorkerManifest, + ) -> Result<(), RuntimeError> { + self.validate_shape()?; + check_request(&request, true)?; + if manifest.worker.name != self.worker_name() { + return Err(reject("resolved Manifest Worker name mismatch")); + } + let mut metadata = WorkerMetadata::new( + self.worker_name(), + Some(WorkerActiveSegmentRef { + session_id: self.session_id, + segment_id: Some(self.segment_id), + }), + ) + .with_workspace_id(&self.workspace_id); + metadata.resolved_manifest_snapshot = Some( + manifest::write_persisted_worker_manifest_snapshot(manifest) + .map_err(|_| reject("invalid resolved Manifest"))?, + ); + crate::worker_backend::validate_fixture_restore_metadata(&metadata, &request) + .map_err(|_| reject("resolved Manifest does not match fresh request settings"))?; + if options.runtime_id != self.runtime_id + || request.worker_id != self.worker_id + || request + .workspace_api + .as_ref() + .map(|api| api.workspace_id.as_str()) + != Some(&self.workspace_id) + { + return Err(reject("installation identity mismatch")); + } + no_symlink_components(&options.root)?; + check_tree(&options.root, &mut 256)?; + // Validate the lock location before the ordinary owner-lock opener. + let lock_path = options.root.with_file_name(format!( + ".{}.runtime-owner.lock", + options + .root + .file_name() + .and_then(|s| s.to_str()) + .ok_or_else(|| reject("invalid Runtime path"))? + )); + if fs::symlink_metadata(&lock_path).is_ok() { + regular_file(&lock_path)?; + } + let lock = acquire_runtime_store_owner_lock(&options.root)?; + validate_fresh_runtime_store_layout(&options.root) + .map_err(|_| reject("destination is not fresh; refusing overwrite"))?; + let auth: InitializedAuth = toml::from_str(&text(&options.root.join("auth.toml"))?) + .map_err(|_| reject("initialize Runtime identity and trust before install"))?; + if auth.identity.identity_id != self.runtime_id + || !auth.archived_legacy_issuers.is_empty() + || auth.identity.signing_key().is_err() + { + return Err(reject("invalid destination Runtime identity")); + } + validate_workspace_issuer_trust_records(&auth.workspace_issuers) + .map_err(|_| reject("invalid destination Workspace trust"))?; + let api = request + .workspace_api + .as_ref() + .expect("validated Workspace binding"); + if !auth.workspace_issuers.iter().any(|trust| { + trust.workspace_id == self.workspace_id + && trust.backend_url == api.base_url + && trust.state == WorkspaceIssuerTrustState::Active + }) { + return Err(reject("request does not match initialized Workspace trust")); + } + let stage = options + .root + .join(format!(".fixture-{}", uuid::Uuid::now_v7())); + io(fs::create_dir(&stage))?; + let store = FsRuntimeStore { + root: stage.clone(), + _owner_lock: Some(lock), + worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), + }; + let result = (|| { + let worker = PersistedWorkerRecord { + restore_guard: crate::runtime::RestoreGuard::default(), + worker_ref: WorkerRef::new(self.worker_id), + worker_id: self.worker_id, + profile: request.profile.clone(), + display_name: request.display_name.clone(), + profile_source: request.profile_source.reference(), + config_bundle: request.config_bundle.clone(), + created_at_ms: self.created_at_ms, + status: WorkerStatus::Stopped, + last_finished_submission_request_id: None, + workspace_id: Some(self.workspace_id.clone()), + workdir_attachments: Vec::new(), + logical_workdir_attachments: Vec::new(), + execution_state: PersistedWorkerExecutionState::Available( + PersistedWorkerExecution { + request, + binding: None, + restore_intent: WorkerRestoreIntent::Explicit, + }, + ), + }; + WorkerAggregateRecord::from_persisted(&worker) + .validate(&store.worker_dir(&self.worker_id).join(WORKER_FILE)) + .map_err(|_| reject("invalid installation request"))?; + store.write_worker_record(&worker)?; + let worker_dir = store.worker_dir(&self.worker_id); + atomic_write_json( + &worker_dir.join(WORKER_METADATA_FILE), + &metadata, + "write fixture metadata", + )?; + let session = worker_dir.join("session"); + io(fs::create_dir_all(session.join("segments")))?; + write_new(&session.join("session.json"), &self.session_manifest)?; + write_new( + &session + .join("segments") + .join(format!("{}.jsonl", self.segment_id)), + &self.segment_jsonl, + )?; + self.validate_session(&session)?; + store.write_runtime_snapshot(&PersistedRuntimeState { + subject_create_recovery_fences: BTreeMap::new(), + subject_releases: Vec::new(), + // Match normal store initialization: this is catalog availability, + // not a process or Worker execution start. + display_name: options.display_name.clone(), + status: RuntimeStatus::Running, + next_diagnostic_id: 1, + workers: BTreeMap::new(), + workspace_owners: BTreeMap::new(), + diagnostics: Vec::new(), + })?; + let loaded = store.load_runtime_state()?; + if loaded.workers.len() != 1 || !loaded.diagnostics.is_empty() { + return Err(reject("staged aggregate failed ordinary store roundtrip")); + } + sync_runtime_tree(&stage)?; + // No destination existed when checked under the owner lock. Hard-link + // the snapshot last with create-new semantics; never replace a file. + // Reserve the destination catalog with create-new semantics. Even + // an empty preexisting workers directory must never be replaced. + let destination_workers = options.root.join(WORKERS_DIR); + io(fs::create_dir(&destination_workers))?; + io(fs::rename( + store.worker_dir(&self.worker_id), + destination_workers.join(self.worker_id.to_string()), + ))?; + sync_directory(&destination_workers, "publish stopped Worker aggregate")?; + io(fs::hard_link( + stage.join(RUNTIME_FILE), + options.root.join(RUNTIME_FILE), + ))?; + sync_directory(&options.root, "publish stopped fixture")?; + Ok(()) + })(); + // Only remove our private staging directory, never existing destination data. + let cleanup = fs::remove_dir_all(&stage); + result?; + io(cleanup)?; + sync_directory(&options.root, "finish stopped fixture") + } + + fn validate_shape(&self) -> Result<(), RuntimeError> { + if self.template_version != TEMPLATE_VERSION + || self.runtime_schema_version != SCHEMA_VERSION + { + return Err(reject("unsupported template schema")); + } + if !portable_id(&self.runtime_id) + || !portable_id(&self.workspace_id) + || self.session_id.get_version_num() != 7 + || self.segment_id.get_version_num() != 7 + || self.session_manifest.len() > MAX_SESSION_BYTES + || self.segment_jsonl.len() > MAX_SESSION_BYTES + { + return Err(reject("invalid portable identity or size")); + } + let manifest: SessionManifest = decode(&self.session_manifest)?; + if manifest.session_id != self.session_id + || manifest.schema_version == 0 + || !self.segment_jsonl.ends_with(b"\n") + { + return Err(reject("Session identity or committed boundary mismatch")); + } + // Read-only JSON traversal, never mutation/reconstruction of Session bytes. + // Reject known secret/host markers even in otherwise valid typed records. + for line in self + .segment_jsonl + .split(|b| *b == b'\n') + .filter(|line| !line.is_empty()) + { + let _: PortableJson = decode(line)?; + } + Ok(()) + } + + fn validate_history_entry( + &self, + entry: &session_store::LoggedHistoryEntry, + ids: &mut std::collections::BTreeSet, + ) -> Result<(), RuntimeError> { + if !portable_id(&entry.metadata.entry_id.0) + || !ids.insert(entry.metadata.entry_id.0.clone()) + { + return Err(reject("duplicate or invalid entry identity")); + } + if !matches!( + entry.item, + session_store::LoggedItem::Message { .. } + | session_store::LoggedItem::AssistantResponseBoundary { .. } + ) { + return Err(reject("only text conversation records are supported")); + } + if let session_store::LoggedSessionHistoryOrigin::ModelOutput { worker } = + &entry.metadata.origin + { + // Session producer IDs identify the original execution, not the + // Workspace catalog Worker. Preserve them; do not rebind/rewrite them. + if !portable_id(&worker.worker_id) + || worker + .workspace_id + .as_ref() + .is_some_and(|id| id != &self.workspace_id) + || worker + .runtime_id + .as_ref() + .is_some_and(|id| id != &self.runtime_id) + { + return Err(reject("Session origin identity mismatch")); + } + } + Ok(()) + } + + fn validate_session(&self, root: &Path) -> Result<(), RuntimeError> { + for entry in io(fs::read_dir(root))? { + let entry = io(entry)?; + match entry.file_name().to_str() { + Some("session.json" | "segments") => {} + Some("artifacts") => check_empty_artifacts(&entry.path())?, + _ => return Err(reject("unsupported Session files")), + } + } + let entries = io(fs::read_dir(root.join("segments")))?.collect::, _>>(); + let entries = io(entries)?; + if entries.len() != 1 + || entries[0].file_name() != format!("{}.jsonl", self.segment_id).as_str() + { + return Err(reject("only the fixed single-Segment fixture is supported")); + } + let store = WorkerSessionStore::open_read_only(root) + .map_err(|_| reject("unsupported Session schema"))?; + let logs = store + .read_all_read_only_bounded(self.session_id, self.segment_id, MAX_SESSION_BYTES as u64) + .map_err(|_| reject("invalid canonical Session records"))? + .0; + if !matches!(logs.first(), Some(session_store::LogEntry::AnnotatedSegmentStart { + session_id, forked_from: None, compacted_from: None, .. + }) if *session_id == self.session_id) + { + return Err(reject("invalid Segment seed or unsupported lineage")); + } + let mut ids = std::collections::BTreeSet::new(); + for (index, log) in logs.iter().enumerate() { + use session_store::LogEntry; + match log { + LogEntry::AnnotatedSegmentStart { history, .. } + if index == 0 && history.is_empty() => {} + LogEntry::AnnotatedUserInput { + history, + extensions, + .. + } => { + for extension in extensions { + validate_settled_activation(extension)?; + } + for entry in history { + self.validate_history_entry(entry, &mut ids)?; + } + } + LogEntry::AnnotatedAssistantItem { entry, .. } => { + self.validate_history_entry(entry, &mut ids)? + } + LogEntry::RunCompleted { + entry_id: Some(id), .. + } => { + if !portable_id(&id.0) || !ids.insert(id.0.clone()) { + return Err(reject("duplicate or invalid entry identity")); + } + } + LogEntry::Invoke { .. } | LogEntry::LlmUsage { .. } | LogEntry::TurnEnd { .. } => {} + _ => { + return Err(reject( + "unsupported conversation record or pending operation", + )); + } + } + } + let state = session_store::collect_state(&logs); + if state.last_run_interrupted || state.active_run_turn_count.is_some() { + return Err(reject("Session has pending execution")); + } + for segments in &state.user_segments { + if segments + .iter() + .any(|segment| !matches!(segment, protocol::Segment::Text { .. })) + { + return Err(reject("Session attachments are unsupported")); + } + } + Ok(()) + } +} + +// The approved conversation has committed submission checkpoints in its original +// Session records. Keep their receipt bytes, but reject any resumable queue or +// Feature invocation history. This projection confers no execution authority. +fn validate_settled_activation( + extension: &session_store::SessionExtension, +) -> Result<(), RuntimeError> { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Checkpoint { + #[serde(rename = "next_activation_sequence")] + _next_activation_sequence: u64, + activating: Option, + activating_notification: Vec, + pending: Vec, + pending_notifications: Vec, + #[serde(rename = "receipts")] + _receipts: Vec, + #[serde(rename = "notification_receipts")] + _notification_receipts: Vec, + invocation_receipts: Vec, + invocation_recovery_blocked: bool, + } + if extension.domain != "worker.pending_activations.v1" { + return Err(reject("unsupported Session extension")); + } + let checkpoint: Checkpoint = serde_json::from_value(extension.payload.clone()) + .map_err(|_| reject("invalid Session activation checkpoint"))?; + if checkpoint.activating.is_some() + || !checkpoint.activating_notification.is_empty() + || !checkpoint.pending.is_empty() + || !checkpoint.pending_notifications.is_empty() + || !checkpoint.invocation_receipts.is_empty() + || checkpoint.invocation_recovery_blocked + { + return Err(reject( + "Session has pending activation or invocation history", + )); + } + Ok(()) +} + +fn check_request(request: &CreateWorkerRequest, install: bool) -> Result<(), RuntimeError> { + if !request.workdir_attachment_requests.is_empty() + || !request.workdir_attachments.is_empty() + || request.backend_job.is_some() + || request.subjektiv_attached + || request.worker_observation_enabled + || !request.worker_observation_grants.is_empty() + || (install && request.initial_input.is_some()) + { + return Err(reject( + "unsupported capabilities, attachments, or initial input", + )); + } + if install + && (request.create_fingerprint.trim().is_empty() + || request.config_bundle.is_none() + || !matches!( + request.profile_source, + ProfileSourceArchiveSource::WorkspaceConfig { .. } + )) + { + return Err(reject( + "fresh config/profile references and create fingerprint are required", + )); + } + Ok(()) +} + +fn check_empty_artifacts(root: &Path) -> Result<(), RuntimeError> { + for entry in io(fs::read_dir(root))? { + let entry = io(entry)?; + let metadata = io(entry.metadata())?; + if metadata.is_dir() { + check_empty_artifacts(&entry.path())?; + } else if entry.file_name() != ".aggregate.lock" || metadata.len() != 0 { + return Err(reject("Session artifacts are unsupported")); + } + } + Ok(()) +} + +fn write_new(path: &Path, bytes: &[u8]) -> Result<(), RuntimeError> { + let mut file = io(OpenOptions::new().write(true).create_new(true).open(path))?; + io(file.write_all(bytes))?; + io(file.sync_all()) +} + +// A streaming validation visitor, not a generic document available to callers. +struct PortableJson; +impl<'de> Deserialize<'de> for PortableJson { + fn deserialize>(deserializer: D) -> Result { + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = PortableJson; + fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("portable fixture JSON") + } + fn visit_bool(self, _: bool) -> Result { + Ok(PortableJson) + } + fn visit_i64(self, _: i64) -> Result { + Ok(PortableJson) + } + fn visit_u64(self, _: u64) -> Result { + Ok(PortableJson) + } + fn visit_f64(self, _: f64) -> Result { + Ok(PortableJson) + } + fn visit_unit(self) -> Result { + Ok(PortableJson) + } + fn visit_str(self, value: &str) -> Result { + let lower = value.to_ascii_lowercase(); + if [ + "://", + "private key", + "bearer ", + "sk-", + "api_key", + "access_token", + "refresh_token", + "private_key", + "authorization", + ] + .iter() + .any(|marker| lower.contains(marker)) + || value + .split_whitespace() + .any(|word| word.starts_with('/') || word.contains(":\\")) + { + return Err(E::custom("nonportable or sensitive fixture material")); + } + Ok(PortableJson) + } + fn visit_seq>( + self, + mut seq: A, + ) -> Result { + while seq.next_element::()?.is_some() {} + Ok(PortableJson) + } + fn visit_map>( + self, + mut map: A, + ) -> Result { + while map.next_entry::()?.is_some() {} + Ok(PortableJson) + } + } + deserializer.deserialize_any(Visitor) + } +} diff --git a/crates/worker-runtime/src/test_support_tests.rs b/crates/worker-runtime/src/test_support_tests.rs new file mode 100644 index 000000000..a6f24a2c9 --- /dev/null +++ b/crates/worker-runtime/src/test_support_tests.rs @@ -0,0 +1,636 @@ +use super::*; +use crate::auth::RuntimeIdentityMaterial; +use crate::catalog::{ProfileSelector, WorkspaceApiRef}; +use crate::profile_archive::ProfileSourceGraphSummary; +use session_store::{LogEntry, Store}; +use sha2::{Digest, Sha256}; + +const RUNTIME: &str = "fixture-runtime"; +const WORKSPACE: &str = "fixture-workspace"; +const NEW_URL: &str = "http://127.0.0.1:34567"; + +fn worker_id() -> WorkerId { + "01a1657d-b762-7791-87ae-a49f28cadbce".parse().unwrap() +} + +fn resolved() -> manifest::WorkerManifest { + let mut config = manifest::WorkerManifestConfig::resolution_defaults(); + config.worker.name = Some(runtime_worker_name(worker_id())); + manifest::WorkerManifest::try_from(config).unwrap() +} + +fn request() -> CreateWorkerRequest { + CreateWorkerRequest { + worker_id: worker_id(), + create_fingerprint: "new-create-fingerprint".into(), + profile: ProfileSelector::Builtin("default".into()), + display_name: Some("fixture".into()), + profile_source: ProfileSourceArchiveSource::WorkspaceConfig { + archive: ProfileSourceArchiveRef { + id: "new-profile".into(), + digest: "new-profile-digest".into(), + size_bytes: 1, + source_graph: ProfileSourceGraphSummary { + source_count: 1, + total_source_bytes: 1, + entrypoints: BTreeMap::from([("builtin:default".into(), "default.dcl".into())]), + import_count: 0, + }, + }, + }, + config_bundle: Some(ConfigBundleRef { + id: "new-config".into(), + digest: "new-config-digest".into(), + }), + workspace_api: Some(WorkspaceApiRef { + workspace_id: WORKSPACE.into(), + base_url: NEW_URL.into(), + }), + initial_input: None, + workdir_attachment_requests: Vec::new(), + workdir_attachments: Vec::new(), + worker_observation_enabled: false, + worker_observation_grants: Vec::new(), + memory_settings: Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: WORKSPACE.into(), + language: "en".into(), + }), + subjektiv_attached: false, + backend_job: None, + } +} + +fn initialize(root: &Path, runtime_id: &str, workspace_id: &str, url: &str) { + #[derive(Serialize)] + struct Auth { + identity: RuntimeIdentityMaterial, + workspace_issuers: Vec, + } + let issuer = RuntimeIdentityMaterial::generate("fixture-issuer").unwrap(); + let digest = Sha256::digest(crate::auth::decode_public_key(&issuer.public_key).unwrap()); + let fingerprint = format!( + "sha256:{}", + digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::() + ); + let trust = WorkspaceIssuerTrustRecord::from_bundle( + server_api::WorkspacePublicIdentityBundle { + workspace_id: workspace_id.into(), + backend_url: url.into(), + key_id: "fixture-issuer".into(), + algorithm: "ed25519".into(), + public_key: issuer.public_key, + public_key_fingerprint: fingerprint, + }, + "fixture-trust".into(), + 100, + ) + .unwrap(); + fs::create_dir_all(root).unwrap(); + fs::write( + root.join("auth.toml"), + toml::to_string(&Auth { + identity: RuntimeIdentityMaterial::generate(runtime_id).unwrap(), + workspace_issuers: vec![trust], + }) + .unwrap(), + ) + .unwrap(); +} + +fn record(request: CreateWorkerRequest) -> PersistedWorkerRecord { + PersistedWorkerRecord { + restore_guard: crate::runtime::RestoreGuard::default(), + worker_ref: WorkerRef::new(request.worker_id), + worker_id: request.worker_id, + profile: request.profile.clone(), + display_name: request.display_name.clone(), + profile_source: request.profile_source.reference(), + config_bundle: request.config_bundle.clone(), + created_at_ms: Some(123456), + status: WorkerStatus::Stopped, + last_finished_submission_request_id: Some("old-submission-operation".into()), + workspace_id: Some(WORKSPACE.into()), + workdir_attachments: vec![], + logical_workdir_attachments: vec![], + execution_state: PersistedWorkerExecutionState::Available(PersistedWorkerExecution { + request, + binding: Some(PersistedWorkerExecutionBinding {}), + restore_intent: WorkerRestoreIntent::Explicit, + }), + } +} + +fn source() -> (tempfile::TempDir, PathBuf, StoppedWorkerTemplate) { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + initialize(&root, RUNTIME, WORKSPACE, "http://127.0.0.1:23456"); + drop( + crate::Runtime::with_fs_store(FsRuntimeStoreOptions::new(&root).with_runtime_id(RUNTIME)) + .unwrap(), + ); + let opened = FsRuntimeStore::open_or_create(root.clone(), RUNTIME).unwrap(); + let mut old = request(); + old.workspace_api.as_mut().unwrap().base_url = "http://127.0.0.1:23456".into(); + old.config_bundle.as_mut().unwrap().id = "old-config".into(); + opened.store.write_worker_record(&record(old)).unwrap(); + let worker = opened.store.worker_dir(&worker_id()); + let session_id = "01a1657d-b800-7777-8888-111111111111".parse().unwrap(); + let segment_id = "01a1657d-b818-77c1-bd73-176172a4bd82".parse().unwrap(); + let mut metadata = WorkerMetadata::new( + runtime_worker_name(worker_id()), + Some(WorkerActiveSegmentRef { + session_id, + segment_id: Some(segment_id), + }), + ) + .with_workspace_id(WORKSPACE) + .with_workspace_root(PathBuf::from("/old/host/path")); + metadata.resolved_manifest_snapshot = + Some(manifest::write_persisted_worker_manifest_snapshot(&resolved()).unwrap()); + atomic_write_json( + &worker.join(WORKER_METADATA_FILE), + &metadata, + "test metadata", + ) + .unwrap(); + let sessions = WorkerSessionStore::new(worker.join("session")).unwrap(); + let history = session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::Message { + role: session_store::LoggedRole::Assistant, + content: vec![session_store::LoggedContentPart::Text { + text: "artificial fixture reply".into(), + }], + }, + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId("fixed-entry-id".into()), + origin: session_store::LoggedSessionHistoryOrigin::ModelOutput { + worker: session_store::LoggedWorkerSubject { + workspace_id: Some(WORKSPACE.into()), + runtime_id: Some(RUNTIME.into()), + worker_id: worker_id().to_string(), + }, + }, + derivation: None, + input_metadata: None, + }, + }; + sessions + .create_segment( + session_id, + segment_id, + &[ + LogEntry::AnnotatedSegmentStart { + ts: 123, + session_id, + system_prompt: None, + config: Default::default(), + history: vec![], + forked_from: None, + compacted_from: None, + }, + LogEntry::AnnotatedAssistantItem { + ts: 124, + entry: history, + }, + ], + ) + .unwrap(); + fs::create_dir_all(worker.join("runs/old-run")).unwrap(); + fs::write(worker.join("runs/old-run/private.log"), "not exported").unwrap(); + drop(opened); + let template = StoppedWorkerTemplate::extract(&worker).unwrap(); + (temp, worker, template) +} + +fn destination(template: &StoppedWorkerTemplate) -> (tempfile::TempDir, FsRuntimeStoreOptions) { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + initialize( + &root, + template.runtime_id(), + template.workspace_id(), + NEW_URL, + ); + ( + temp, + FsRuntimeStoreOptions::new(root).with_runtime_id(template.runtime_id()), + ) +} + +fn assert_fresh(options: &FsRuntimeStoreOptions) { + assert_eq!(fs::read_dir(&options.root).unwrap().count(), 1); + assert!(options.root.join("auth.toml").is_file()); +} + +#[test] +fn extraction_install_roundtrip_preserves_bytes_and_is_inactive() { + let (_source, worker, template) = source(); + let original_worker = fs::read(worker.join(WORKER_FILE)).unwrap(); + let original_metadata = fs::read(worker.join(WORKER_METADATA_FILE)).unwrap(); + let exported = serde_json::to_vec(&template).unwrap(); + let exported_text = String::from_utf8(exported.clone()).unwrap(); + for forbidden in [ + "old-config", + "old-submission-operation", + "restore_guard", + "private_key", + "http://", + "/old/host", + "private.log", + "resolved_manifest_snapshot", + ] { + assert!(!exported_text.contains(forbidden)); + } + let template: StoppedWorkerTemplate = serde_json::from_slice(&exported).unwrap(); + let (_target, options) = destination(&template); + let auth_before = fs::read(options.root.join("auth.toml")).unwrap(); + template.install(&options, request(), &resolved()).unwrap(); + assert_eq!( + fs::read(options.root.join("auth.toml")).unwrap(), + auth_before + ); + assert_eq!(fs::read(worker.join(WORKER_FILE)).unwrap(), original_worker); + assert_eq!( + fs::read(worker.join(WORKER_METADATA_FILE)).unwrap(), + original_metadata + ); + let installed = options + .root + .join(WORKERS_DIR) + .join(template.worker_id().to_string()); + assert_eq!( + fs::read(installed.join("session/session.json")).unwrap(), + template.session_manifest + ); + assert_eq!( + fs::read(installed.join(format!("session/segments/{}.jsonl", template.segment_id()))) + .unwrap(), + template.segment_jsonl + ); + assert!(!installed.join("runs").exists()); + let metadata: WorkerMetadata = + decode(&fs::read(installed.join(WORKER_METADATA_FILE)).unwrap()).unwrap(); + assert_eq!(metadata.active.unwrap().session_id, template.session_id()); + assert!(metadata.workspace_root.is_none() && metadata.resolved_manifest_snapshot.is_some()); + + let opened = FsRuntimeStore::open_or_create(options.root.clone(), RUNTIME).unwrap(); + let state = opened.state.unwrap(); + assert!(state.diagnostics.is_empty()); + let saved = state.workers.get(&worker_id()).unwrap(); + assert_eq!(saved.status, WorkerStatus::Stopped); + assert!(saved.last_finished_submission_request_id.is_none()); + let PersistedWorkerExecutionState::Available(execution) = &saved.execution_state else { + panic!("not available") + }; + assert_eq!(execution.restore_intent, WorkerRestoreIntent::Explicit); + assert!(execution.binding.is_none()); + assert_eq!(execution.request, request()); + // Fresh restore guard, no operation owner/receipt history. + assert_ne!( + serde_json::to_vec(&saved.restore_guard).unwrap(), + serde_json::to_vec( + &decode::(&original_worker) + .unwrap() + .restore_guard + ) + .unwrap() + ); + let guard: GuardState = decode(&serde_json::to_vec(&saved.restore_guard).unwrap()).unwrap(); + assert!(guard.active_request_id.is_none() && guard.owners.is_empty()); + drop(opened.store); + + struct NoExecution; + impl crate::execution::WorkerExecutionBackend for NoExecution { + fn backend_id(&self) -> &str { + "fixture-no-execution" + } + fn spawn_worker( + &self, + _: crate::execution::WorkerExecutionSpawnRequest, + ) -> crate::execution::WorkerExecutionSpawnResult { + panic!("fixture executed") + } + fn dispatch_input( + &self, + _: &WorkerRef, + _: crate::interaction::WorkerInput, + ) -> crate::execution::WorkerExecutionResult { + panic!("fixture submitted") + } + fn restore_worker( + &self, + _: crate::execution::WorkerExecutionRestoreRequest, + ) -> crate::execution::WorkerExecutionSpawnResult { + panic!("fixture auto-restored") + } + } + let runtime = + crate::Runtime::with_fs_store_and_execution_backend(options.clone(), Arc::new(NoExecution)) + .unwrap(); + assert_eq!(runtime.list_workers().unwrap().len(), 1); + assert_eq!(runtime.summary().unwrap().active_worker_count, 0); + assert_eq!( + runtime + .worker_detail(&WorkerRef::new(worker_id())) + .unwrap() + .status, + WorkerStatus::Stopped + ); + drop(runtime); + let runtime = crate::Runtime::with_fs_store(options).unwrap(); + assert_eq!( + runtime.list_workers().unwrap()[0].status, + WorkerStatus::Stopped + ); +} + +#[test] +fn install_rejects_schema_identity_input_and_overwrite() { + let (_source, _, template) = source(); + for change in [ + "template", + "runtime-schema", + "session-schema", + "session-id", + "segment-id", + "runtime-id", + "workspace-id", + "worker-id", + "origin-id", + "duplicate-id", + "truncated", + ] { + let (_target, options) = destination(&template); + let mut invalid = template.clone(); + match change { + "template" => invalid.template_version += 1, + "runtime-schema" => invalid.runtime_schema_version += 1, + "session-schema" => { + invalid.session_manifest = format!( + "{{\"schema_version\":999,\"session_id\":\"{}\"}}", + invalid.session_id + ) + .into_bytes() + } + "session-id" => invalid.session_id = uuid::Uuid::now_v7(), + "segment-id" => invalid.segment_id = uuid::Uuid::nil(), // no external attestation: valid new Segment ID is otherwise portable + "runtime-id" => invalid.runtime_id = "other-runtime".into(), + "workspace-id" => invalid.workspace_id = "other-workspace".into(), + "worker-id" => invalid.worker_id = WorkerId::now_v7(), + "origin-id" => { + invalid.segment_jsonl = String::from_utf8(invalid.segment_jsonl) + .unwrap() + .replace(RUNTIME, "other-runtime") + .into_bytes() + } + "duplicate-id" => { + let last = invalid + .segment_jsonl + .split_inclusive(|b| *b == b'\n') + .last() + .unwrap() + .to_vec(); + invalid.segment_jsonl.extend(last); + } + "truncated" => { + invalid.segment_jsonl.pop(); + } + _ => unreachable!(), + } + assert!( + invalid.install(&options, request(), &resolved()).is_err(), + "accepted {change}" + ); + assert_fresh(&options); + } + let (_target, options) = destination(&template); + let mut launch = request(); + launch.initial_input = Some(crate::interaction::WorkerInput::user("do not execute")); + assert!(template.install(&options, launch, &resolved()).is_err()); + assert_fresh(&options); + template.install(&options, request(), &resolved()).unwrap(); + let worker_path = options + .root + .join(WORKERS_DIR) + .join(worker_id().to_string()) + .join(WORKER_FILE); + let before = fs::read(&worker_path).unwrap(); + assert!(template.install(&options, request(), &resolved()).is_err()); + assert_eq!(fs::read(worker_path).unwrap(), before); +} + +#[test] +fn rejects_uninitialized_untrusted_live_and_partial_destination() { + let (_source, _, template) = source(); + let (_target, options) = destination(&template); + let held = acquire_runtime_store_owner_lock(&options.root).unwrap(); + assert!(template.install(&options, request(), &resolved()).is_err()); + assert_fresh(&options); + drop(held); + initialize(&options.root, "wrong-runtime", WORKSPACE, NEW_URL); + assert!(template.install(&options, request(), &resolved()).is_err()); + initialize(&options.root, RUNTIME, WORKSPACE, "http://127.0.0.1:23456"); + assert!(template.install(&options, request(), &resolved()).is_err()); + fs::remove_file(options.root.join("auth.toml")).unwrap(); + assert!(template.install(&options, request(), &resolved()).is_err()); + initialize(&options.root, RUNTIME, WORKSPACE, NEW_URL); + fs::create_dir(options.root.join(WORKERS_DIR)).unwrap(); + assert!(template.install(&options, request(), &resolved()).is_err()); +} + +#[test] +fn extraction_rejects_nonstopped_pending_schema_and_metadata_mismatch() { + for change in [ + "running", + "pending", + "guard-pending", + "attachment", + "schema", + "worker-id", + "metadata-workspace", + "metadata-session", + "artifact", + "secret", + "url", + "path", + ] { + let (_source, worker, _) = source(); + let mut aggregate: WorkerAggregateRecord = + decode(&fs::read(worker.join(WORKER_FILE)).unwrap()).unwrap(); + match change { + "running" => aggregate.status = WorkerStatus::Running, + "pending" => { + aggregate.execution_state = WorkerExecutionStateRecord::ReconciliationRequired( + PersistedWorkerLifecycleOperation::Restore(PersistedWorkerRestoreOperation { + operation_id: WorkerLifecycleOperationId::new(), + mode: PersistedWorkerRestoreMode::Explicit, + request: request(), + binding: None, + last_settled_status: WorkerStatus::Stopped, + }), + ) + } + "guard-pending" => aggregate.restore_guard = serde_json::from_str(r#"{"observation_token":"old-token","active_request_id":"pending-restore","owners":{}}"#).unwrap(), + "attachment" => aggregate.logical_workdir_attachments.push(LogicalWorkdirAttachment { + alias: workdir::WorkdirAttachmentAlias::new("source").unwrap(), + working_directory_id: "old-workdir".into(), + capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, + }), + "schema" => aggregate.schema_version += 1, + "worker-id" => aggregate.worker_id = WorkerId::now_v7(), + "metadata-workspace" | "metadata-session" => { + let mut metadata: WorkerMetadata = + decode(&fs::read(worker.join(WORKER_METADATA_FILE)).unwrap()).unwrap(); + if change == "metadata-workspace" { + metadata.workspace_id = Some("wrong-workspace".into()); + } else { + metadata.active.as_mut().unwrap().session_id = uuid::Uuid::now_v7(); + } + atomic_write_json( + &worker.join(WORKER_METADATA_FILE), + &metadata, + "test mismatch", + ) + .unwrap(); + } + "artifact" => { + fs::create_dir_all(worker.join("session/artifacts/paste")).unwrap(); + fs::write( + worker.join("session/artifacts/paste/payload"), + "unsupported", + ) + .unwrap(); + } + "secret" | "url" | "path" => { + let segment = fs::read_dir(worker.join("session/segments")) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let replacement = match change { + "secret" => "Bearer secret", + "url" => "https://old-host.invalid", + _ => "/old/host/path", + }; + let text = fs::read_to_string(&segment) + .unwrap() + .replace("artificial fixture reply", replacement); + fs::write(segment, text).unwrap(); + } + _ => unreachable!(), + } + atomic_write_json(&worker.join(WORKER_FILE), &aggregate, "test aggregate").unwrap(); + assert!( + StoppedWorkerTemplate::extract(&worker).is_err(), + "accepted {change}" + ); + } + let (_source, worker, _) = source(); + let held = + acquire_runtime_store_owner_lock(worker.parent().unwrap().parent().unwrap()).unwrap(); + assert!(StoppedWorkerTemplate::extract(&worker).is_err()); + drop(held); +} + +#[test] +fn fresh_manifest_and_memory_settings_are_required() { + let (_source, _, template) = source(); + let (_target, options) = destination(&template); + let mut manifest = resolved(); + manifest.worker.name = "wrong-worker".into(); + assert!(template.install(&options, request(), &manifest).is_err()); + let mut binding = request(); + binding.memory_settings = None; + assert!(template.install(&options, binding, &resolved()).is_err()); + let mut binding = request(); + binding.memory_settings.as_mut().unwrap().workspace_id = "other-workspace".into(); + assert!(template.install(&options, binding, &resolved()).is_err()); + assert_fresh(&options); +} + +#[test] +fn settled_checkpoints_preserve_receipts_but_refuse_pending_work() { + #[derive(Default, Serialize)] + struct Checkpoint { + next_activation_sequence: u64, + activating: Option, + activating_notification: Vec, + pending: Vec, + pending_notifications: Vec, + receipts: Vec, + notification_receipts: Vec, + invocation_receipts: Vec, + invocation_recovery_blocked: bool, + } + let extension = |checkpoint: &Checkpoint| { + session_store::SessionExtension::new( + "worker.pending_activations.v1", + serde_json::to_value(checkpoint).unwrap(), + ) + }; + let mut checkpoint = Checkpoint { + receipts: vec![1, 2], + ..Default::default() + }; + assert!(validate_settled_activation(&extension(&checkpoint)).is_ok()); + checkpoint.pending.push(1); + assert!(validate_settled_activation(&extension(&checkpoint)).is_err()); + checkpoint.pending.clear(); + checkpoint.activating = Some(1); + assert!(validate_settled_activation(&extension(&checkpoint)).is_err()); + checkpoint.activating = None; + checkpoint.invocation_receipts.push(1); + assert!(validate_settled_activation(&extension(&checkpoint)).is_err()); + checkpoint.invocation_receipts.clear(); + checkpoint.invocation_recovery_blocked = true; + assert!(validate_settled_activation(&extension(&checkpoint)).is_err()); +} + +#[cfg(unix)] +#[test] +fn refuses_source_and_destination_symlinks() { + let (_source, worker, template) = source(); + let (_target, options) = destination(&template); + std::os::unix::fs::symlink(worker.join(WORKER_FILE), options.root.join("linked")).unwrap(); + assert!(template.install(&options, request(), &resolved()).is_err()); + fs::remove_file(options.root.join("linked")).unwrap(); + std::os::unix::fs::symlink(worker.join(WORKER_FILE), worker.join("runs/link")).unwrap(); + assert!(StoppedWorkerTemplate::extract(&worker).is_err()); +} + +/// Opt-in only: source path supplied explicitly; assertions never display content. +#[test] +#[ignore = "requires the approved artificial lab aggregate via YOI_STOPPED_FIXTURE_SOURCE"] +fn approved_lab_extract_install() { + let path = PathBuf::from( + std::env::var_os("YOI_STOPPED_FIXTURE_SOURCE").expect("explicit source required"), + ); + let template = StoppedWorkerTemplate::extract(&path).unwrap(); + let (_target, options) = destination(&template); + let mut bind = request(); + bind.worker_id = template.worker_id(); + bind.workspace_api.as_mut().unwrap().workspace_id = template.workspace_id().into(); + bind.memory_settings.as_mut().unwrap().workspace_id = template.workspace_id().into(); + let mut manifest = resolved(); + manifest.worker.name = template.worker_name(); + template.install(&options, bind, &manifest).unwrap(); + let installed = options + .root + .join(WORKERS_DIR) + .join(template.worker_id().to_string()); + for relative in [ + "session/session.json".to_string(), + format!("session/segments/{}.jsonl", template.segment_id()), + ] { + assert!( + fs::read(path.join(&relative)).unwrap() == fs::read(installed.join(relative)).unwrap() + ); + } + let runtime = crate::Runtime::with_fs_store(options).unwrap(); + assert_eq!(runtime.list_workers().unwrap().len(), 1); + assert_eq!(runtime.summary().unwrap().active_worker_count, 0); +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index a482902c6..9fc0dabdd 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1389,6 +1389,15 @@ fn bind_workspace_memory_settings( Ok(()) } +/// Read-only validation shared with the offline, feature-gated fixture installer. +#[cfg(feature = "test-support")] +pub(crate) fn validate_fixture_restore_metadata( + metadata: &session_store::WorkerMetadata, + request: &CreateWorkerRequest, +) -> Result<(), String> { + ProfileRuntimeWorkerFactory::manifest_for_restore(metadata, request).map(|_| ()) +} + fn validate_worker_memory_settings( manifest: &manifest::WorkerManifest, request: &CreateWorkerRequest,