test: adapt fixed Worker restore fixtures to integration lane

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"]
 

crates/worker-runtime/README.md

@@ -8,3 +8,34 @@ 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. A saved system prompt is required for restore;
+checkpoint receipts must match the Worker's typed receipt schema. Preserved records
+must use canonical writer field shapes: unknown/duplicate fields (including nested
+fields), discarded optional fields, and opaque Feature history payloads are rejected,
+not stripped or rewritten. Portability/secret-marker checks do not replace
+review of the supplied artificial conversation or its independently verified hashes.
+The installer refuses any destination content other than initialized `auth.toml`;
+interrupted publication is fail-closed, not an overwrite/recovery facility.

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;

crates/worker-runtime/src/lib.rs

@@ -35,6 +35,8 @@ pub mod workspace_issuer;
 #[cfg(feature = "http-server")]
 pub mod workspace_request;
 
+#[cfg(feature = "test-support")]
+pub use fs_store::test_support;
 #[cfg(feature = "fs-store")]
 pub use fs_store::{FsRuntimeStore, FsRuntimeStoreOptions};
 pub use management::RuntimeOptions;

crates/worker-runtime/src/test_support.rs

@@ -0,0 +1,859 @@
+//! 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<u64>,
+    session_id: SessionId,
+    segment_id: SegmentId,
+    session_manifest: Vec<u8>,
+    segment_jsonl: Vec<u8>,
+}
+
+#[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<WorkspaceIssuerTrustRecord>,
+    #[serde(default)]
+    archived_legacy_issuers: Vec<String>,
+}
+
+#[derive(Deserialize)]
+struct GuardState {
+    active_request_id: Option<String>,
+    owners: BTreeMap<String, GuardOwner>,
+}
+
+#[derive(Deserialize)]
+struct GuardOwner {
+    receipt: Option<serde::de::IgnoredAny>,
+}
+
+fn reject(message: &'static str) -> RuntimeError {
+    RuntimeError::InvalidRequest(format!("stopped Worker fixture: {message}"))
+}
+
+fn io<T>(result: std::io::Result<T>) -> Result<T, RuntimeError> {
+    result.map_err(|_| reject("filesystem operation failed"))
+}
+
+fn decode<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, RuntimeError> {
+    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(&current))?;
+        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<Vec<u8>, 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, RuntimeError> {
+    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 `<runtime>/workers/<WorkerId>`.
+    pub fn extract(worker_dir: &Path) -> Result<Self, RuntimeError> {
+        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)?;
+            validate_preserved_record(line)?;
+        }
+        Ok(())
+    }
+
+    fn validate_history_entry(
+        &self,
+        entry: &session_store::LoggedHistoryEntry,
+        ids: &mut std::collections::BTreeSet<String>,
+    ) -> 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 entry
+            .metadata
+            .input_metadata
+            .as_ref()
+            .is_some_and(|metadata| {
+                metadata.feature_payload().is_some() || metadata.public_item().is_some()
+            })
+        {
+            // These are opaque owner-specific payloads, not part of the narrow
+            // artificial conversation contract. Never preserve unchecked JSON.
+            return Err(reject("Feature history payloads are unsupported"));
+        }
+        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::<Result<Vec<_>, _>>();
+        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, system_prompt: Some(_), forked_from: None, compacted_from: None, ..
+        }) if *session_id == self.session_id)
+        {
+            return Err(reject(
+                "missing system prompt, 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(())
+    }
+}
+
+// Compare *fields*, not re-encoded bytes or scalar values (e.g. f32 JSON
+// normalization). Serde's permissive/untagged/flattened compatibility readers
+// can drop unknown data without surfacing it to an ignored-field callback. A
+// retained field must survive the owner's typed projection at every depth.
+// This intentionally accepts only canonical writer shapes: even an explicit
+// empty optional field omitted by that writer is rejected, not repaired.
+// The projection is validation-only; installation always writes original bytes.
+fn validate_preserved_record(bytes: &[u8]) -> Result<(), RuntimeError> {
+    let original: serde_json::Value = decode(bytes)?;
+    let record: session_store::LogEntry = decode(bytes)?;
+    let projection =
+        serde_json::to_value(&record).map_err(|_| reject("invalid typed Session record"))?;
+    require_preserved_fields(&original, &projection)
+}
+
+fn require_preserved_fields(
+    original: &serde_json::Value,
+    projection: &serde_json::Value,
+) -> Result<(), RuntimeError> {
+    use serde_json::Value;
+    match (original, projection) {
+        (Value::Object(original), Value::Object(projection)) => {
+            for (key, value) in original {
+                let projected = projection
+                    .get(key)
+                    .ok_or_else(|| reject("unknown or noncanonical Session field"))?;
+                require_preserved_fields(value, projected)?;
+            }
+        }
+        (Value::Array(original), Value::Array(projection))
+            if original.len() == projection.len() =>
+        {
+            for (value, projected) in original.iter().zip(projection) {
+                require_preserved_fields(value, projected)?;
+            }
+        }
+        (Value::Null, Value::Null)
+        | (Value::Bool(_), Value::Bool(_))
+        | (Value::Number(_), Value::Number(_))
+        | (Value::String(_), Value::String(_)) => {}
+        _ => return Err(reject("noncanonical Session field shape")),
+    }
+    Ok(())
+}
+
+// Exact receipt wire fields owned by worker::PendingActivationState. Keep these
+// fixture-only mirrors strict; do not change the production compatibility reader.
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SubmissionReceipt {
+    submission_request_id: String,
+    source_namespace: String,
+    submission_id: String,
+    payload_digest: String,
+    disposition: protocol::SubmissionDisposition,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct NotificationReceipt {
+    notification_request_id: String,
+    source_namespace: String,
+    payload_digest: String,
+}
+
+// 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<serde::de::IgnoredAny>,
+        activating_notification: Vec<serde::de::IgnoredAny>,
+        pending: Vec<serde::de::IgnoredAny>,
+        pending_notifications: Vec<serde::de::IgnoredAny>,
+        #[serde(rename = "receipts")]
+        _receipts: Vec<SubmissionReceipt>,
+        #[serde(rename = "notification_receipts")]
+        _notification_receipts: Vec<NotificationReceipt>,
+        invocation_receipts: Vec<serde::de::IgnoredAny>,
+        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<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        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<E: serde::de::Error>(self, _: bool) -> Result<PortableJson, E> {
+                Ok(PortableJson)
+            }
+            fn visit_i64<E: serde::de::Error>(self, _: i64) -> Result<PortableJson, E> {
+                Ok(PortableJson)
+            }
+            fn visit_u64<E: serde::de::Error>(self, _: u64) -> Result<PortableJson, E> {
+                Ok(PortableJson)
+            }
+            fn visit_f64<E: serde::de::Error>(self, _: f64) -> Result<PortableJson, E> {
+                Ok(PortableJson)
+            }
+            fn visit_unit<E: serde::de::Error>(self) -> Result<PortableJson, E> {
+                Ok(PortableJson)
+            }
+            fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<PortableJson, E> {
+                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<A: serde::de::SeqAccess<'de>>(
+                self,
+                mut seq: A,
+            ) -> Result<PortableJson, A::Error> {
+                while seq.next_element::<PortableJson>()?.is_some() {}
+                Ok(PortableJson)
+            }
+            fn visit_map<A: serde::de::MapAccess<'de>>(
+                self,
+                mut map: A,
+            ) -> Result<PortableJson, A::Error> {
+                let mut keys = std::collections::BTreeSet::new();
+                while let Some(key) = map.next_key::<String>()? {
+                    // A duplicate key can hide discarded data even inside an
+                    // otherwise opaque extension payload. Never retain it.
+                    if !keys.insert(key.clone()) {
+                        return Err(serde::de::Error::custom("duplicate fixture JSON field"));
+                    }
+                    serde::de::Visitor::visit_str::<A::Error>(Visitor, &key)?;
+                    map.next_value::<PortableJson>()?;
+                }
+                Ok(PortableJson)
+            }
+        }
+        deserializer.deserialize_any(Visitor)
+    }
+}

crates/worker-runtime/src/test_support_tests.rs

@@ -0,0 +1,885 @@
+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";
+
+#[derive(Serialize)]
+#[serde(untagged)]
+enum ReceiptFixture<T> {
+    Valid(T),
+    Malformed(u8),
+}
+
+#[derive(Default, Serialize)]
+struct CheckpointFixture {
+    next_activation_sequence: u64,
+    activating: Option<u8>,
+    activating_notification: Vec<u8>,
+    pending: Vec<u8>,
+    pending_notifications: Vec<u8>,
+    receipts: Vec<ReceiptFixture<SubmissionReceipt>>,
+    notification_receipts: Vec<ReceiptFixture<NotificationReceipt>>,
+    invocation_receipts: Vec<u8>,
+    invocation_recovery_blocked: bool,
+}
+
+fn settled_checkpoint() -> CheckpointFixture {
+    CheckpointFixture {
+        next_activation_sequence: 2,
+        receipts: vec![ReceiptFixture::Valid(SubmissionReceipt {
+            submission_request_id: "completed-submission-request".into(),
+            source_namespace: "fixture-source".into(),
+            submission_id: "completed-submission".into(),
+            payload_digest: "fixture-payload-digest".into(),
+            disposition: protocol::SubmissionDisposition::Started,
+        })],
+        notification_receipts: vec![ReceiptFixture::Valid(NotificationReceipt {
+            notification_request_id: "completed-notification-request".into(),
+            source_namespace: "fixture-source".into(),
+            payload_digest: "fixture-notification-digest".into(),
+        })],
+        ..Default::default()
+    }
+}
+
+fn checkpoint_extension(checkpoint: &CheckpointFixture) -> session_store::SessionExtension {
+    session_store::SessionExtension::new(
+        "worker.pending_activations.v1",
+        serde_json::to_value(checkpoint).unwrap(),
+    )
+}
+
+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<WorkspaceIssuerTrustRecord>,
+    }
+    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::<String>()
+    );
+    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: Some("Harmless artificial fixture instructions.".into()),
+                    config: Default::default(),
+                    history: vec![],
+                    forked_from: None,
+                    compacted_from: None,
+                },
+                LogEntry::Invoke {
+                    ts: 124,
+                    trigger: protocol::InvokeKind::UserSend,
+                },
+                LogEntry::AnnotatedUserInput {
+                    ts: 124,
+                    segments: vec![protocol::Segment::text(
+                        "Harmless artificial fixture input.",
+                    )],
+                    extensions: vec![checkpoint_extension(&settled_checkpoint())],
+                    history: vec![session_store::LoggedHistoryEntry {
+                        item: session_store::LoggedItem::Message {
+                            role: session_store::LoggedRole::User,
+                            content: vec![session_store::LoggedContentPart::Text {
+                                text: "Harmless artificial fixture input.".into(),
+                            }],
+                        },
+                        metadata: session_store::LoggedSessionHistoryMetadata {
+                            entry_id: session_store::LoggedSessionHistoryEntryId(
+                                "fixed-user-id".into(),
+                            ),
+                            origin: session_store::LoggedSessionHistoryOrigin::BackendInstruction {
+                                operation_id: None,
+                            },
+                            derivation: None,
+                            input_metadata: None,
+                        },
+                    }],
+                },
+                LogEntry::AnnotatedAssistantItem {
+                    ts: 125,
+                    entry: history,
+                },
+                LogEntry::LlmUsage {
+                    ts: 125,
+                    history_len: 2,
+                    input_total_tokens: 1,
+                    cache_read_tokens: 0,
+                    cache_write_tokens: 0,
+                    output_tokens: 1,
+                },
+                LogEntry::TurnEnd {
+                    ts: 126,
+                    turn_count: 1,
+                },
+                LogEntry::RunCompleted {
+                    entry_id: Some(session_store::LoggedSessionHistoryEntryId(
+                        "fixed-terminal-id".into(),
+                    )),
+                    ts: 126,
+                    interrupted: false,
+                    result: agen::EngineResult::Finished,
+                    active_run_turn_count: None,
+                },
+            ],
+        )
+        .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::<WorkerAggregateRecord>(&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() {
+    let mut checkpoint = settled_checkpoint();
+    assert!(validate_settled_activation(&checkpoint_extension(&checkpoint)).is_ok());
+    checkpoint.pending.push(1);
+    assert!(validate_settled_activation(&checkpoint_extension(&checkpoint)).is_err());
+    checkpoint.pending.clear();
+    checkpoint.activating = Some(1);
+    assert!(validate_settled_activation(&checkpoint_extension(&checkpoint)).is_err());
+    checkpoint.activating = None;
+    checkpoint.invocation_receipts.push(1);
+    assert!(validate_settled_activation(&checkpoint_extension(&checkpoint)).is_err());
+    checkpoint.invocation_receipts.clear();
+    checkpoint.invocation_recovery_blocked = true;
+    assert!(validate_settled_activation(&checkpoint_extension(&checkpoint)).is_err());
+}
+
+// Deliberately corrupt only test-owned synthetic source files, never the lab.
+fn assert_rejected_session_bytes(worker: &Path, template: &StoppedWorkerTemplate, bytes: Vec<u8>) {
+    let path = worker.join(format!("session/segments/{}.jsonl", template.segment_id()));
+    fs::write(path, &bytes).unwrap();
+    let extraction_error = StoppedWorkerTemplate::extract(worker)
+        .err()
+        .expect("invalid source accepted");
+    assert!(
+        !extraction_error
+            .to_string()
+            .contains("synthetic-discarded-material")
+    );
+    let (_target, options) = destination(template);
+    let mut invalid = template.clone();
+    invalid.segment_jsonl = bytes;
+    let installation_error = invalid
+        .install(&options, request(), &resolved())
+        .unwrap_err();
+    assert!(
+        !installation_error
+            .to_string()
+            .contains("synthetic-discarded-material")
+    );
+    assert_fresh(&options);
+}
+
+fn synthetic_logs(template: &StoppedWorkerTemplate) -> Vec<LogEntry> {
+    template
+        .segment_jsonl
+        .split(|byte| *byte == b'\n')
+        .filter(|line| !line.is_empty())
+        .map(|line| serde_json::from_slice(line).unwrap())
+        .collect()
+}
+
+fn encode_synthetic_logs(logs: &[LogEntry]) -> Vec<u8> {
+    let mut bytes = Vec::new();
+    for log in logs {
+        serde_json::to_writer(&mut bytes, log).unwrap();
+        bytes.push(b'\n');
+    }
+    bytes
+}
+
+#[tokio::test]
+async fn missing_system_prompt_matches_worker_restore_rejection() {
+    let (_source, worker_dir, template) = source();
+    let mut logs = synthetic_logs(&template);
+    assert!(session_store::collect_state(&logs).system_prompt.is_some());
+    let LogEntry::AnnotatedSegmentStart { system_prompt, .. } = &mut logs[0] else {
+        panic!("missing seed")
+    };
+    *system_prompt = None;
+    assert_rejected_session_bytes(&worker_dir, &template, encode_synthetic_logs(&logs));
+    // The owning Worker constructor rejects this before model-client creation,
+    // allocation, or execution. Do not start a controller or submit any input.
+    let store = session_store::CombinedStore::new(
+        WorkerSessionStore::open_read_only(worker_dir.join("session")).unwrap(),
+        session_store::WorkerAggregateStore::open_read_only(&worker_dir, template.worker_name())
+            .unwrap(),
+    );
+    let restored = worker::Worker::restore_from_manifest(
+        template.session_id(),
+        template.segment_id(),
+        resolved(),
+        store,
+        worker::PromptCatalogSource::builtins_only(),
+    )
+    .await;
+    assert!(matches!(
+        restored,
+        Err(worker::WorkerError::SegmentSystemPromptMissing { .. })
+    ));
+}
+
+#[test]
+fn malformed_submission_and_notification_receipts_fail_extraction_and_install() {
+    let (_source, worker, template) = source();
+    for notification in [false, true] {
+        let mut checkpoint = settled_checkpoint();
+        if notification {
+            checkpoint.notification_receipts =
+                vec![ReceiptFixture::Malformed(1), ReceiptFixture::Malformed(2)];
+        } else {
+            checkpoint.receipts = vec![ReceiptFixture::Malformed(1), ReceiptFixture::Malformed(2)];
+        }
+        let mut logs = synthetic_logs(&template);
+        for log in &mut logs {
+            if let LogEntry::AnnotatedUserInput { extensions, .. } = log {
+                *extensions = vec![checkpoint_extension(&checkpoint)];
+            }
+        }
+        assert_rejected_session_bytes(&worker, &template, encode_synthetic_logs(&logs));
+    }
+}
+
+#[test]
+fn unknown_fields_at_every_retained_record_depth_are_rejected_without_secret_diagnostics() {
+    let (_source, worker, template) = source();
+    let original = String::from_utf8(template.segment_jsonl.clone()).unwrap();
+    // These values are harmless synthetic markers, not lab/private material.
+    // The compatibility decoder normally discards many of these fields.
+    let locations = [
+        r#""kind":"annotated_segment_start""#,
+        r#""kind":"invoke""#,
+        r#""kind":"annotated_user_input""#,
+        r#""kind":"annotated_assistant_item""#,
+        r#""kind":"llm_usage""#,
+        r#""kind":"turn_end""#,
+        r#""kind":"run_completed""#,
+        r#""max_tokens":null"#,
+        r#""kind":"message","role":"assistant""#,
+        r#""kind":"text","text":"artificial fixture reply""#,
+        r#""kind":"text","content":"Harmless artificial fixture input.""#,
+        r#""entry_id":"fixed-entry-id""#,
+        r#""kind":"model_output""#,
+        r#""worker_id":"01a1657d-b762-7791-87ae-a49f28cadbce""#,
+        r#""domain":"worker.pending_activations.v1""#,
+        r#""next_activation_sequence":2"#,
+        r#""submission_request_id":"completed-submission-request""#,
+        r#""notification_request_id":"completed-notification-request""#,
+    ];
+    for location in locations {
+        assert!(original.contains(location));
+        for extra in [
+            r#", "reasoning":{"text":"synthetic-discarded-material"}"#,
+            r#", "encrypted_content":"synthetic-discarded-material""#,
+        ] {
+            let altered = original.replacen(location, &format!("{location}{extra}"), 1);
+            assert_rejected_session_bytes(&worker, &template, altered.into_bytes());
+        }
+    }
+    let duplicate = original.replacen(
+        r#""role":"assistant""#,
+        r#""role":"assistant","role":"assistant""#,
+        1,
+    );
+    assert_rejected_session_bytes(&worker, &template, duplicate.into_bytes());
+}
+
+#[test]
+fn opaque_feature_payloads_cannot_bypass_strict_session_validation() {
+    let (_source, worker, template) = source();
+    let mut logs = synthetic_logs(&template);
+    for log in &mut logs {
+        if let LogEntry::AnnotatedUserInput { history, .. } = log {
+            if let session_store::LoggedItem::Message { role, .. } = &mut history[0].item {
+                *role = session_store::LoggedRole::System;
+            }
+            history[0].metadata.input_metadata = Some(
+                session_store::HistoryInputMetadata::new(
+                    session_store::HistoryInputUsePolicy {
+                        recording_material: session_store::InputEvidenceUse::Exclude,
+                        memory_evidence: session_store::InputEvidenceUse::Exclude,
+                        observation: session_store::InputObservationUse::Hidden,
+                        compaction: session_store::InputCompactionUse::Retain,
+                    },
+                    Some(
+                        session_store::FeatureInputPayload::encode(
+                            "fixture.test",
+                            1,
+                            &BTreeMap::from([("reasoning", "harmless synthetic payload")]),
+                        )
+                        .unwrap(),
+                    ),
+                    None,
+                )
+                .unwrap(),
+            );
+        }
+    }
+    assert_rejected_session_bytes(&worker, &template, encode_synthetic_logs(&logs));
+}
+
+#[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);
+}

crates/worker-runtime/src/worker_backend.rs

@@ -1682,6 +1682,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,

crates/workspace-server/Cargo.toml

@@ -10,6 +10,10 @@ autobins = false
 name = "yoi-server"
 path = "src/main.rs"
 
+[features]
+# Offline fixture preparation only; build product binaries without this feature.
+test-support = []
+
 [dependencies]
 api-macros = { workspace = true, features = ["axum"] }
 async-trait.workspace = true

crates/workspace-server/src/lib.rs

@@ -28,6 +28,8 @@ pub mod runtime_subscription;
 pub mod server;
 pub mod skills;
 pub mod store;
+#[cfg(feature = "test-support")]
+pub use store::fixture_support as test_support;
 pub mod subjektiv;
 #[cfg(test)]
 mod ticket_browse_performance_tests;

crates/workspace-server/src/store.rs

@@ -2589,6 +2589,10 @@ pub struct SqliteWorkspaceStore {
     db_executor: crate::db_executor::DbExecutor,
 }
 
+#[cfg(feature = "test-support")]
+#[path = "test_support.rs"]
+pub mod fixture_support;
+
 impl SqliteWorkspaceStore {
     pub fn open(path: impl AsRef<Path>) -> Result<Self> {
         let conn = Connection::open(path)?;
@@ -2900,6 +2904,33 @@ impl SqliteWorkspaceStore {
         request_fingerprint: &str,
         singleton_key: Option<&str>,
         current_memory_settings: &WorkspaceMemorySettingsRecord,
+    ) -> Result<WorkerCreateReservation> {
+        self.reserve_worker_create_with(
+            workspace_id,
+            runtime_id,
+            allocation_key,
+            request_fingerprint,
+            singleton_key,
+            current_memory_settings,
+            |_| Ok(None),
+            |_, _| Ok(()),
+        )
+    }
+
+    // Keep reservation validation, Memory binding and writes in one transaction. The
+    // nondefault test-support module supplies fixed identity / offline catalog hooks;
+    // ordinary creation always supplies a newly generated ID and no completion hook.
+    #[allow(clippy::too_many_arguments)]
+    fn reserve_worker_create_with(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        allocation_key: &str,
+        request_fingerprint: &str,
+        singleton_key: Option<&str>,
+        current_memory_settings: &WorkspaceMemorySettingsRecord,
+        prepare: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<Option<WorkerId>>,
+        complete: impl FnOnce(&rusqlite::Transaction<'_>, &WorkerCreateReservation) -> Result<()>,
     ) -> Result<WorkerCreateReservation> {
         if allocation_key.trim().is_empty() || request_fingerprint.trim().is_empty() {
             return Err(Error::InvalidInput(
@@ -2912,6 +2943,7 @@ impl SqliteWorkspaceStore {
         validate_workspace_memory_settings_record(current_memory_settings, workspace_id)?;
         self.with_conn_mut(|conn| {
             let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let fixed_worker_id = prepare(&tx)?;
             let existing = tx
                 .query_row(
                     "SELECT worker_id, runtime_id, request_fingerprint, create_fingerprint, \
@@ -3094,7 +3126,7 @@ impl SqliteWorkspaceStore {
             validate_workspace_memory_settings_snapshot(&snapshot, workspace_id)?;
             let create_fingerprint =
                 bound_worker_create_fingerprint(request_fingerprint, &snapshot);
-            let worker_id = WorkerId::now_v7();
+            let worker_id = fixed_worker_id.unwrap_or_else(WorkerId::now_v7);
             let worker_id_text = worker_id.to_string();
             let now = chrono::Utc::now().to_rfc3339();
             let singleton_generation = match singleton_owner.as_ref() {
@@ -3183,13 +3215,15 @@ impl SqliteWorkspaceStore {
                     ));
                 }
             }
-            tx.commit()?;
-            Ok(WorkerCreateReservation {
+            let reservation = WorkerCreateReservation {
                 worker_id,
                 create_fingerprint,
                 memory_settings: snapshot,
                 singleton,
-            })
+            };
+            complete(&tx, &reservation)?;
+            tx.commit()?;
+            Ok(reservation)
         })
     }
 

crates/workspace-server/src/test_support.rs

@@ -0,0 +1,688 @@
+//! Offline Backend fixture preparation, available only with `test-support`.
+//!
+//! Use a new isolated database and the existing account, Workspace signing-identity /
+//! `ControlPlaneStore::create_workspace_bootstrap`, and Runtime binding APIs first.
+//! Generate fresh account credentials and signing keys for each environment; no fixture
+//! credentials are accepted here. Install Profile configuration through the existing
+//! config-source mechanisms before starting the product processes.
+//!
+//! This module seeds Backend ownership only. It neither writes Runtime storage nor
+//! launches a Worker, fabricates Runtime observations, changes authorization, or sends
+//! initial input. The caller must separately install the matching stopped Runtime
+//! fixture. Build product binaries in a separate Cargo invocation **without** this
+//! feature (Cargo features unify within an invocation).
+
+use super::*;
+use crate::hosts::{WorkerSpawnIntent, WorkerSpawnRequest, worker_spawn_create_fingerprint};
+use worker_runtime::catalog::ProfileSelector;
+
+/// One ordinary, attachment-free Worker from an externally validated fixture.
+///
+/// IDs must match the Runtime fixture. `request` is the unresolved canonical create
+/// intent, not a Runtime launch request containing credentials or execution grants.
+/// The Profile selector must already be installed by the harness. `initial_submit`
+/// must be empty: saved Session contents belong to the Runtime fixture, not new input.
+/// Singleton/service, Ticket-assigned and attachment-bearing Workers are deliberately
+/// unsupported.
+pub struct FixedWorkerSeed<'a> {
+    pub workspace_id: &'a str,
+    pub runtime_id: &'a str,
+    pub worker_id: WorkerId,
+    pub allocation_key: &'a str,
+    pub request: &'a WorkerSpawnRequest,
+}
+
+/// Insert a fixed Worker catalog entry and completed creation reservation atomically.
+///
+/// Uses the normal reservation validators, authoritative Memory snapshot, fingerprint
+/// binding, resource-key allocator and immediate transaction. Even an exact repeat is
+/// rejected: this is not an upsert/import/recovery API. Existing allocations, Worker
+/// identities and residual lifecycle/attachment records are never adopted or repaired.
+/// Returns the creation binding needed to check the corresponding Runtime fixture.
+pub fn seed_fixed_worker(
+    store: &SqliteWorkspaceStore,
+    seed: FixedWorkerSeed<'_>,
+) -> Result<WorkerCreateReservation> {
+    validate_identifier("workspace_id", seed.workspace_id)?;
+    validate_identifier("runtime_id", seed.runtime_id)?;
+    let request = seed.request;
+    if !request.initial_submit.is_empty() {
+        return Err(Error::InvalidInput(
+            "fixture Worker cannot contain initial_submit; offline seeding accepts no new input"
+                .into(),
+        ));
+    }
+    if !matches!(request.intent, WorkerSpawnIntent::WorkspaceCoding)
+        || request.singleton_key.is_some()
+        || request.ticket_assignment.is_some()
+        || request.resolved_subjektiv_attached
+        || !request.workdir_attachment_requests.is_empty()
+        || !request.resolved_workdir_attachment_requests.is_empty()
+        || !request.resolved_workdir_attachments.is_empty()
+        || request.resolved_config_bundle.is_some()
+        || request.resolved_workspace_api.is_some()
+        || request.resolved_memory_settings.is_some()
+        || request.resolved_worker_observation_enabled
+        || !request.resolved_worker_observation_grants.is_empty()
+        || request.resolved_control_operation.is_some()
+        || request.resolved_execution_grant.is_some()
+    {
+        return Err(Error::InvalidInput(
+            "fixture Worker must be ordinary unresolved create intent without singleton, assignments, attachments or grants".into(),
+        ));
+    }
+    // Builtin selectors already carry their canonical `builtin:` prefix.
+    let profile = match &request.profile {
+        ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => name,
+    };
+    validate_non_empty("fixture Profile selector", profile)?;
+    let worker_id = seed.worker_id.to_string();
+    let display_name = request
+        .requested_worker_name
+        .as_deref()
+        .unwrap_or(&worker_id);
+    validate_non_empty("fixture Worker display name", display_name)?;
+    let fingerprint = worker_spawn_create_fingerprint(request).map_err(Error::InvalidInput)?;
+    let memory = store.get_workspace_memory_settings(seed.workspace_id)?;
+    store.reserve_worker_create_with(
+        seed.workspace_id,
+        seed.runtime_id,
+        seed.allocation_key,
+        &fingerprint,
+        None,
+        &memory,
+        |tx| {
+            require_seed_target(tx, &seed)?;
+            Ok(Some(seed.worker_id))
+        },
+        |tx, reservation| {
+            let now = chrono::Utc::now().to_rfc3339();
+            tx.execute(
+                "INSERT INTO worker_registry (
+                    workspace_id, runtime_id, worker_id, display_name, profile,
+                    retention_state, created_at, updated_at
+                 ) VALUES (?1, ?2, ?3, ?4, ?5, 'normal', ?6, ?6)",
+                params![seed.workspace_id, seed.runtime_id, worker_id, display_name, profile, now],
+            )?;
+            let changed = tx.execute(
+                "UPDATE worker_create_reservations SET state = 'created', updated_at = ?4
+                 WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND state = 'reserved'",
+                params![seed.workspace_id, seed.runtime_id, reservation.worker_id.to_string(), now],
+            )?;
+            if changed != 1 {
+                return Err(Error::Store("fixture creation reservation disappeared".into()));
+            }
+            Ok(())
+        },
+    )
+}
+
+fn require_seed_target(tx: &rusqlite::Transaction<'_>, seed: &FixedWorkerSeed<'_>) -> Result<()> {
+    let active: bool = tx.query_row(
+        "SELECT EXISTS(SELECT 1 FROM workspaces w
+         JOIN workspace_runtime_bindings r ON r.workspace_id = w.workspace_id
+         JOIN workspace_signing_identities s ON s.workspace_id = w.workspace_id
+         WHERE w.workspace_id = ?1 AND w.state = 'active'
+           AND r.runtime_id = ?2 AND r.revoked_at IS NULL
+           AND r.authentication_mode = 'workspace_identity' AND s.state = 'active')",
+        params![seed.workspace_id, seed.runtime_id],
+        |row| row.get(0),
+    )?;
+    if !active {
+        return Err(Error::InvalidInput(
+            "fixture requires an active bootstrapped Workspace and authenticated Runtime binding"
+                .into(),
+        ));
+    }
+    let worker_id = seed.worker_id.to_string();
+    let allocation_exists: bool = tx.query_row(
+        "SELECT EXISTS(SELECT 1 FROM worker_create_reservations
+         WHERE workspace_id = ?1 AND allocation_key = ?2)",
+        params![seed.workspace_id, seed.allocation_key],
+        |row| row.get(0),
+    )?;
+    if allocation_exists {
+        return Err(Error::InvalidInput(
+            "fixture Worker allocation already exists".into(),
+        ));
+    }
+    // Check by WorkerId, not just the proposed Runtime tuple: orphaned and terminal
+    // ownership is still a collision. These tables are owned by this store schema.
+    for table in [
+        "worker_create_reservations",
+        "worker_registry",
+        "worker_registry_observations",
+        "worker_registry_projection_removals",
+        "worker_removal_operations",
+        "worker_singleton_owners",
+        "worker_workdir_links",
+        "worker_restore_intents",
+        "worker_tombstones",
+        "worker_orphan_diagnostics",
+        "worker_registry_projection_diagnostics",
+        "worker_session_archives",
+        "worker_diagnostics_archives",
+        "ticket_assignment_worker_tombstones",
+    ] {
+        let exists: bool = tx.query_row(
+            &format!("SELECT EXISTS(SELECT 1 FROM {table} WHERE worker_id = ?1)"),
+            [&worker_id],
+            |row| row.get(0),
+        )?;
+        if exists {
+            return Err(Error::InvalidInput(format!(
+                "fixture Worker identity already exists in {table}"
+            )));
+        }
+    }
+    // A resource key without a catalog/reservation must not be silently reused.
+    let mut statement = tx.prepare(
+        "SELECT resource_id FROM workspace_resource_keys WHERE resource_kind = 'worker'",
+    )?;
+    let resources = statement.query_map([], |row| row.get::<_, String>(0))?;
+    for resource in resources {
+        let resource = resource?;
+        let existing = resource
+            .is_ascii()
+            .then(|| parse_worker_resource_id(&resource))
+            .flatten()
+            .ok_or_else(|| {
+                Error::InvalidInput("unknown fixture Worker resource identity collision".into())
+            })?;
+        if existing.worker_id == worker_id {
+            return Err(Error::InvalidInput(
+                "fixture Worker resource identity already exists".into(),
+            ));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::hosts::{
+        WorkerSpawnAcceptanceRequirement, WorkerSpawnWorkingDirectoryAttachmentRequest,
+        WorkerSpawnWorkingDirectoryRequest,
+    };
+    use crate::workspace_signing_identity::{
+        InMemoryWorkspaceSigningMaterialStore, WorkspaceSigningIdentityService,
+    };
+
+    const WORKSPACE: &str = "0197a949-4b6b-7f2a-9d9a-1f87e3a4c501";
+    const RUNTIME: &str = "fixture-runtime";
+    const WORKER: &str = "0197a949-4b6b-7f2a-9d9a-1f87e3a4c502";
+
+    // Exercises the same public bootstrap APIs the offline harness uses. All account
+    // and key identities are newly generated, even across stores with fixed fixture IDs.
+    fn bootstrap() -> SqliteWorkspaceStore {
+        let store = SqliteWorkspaceStore::in_memory_for_test().unwrap();
+        let account = AccountRecord {
+            account_id: format!("fixture-account-{}", Uuid::now_v7()),
+            kind: "user".into(),
+            handle: format!("fixture-{}", Uuid::now_v7()),
+            display_name: "Fixture owner".into(),
+            created_at: "2026-10-24T00:00:00Z".into(),
+            updated_at: "2026-10-24T00:00:00Z".into(),
+        };
+        store.upsert_account(&account).unwrap();
+        let identities = WorkspaceSigningIdentityService::new(
+            Arc::new(store.clone()),
+            Arc::new(InMemoryWorkspaceSigningMaterialStore::default()),
+        );
+        let (identity, operation) = identities
+            .prepare_workspace_creation(
+                "fixture-bootstrap",
+                "fixture-bootstrap",
+                WORKSPACE,
+                &account.account_id,
+            )
+            .unwrap();
+        store
+            .create_workspace_bootstrap(
+                &WorkspaceBootstrapRecord {
+                    operation_key: "fixture-bootstrap".into(),
+                    request_fingerprint: "fixture-bootstrap".into(),
+                    workspace: WorkspaceRecord {
+                        workspace_id: WORKSPACE.into(),
+                        owner_account_id: account.account_id,
+                        display_name: "Fixture".into(),
+                        state: "active".into(),
+                        created_at: account.created_at,
+                        updated_at: account.updated_at,
+                    },
+                    repository: None,
+                },
+                &identity,
+                &operation,
+            )
+            .unwrap();
+        let runtime_key = worker_runtime::auth::RuntimeIdentityMaterial::generate(RUNTIME).unwrap();
+        store
+            .upsert_workspace_runtime_binding(
+                WorkspaceRuntimeBinding {
+                    workspace_id: WORKSPACE.into(),
+                    runtime_id: RUNTIME.into(),
+                    display_name: "Fixture Runtime".into(),
+                    base_url: "http://127.0.0.1:12345".into(),
+                    public_key: runtime_key.public_key,
+                    public_key_fingerprint: String::new(),
+                    binding_id: String::new(),
+                    authentication_mode: WorkspaceRuntimeAuthenticationMode::WorkspaceIdentity,
+                    created_at: "2026-10-24T00:00:00Z".into(),
+                    updated_at: "2026-10-24T00:00:00Z".into(),
+                    revoked_at: None,
+                },
+                false,
+            )
+            .unwrap();
+        store
+    }
+
+    fn request() -> WorkerSpawnRequest {
+        WorkerSpawnRequest {
+            intent: WorkerSpawnIntent::WorkspaceCoding,
+            requested_worker_name: Some("Fixture Worker".into()),
+            singleton_key: None,
+            acceptance: WorkerSpawnAcceptanceRequirement::SocketReady,
+            profile: ProfileSelector::Builtin("builtin:coder".into()),
+            ticket_assignment: None,
+            initial_submit: Vec::new(),
+            workdir_attachment_requests: Vec::new(),
+            resolved_workdir_attachment_requests: Vec::new(),
+            resolved_workdir_attachments: Vec::new(),
+            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::new(),
+            resolved_control_operation: None,
+            resolved_execution_grant: None,
+        }
+    }
+
+    fn seed(request: &WorkerSpawnRequest) -> FixedWorkerSeed<'_> {
+        FixedWorkerSeed {
+            workspace_id: WORKSPACE,
+            runtime_id: RUNTIME,
+            worker_id: WORKER.parse().unwrap(),
+            allocation_key: "fixture-allocation",
+            request,
+        }
+    }
+
+    fn worker() -> RuntimeWorkerRef {
+        RuntimeWorkerRef::new(RUNTIME, WORKER)
+    }
+
+    #[test]
+    fn fixed_ids_are_isolated_and_seed_has_no_runtime_observation() {
+        let a = bootstrap();
+        let b = bootstrap();
+        let request = request();
+        let first = seed_fixed_worker(&a, seed(&request)).unwrap();
+        let second = seed_fixed_worker(&b, seed(&request)).unwrap();
+        assert_eq!(first, second);
+        assert_eq!(first.worker_id.to_string(), WORKER);
+        assert_eq!(first.memory_settings.workspace_id, WORKSPACE);
+        assert_eq!(first.memory_settings.language, "English");
+        assert!(first.singleton.is_none());
+        assert_ne!(
+            a.get_workspace_signing_identity(WORKSPACE)
+                .unwrap()
+                .unwrap()
+                .public_key,
+            b.get_workspace_signing_identity(WORKSPACE)
+                .unwrap()
+                .unwrap()
+                .public_key
+        );
+        assert_ne!(
+            a.get_workspace_runtime_binding(WORKSPACE, RUNTIME)
+                .unwrap()
+                .unwrap()
+                .public_key,
+            b.get_workspace_runtime_binding(WORKSPACE, RUNTIME)
+                .unwrap()
+                .unwrap()
+                .public_key
+        );
+        for store in [&a, &b] {
+            assert_eq!(
+                store
+                    .worker_resource_key(WORKSPACE, &worker())
+                    .unwrap()
+                    .as_deref(),
+                Some("W-1")
+            );
+            assert_eq!(
+                store
+                    .resolve_worker_resource_reference(WORKSPACE, "W-1")
+                    .unwrap(),
+                Some(worker())
+            );
+            assert!(
+                !store
+                    .has_active_worker_create_reservation(WORKSPACE, &worker())
+                    .unwrap()
+            );
+            let registry = store
+                .get_worker_registry(WORKSPACE, &worker())
+                .unwrap()
+                .unwrap();
+            assert_eq!(registry.profile.as_deref(), Some("builtin:coder"));
+            assert_eq!(registry.display_name, "Fixture Worker");
+            assert_eq!(registry.retention_state, "normal");
+            let projection = store
+                .with_conn(|conn| read_worker_registry_projection(conn, WORKSPACE, &worker()))
+                .unwrap()
+                .unwrap();
+            assert!(projection.observation.is_none());
+            assert!(projection.job.is_none());
+        }
+    }
+
+    #[test]
+    fn initial_submit_is_refused_without_registration_or_allocation() {
+        let store = bootstrap();
+        let mut request = request();
+        // Even an empty text segment is new input; the vector itself must be empty.
+        for text in ["new input must not be seeded", ""] {
+            request.initial_submit = vec![protocol::Segment::text(text)];
+            let error = seed_fixed_worker(&store, seed(&request)).unwrap_err();
+            assert!(
+                matches!(error, Error::InvalidInput(ref message) if message.contains("initial_submit"))
+            );
+            assert!(
+                store
+                    .get_worker_registry(WORKSPACE, &worker())
+                    .unwrap()
+                    .is_none()
+            );
+            assert!(
+                store
+                    .worker_resource_key(WORKSPACE, &worker())
+                    .unwrap()
+                    .is_none()
+            );
+            assert!(
+                !store
+                    .has_active_worker_create_reservation(WORKSPACE, &worker())
+                    .unwrap()
+            );
+        }
+        // The rejected attempts must not consume the allocation or resource counter.
+        request.initial_submit.clear();
+        seed_fixed_worker(&store, seed(&request)).unwrap();
+        assert_eq!(
+            store
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .as_deref(),
+            Some("W-1")
+        );
+    }
+
+    #[test]
+    fn exact_replay_changed_input_and_cross_runtime_identity_are_refused() {
+        let store = bootstrap();
+        let mut request = request();
+        seed_fixed_worker(&store, seed(&request)).unwrap();
+        let before = store.get_worker_registry(WORKSPACE, &worker()).unwrap();
+        assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        request.requested_worker_name = Some("Must not overwrite".into());
+        assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        let mut new_allocation = seed(&request);
+        new_allocation.allocation_key = "another-allocation";
+        assert!(seed_fixed_worker(&store, new_allocation).is_err());
+        let mut same_allocation = seed(&request);
+        same_allocation.worker_id = WorkerId::now_v7();
+        assert!(seed_fixed_worker(&store, same_allocation).is_err());
+        assert_eq!(
+            store.get_worker_registry(WORKSPACE, &worker()).unwrap(),
+            before
+        );
+
+        let other = bootstrap();
+        let mut record = before.unwrap();
+        record.worker.runtime_id = "different-runtime".into();
+        other.upsert_worker_registry(&record).unwrap();
+        assert!(seed_fixed_worker(&other, seed(&request)).is_err());
+        assert!(
+            other
+                .get_worker_registry(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn reserved_and_orphaned_resource_identities_are_not_adopted() {
+        let store = bootstrap();
+        let request = request();
+        let memory = store.get_workspace_memory_settings(WORKSPACE).unwrap();
+        let ordinary = store
+            .reserve_worker_create(WORKSPACE, RUNTIME, "ordinary", "fingerprint", None, &memory)
+            .unwrap();
+        let mut collision = seed(&request);
+        collision.worker_id = ordinary.worker_id;
+        assert!(seed_fixed_worker(&store, collision).is_err());
+        // Store-level fault setup only: the external harness never writes SQL.
+        store
+            .with_conn(|conn| {
+                allocate_resource_key(
+                    conn,
+                    WORKSPACE,
+                    WorkspaceResourceKind::Worker,
+                    &worker_resource_id(&worker()),
+                    "now",
+                )?;
+                Ok(())
+            })
+            .unwrap();
+        assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        assert!(
+            store
+                .get_worker_registry(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn unknown_resource_identity_and_retained_removal_fence_are_refused() {
+        let request = request();
+        let malformed = bootstrap();
+        malformed
+            .with_conn(|conn| {
+                allocate_resource_key(
+                    conn,
+                    WORKSPACE,
+                    WorkspaceResourceKind::Worker,
+                    "unknown-identity",
+                    "now",
+                )?;
+                Ok(())
+            })
+            .unwrap();
+        assert!(matches!(
+            seed_fixed_worker(&malformed, seed(&request)),
+            Err(Error::InvalidInput(_))
+        ));
+        assert!(
+            malformed
+                .get_worker_registry(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+
+        let removed = bootstrap();
+        removed.with_conn(|conn| {
+            conn.execute(
+                "INSERT INTO worker_registry_projection_removals(workspace_id, runtime_id, worker_id) VALUES (?1, ?2, ?3)",
+                params![WORKSPACE, RUNTIME, WORKER],
+            )?;
+            Ok(())
+        }).unwrap();
+        assert!(matches!(
+            seed_fixed_worker(&removed, seed(&request)),
+            Err(Error::InvalidInput(_))
+        ));
+        assert!(
+            removed
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn unknown_workspace_runtime_and_revoked_binding_are_refused() {
+        let store = bootstrap();
+        let request = request();
+        let mut unknown = seed(&request);
+        unknown.workspace_id = "unknown";
+        assert!(seed_fixed_worker(&store, unknown).is_err());
+        let mut unknown = seed(&request);
+        unknown.runtime_id = "unknown";
+        assert!(seed_fixed_worker(&store, unknown).is_err());
+        store
+            .revoke_workspace_runtime_binding(WORKSPACE, RUNTIME, "2026-10-24T01:00:00Z")
+            .unwrap();
+        assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        assert!(
+            store
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn unsupported_intent_is_refused_without_partial_registration() {
+        let store = bootstrap();
+        let base = request();
+        let mut requests = Vec::new();
+        let mut request = base.clone();
+        request.singleton_key = Some("fixture-singleton".into());
+        requests.push(request);
+        let mut request = base.clone();
+        request.intent = WorkerSpawnIntent::WorkspaceOrchestrator;
+        requests.push(request);
+        let mut request = base.clone();
+        request.resolved_subjektiv_attached = true;
+        requests.push(request);
+        let mut request = base.clone();
+        request
+            .workdir_attachment_requests
+            .push(WorkerSpawnWorkingDirectoryAttachmentRequest {
+                alias: workdir::WorkdirAttachmentAlias::new("repo").unwrap(),
+                working_directory: WorkerSpawnWorkingDirectoryRequest {
+                    repository_key: "repo".into(),
+                    selector: None,
+                },
+            });
+        requests.push(request);
+        for request in requests {
+            assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        }
+        assert!(
+            store
+                .get_worker_registry(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+        assert!(
+            store
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+        assert_eq!(
+            seed_fixed_worker(&store, seed(&base))
+                .unwrap()
+                .worker_id
+                .to_string(),
+            WORKER
+        );
+    }
+
+    #[test]
+    fn late_catalog_failure_rolls_back_reservation_and_resource_allocation() {
+        let store = bootstrap();
+        let request = request();
+        store
+            .with_conn(|conn| {
+                conn.execute_batch(
+                    "CREATE TEMP TRIGGER fail_fixture_catalog BEFORE INSERT ON worker_registry
+                                BEGIN SELECT RAISE(ABORT, 'injected catalog failure'); END",
+                )?;
+                Ok(())
+            })
+            .unwrap();
+        assert!(seed_fixed_worker(&store, seed(&request)).is_err());
+        assert!(
+            !store
+                .has_active_worker_create_reservation(WORKSPACE, &worker())
+                .unwrap()
+        );
+        assert!(
+            store
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .is_none()
+        );
+        store
+            .with_conn(|conn| {
+                conn.execute_batch("DROP TRIGGER fail_fixture_catalog")?;
+                Ok(())
+            })
+            .unwrap();
+        seed_fixed_worker(&store, seed(&request)).unwrap();
+        assert_eq!(
+            store
+                .worker_resource_key(WORKSPACE, &worker())
+                .unwrap()
+                .as_deref(),
+            Some("W-1")
+        );
+    }
+
+    #[test]
+    fn ordinary_creation_still_allocates_and_replays_without_catalog_insertion() {
+        let store = bootstrap();
+        let memory = store.get_workspace_memory_settings(WORKSPACE).unwrap();
+        let reserve = |key| {
+            store
+                .reserve_worker_create(WORKSPACE, RUNTIME, key, "fingerprint", None, &memory)
+                .unwrap()
+        };
+        let first = reserve("ordinary-one");
+        assert_eq!(first, reserve("ordinary-one"));
+        let second = reserve("ordinary-two");
+        assert_ne!(first.worker_id, second.worker_id);
+        assert_ne!(first.worker_id.to_string(), WORKER);
+        assert_eq!(
+            first.worker_id.as_uuid().get_version(),
+            Some(uuid::Version::SortRand)
+        );
+        let worker = RuntimeWorkerRef::new(RUNTIME, first.worker_id.to_string());
+        assert!(
+            store
+                .has_active_worker_create_reservation(WORKSPACE, &worker)
+                .unwrap()
+        );
+        assert!(
+            store
+                .get_worker_registry(WORKSPACE, &worker)
+                .unwrap()
+                .is_none()
+        );
+        assert!(
+            store
+                .reserve_worker_create(WORKSPACE, RUNTIME, "ordinary-one", "changed", None, &memory)
+                .is_err()
+        );
+    }
+}