fix: adapt T-769 recovery onto current restore lane checkpoint

UNCOMPILED TRANSPORT / PRESERVATION CHECKPOINT, NOT A VALIDATED CANDIDATE. One-time recovery authorization permits publication before validation so the parent can validate this exact source in a separate Workdir. Prior compilation attempts were interrupted by Workdir-session unavailability; cause and test outcome remain unknown. No Cargo build/check/test was started in this bounded checkpoint turn.

Merge current lane 4ef7b03651 into existing source b7a22146a0. Selectively adapt typed factory failure and same-operation cleanup/replay while preserving current saved Feature restore and operation-time authorization, logical/deferred Workdirs, T765 Host/schema and T768 replay-safe Subject grant. No T770/T771 implementation.

All 19 target-relative paths and added test bodies were source-audited. Workspace fmt check, scoped rustfmt, whitespace/diff checks passed; conflict markers zero. Excluded unrelated formatter changes and untracked compiler artifacts. Local audit: target/t769-checkpoint/AUDIT.md.

Handoff to parent: compile and run focused and package/API regressions on this exact source, fix as necessary, then obtain fresh independent review. Existing MR 01a15606-a6ba-7d23-8d60-0874607399e7 only; no new review requested here. No acceptance, approval, adoption, target/develop push, integration, Ticket closure or service update is asserted.

crates/worker-runtime/Cargo.toml

@@ -56,6 +56,7 @@ server-api = { path = "../server-api" }
 workdir.workspace = true
 
 [dev-dependencies]
+workdir = { workspace = true, features = ["http-client"] }
 futures.workspace = true
 agen.workspace = true
 serial_test = "3.4.0"

crates/worker-runtime/src/execution.rs

@@ -432,6 +432,9 @@ pub struct WorkerExecutionRestoreRequest {
     pub previous_workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
     pub logical_workdir_attachments: Vec<crate::catalog::LogicalWorkdirAttachment>,
     pub workdir_attachments: BTreeMap<workdir::WorkdirAttachmentAlias, WorkingDirectoryBinding>,
+    /// Process-owned sessions that open providers only on a Workdir operation.
+    pub deferred_workdir_sessions:
+        BTreeMap<workdir::WorkdirAttachmentAlias, workdir::WorkdirSessionHandle>,
     pub config_bundle: Option<ConfigBundle>,
 }
 

crates/worker-runtime/src/fs_store.rs

@@ -657,18 +657,32 @@ fn plan_runtime_store_migration(
         ));
     }
     if current_schema_version == SCHEMA_VERSION {
+        // Runtime and Session schemas evolve independently. Probe only through
+        // Session Store's non-mutating current-schema admission boundary; any
+        // conversion still runs on the existing owner-locked staging tree.
+        let migration_required = crate::retention::sessions_require_migration(root)?;
+        let excluded_ephemeral_paths = if migration_required {
+            runtime_tree_exclusions(root)?
+        } else {
+            Vec::new()
+        };
+        let archived_orphan_worker_count = if migration_required {
+            orphaned_worker_directories(root)?.len()
+        } else {
+            0
+        };
         let plan = FsRuntimeStoreMigrationPlan {
             current_schema_version,
             target_schema_version: SCHEMA_VERSION,
-            migration_required: false,
+            migration_required,
             worker_count: 0,
-            archived_orphan_worker_count: 0,
+            archived_orphan_worker_count,
             migrated_worker_aggregate_count: 0,
             migrated_diagnostic_worker_ref_count: 0,
             cleared_diagnostic_worker_ref_count: 0,
             mapping_digest: legacy_worker_identity_mapping_digest(&[]),
             mappings: Vec::new(),
-            excluded_ephemeral_paths: Vec::new(),
+            excluded_ephemeral_paths,
         };
         return Ok((plan, Vec::new()));
     }
@@ -1808,7 +1822,9 @@ fn migrate_runtime_store(
     Ok(plan)
 }
 
-fn archive_orphaned_worker_directories(root: &Path) -> Result<usize, RuntimeError> {
+fn orphaned_worker_directories(
+    root: &Path,
+) -> Result<Vec<(std::ffi::OsString, PathBuf)>, RuntimeError> {
     let workers_dir = root.join(WORKERS_DIR);
     let mut orphaned = fs::read_dir(&workers_dir)
         .map_err(|source| runtime_io_error("read Worker records", &workers_dir, source))?
@@ -1821,6 +1837,11 @@ fn archive_orphaned_worker_directories(root: &Path) -> Result<usize, RuntimeErro
         })
         .collect::<Vec<_>>();
     orphaned.sort_by(|(left, _), (right, _)| left.cmp(right));
+    Ok(orphaned)
+}
+
+fn archive_orphaned_worker_directories(root: &Path) -> Result<usize, RuntimeError> {
+    let orphaned = orphaned_worker_directories(root)?;
     if orphaned.is_empty() {
         return Ok(0);
     }
@@ -1855,6 +1876,21 @@ fn migrate_runtime_store_in_place(
     if !plan.migration_required {
         return Ok(plan);
     }
+    if plan.current_schema_version == SCHEMA_VERSION {
+        // Session-only migration: do not reserialize Runtime/Worker identity,
+        // Restore guards, recovery fences, or retained execution receipts.
+        // As in a Runtime schema migration, retain orphan trees outside the
+        // Worker catalog before validating the completed staging tree.
+        let archived = archive_orphaned_worker_directories(root)?;
+        if archived != plan.archived_orphan_worker_count {
+            return Err(runtime_store_corrupt(
+                root,
+                "orphan count changed during Session migration".into(),
+            ));
+        }
+        crate::retention::migrate_staged_sessions(root)?;
+        return Ok(plan);
+    }
     let runtime_path = root.join(RUNTIME_FILE);
     let bytes =
         fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?;
@@ -1948,11 +1984,9 @@ fn migrate_runtime_store_in_place(
             ),
         ));
     }
-    // Schema 11 already migrated Sessions; the recovery-fence bump must not
-    // rewrite retained history or archive attestations.
-    if plan.current_schema_version != SOURCE_SCHEMA_VERSION {
-        crate::retention::migrate_staged_sessions(root)?;
-    }
+    // A Runtime schema bump says nothing about the retained Session schema.
+    // Session Store owns versioned conversion, including v4 -> generic v5.
+    crate::retention::migrate_staged_sessions(root)?;
     let (document, _) = migrate_runtime_document(
         document,
         plan.current_schema_version,
@@ -3011,9 +3045,8 @@ mod tests {
             } else {
                 drop(result.unwrap());
                 assert_eq!(stored_schema_version(&root), u64::from(SCHEMA_VERSION));
-                let manifest: serde_json::Value =
-                    read_json(&session.join("session.json"), "test").unwrap();
-                assert_eq!(manifest["schema_version"], 4);
+                session_store::WorkerSessionStore::open_read_only(&session)
+                    .expect("startup migration must admit the current Session schema");
                 let once = fs::read(session.join("session.json")).unwrap();
                 drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap());
                 assert_eq!(fs::read(session.join("session.json")).unwrap(), once);

crates/worker-runtime/src/http_server.rs

@@ -67,7 +67,7 @@ use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use tokio::net::TcpListener;
 use workdir::{
-    CommandOutput, CommandStatus, WorkdirSessionHandle, dispatch_workdir_session_operation,
+    CommandOutput, WorkdirSessionHandle, dispatch_workdir_session_operation,
     http::{
         OpenWorkdirSessionRequest, OpenWorkdirSessionResponse, WorkdirSessionId,
         WorkdirSessionOperation, WorkdirSessionOperationRequest, WorkdirSessionOperationResult,
@@ -914,13 +914,9 @@ async fn run_workdir_session_operation(
             {
                 Ok(result) => result?,
                 Err(_) => CommandOutput {
-                    exit_code: None,
-                    timed_out: false,
-                    status: CommandStatus::Running,
-                    content: String::new(),
-                    next_cursor: Some(cursor),
-                    truncated: false,
-                    output_path: None,
+                    chunks: Vec::new(),
+                    next_cursor: cursor,
+                    terminal: None,
                 },
             };
             WorkdirSessionOperationResult::CommandOutput(output)
@@ -2426,6 +2422,7 @@ pub enum RuntimeHttpServerError {
 
 #[cfg(test)]
 mod tests {
+    mod command_transport_tests;
     #[cfg(feature = "fs-store")]
     mod session_material_tests;
 

crates/worker-runtime/src/http_server/tests/command_transport_tests.rs

@@ -0,0 +1,408 @@
+//! Loopback integration: production RemoteWorkdirSession, signed HTTP middleware,
+//! command operation handler, local process pipes, and JSON DTOs. Only opening the
+//! pre-seeded session is a fixture; no Runtime Worker/materializer is claimed here.
+
+use super::*;
+use std::time::Duration;
+use workdir::http::{
+    RemoteWorkdirSession, WorkdirHttpAuthorization, WorkdirHttpAuthorizationFuture,
+};
+use workdir::{CommandOutputRequest, CommandRequest, CommandStream, WorkdirSession};
+
+struct SignedAuthorization {
+    identity: RuntimeIdentityMaterial,
+    trust: WorkspaceIssuerTrustRecord,
+}
+impl std::fmt::Debug for SignedAuthorization {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("SignedAuthorization")
+    }
+}
+impl WorkdirHttpAuthorization for SignedAuthorization {
+    fn bearer_token(
+        &self,
+        method: String,
+        path: String,
+        body: Vec<u8>,
+    ) -> WorkdirHttpAuthorizationFuture {
+        let claims = signed_claims(&self.trust, &method, &path, &body);
+        let token =
+            issue_workspace_capability_token(&self.identity.signing_key().unwrap(), &claims)
+                .unwrap();
+        Box::pin(async move { Ok(token) })
+    }
+}
+
+struct Harness {
+    server: tokio::task::JoinHandle<()>,
+    session: WorkdirSessionHandle,
+    _root: tempfile::TempDir,
+    remote: RemoteWorkdirSession,
+    client: reqwest::Client,
+    operation_url: url::Url,
+    authorization: Arc<SignedAuthorization>,
+}
+impl Drop for Harness {
+    fn drop(&mut self) {
+        self.server.abort();
+    }
+}
+impl Harness {
+    async fn raw_operation(&self, operation: serde_json::Value) -> reqwest::Response {
+        let body = serde_json::to_vec(&serde_json::json!({"operation": operation})).unwrap();
+        let token = self
+            .authorization
+            .bearer_token(
+                "POST".into(),
+                self.operation_url.path().into(),
+                body.clone(),
+            )
+            .await
+            .unwrap();
+        self.client
+            .post(self.operation_url.clone())
+            .bearer_auth(token)
+            .header("content-type", "application/json")
+            .body(body)
+            .send()
+            .await
+            .unwrap()
+    }
+
+    async fn new() -> Self {
+        let root = tempfile::tempdir().unwrap();
+        let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
+            Workdir::new("wd-transport"),
+            root.path().to_path_buf(),
+            root.path().to_path_buf(),
+            SharedScope::new(Scope::writable(root.path()).unwrap()),
+            WorkdirSessionCapabilities::ALL,
+        ));
+        let identity = RuntimeIdentityMaterial::generate("transport-key").unwrap();
+        let trust = workspace_trust(&identity);
+        let state = RuntimeHttpState {
+            runtime: Runtime::new_memory(),
+            local_token: None,
+            workspace_auth: Some(Arc::new(workspace_test_auth(vec![trust.clone()]))),
+            workdir_sessions: Arc::new(Mutex::new(HashMap::from([(
+                "session-transport".into(),
+                RuntimeHttpWorkdirSession {
+                    owner: RuntimeWorkspaceScope::new("workspace-a", "https://backend.test"),
+                    session: session.clone(),
+                },
+            )]))),
+            ssh_keyscan_program: Arc::new(PathBuf::from("ssh-keyscan")),
+        };
+        let app = Router::new()
+            .route(
+                "/v1/working-directories/{id}/sessions",
+                post(|| async {
+                    Json(OpenWorkdirSessionResponse {
+                        session_id: WorkdirSessionId::new("session-transport").unwrap(),
+                        workdir_id: Workdir::new("wd-transport").id().clone(),
+                        capabilities: WorkdirSessionCapabilities::ALL,
+                    })
+                }),
+            )
+            .route(
+                "/v1/workdir-sessions/{session_id}/operations",
+                post(run_workdir_session_operation),
+            )
+            .route(
+                "/v1/workdir-sessions/{session_id}",
+                delete(close_workdir_session),
+            )
+            .with_state(state.clone())
+            .layer(middleware::from_fn_with_state(state, require_runtime_auth));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base: url::Url = format!("http://{}", listener.local_addr().unwrap())
+            .parse()
+            .unwrap();
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        let mut headers = reqwest::header::HeaderMap::new();
+        headers.insert(
+            RUNTIME_WORKSPACE_SCOPE_HEADER,
+            "workspace-a".parse().unwrap(),
+        );
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .default_headers(headers)
+            .build()
+            .unwrap();
+        let authorization = Arc::new(SignedAuthorization { identity, trust });
+        let operation_url = base
+            .join("/v1/workdir-sessions/session-transport/operations")
+            .unwrap();
+        let remote = RemoteWorkdirSession::open_with_authorization(
+            client.clone(),
+            base,
+            authorization.clone(),
+            Workdir::new("wd-transport").id().clone(),
+            OpenWorkdirSessionRequest::default(),
+        )
+        .await
+        .unwrap();
+        Self {
+            server,
+            session,
+            _root: root,
+            remote,
+            client,
+            operation_url,
+            authorization,
+        }
+    }
+}
+
+#[tokio::test]
+async fn real_http_command_pages_preserve_binary_streams_replay_and_nonzero_exit() {
+    tokio::time::timeout(Duration::from_secs(20), async {
+        let harness = Harness::new().await;
+        let handle = harness.remote.start_command(CommandRequest {
+            transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES,
+            command: "head -c 2097152 /dev/zero | tr '\\0' x; printf '\\342\\202\\254\\377\\000'; printf 'stderr\\346\\227\\245' >&2; exit 9".into(),
+            timeout_secs: 15, cwd: WorkdirPath::root(), tool_call_id: Some("http-tool-call".into()),
+        }).await.unwrap();
+        let mut cursor = 0;
+        let mut stdout = Vec::new();
+        let mut stderr = Vec::new();
+        let terminal = loop {
+            let request = CommandOutputRequest { handle: handle.clone(), cursor, limit: 16384, wait: true };
+            let page = harness.remote.command_output(request.clone()).await.unwrap();
+            assert_eq!(harness.remote.command_output(request).await.unwrap(), page);
+            assert!(page.chunks.iter().map(|chunk| chunk.bytes.len()).sum::<usize>() <= 16384);
+            for chunk in page.chunks {
+                assert_eq!(chunk.sequence, cursor);
+                cursor += chunk.bytes.len() as u64;
+                let target = match chunk.stream { CommandStream::Stdout => &mut stdout, CommandStream::Stderr => &mut stderr };
+                assert_eq!(chunk.offset, target.len() as u64);
+                target.extend(chunk.bytes);
+            }
+            assert_eq!(cursor, page.next_cursor);
+            if let Some(terminal) = page.terminal { break terminal; }
+        };
+        let mut expected = vec![b'x'; 2097152];
+        expected.extend_from_slice(b"\xe2\x82\xac\xff\0");
+        assert_eq!(stdout, expected);
+        assert_eq!(stderr, b"stderr\xe6\x97\xa5");
+        assert_eq!(terminal.status, workdir::CommandStatus::Failed);
+        assert_eq!(terminal.exit_code, Some(9));
+        assert!(harness.remote.command_output(CommandOutputRequest { handle, cursor: 0, limit: 1, wait: false }).await.is_err());
+        harness.remote.close().await.unwrap();
+        assert!(harness.session.command_snapshot().is_empty());
+    }).await.expect("bounded HTTP command transport");
+}
+
+#[tokio::test]
+async fn real_http_cancel_drains_tail_before_terminal_and_timeout_is_distinct() {
+    tokio::time::timeout(Duration::from_secs(15), async {
+        let harness = Harness::new().await;
+        for (command, timeout, cancel, expected) in [
+            (
+                "printf ready; exec yes x",
+                10,
+                true,
+                workdir::CommandStatus::Cancelled,
+            ),
+            (
+                "printf ready; exec tail -f /dev/null",
+                1,
+                false,
+                workdir::CommandStatus::TimedOut,
+            ),
+        ] {
+            let handle = harness
+                .remote
+                .start_command(CommandRequest {
+                    transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES,
+                    command: command.into(),
+                    timeout_secs: timeout,
+                    cwd: WorkdirPath::root(),
+                    tool_call_id: None,
+                })
+                .await
+                .unwrap();
+            let mut cursor = 0;
+            let first = harness
+                .remote
+                .command_output(CommandOutputRequest {
+                    handle: handle.clone(),
+                    cursor,
+                    limit: 5,
+                    wait: true,
+                })
+                .await
+                .unwrap();
+            assert_eq!(
+                first
+                    .chunks
+                    .iter()
+                    .flat_map(|chunk| chunk.bytes.iter().copied())
+                    .collect::<Vec<_>>(),
+                b"ready"
+            );
+            cursor = first.next_cursor;
+            if cancel {
+                harness.remote.cancel_command(handle.clone()).await.unwrap();
+            }
+            loop {
+                let page = harness
+                    .remote
+                    .command_output(CommandOutputRequest {
+                        handle: handle.clone(),
+                        cursor,
+                        limit: workdir::MAX_COMMAND_READ_BYTES,
+                        wait: true,
+                    })
+                    .await
+                    .unwrap();
+                cursor = page.next_cursor;
+                if let Some(terminal) = page.terminal {
+                    assert_eq!(terminal.status, expected);
+                    break;
+                }
+            }
+        }
+        harness.remote.close().await.unwrap();
+    })
+    .await
+    .expect("bounded cancellation and timeout over HTTP");
+}
+
+#[tokio::test]
+async fn real_http_command_budget_bounds_unique_payload_and_reports_larger_seen_totals() {
+    tokio::time::timeout(Duration::from_secs(20), async {
+        let harness = Harness::new().await;
+        for budget in [0, 12345] {
+            let handle = harness
+                .remote
+                .start_command(CommandRequest {
+                    command: "head -c 1048576 /dev/zero; head -c 65536 /dev/zero >&2; exit 9"
+                        .into(),
+                    timeout_secs: 10,
+                    transport_byte_budget: budget,
+                    cwd: WorkdirPath::root(),
+                    tool_call_id: None,
+                })
+                .await
+                .unwrap();
+            let mut cursor = 0;
+            let mut delivered = [0_u64; 2];
+            let terminal = loop {
+                let request = CommandOutputRequest {
+                    handle: handle.clone(),
+                    cursor,
+                    limit: 997,
+                    wait: true,
+                };
+                let page = harness
+                    .remote
+                    .command_output(request.clone())
+                    .await
+                    .unwrap();
+                assert_eq!(harness.remote.command_output(request).await.unwrap(), page);
+                for chunk in page.chunks {
+                    assert_eq!(chunk.sequence, cursor);
+                    cursor += chunk.bytes.len() as u64;
+                    let index = match chunk.stream {
+                        CommandStream::Stdout => 0,
+                        CommandStream::Stderr => 1,
+                    };
+                    assert_eq!(chunk.offset, delivered[index]);
+                    delivered[index] += chunk.bytes.len() as u64;
+                }
+                assert_eq!(cursor, page.next_cursor);
+                assert!(
+                    cursor <= budget,
+                    "RemoteWorkdirSession must not receive bytes beyond budget"
+                );
+                if let Some(terminal) = page.terminal {
+                    break terminal;
+                }
+            };
+            assert_eq!(cursor, budget);
+            assert_eq!(terminal.stdout_bytes, 1048576);
+            assert_eq!(terminal.stderr_bytes, 65536);
+            assert_eq!(terminal.stdout_delivered_bytes, delivered[0]);
+            assert_eq!(terminal.stderr_delivered_bytes, delivered[1]);
+            assert_eq!(
+                terminal.truncation_reason,
+                Some(workdir::CommandOutputTruncation::ByteBudget)
+            );
+            assert_eq!(terminal.status, workdir::CommandStatus::Failed);
+            assert_eq!(terminal.exit_code, Some(9));
+            assert!(
+                harness
+                    .remote
+                    .command_output(CommandOutputRequest {
+                        handle: handle.clone(),
+                        cursor: budget + 1,
+                        limit: 1,
+                        wait: false,
+                    })
+                    .await
+                    .is_err()
+            );
+            if budget > 0 {
+                assert!(
+                    harness
+                        .remote
+                        .command_output(CommandOutputRequest {
+                            handle,
+                            cursor: 0,
+                            limit: 1,
+                            wait: false,
+                        })
+                        .await
+                        .is_err()
+                );
+            }
+        }
+        harness.remote.close().await.unwrap();
+    })
+    .await
+    .expect("bounded HTTP byte-budget transport");
+}
+
+#[tokio::test]
+async fn real_http_rejects_invalid_or_legacy_budget_requests_before_process_start() {
+    tokio::time::timeout(Duration::from_secs(10), async {
+        let harness = Harness::new().await;
+        let request = CommandRequest {
+            command: "touch must-not-exist".into(),
+            timeout_secs: 1,
+            transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES + 1,
+            cwd: WorkdirPath::root(),
+            tool_call_id: None,
+        };
+        let oversize =
+            serde_json::to_value(WorkdirSessionOperation::CommandStart(request)).unwrap();
+        assert_eq!(
+            harness.raw_operation(oversize.clone()).await.status(),
+            reqwest::StatusCode::BAD_REQUEST
+        );
+        let mut missing = oversize.clone();
+        // CommandStart is externally tagged by operation / request.
+        let object = missing.as_object_mut().unwrap();
+        let request = object.get_mut("request").unwrap().as_object_mut().unwrap();
+        request.remove("transport_byte_budget");
+        assert_eq!(
+            harness.raw_operation(missing).await.status(),
+            reqwest::StatusCode::BAD_REQUEST
+        );
+        let mut legacy = oversize;
+        let request = legacy["request"].as_object_mut().unwrap();
+        request.insert("transport_byte_budget".into(), serde_json::json!(0));
+        request.insert("spill_dir".into(), serde_json::json!("/private/host"));
+        assert_eq!(
+            harness.raw_operation(legacy).await.status(),
+            reqwest::StatusCode::BAD_REQUEST
+        );
+        assert!(!harness._root.path().join("must-not-exist").exists());
+        harness.remote.close().await.unwrap();
+    })
+    .await
+    .expect("bounded HTTP budget admission");
+}

crates/worker-runtime/src/http_server/tests/session_material_tests.rs

@@ -35,6 +35,7 @@ fn message(id: &str, role: LoggedRole, text: &str) -> LoggedHistoryEntry {
             content: vec![LoggedContentPart::Text { text: text.into() }],
         },
         metadata: LoggedSessionHistoryMetadata {
+            input_metadata: None,
             entry_id: LoggedSessionHistoryEntryId(id.into()),
             origin: LoggedSessionHistoryOrigin::LocalHumanInput,
             derivation: None,

crates/worker-runtime/src/retention.rs

@@ -641,6 +641,13 @@ impl FsWorkerRetentionProvider {
         let (manifest, session_root) = self.session_archive_identity(archive_id)?;
         let archive_dir = self.archive_dir(archive_id)?;
         let manifest = validate_existing_archive(&archive_dir, &manifest, false)?;
+        session_store::WorkerSessionStore::open_read_only(&session_root).map_err(|_| {
+            RuntimeError::StoreCorrupt {
+                operation: "read Worker Session archive",
+                path: session_root.clone(),
+                message: "archive Session is unavailable in the current schema".into(),
+            }
+        })?;
         Ok((manifest, session_root))
     }
 
@@ -653,13 +660,7 @@ impl FsWorkerRetentionProvider {
             &archive_dir.join("manifest.json"),
             "read Worker Session archive",
         )?;
-        if manifest.schema_version != ARCHIVE_SCHEMA_VERSION || manifest.archive_id != archive_id {
-            return Err(RuntimeError::StoreCorrupt {
-                operation: "read Worker Session archive",
-                path: archive_dir.join("manifest.json"),
-                message: "archive manifest identity mismatch".to_string(),
-            });
-        }
+        validate_archive_identity(&archive_dir, &manifest)?;
         Ok((manifest, archive_dir.join("session")))
     }
 
@@ -1204,9 +1205,14 @@ fn archive_storage(
     }
 }
 
-/// Called only on the private Runtime migration staging tree under its owner
-/// lock. Failure discards the whole tree, including manifests and attestations.
-pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
+// Share discovery and erasure fencing between Runtime preflight and staging.
+// Unreadable archives are retained but excluded before any conversion writes;
+// one historical archive must not block the live Runtime's startup. Session
+// Store owns conversion admission, including the current resource limits.
+fn for_each_live_session(
+    root: &Path,
+    mut visit: impl FnMut(&Path, bool) -> Result<(), RuntimeError>,
+) -> Result<(), RuntimeError> {
     for directory in ["workers", "orphaned-workers", "archives/workers"] {
         let directory = root.join(directory);
         let entries = match fs::read_dir(&directory) {
@@ -1228,69 +1234,232 @@ pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
             })?;
             let aggregate = entry.path();
             let session = aggregate.join("session");
-            if !session
-                .join("session.json")
-                .try_exists()
-                .map_err(|source| RuntimeError::StoreIo {
-                    operation: "inspect Session migration",
-                    path: session.clone(),
-                    source,
-                })?
-            {
-                continue;
-            }
-            let identity: CanonicalSessionManifest = read_json(
-                &session.join("session.json"),
-                "read Session migration identity",
-            )?;
-            let session_id = identity.session_id.parse().map_err(|_| {
-                RuntimeError::InvalidRequest("Session migration identity is invalid".into())
-            })?;
+            let is_archive = directory == root.join("archives/workers");
+            let identity = (|| {
+                if !session
+                    .join("session.json")
+                    .try_exists()
+                    .map_err(|source| RuntimeError::StoreIo {
+                        operation: "inspect Session migration",
+                        path: session.clone(),
+                        source,
+                    })?
+                {
+                    return Ok(None);
+                }
+                let identity: CanonicalSessionManifest = read_json(
+                    &session.join("session.json"),
+                    "read Session migration identity",
+                )?;
+                identity.session_id.parse().map(Some).map_err(|_| {
+                    RuntimeError::InvalidRequest("Session migration identity is invalid".into())
+                })
+            })();
+            let session_id = match identity {
+                Ok(Some(session_id)) => session_id,
+                Ok(None) => continue,
+                Err(error) if is_archive => {
+                    warn_unavailable_archive(
+                        &aggregate,
+                        ArchiveMigrationFailure::runtime("session_identity", &error),
+                    );
+                    continue;
+                }
+                Err(error) => return Err(error),
+            };
             match runtime_session_retention(root).ensure_live(session_id) {
                 Ok(()) => {}
-                // Erasure fences remain authoritative even if physical removal
-                // was interrupted. Do not inspect or migrate fenced originals.
+                // Retention-ledger failures are not archive-local. Never bypass
+                // erasure authority while isolating an unreadable archive.
                 Err(session_store::StoreError::SessionErased(_)) => continue,
                 Err(error) => return Err(retention_error(error)),
             }
-            let archive = if directory == root.join("archives/workers") {
-                let manifest: WorkerSessionArchiveManifest = read_json(
-                    &aggregate.join("manifest.json"),
-                    "read archive before Session migration",
-                )?;
-                Some(validate_existing_archive(&aggregate, &manifest, false)?)
-            } else {
-                None
-            };
-            session_store::WorkerSessionStore::new(&session).map_err(|error| {
-                RuntimeError::StoreCorrupt {
-                    operation: "migrate Session schema",
-                    path: session.clone(),
-                    message: error.to_string(),
+            if is_archive {
+                if let Err(failure) = validate_archive_migration(&aggregate) {
+                    warn_unavailable_archive(&aggregate, failure);
+                    continue;
                 }
-            })?;
-            if let Some(ingestion) = archive {
-                let (checksum_sha256, bytes, file_count) = checksum_files(
-                    &collect_files(&session, "migrate archive storage")?,
-                    "migrate archive storage",
-                )?;
-                atomic_write_json(
-                    &aggregate.join(ARCHIVE_STORAGE_FILE),
-                    &MigratedArchiveStorage {
-                        schema_version: 1,
-                        ingestion,
-                        checksum_sha256,
-                        bytes,
-                        file_count,
-                    },
-                    "commit migrated archive storage",
-                )?;
             }
+            // Publication failures still abort the outer staging transaction.
+            visit(&aggregate, is_archive)?;
         }
     }
     Ok(())
 }
 
+struct ArchiveMigrationFailure {
+    stage: &'static str,
+    reason: &'static str,
+}
+
+impl ArchiveMigrationFailure {
+    fn runtime(stage: &'static str, error: &RuntimeError) -> Self {
+        let reason = match error {
+            RuntimeError::StoreIo { source, .. } => safe_io_reason(source),
+            RuntimeError::StoreCorrupt { message, .. } => match message.as_str() {
+                "archive manifest identity mismatch" => "archive_identity_mismatch",
+                "archive id collision or manifest mismatch" => "archive_manifest_mismatch",
+                "archive checksum or content summary mismatch" => "archive_checksum_mismatch",
+                "archive storage attestation does not match ingestion receipt" => {
+                    "archive_storage_attestation_mismatch"
+                }
+                _ => "invalid_stored_data",
+            },
+            RuntimeError::InvalidRequest(_) => "invalid_identity",
+            _ => "archive_validation_failed",
+        };
+        Self { stage, reason }
+    }
+
+    fn conversion(error: session_store::StoreError) -> Self {
+        use session_store::StoreError;
+        let reason = match error {
+            StoreError::Io(error) => safe_io_reason(&error),
+            StoreError::Serde(_) => "invalid_session_manifest",
+            StoreError::Corrupt { message, .. }
+                if message.starts_with("unsupported Worker Session schema version ") =>
+            {
+                "unsupported_session_schema"
+            }
+            // The legacy decoder wraps InputMetadataError in a path-bearing Corrupt
+            // message. Recognize only its fixed cause; never publish that wrapper.
+            StoreError::Corrupt { message, .. }
+                if message.ends_with(&format!(
+                    ": {}",
+                    session_store::InputMetadataError::ResourceLimit
+                )) =>
+            {
+                "input_resource_limit_exceeded"
+            }
+            StoreError::Corrupt { .. } => "invalid_session_log",
+            StoreError::InvalidInputMetadata(session_store::InputMetadataError::ResourceLimit) => {
+                "input_resource_limit_exceeded"
+            }
+            StoreError::InvalidInputMetadata(_) => "invalid_input_metadata",
+            StoreError::ReadLimitExceeded => "read_limit_exceeded",
+            _ => "session_conversion_failed",
+        };
+        Self {
+            stage: "session_conversion",
+            reason,
+        }
+    }
+}
+
+fn safe_io_reason(error: &std::io::Error) -> &'static str {
+    match error.kind() {
+        std::io::ErrorKind::NotFound => "not_found",
+        std::io::ErrorKind::PermissionDenied => "permission_denied",
+        std::io::ErrorKind::InvalidData => "invalid_data",
+        _ => "io_error",
+    }
+}
+
+fn validate_archive_migration(aggregate: &Path) -> Result<(), ArchiveMigrationFailure> {
+    let session = aggregate.join("session");
+    // Current archives need no conversion. Their payload integrity is checked
+    // on access, not by scanning every retained byte during Runtime startup.
+    if session_store::WorkerSessionStore::open_read_only(&session).is_ok() {
+        return Ok(());
+    }
+    let manifest: WorkerSessionArchiveManifest = read_json(
+        &aggregate.join("manifest.json"),
+        "read archive before Session migration",
+    )
+    .map_err(|error| ArchiveMigrationFailure::runtime("archive_manifest", &error))?;
+    validate_existing_archive(aggregate, &manifest, false)
+        .map_err(|error| ArchiveMigrationFailure::runtime("archive_integrity", &error))?;
+    session_store::WorkerSessionStore::validate_migration(&session)
+        .map_err(ArchiveMigrationFailure::conversion)
+}
+
+fn warn_unavailable_archive(aggregate: &Path, failure: ArchiveMigrationFailure) {
+    let archive_id = aggregate.file_name().unwrap_or_default().to_string_lossy();
+    tracing::warn!(
+        archive_id = %bounded_diagnostic_id(&archive_id),
+        code = "session_archive_unavailable",
+        stage = failure.stage,
+        reason = failure.reason,
+        "Retained Session archive could not be admitted for migration; original bytes retained"
+    );
+}
+
+/// Read-only admission probe. Only convertible archives request a staged pass;
+/// unreadable originals remain available for archive-local diagnostics/recovery.
+/// Never create a writable Session store or rewrite archive attestations here.
+pub(crate) fn sessions_require_migration(root: &Path) -> Result<bool, RuntimeError> {
+    let mut required = false;
+    for_each_live_session(root, |aggregate, _| {
+        required |=
+            session_store::WorkerSessionStore::open_read_only(aggregate.join("session")).is_err();
+        Ok(())
+    })?;
+    Ok(required)
+}
+
+/// Called only on the private Runtime migration staging tree under its owner
+/// lock. Failure discards the whole tree, including manifests and attestations.
+pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
+    for_each_live_session(root, |aggregate, is_archive| {
+        let session = aggregate.join("session");
+        // Current Sessions need no writable admission, and their storage
+        // attestations must remain byte-for-byte unchanged.
+        if session_store::WorkerSessionStore::open_read_only(&session).is_ok() {
+            return Ok(());
+        }
+        let archive = if is_archive {
+            let manifest: WorkerSessionArchiveManifest = read_json(
+                &aggregate.join("manifest.json"),
+                "read archive before Session migration",
+            )?;
+            Some(validate_existing_archive(aggregate, &manifest, false)?)
+        } else {
+            None
+        };
+        session_store::WorkerSessionStore::new(&session).map_err(|error| {
+            RuntimeError::StoreCorrupt {
+                operation: "migrate Session schema",
+                path: session.clone(),
+                message: error.to_string(),
+            }
+        })?;
+        if let Some(ingestion) = archive {
+            let (checksum_sha256, bytes, file_count) = checksum_files(
+                &collect_files(&session, "migrate archive storage")?,
+                "migrate archive storage",
+            )?;
+            atomic_write_json(
+                &aggregate.join(ARCHIVE_STORAGE_FILE),
+                &MigratedArchiveStorage {
+                    schema_version: 1,
+                    ingestion,
+                    checksum_sha256,
+                    bytes,
+                    file_count,
+                },
+                "commit migrated archive storage",
+            )?;
+        }
+        Ok(())
+    })
+}
+
+fn validate_archive_identity(
+    archive_dir: &Path,
+    manifest: &WorkerSessionArchiveManifest,
+) -> Result<(), RuntimeError> {
+    if manifest.schema_version != ARCHIVE_SCHEMA_VERSION
+        || archive_dir.file_name().and_then(|name| name.to_str()) != Some(&manifest.archive_id)
+    {
+        return Err(RuntimeError::StoreCorrupt {
+            operation: "read Worker Session archive",
+            path: archive_dir.join("manifest.json"),
+            message: "archive manifest identity mismatch".into(),
+        });
+    }
+    Ok(())
+}
+
 fn validate_existing_archive(
     archive_dir: &Path,
     expected: &WorkerSessionArchiveManifest,
@@ -1300,6 +1469,7 @@ fn validate_existing_archive(
         &archive_dir.join("manifest.json"),
         "verify Worker Session archive",
     )?;
+    validate_archive_identity(archive_dir, &existing)?;
     let storage = archive_storage(archive_dir, &existing)?;
     let mut comparable_expected = expected.clone();
     comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds;
@@ -1753,12 +1923,16 @@ fn temporary_path(path: &Path) -> PathBuf {
     path.with_file_name(format!(".{name}.tmp-{}-{sequence}", std::process::id()))
 }
 
+#[cfg(all(test, feature = "fs-store"))]
+#[path = "retention/runtime_v5_tests.rs"]
+mod runtime_v5_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::sync::{Arc, Barrier};
 
-    fn write_json(path: &Path, value: &impl Serialize) {
+    pub(super) fn write_json(path: &Path, value: &impl Serialize) {
         fs::create_dir_all(path.parent().unwrap()).unwrap();
         fs::write(path, serde_json::to_vec_pretty(value).unwrap()).unwrap();
     }
@@ -1786,7 +1960,7 @@ mod tests {
         fs::write(worker.join("runs/attempt-a/worker.sock"), b"not retained").unwrap();
     }
 
-    fn request(
+    pub(super) fn request(
         worker_id: WorkerId,
         disposition: SessionDisposition,
     ) -> WorkerRetentionExecutionRequest {
@@ -1831,6 +2005,7 @@ mod tests {
                 }],
             },
             metadata: LoggedSessionHistoryMetadata {
+                input_metadata: None,
                 entry_id: LoggedSessionHistoryEntryId("original".into()),
                 origin: LoggedSessionHistoryOrigin::LocalHumanInput,
                 derivation: None,
@@ -1844,6 +2019,7 @@ mod tests {
                 }],
             },
             metadata: LoggedSessionHistoryMetadata {
+                input_metadata: None,
                 entry_id: LoggedSessionHistoryEntryId("summary".into()),
                 origin: LoggedSessionHistoryOrigin::DerivedSummary,
                 derivation: Some(session_store::LoggedHistoryDerivation {
@@ -2002,7 +2178,7 @@ mod tests {
     }
 
     #[test]
-    fn session_migration_rejects_corrupt_archive_before_replacing_its_bytes() {
+    fn session_migration_preserves_corrupt_archive_without_blocking_other_sessions() {
         let temp = tempfile::tempdir().unwrap();
         let worker = WorkerId::now_v7();
         let compact = old_compacted_source(temp.path(), worker);
@@ -2014,11 +2190,165 @@ mod tests {
             .join("session/segments")
             .join(format!("{compact}.jsonl"));
         fs::write(&path, b"corrupt\n").unwrap();
-        assert!(migrate_staged_sessions(temp.path()).is_err());
+        migrate_staged_sessions(temp.path()).unwrap();
         assert_eq!(fs::read(&path).unwrap(), b"corrupt\n");
         assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists());
     }
 
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn archive_admission_logs_safe_causes_and_startup_preserves_original_bytes() {
+        // Exercise FsRuntimeStore startup, real archive admission and Session conversion;
+        // only the tracing sink is substituted, and all storage is temporary.
+        #[derive(Clone)]
+        struct Capture(Arc<std::sync::Mutex<Vec<u8>>>);
+        impl std::io::Write for Capture {
+            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                self.0.lock().unwrap().extend_from_slice(bytes);
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        use crate::fs_store::FsRuntimeStore;
+        const PRIVATE: &str = "private-session-content-token=/private/host/session";
+        for (case, stage, reason) in [
+            ("identity_json", "session_identity", "invalid_stored_data"),
+            ("identity_value", "session_identity", "invalid_identity"),
+            ("manifest_json", "archive_manifest", "invalid_stored_data"),
+            (
+                "manifest_identity",
+                "archive_integrity",
+                "archive_identity_mismatch",
+            ),
+            ("storage_json", "archive_integrity", "invalid_stored_data"),
+            ("checksum", "archive_integrity", "archive_checksum_mismatch"),
+            ("schema", "session_conversion", "unsupported_session_schema"),
+            ("log", "session_conversion", "invalid_session_log"),
+            (
+                "resource_limit",
+                "session_conversion",
+                "input_resource_limit_exceeded",
+            ),
+        ] {
+            let temp = tempfile::tempdir().unwrap();
+            let root = temp.path().join("private-runtime-root");
+            drop(
+                crate::Runtime::with_fs_store(
+                    crate::fs_store::FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"),
+                )
+                .unwrap(),
+            );
+            let worker = WorkerId::now_v7();
+            let segment = old_compacted_source(&root, worker);
+            let provider = FsWorkerRetentionProvider::new(&root);
+            let worker_dir = provider.worker_dir(worker);
+            let log = worker_dir
+                .join("session/segments")
+                .join(format!("{segment}.jsonl"));
+            if case == "log" {
+                fs::write(&log, format!("{{\"type\":\"{PRIVATE}\"}}\n")).unwrap();
+            } else if case == "schema" {
+                let path = worker_dir.join("session/session.json");
+                let mut manifest: serde_json::Value = read_json(&path, "test").unwrap();
+                manifest["schema_version"] = serde_json::json!(999);
+                write_json(&path, &manifest);
+            } else if case == "resource_limit" {
+                let mut bytes = fs::read_to_string(&log).unwrap();
+                for line in include_str!("retention/fixtures/system-input-v4.jsonl").lines() {
+                    let mut entry: serde_json::Value = serde_json::from_str(line).unwrap();
+                    if let Some(item) = entry.pointer_mut("/entry/item")
+                        && item.get("snapshot").is_some()
+                    {
+                        item["snapshot"] = serde_json::json!({
+                            "private": PRIVATE.repeat(128 * 1024 / PRIVATE.len() + 1)
+                        });
+                    }
+                    bytes.push_str(&serde_json::to_string(&entry).unwrap());
+                    bytes.push('\n');
+                }
+                fs::write(&log, bytes).unwrap();
+            }
+            commit_session_archive(
+                &provider,
+                &request(worker, SessionDisposition::Archive),
+                &worker_dir,
+            )
+            .unwrap();
+            fs::remove_dir_all(worker_dir).unwrap();
+            let archive = provider.archive_dir("archive-a").unwrap();
+            match case {
+                "identity_json" => write_json(
+                    &archive.join("session/session.json"),
+                    &serde_json::json!({"session_id": [PRIVATE]}),
+                ),
+                "identity_value" => write_json(
+                    &archive.join("session/session.json"),
+                    &serde_json::json!({"session_id": PRIVATE}),
+                ),
+                "manifest_json" => write_json(
+                    &archive.join("manifest.json"),
+                    &serde_json::json!({"schema_version": PRIVATE}),
+                ),
+                "manifest_identity" => {
+                    let path = archive.join("manifest.json");
+                    let mut manifest: WorkerSessionArchiveManifest =
+                        read_json(&path, "test").unwrap();
+                    manifest.archive_id = PRIVATE.into();
+                    write_json(&path, &manifest);
+                }
+                "storage_json" => write_json(
+                    &archive.join(ARCHIVE_STORAGE_FILE),
+                    &serde_json::json!({"schema_version": PRIVATE}),
+                ),
+                "checksum" => fs::write(
+                    archive
+                        .join("session/segments")
+                        .join(format!("{segment}.jsonl")),
+                    PRIVATE,
+                )
+                .unwrap(),
+                _ => {}
+            }
+            let snapshot = || {
+                collect_files(&archive, "test")
+                    .unwrap()
+                    .into_iter()
+                    .map(|(relative, path)| (relative, fs::read(path).unwrap()))
+                    .collect::<Vec<_>>()
+            };
+            let before = snapshot();
+            for _ in 0..2 {
+                let output = Arc::new(std::sync::Mutex::new(Vec::new()));
+                let writer = Capture(output.clone());
+                let subscriber = tracing_subscriber::fmt()
+                    .with_ansi(false)
+                    .without_time()
+                    .with_writer(move || writer.clone())
+                    .finish();
+                tracing::subscriber::with_default(subscriber, || {
+                    tracing::callsite::rebuild_interest_cache();
+                    drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
+                });
+                let log = String::from_utf8(output.lock().unwrap().clone()).unwrap();
+                assert!(log.contains("session_archive_unavailable"), "{case}: {log}");
+                assert!(log.contains("archive_id=archive-a"), "{case}: {log}");
+                assert!(log.contains(&format!("stage=\"{stage}\"")), "{case}: {log}");
+                assert!(
+                    log.contains(&format!("reason=\"{reason}\"")),
+                    "{case}: {log}"
+                );
+                assert!(log.contains("original bytes retained"), "{case}: {log}");
+                assert!(!log.contains("private-session-content"), "{case}: {log}");
+                assert!(!log.contains("/private"), "{case}: {log}");
+                assert!(!log.contains("private-runtime-root"), "{case}: {log}");
+                assert_eq!(snapshot(), before, "{case}");
+                assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists() || case == "storage_json");
+            }
+        }
+    }
+
     #[test]
     fn session_migration_does_not_read_or_upgrade_erased_originals() {
         let temp = tempfile::tempdir().unwrap();

crates/worker-runtime/src/retention/fixtures/system-input-v4.jsonl

@@ -0,0 +1,2 @@
+{"kind":"annotated_system_item","ts":101,"entry":{"item":{"kind":"journal_startup","body":"journal body\nunchanged","snapshot":{"entries":[{"text":"private snapshot"}]},"snapshot_sha256":"snapshot-digest","prompt_provenance":{"workspace_id":"workspace","source_digest":"source","projection_digest":"projection","logical_name":"journal_startup"}},"metadata":{"entry_id":"journal-id","origin":{"kind":"legacy_unknown"},"derivation":{"sources":["source-id"]}}},"extensions":[{"domain":"migration-fixture","payload":{"kept":true}}]}
+{"kind":"annotated_system_item","ts":103,"entry":{"item":{"kind":"subject_behavior_refresh","body":"rendered subject body","behavior_md":"exact behavior","prompt_provenance":{"source_digest":"subject-source","projection_digest":"subject-projection","logical_name":"subject_behavior"}},"metadata":{"entry_id":"subject-id","origin":{"kind":"backend_instruction","operation_id":"op"}}}}

crates/worker-runtime/src/retention/runtime_v5_tests.rs

@@ -0,0 +1,774 @@
+//! Runtime integration, not a second implementation of Session migration.
+//! The Feature records are frozen v4 wire copied from the Session Store fixture;
+//! only the empty segment envelope is built with current APIs.
+use super::tests::{request, write_json};
+use super::*;
+use crate::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions};
+use serde_json::{Value, json};
+use session_store::{LogEntry, SegmentId, SessionId, Store, WorkerSessionStore};
+use std::collections::BTreeMap;
+
+const V4_SYSTEM_INPUTS: &str = include_str!("fixtures/system-input-v4.jsonl");
+
+fn write_v4_session(aggregate: &Path) -> (SessionId, SegmentId) {
+    let root = aggregate.join("session");
+    let session = session_store::new_session_id();
+    let segment = session_store::new_segment_id();
+    let store = WorkerSessionStore::new(&root).unwrap();
+    store
+        .create_segment(
+            session,
+            segment,
+            &[LogEntry::AnnotatedSegmentStart {
+                ts: 1,
+                session_id: session,
+                system_prompt: None,
+                config: Default::default(),
+                history: vec![],
+                forked_from: None,
+                compacted_from: None,
+            }],
+        )
+        .unwrap();
+    let path = root.join("segments").join(format!("{segment}.jsonl"));
+    let mut bytes = fs::read(&path).unwrap();
+    bytes.extend_from_slice(V4_SYSTEM_INPUTS.as_bytes());
+    fs::write(path, bytes).unwrap();
+    write_json(
+        &root.join("session.json"),
+        &json!({
+            "schema_version": 4, "session_id": session,
+        }),
+    );
+    (session, segment)
+}
+
+fn stopped_worker(root: &Path, schema: u32) -> (WorkerId, SessionId, SegmentId) {
+    let worker = WorkerId::now_v7();
+    let aggregate = root.join("workers").join(worker.to_string());
+    // Serialize request defaults just as a production aggregate writer does.
+    let request: crate::catalog::CreateWorkerRequest = serde_json::from_value(json!({
+        "worker_id": worker,
+        "workspace_api": {"workspace_id": "workspace-a", "base_url": "https://workspace.invalid"},
+        "create_fingerprint": "runtime-v5-fixture",
+        "profile": {"kind": "builtin", "value": "builtin:coder"},
+        "profile_source": {"kind": "workspace_config", "archive": {
+            "id": "archive-1", "digest": "sha256:archive", "size_bytes": 0,
+            "source_graph": {"source_count": 0, "total_source_bytes": 0,
+                "entrypoints": {}, "import_count": 0}
+        }}
+    }))
+    .unwrap();
+    write_json(
+        &aggregate.join("worker.json"),
+        &json!({
+            "schema_version": schema,
+            "worker_ref": {"worker_id": worker}, "worker_id": worker,
+            "profile": request.profile,
+            "profile_source": request.profile_source.reference(),
+            "display_name": null, "config_bundle": null,
+            "created_at_ms": 1234, "status": "stopped", "workspace_id": "workspace-a",
+            "last_finished_submission_request_id": "retained-submission",
+            "restore_guard": {"observation_token": "retained-token",
+                "active_request_id": null, "owners": {}},
+            "execution_state": {"state": "available", "execution": {
+                "request": request, "binding": {}, "restore_intent": "explicit"
+            }}
+        }),
+    );
+    let (session, segment) = write_v4_session(&aggregate);
+    let mut metadata = session_store::WorkerMetadata::new(
+        format!("worker-runtime-{worker}"),
+        Some(session_store::WorkerActiveSegmentRef::active_segment(
+            session, segment,
+        )),
+    );
+    metadata.workspace_id = Some("workspace-a".into());
+    write_json(&aggregate.join("metadata.json"), &metadata);
+    (worker, session, segment)
+}
+
+fn runtime_fixture(root: &Path, schema: u32) -> (WorkerId, SessionId, SegmentId) {
+    fs::create_dir_all(root.join("workers")).unwrap();
+    let fenced_worker = WorkerId::now_v7();
+    write_json(
+        &root.join("runtime.json"),
+        &json!({
+            "schema_version": schema, "display_name": "retained Runtime",
+            "backend": "fs_store", "status": "running", "next_diagnostic_id": 1,
+            "config_bundles": {}, "workspace_owners": {"workspace-a": "owner-a"},
+            "diagnostics": [], "subject_releases": [],
+            "subject_create_recovery_fences": {fenced_worker.to_string(): {
+                "request": {"workspace_id": "workspace-a", "worker_id": fenced_worker,
+                    "create_fingerprint": "retained-create", "operation_id": "retained-recovery"},
+                "response": null
+            }}
+        }),
+    );
+    stopped_worker(root, schema)
+}
+
+// Include every file, not just schema markers: failures must preserve original
+// history, metadata, archive receipts, checksums and retention evidence together.
+fn tree_bytes(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
+    fn walk(root: &Path, at: &Path, result: &mut BTreeMap<PathBuf, Vec<u8>>) {
+        for entry in fs::read_dir(at).unwrap() {
+            let entry = entry.unwrap();
+            if entry.file_type().unwrap().is_dir() {
+                walk(root, &entry.path(), result);
+            } else {
+                result.insert(
+                    entry.path().strip_prefix(root).unwrap().to_path_buf(),
+                    fs::read(entry.path()).unwrap(),
+                );
+            }
+        }
+    }
+    let mut result = BTreeMap::new();
+    walk(root, root, &mut result);
+    result
+}
+
+fn assert_generic(root: &Path, session: SessionId, segment: SegmentId) {
+    use session_store::{InputCompactionUse, InputEvidenceUse, InputObservationUse};
+    let store = WorkerSessionStore::open_read_only(root).unwrap();
+    let entries = store.read_all_read_only(session, segment).unwrap();
+    assert_eq!(entries.len(), 3);
+    let originals: Vec<Value> = V4_SYSTEM_INPUTS
+        .lines()
+        .map(|line| serde_json::from_str(line).unwrap())
+        .collect();
+    for (log, old) in entries[1..].iter().zip(originals) {
+        let LogEntry::AnnotatedSystemItem { entry, .. } = log else {
+            panic!("lost System record: {log:?}");
+        };
+        let current = serde_json::to_value(log).unwrap();
+        assert_eq!(current["ts"], old["ts"]);
+        for field in ["entry_id", "origin", "derivation"] {
+            assert_eq!(
+                current["entry"]["metadata"][field],
+                old["entry"]["metadata"][field]
+            );
+        }
+        assert_eq!(current["entry"]["item"]["kind"], "input");
+        assert_eq!(
+            entry.item.history_text(),
+            old["entry"]["item"]["body"].as_str().unwrap()
+        );
+        let input = entry.metadata.input_metadata.as_ref().unwrap();
+        let usage = input.usage();
+        assert_eq!(usage.recording_material, InputEvidenceUse::Exclude);
+        assert_eq!(usage.memory_evidence, InputEvidenceUse::Exclude);
+        assert_eq!(usage.observation, InputObservationUse::Hidden);
+        let kind = old["entry"]["item"]["kind"].as_str().unwrap();
+        let schema = if kind == "journal_startup" {
+            assert_eq!(usage.compaction, InputCompactionUse::ReferenceInput);
+            assert_eq!(current["extensions"], old["extensions"]);
+            "journal:startup"
+        } else {
+            assert_eq!(usage.compaction, InputCompactionUse::Retain);
+            "subject:behavior"
+        };
+        let payload = input.feature_payload().unwrap();
+        assert_eq!(payload.schema(), schema);
+        assert_eq!(payload.version(), 1);
+        let mut expected_payload = old["entry"]["item"].clone();
+        expected_payload.as_object_mut().unwrap().remove("kind");
+        expected_payload.as_object_mut().unwrap().remove("body");
+        assert_eq!(
+            payload.decode::<Value>(schema, 1).unwrap(),
+            expected_payload
+        );
+        let public = serde_json::to_value(input.public_item().unwrap()).unwrap();
+        assert_eq!(public["kind"], kind);
+        assert_eq!(public["body"], old["entry"]["item"]["body"]);
+        assert_eq!(public["presentation"], json!({"kind": "context"}));
+    }
+}
+
+fn completed_v4_archive(
+    root: &Path,
+    schema: u32,
+) -> (
+    SessionId,
+    SegmentId,
+    WorkerRetentionExecutionRequest,
+    WorkerRetentionExecutionResult,
+) {
+    let (worker, session, segment) = stopped_worker(root, schema);
+    let provider = FsWorkerRetentionProvider::new(root);
+    let request = request(worker, SessionDisposition::Archive);
+    // Frozen old-runtime receipt: ingestion and removal completed before the
+    // current strict decoder existed. Do not ask the v5 reader to mint it.
+    let manifest =
+        commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap();
+    let completed = WorkerRetentionExecutionResult {
+        operation_id: request.operation_id.clone(),
+        input_fingerprint: request.input_fingerprint.clone(),
+        worker_id: worker,
+        session_disposition: SessionDisposition::Archive,
+        diagnostics_disposition: DiagnosticsDisposition::Purge,
+        archive: Some(manifest),
+        source_removed: true,
+        diagnostics_retained: false,
+    };
+    write_json(
+        &provider.operation_path(&request.operation_id).unwrap(),
+        &RetentionOperationReceipt {
+            schema_version: OPERATION_SCHEMA_VERSION,
+            request: request.clone(),
+            result: completed.clone(),
+        },
+    );
+    fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
+    (session, segment, request, completed)
+}
+
+#[test]
+fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
+    for runtime_schema in [11, 12] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        let (worker, session, segment) = runtime_fixture(&root, runtime_schema);
+        let orphan = root
+            .join("orphaned-workers")
+            .join(WorkerId::now_v7().to_string());
+        let (orphan_session, orphan_segment) = write_v4_session(&orphan);
+        // Include an unregistered tree still under workers, not just an
+        // already-archived orphan. Both Runtime source schemas must fence it
+        // out of the live catalog without discarding its Session.
+        let loose_id = WorkerId::now_v7();
+        let loose = root.join("workers").join(loose_id.to_string());
+        let (loose_session, loose_segment) = write_v4_session(&loose);
+        let retained_loose = root.join("orphaned-workers").join(loose_id.to_string());
+        let (archived_session, archived_segment, request, completed) =
+            completed_v4_archive(&root, runtime_schema);
+        let provider = FsWorkerRetentionProvider::new(&root);
+        let archive = provider.archive_dir("archive-a").unwrap();
+        // Also cover an archive which already has a local storage attestation
+        // from an earlier schema migration. Keep ingestion bytes immutable;
+        // harmless marker whitespace makes current storage differ from them.
+        if runtime_schema == 12 {
+            let marker = archive.join("session/session.json");
+            let mut bytes = fs::read(&marker).unwrap();
+            bytes.push(b'\n');
+            fs::write(marker, bytes).unwrap();
+            let ingestion = completed.archive.clone().unwrap();
+            let (checksum_sha256, bytes, file_count) = checksum_files(
+                &collect_files(&archive.join("session"), "test").unwrap(),
+                "test",
+            )
+            .unwrap();
+            write_json(
+                &archive.join(ARCHIVE_STORAGE_FILE),
+                &MigratedArchiveStorage {
+                    schema_version: 1,
+                    ingestion,
+                    checksum_sha256,
+                    bytes,
+                    file_count,
+                },
+            );
+        }
+        let storage_before = fs::read(archive.join(ARCHIVE_STORAGE_FILE)).ok();
+        let manifest_before = fs::read(archive.join("manifest.json")).unwrap();
+        let receipt_path = provider.operation_path(&request.operation_id).unwrap();
+        let receipt_before = fs::read(&receipt_path).unwrap();
+        let stopped = root.join("workers").join(worker.to_string());
+        assert!(WorkerSessionStore::open_read_only(stopped.join("session")).is_err());
+        // Current decode itself must not be made permissive to fix routing.
+        for line in V4_SYSTEM_INPUTS.lines() {
+            assert!(serde_json::from_str::<LogEntry>(line).is_err());
+        }
+        let before = tree_bytes(&root);
+        let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+        let plan = FsRuntimeStore::migration_plan(&options).unwrap();
+        assert!(plan.migration_required);
+        assert_eq!(plan.archived_orphan_worker_count, 1);
+        assert_eq!(plan.current_schema_version, runtime_schema);
+        assert_eq!(tree_bytes(&root), before, "preflight mutated source");
+        // Exercise normal startup, not just the private migration helper.
+        let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap();
+        assert!(
+            matches!(
+                FsRuntimeStore::migrate(&options),
+                Err(RuntimeError::RuntimeStoreAlreadyOpen { .. })
+            ),
+            "owner lock not fenced"
+        );
+        drop(opened);
+        assert_generic(&stopped.join("session"), session, segment);
+        assert_generic(&orphan.join("session"), orphan_session, orphan_segment);
+        assert!(!loose.exists());
+        assert_generic(
+            &retained_loose.join("session"),
+            loose_session,
+            loose_segment,
+        );
+        assert_generic(&archive.join("session"), archived_session, archived_segment);
+        assert_eq!(
+            fs::read(archive.join("manifest.json")).unwrap(),
+            manifest_before
+        );
+        assert_eq!(fs::read(&receipt_path).unwrap(), receipt_before);
+        let (manifest, archive_session) = provider.session_archive("archive-a").unwrap();
+        assert_eq!(completed.archive.as_ref(), Some(&manifest));
+        assert_eq!(provider.execute(&request).unwrap(), completed);
+        if let Some(storage_before) = storage_before {
+            assert_ne!(
+                fs::read(archive.join(ARCHIVE_STORAGE_FILE)).unwrap(),
+                storage_before
+            );
+        }
+        let storage: MigratedArchiveStorage =
+            read_json(&archive.join(ARCHIVE_STORAGE_FILE), "test").unwrap();
+        assert_eq!(storage.ingestion, manifest);
+        let (checksum, bytes, count) =
+            checksum_files(&collect_files(&archive_session, "test").unwrap(), "test").unwrap();
+        assert_eq!(
+            (storage.checksum_sha256, storage.bytes, storage.file_count),
+            (checksum, bytes, count)
+        );
+        if runtime_schema == 12 {
+            for name in [
+                PathBuf::from("runtime.json"),
+                PathBuf::from(format!("workers/{worker}/worker.json")),
+                PathBuf::from(format!("workers/{worker}/metadata.json")),
+            ] {
+                assert_eq!(fs::read(root.join(&name)).unwrap(), before[&name]);
+            }
+        }
+        let once = tree_bytes(&root);
+        assert!(
+            !FsRuntimeStore::migrate(&options)
+                .unwrap()
+                .migration_required
+        );
+        drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
+        assert_eq!(tree_bytes(&root), once, "retry rewrote migrated evidence");
+        assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
+        assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists());
+        // session-storage, not the immutable receipt's old checksum, attests
+        // current bytes. Tampering with those bytes must remain detectable.
+        let path = archive_session
+            .join("segments")
+            .join(format!("{archived_segment}.jsonl"));
+        fs::write(path, b"tampered\n").unwrap();
+        assert!(provider.session_archive("archive-a").is_err());
+    }
+}
+
+#[test]
+fn corrupt_archive_does_not_block_startup_or_healthy_session_migration_and_repair_converges() {
+    for runtime_schema in [11, 12] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        let (worker, session, segment) = runtime_fixture(&root, runtime_schema);
+        let (archived_session, archived_segment, request, completed) =
+            completed_v4_archive(&root, runtime_schema);
+        let provider = FsWorkerRetentionProvider::new(&root);
+        let archive = provider.archive_dir("archive-a").unwrap();
+        let path = archive
+            .join("session/segments")
+            .join(format!("{archived_segment}.jsonl"));
+        let original = fs::read(&path).unwrap();
+        fs::write(&path, b"corrupt archive\n").unwrap();
+        let (healthy_worker, healthy_session, healthy_segment) =
+            stopped_worker(&root, runtime_schema);
+        let mut healthy_request =
+            super::tests::request(healthy_worker, SessionDisposition::Archive);
+        healthy_request.operation_id = "healthy-operation".into();
+        healthy_request.archive_id = Some("healthy-archive".into());
+        let healthy_manifest = commit_session_archive(
+            &provider,
+            &healthy_request,
+            &provider.worker_dir(healthy_worker),
+        )
+        .unwrap();
+        fs::remove_dir_all(provider.worker_dir(healthy_worker)).unwrap();
+        let before = tree_bytes(&archive);
+        let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+        assert!(
+            FsRuntimeStore::migration_plan(&options)
+                .unwrap()
+                .migration_required
+        );
+        drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
+        assert_eq!(tree_bytes(&archive), before);
+        assert!(provider.session_archive("archive-a").is_err());
+        let (manifest, healthy_root) = provider.session_archive("healthy-archive").unwrap();
+        assert_eq!(manifest, healthy_manifest);
+        assert_generic(&healthy_root, healthy_session, healthy_segment);
+        assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists());
+        assert_generic(
+            &root
+                .join("workers")
+                .join(worker.to_string())
+                .join("session"),
+            session,
+            segment,
+        );
+        assert!(
+            !FsRuntimeStore::migration_plan(&options)
+                .unwrap()
+                .migration_required
+        );
+        drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
+        assert_eq!(tree_bytes(&archive), before);
+        fs::write(&path, original).unwrap();
+        FsRuntimeStore::migrate(&options).unwrap();
+        assert_generic(
+            &root
+                .join("workers")
+                .join(worker.to_string())
+                .join("session"),
+            session,
+            segment,
+        );
+        assert_generic(&archive.join("session"), archived_session, archived_segment);
+        assert_eq!(provider.execute(&request).unwrap(), completed);
+    }
+}
+
+#[test]
+fn unmigratable_archive_preserves_evidence_across_startup_and_reopen() {
+    for (field, value) in [
+        ("snapshot_sha256", json!(42)),
+        ("behavior_md", json!([])),
+        ("snapshot", json!({"large": "x".repeat(128 * 1024 + 1)})),
+        ("body", json!("x".repeat(256 * 1024 + 1))),
+    ] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        let (live_worker, live_session, live_segment) = runtime_fixture(&root, 12);
+        let (worker, _, segment) = stopped_worker(&root, 12);
+        let provider = FsWorkerRetentionProvider::new(&root);
+        let path = provider
+            .worker_dir(worker)
+            .join("session/segments")
+            .join(format!("{segment}.jsonl"));
+        let source = fs::read_to_string(&path).unwrap();
+        let mut replaced = false;
+        let mut altered = String::new();
+        for line in source.lines() {
+            let mut entry: Value = serde_json::from_str(line).unwrap();
+            if let Some(item) = entry.pointer_mut("/entry/item")
+                && item.get(field).is_some()
+            {
+                item[field] = value.clone();
+                replaced = true;
+            }
+            altered.push_str(&serde_json::to_string(&entry).unwrap());
+            altered.push('\n');
+        }
+        assert!(replaced);
+        fs::write(&path, altered).unwrap();
+        let request = request(worker, SessionDisposition::Archive);
+        let manifest =
+            commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap();
+        fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
+        let archive = provider.archive_dir("archive-a").unwrap();
+        validate_existing_archive(&archive, &manifest, false).unwrap();
+        let before = tree_bytes(&archive);
+        let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+        for _ in 0..2 {
+            let runtime = crate::Runtime::with_fs_store(options.clone()).unwrap();
+            assert_eq!(runtime.summary().unwrap().worker_count, 1);
+            runtime.bind_runtime_identity("runtime-a").unwrap();
+            let search = runtime_api::SessionPublicSearchRequest {
+                workspace_id: "workspace-a".into(),
+                source: runtime_api::SessionPublicSource::Archived {
+                    archive_id: "archive-a".into(),
+                },
+                expected_session_id: manifest.source_session_id.clone(),
+                expected_generation: None,
+                query: None,
+                kind: None,
+                tool_name: None,
+                tool_part: runtime_api::SessionPublicToolPart::Both,
+                scan_cursor: None,
+                limit: 10,
+                max_scan_bytes: 1024 * 1024,
+                max_segments: 10,
+                max_entries: 100,
+            };
+            let availability = runtime
+                .session_public_search_scoped(
+                    &crate::runtime::RuntimeWorkspaceScope::new("workspace-a", "owner-a"),
+                    &search,
+                )
+                .unwrap();
+            assert!(matches!(
+                availability,
+                runtime_api::SessionPublicSearchAvailability::Unavailable {
+                    reason: runtime_api::SessionPublicUnavailableReason::ArchiveIncomplete,
+                    ..
+                }
+            ));
+            drop(runtime);
+            assert_eq!(tree_bytes(&archive), before);
+            validate_existing_archive(&archive, &manifest, false).unwrap();
+            assert!(matches!(
+                provider.session_archive("archive-a"),
+                Err(RuntimeError::StoreCorrupt { .. })
+            ));
+            assert!(
+                !FsRuntimeStore::migration_plan(&options)
+                    .unwrap()
+                    .migration_required
+            );
+            assert_generic(
+                &provider.worker_dir(live_worker).join("session"),
+                live_session,
+                live_segment,
+            );
+        }
+    }
+}
+
+#[test]
+fn unreadable_archive_metadata_does_not_block_archive_only_runtime_startup() {
+    for (path, field, value) in [
+        ("session/session.json", None, Value::Null),
+        ("manifest.json", None, Value::Null),
+        (ARCHIVE_STORAGE_FILE, None, Value::Null),
+        ("session/session.json", Some("schema_version"), json!(999)),
+        ("manifest.json", Some("schema_version"), json!(999)),
+        (
+            "manifest.json",
+            Some("archive_id"),
+            json!("another-archive"),
+        ),
+    ] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        let (worker, _, _) = runtime_fixture(&root, 12);
+        completed_v4_archive(&root, 12);
+        let provider = FsWorkerRetentionProvider::new(&root);
+        fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
+        let archive = provider.archive_dir("archive-a").unwrap();
+        if let Some(field) = field {
+            let mut record: Value = read_json(&archive.join(path), "test").unwrap();
+            record[field] = value;
+            write_json(&archive.join(path), &record);
+        } else {
+            fs::write(archive.join(path), b"{broken\n").unwrap();
+        }
+        let before = tree_bytes(&archive);
+        let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+        for _ in 0..2 {
+            assert!(
+                !FsRuntimeStore::migration_plan(&options)
+                    .unwrap()
+                    .migration_required
+            );
+            let runtime = crate::Runtime::with_fs_store(options.clone()).unwrap();
+            assert_eq!(runtime.summary().unwrap().worker_count, 0);
+            drop(runtime);
+            assert!(provider.session_archive("archive-a").is_err());
+            assert_eq!(tree_bytes(&archive), before);
+        }
+    }
+}
+
+#[test]
+fn archive_isolation_does_not_bypass_corrupt_erasure_authority() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("runtime");
+    runtime_fixture(&root, 12);
+    let (session, _, _, _) = completed_v4_archive(&root, 12);
+    let ledger = root.join("retention/session-sources");
+    fs::create_dir_all(&ledger).unwrap();
+    fs::write(ledger.join(format!("{session}.json")), b"{broken\n").unwrap();
+    let before = tree_bytes(&root);
+    let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+    assert!(FsRuntimeStore::migration_plan(&options).is_err());
+    assert!(crate::Runtime::with_fs_store(options).is_err());
+    assert_eq!(tree_bytes(&root), before);
+}
+
+#[test]
+fn runtime_v5_probe_and_staging_skip_erased_originals() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("runtime");
+    let (worker, live_session, live_segment) = runtime_fixture(&root, 12);
+    let erased = root
+        .join("orphaned-workers")
+        .join(WorkerId::now_v7().to_string());
+    let (session, segment) = write_v4_session(&erased);
+    runtime_session_retention(&root)
+        .delete(
+            session,
+            session_store::retention::SessionDeletionIntent::ErasePermanently,
+            || Ok(()),
+        )
+        .unwrap();
+    fs::write(
+        erased
+            .join("session/segments")
+            .join(format!("{segment}.jsonl")),
+        b"fenced malformed JSON\n",
+    )
+    .unwrap();
+    let before = tree_bytes(&erased);
+    let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+    assert!(
+        FsRuntimeStore::migration_plan(&options)
+            .unwrap()
+            .migration_required
+    );
+    FsRuntimeStore::migrate(&options).unwrap();
+    assert_generic(
+        &root
+            .join("workers")
+            .join(worker.to_string())
+            .join("session"),
+        live_session,
+        live_segment,
+    );
+    assert_eq!(tree_bytes(&erased), before);
+    assert!(
+        !FsRuntimeStore::migration_plan(&options)
+            .unwrap()
+            .migration_required
+    );
+    assert!(matches!(
+        runtime_session_retention(&root).ensure_live(session),
+        Err(session_store::StoreError::SessionErased(_))
+    ));
+}
+
+#[test]
+fn runtime_v5_rejects_broken_or_future_source_without_rewriting_evidence() {
+    for corruption in ["broken-json", "future-schema"] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        let (worker, session, segment) = runtime_fixture(&root, 12);
+        let session_root = root
+            .join("workers")
+            .join(worker.to_string())
+            .join("session");
+        if corruption == "broken-json" {
+            fs::write(
+                session_root
+                    .join("segments")
+                    .join(format!("{segment}.jsonl")),
+                b"{not valid JSON\n",
+            )
+            .unwrap();
+        } else {
+            write_json(
+                &session_root.join("session.json"),
+                &json!({
+                    "schema_version": 999, "session_id": session,
+                }),
+            );
+        }
+        let before = tree_bytes(&root);
+        let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+        assert!(
+            FsRuntimeStore::migration_plan(&options)
+                .unwrap()
+                .migration_required
+        );
+        assert_eq!(tree_bytes(&root), before);
+        assert!(FsRuntimeStore::migrate(&options).is_err());
+        assert_eq!(tree_bytes(&root), before);
+        assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
+        assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists());
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn current_runtime_session_migration_still_rejects_live_worker_sockets() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("runtime");
+    let (worker, _, _) = runtime_fixture(&root, 12);
+    let before = tree_bytes(&root);
+    let runs = root.join("workers").join(worker.to_string()).join("runs");
+    fs::create_dir_all(runs.join("1")).unwrap();
+    let listener = std::os::unix::net::UnixListener::bind(runs.join("1/worker.sock")).unwrap();
+    let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+    let error = FsRuntimeStore::migrate(&options).unwrap_err();
+    assert!(error.to_string().contains("active Worker socket"));
+    drop(listener);
+    fs::remove_dir_all(runs).unwrap();
+    assert_eq!(tree_bytes(&root), before);
+    assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
+}
+
+#[test]
+fn current_runtime_upgrades_v5_role_metadata_through_staging_without_reserializing_owner_evidence()
+{
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("runtime");
+    let (worker, session, segment) = runtime_fixture(&root, 12);
+    let aggregate = root.join("workers").join(worker.to_string());
+    let path = aggregate
+        .join("session/segments")
+        .join(format!("{segment}.jsonl"));
+    let old = fs::read_to_string(&path).unwrap();
+    let header = old.lines().next().unwrap();
+    let source = include_str!(
+        "../../../session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl"
+    );
+    fs::write(&path, format!("{header}\n{source}")).unwrap();
+    write_json(
+        &aggregate.join("session/session.json"),
+        &json!({"schema_version":5, "session_id": session}),
+    );
+    let owner_before = fs::read(root.join("runtime.json")).unwrap();
+    let worker_before = fs::read(aggregate.join("worker.json")).unwrap();
+    let metadata_before = fs::read(aggregate.join("metadata.json")).unwrap();
+    assert!(WorkerSessionStore::open_read_only(aggregate.join("session")).is_err());
+    let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
+    assert!(
+        FsRuntimeStore::migration_plan(&options)
+            .unwrap()
+            .migration_required
+    );
+    FsRuntimeStore::migrate(&options).unwrap();
+    let store = WorkerSessionStore::open_read_only(aggregate.join("session")).unwrap();
+    let entries = store.read_all_read_only(session, segment).unwrap();
+    assert_eq!(entries.len(), 6);
+    for (current, old) in entries[1..].iter().zip(source.lines()) {
+        let current = serde_json::to_value(current).unwrap();
+        let old: Value = serde_json::from_str(old).unwrap();
+        assert_eq!(current["ts"], old["ts"]);
+        assert_eq!(current["entry"]["item"], old["entry"]["item"]);
+        for field in ["entry_id", "origin", "derivation"] {
+            assert_eq!(
+                current["entry"]["metadata"][field],
+                old["entry"]["metadata"][field]
+            );
+        }
+        if current["entry"]["metadata"]["origin"]["kind"] == "derived_summary" {
+            assert_eq!(
+                current["entry"]["metadata"]["input_metadata"]["usage"]["recording_material"],
+                "exclude"
+            );
+            assert_eq!(
+                current["entry"]["metadata"]["input_metadata"]["usage"]["memory_evidence"],
+                "exclude"
+            );
+        }
+    }
+    assert_eq!(fs::read(root.join("runtime.json")).unwrap(), owner_before);
+    assert_eq!(
+        fs::read(aggregate.join("worker.json")).unwrap(),
+        worker_before
+    );
+    assert_eq!(
+        fs::read(aggregate.join("metadata.json")).unwrap(),
+        metadata_before
+    );
+    let once = tree_bytes(&aggregate.join("session"));
+    assert!(
+        !FsRuntimeStore::migration_plan(&options)
+            .unwrap()
+            .migration_required
+    );
+    FsRuntimeStore::migrate(&options).unwrap();
+    assert_eq!(tree_bytes(&aggregate.join("session")), once);
+}

crates/worker-runtime/src/runtime.rs

@@ -1865,20 +1865,23 @@ impl Runtime {
         if let Some(scope) = scope {
             self.ensure_worker_in_workspace(scope, worker_ref)?;
         }
-        self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments)
+        self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments, false)
     }
 
     fn replace_worker_workdir_attachments_under_lock(
         &self,
         worker_ref: &WorkerRef,
         mut attachments: Vec<LogicalWorkdirAttachment>,
+        restoring: bool,
     ) -> Result<WorkerDetail, RuntimeError> {
         validate_logical_workdir_attachments(&attachments)?;
         attachments.sort_by(|left, right| left.alias.cmp(&right.alias));
 
         let mut state = self.lock()?;
-        for attachment in &attachments {
-            state.ensure_workdir_not_removing(&attachment.working_directory_id)?;
+        if !restoring {
+            for attachment in &attachments {
+                state.ensure_workdir_not_removing(&attachment.working_directory_id)?;
+            }
         }
         let previous = state.worker(worker_ref)?.clone();
         {
@@ -2146,7 +2149,6 @@ impl Runtime {
         ),
         RuntimeError,
     > {
-        let mut preparation_authorized = !coordinate && request.preparation.is_some();
         let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?;
         let _operation_guard = operation_lock
             .lock()
@@ -2191,7 +2193,7 @@ impl Runtime {
             let conflict = || RuntimeError::RestoreObservationConflict {
                 worker_id: worker_ref.worker_id,
             };
-            if let Some(mut owner) = candidate
+            if let Some(owner) = candidate
                 .restore_guard
                 .owners
                 .get(&request.request_id)
@@ -2204,13 +2206,6 @@ impl Runtime {
                 if let Some(receipt) = &owner.receipt {
                     return Ok((Some(receipt.clone()), None));
                 }
-                if let Some(preparation) = &owner.request.preparation {
-                    for id in
-                        preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
-                    {
-                        state.ensure_workdir_not_removing(id)?;
-                    }
-                }
                 if candidate.restore_guard.active_request_id.as_deref() != Some(&request.request_id)
                     || candidate
                         .pending_restore
@@ -2218,151 +2213,6 @@ impl Runtime {
                 {
                     return Err(conflict());
                 }
-                if coordinate && !owner.preparation_applied {
-                    if let Some(preparation) = owner.request.preparation.as_mut() {
-                        if preparation.repository_access_workdirs.is_empty()
-                            && !preparation.repository_access.is_empty()
-                        {
-                            preparation.repository_access_workdirs = preparation
-                                .repository_access
-                                .iter()
-                                .map(|access| runtime_api::RepositoryAccessWorkdirReference {
-                                    runtime_id: access.materialization.runtime_id.clone(),
-                                    working_directory_id: access.working_directory_id.clone(),
-                                })
-                                .collect();
-                            candidate
-                                .restore_guard
-                                .owners
-                                .insert(request.request_id.clone(), owner.clone());
-                            match state.persist_worker_record(&candidate) {
-                                Ok(()) => {
-                                    state
-                                        .workers
-                                        .insert(worker_ref.worker_id, candidate.clone());
-                                }
-                                Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
-                                    state.workers.insert(worker_ref.worker_id, candidate);
-                                    return Ok((
-                                        Some(RuntimeWorkerRestoreResult::failed(
-                                            WorkerRestoreState::ReconciliationRequired,
-                                            "worker_restore_preparation_commit_unknown",
-                                            sanitize_worker_create_failure_message(
-                                                &error.to_string(),
-                                            ),
-                                        )),
-                                        None,
-                                    ));
-                                }
-                                Err(error) => {
-                                    return Ok((
-                                        Some(RuntimeWorkerRestoreResult::failed(
-                                            WorkerRestoreState::ReconciliationRequired,
-                                            "worker_restore_preparation_commit_failed",
-                                            sanitize_worker_create_failure_message(
-                                                &error.to_string(),
-                                            ),
-                                        )),
-                                        None,
-                                    ));
-                                }
-                            }
-                        }
-                    }
-                }
-                if !owner.preparation_applied {
-                    if let (Some(original), Some(incoming)) =
-                        (&owner.request.preparation, &request.preparation)
-                    {
-                        // Server-regenerated bindings are not part of the public fingerprint.
-                        // Reuse admitted bindings and accept only resource completion for those references.
-                        let references_match = incoming.repository_access.len()
-                            == original.repository_access_workdirs.len()
-                            && original.repository_access_workdirs.iter().all(|reference| {
-                                incoming
-                                    .repository_access
-                                    .iter()
-                                    .filter(|access| {
-                                        reference.working_directory_id
-                                            == access.working_directory_id
-                                            && reference.runtime_id
-                                                == access.materialization.runtime_id
-                                    })
-                                    .count()
-                                    == 1
-                            });
-                        let snapshot_matches = original.repository_access.is_empty()
-                            || (original.repository_access.len()
-                                == incoming.repository_access.len()
-                                && original
-                                    .repository_access
-                                    .iter()
-                                    .zip(&incoming.repository_access)
-                                    .all(|(original, incoming)| {
-                                        let mut normalized = incoming.clone();
-                                        if let (Some(old), Some(new)) = (
-                                            &original.materialization.ssh,
-                                            &mut normalized.materialization.ssh,
-                                        ) {
-                                            new.secret_resource = old.secret_resource.clone();
-                                            new.expires_at_epoch_seconds =
-                                                old.expires_at_epoch_seconds;
-                                        }
-                                        &normalized == original
-                                    }));
-                        if !original.repository_access_workdirs.is_empty() {
-                            preparation_authorized =
-                                !coordinate && references_match && snapshot_matches;
-                        }
-                        if references_match
-                            && snapshot_matches
-                            && original.repository_access != incoming.repository_access
-                        {
-                            candidate
-                                .restore_guard
-                                .owners
-                                .get_mut(&request.request_id)
-                                .unwrap()
-                                .request
-                                .preparation
-                                .as_mut()
-                                .unwrap()
-                                .repository_access = incoming.repository_access.clone();
-                            match state.persist_worker_record(&candidate) {
-                                Ok(()) => {
-                                    state
-                                        .workers
-                                        .insert(worker_ref.worker_id, candidate.clone());
-                                }
-                                Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
-                                    state.workers.insert(worker_ref.worker_id, candidate);
-                                    return Ok((
-                                        Some(RuntimeWorkerRestoreResult::failed(
-                                            WorkerRestoreState::ReconciliationRequired,
-                                            "worker_restore_preparation_commit_unknown",
-                                            sanitize_worker_create_failure_message(
-                                                &error.to_string(),
-                                            ),
-                                        )),
-                                        None,
-                                    ));
-                                }
-                                Err(error) => {
-                                    return Ok((
-                                        Some(RuntimeWorkerRestoreResult::failed(
-                                            WorkerRestoreState::ReconciliationRequired,
-                                            "worker_restore_preparation_commit_failed",
-                                            sanitize_worker_create_failure_message(
-                                                &error.to_string(),
-                                            ),
-                                        )),
-                                        None,
-                                    ));
-                                }
-                            }
-                        }
-                    }
-                }
             } else {
                 // Startup/older persisted pending operations have a lifecycle
                 // owner but no public request receipt. Bind recovery to that
@@ -2417,13 +2267,6 @@ impl Runtime {
                 if recovering_pending {
                     request.preparation = None;
                 }
-                if let Some(preparation) = &request.preparation {
-                    for id in
-                        preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
-                    {
-                        state.ensure_workdir_not_removing(id)?;
-                    }
-                }
                 let pending = candidate
                     .pending_restore
                     .unwrap_or_else(|| PendingWorkerRestore {
@@ -2481,12 +2324,8 @@ impl Runtime {
                 return Ok((None, owner.request.preparation.clone()));
             }
         }
-        self.restore_admitted_worker_under_lock_with_preparation(
-            worker_ref,
-            WorkerRestoreMode::Explicit,
-            preparation_authorized,
-        )
-        .map(|result| (Some(result), None))
+        self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit)
+            .map(|result| (Some(result), None))
     }
 
     #[cfg(all(test, feature = "fs-store"))]
@@ -2522,15 +2361,6 @@ impl Runtime {
         &self,
         worker_ref: &WorkerRef,
         mode: WorkerRestoreMode,
-    ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
-        self.restore_admitted_worker_under_lock_with_preparation(worker_ref, mode, false)
-    }
-
-    fn restore_admitted_worker_under_lock_with_preparation(
-        &self,
-        worker_ref: &WorkerRef,
-        mode: WorkerRestoreMode,
-        preparation_authorized: bool,
     ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
         self.lock()?
             .ensure_subject_not_reserved(worker_ref, false)?;
@@ -2595,15 +2425,7 @@ impl Runtime {
                         message,
                     )
                 };
-                if owner.request.preparation.as_ref().is_some_and(|prep| {
-                    prep.repository_access.len() < prep.repository_access_workdirs.len()
-                        || (!prep.repository_access_workdirs.is_empty() && !preparation_authorized)
-                }) {
-                    return Ok(uncertain(
-                        "Admitted Restore is awaiting Workspace resource completion".into(),
-                    ));
-                }
-                // Complete idempotent binding/access effects under the original pending_restore ID.
+                // Apply reference updates under the original pending_restore ID; no provider is opened.
                 let mut effects_started = false;
                 if let Err(error) = self.apply_restore_preparation_under_lock(
                     worker_ref,
@@ -2744,42 +2566,10 @@ impl Runtime {
             .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
             validate_logical_workdir_attachments(&attachments)?;
             *effects_started = true;
-            self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments)?;
-        }
-        for access in preparation.repository_access {
-            if preparation
-                .repository_access_workdirs
-                .iter()
-                .any(|reference| {
-                    reference.working_directory_id == access.working_directory_id
-                        && reference.runtime_id == access.materialization.runtime_id
-                })
-            {
-                // Workspace completed these at the Workdir owner Runtime after snapshot persistence.
-                // Never send cross-Runtime credentials to this Worker's Runtime.
-                continue;
-            }
-            let access: WorkingDirectoryRepositoryAccessRequest = serde_json::from_value(
-                serde_json::to_value(access)
-                    .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?,
-            )
-            .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
-            *effects_started = true;
-            let runtime = self.clone();
-            // The public embedded interface is synchronous. A separate IO executor
-            // avoids nested Tokio block_on while retaining the lifecycle lock.
-            std::thread::spawn(move || {
-                let executor = tokio::runtime::Builder::new_current_thread()
-                    .enable_all()
-                    .build()
-                    .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
-                executor.block_on(
-                    runtime.authorize_working_directory_repository_access_from_resource(access),
-                )
-            })
-            .join()
-            .map_err(|_| RuntimeError::StatePoisoned)??;
+            self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments, true)?;
         }
+        // Retained legacy SSH preparation is not a prerequisite or an effect
+        // of restoring a Worker. Sessions resolve live access when used.
         Ok(())
     }
 
@@ -2864,6 +2654,7 @@ impl Runtime {
                 previous_workdir_attachments: worker.workdir_attachments.clone(),
                 logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
                 workdir_attachments: BTreeMap::new(),
+                deferred_workdir_sessions: BTreeMap::new(),
                 config_bundle: None,
             };
             (
@@ -2876,28 +2667,6 @@ impl Runtime {
             )
         };
 
-        let _workdir_operation = self.reserve_workdir_use(
-            requested_workdir_ids(&request.request)
-                .into_iter()
-                .chain(
-                    request
-                        .previous_workdir_attachments
-                        .iter()
-                        .map(|attachment| {
-                            attachment
-                                .working_directory
-                                .summary
-                                .working_directory_id
-                                .as_str()
-                        }),
-                )
-                .chain(
-                    request
-                        .logical_workdir_attachments
-                        .iter()
-                        .map(|attachment| attachment.working_directory_id.as_str()),
-                ),
-        )?;
         if !reconcile && let Err(result) = backend.preflight_restore(&request) {
             // A preflight may attest an already-connected Controller snapshot.
             // Repeated restore is read-only; do not replace its operation owner.
@@ -3136,7 +2905,7 @@ impl Runtime {
                 Ok(receipt)
             }
             WorkerExecutionSpawnResult::ReconciliationRequired {
-                result: _result,
+                result,
                 worker_state,
                 workdir_attachments,
             } => {
@@ -3165,7 +2934,11 @@ impl Runtime {
                 Ok(RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::ReconciliationRequired,
                     "worker_restore_reconciliation_required",
-                    "Worker restore result is uncertain; reread or retry the Worker restore",
+                    sanitize_worker_create_failure_message(&format!(
+                        "Worker restore requires reconciliation ({:?}): {}",
+                        result.outcome,
+                        result.message_or_default(),
+                    )),
                 ))
             }
             WorkerExecutionSpawnResult::Errored(_result) => {
@@ -8944,7 +8717,7 @@ mod tests {
     }
 
     #[test]
-    fn active_workdir_cleanup_rejects_same_id_admissions_without_blocking_other_workers() {
+    fn active_workdir_cleanup_blocks_access_but_allows_reference_only_restore() {
         let (runtime, backend) = runtime_and_backend();
         let owner = scope("workspace-a", "server-a");
         let worker = runtime
@@ -9004,36 +8777,43 @@ mod tests {
             .stop_worker_scoped(&owner, &worker.worker_ref, None)
             .unwrap();
         let restore = runtime.test_restore_request(&worker.worker_ref);
-        assert_workdir_busy(
-            runtime
-                .coordinate_worker_restore_operation(
-                    &worker.worker_ref,
-                    runtime_api::WorkerRestoreCoordinationRequest {
-                        expected_observation_token: restore.expected_observation_token,
-                        request_id: restore.request_id,
-                        preparation: Some(runtime_api::WorkerRestorePreparation {
-                            workdir_attachments: Some(vec![
-                                runtime_api::LogicalWorkdirAttachment {
-                                    alias: attachment.alias,
-                                    working_directory_id: attachment.working_directory_id,
-                                    capabilities: attachment.capabilities,
-                                },
-                            ]),
-                            ..Default::default()
-                        }),
-                    },
-                    Some(&owner),
-                )
-                .unwrap_err(),
-        );
-        assert!(
+        let (result, preparation) = runtime
+            .coordinate_worker_restore_operation(
+                &worker.worker_ref,
+                runtime_api::WorkerRestoreCoordinationRequest {
+                    expected_observation_token: restore.expected_observation_token.clone(),
+                    request_id: restore.request_id.clone(),
+                    preparation: Some(runtime_api::WorkerRestorePreparation {
+                        workdir_attachments: Some(vec![runtime_api::LogicalWorkdirAttachment {
+                            alias: attachment.alias.clone(),
+                            working_directory_id: attachment.working_directory_id.clone(),
+                            capabilities: attachment.capabilities,
+                        }]),
+                        ..Default::default()
+                    }),
+                },
+                Some(&owner),
+            )
+            .unwrap();
+        assert!(result.is_none());
+        assert!(preparation.is_some());
+        let restored = runtime
+            .restore_worker_operation_scoped(&owner, &worker.worker_ref, restore)
+            .unwrap();
+        assert_eq!(restored.state, WorkerRestoreState::Accepted);
+        assert_eq!(
             runtime
                 .lock()
                 .unwrap()
                 .worker(&worker.worker_ref)
                 .unwrap()
-                .pending_restore
-                .is_none()
+                .logical_workdir_attachments,
+            vec![attachment]
+        );
+        assert_workdir_busy(
+            runtime
+                .open_workdir_session_scoped(&owner, id, None)
+                .unwrap_err(),
         );
         assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1);
         // Different IDs and Worker creation remain available during cleanup.
@@ -10795,102 +10575,68 @@ mod tests {
     }
 
     #[test]
-    fn restore_guard_resource_snapshot_requires_workspace_completion_before_execution() {
-        let (runtime, backend) = runtime_and_backend();
-        let scope = scope("workspace-a", "server-a");
-        let worker = runtime
-            .create_worker_scoped(
-                &scope,
-                scoped_task_request("resource coordination", "workspace-a"),
-            )
-            .unwrap();
-        runtime.stop_worker(&worker.worker_ref, None).unwrap();
-        let mut request = runtime.test_restore_request(&worker.worker_ref);
-        let mut preparation = runtime_api::WorkerRestorePreparation::default();
-        preparation.repository_access_workdirs.push(
-            runtime_api::RepositoryAccessWorkdirReference {
-                runtime_id: "peer-runtime".into(),
-                working_directory_id: "peer-workdir".into(),
-            },
-        );
-        let mut coordination = runtime_api::WorkerRestoreCoordinationRequest {
-            expected_observation_token: request.expected_observation_token.clone(),
-            request_id: request.request_id.clone(),
-            preparation: Some(preparation.clone()),
-        };
-        runtime
-            .coordinate_worker_restore_operation(
-                &worker.worker_ref,
-                coordination.clone(),
-                Some(&scope),
-            )
-            .unwrap();
-        let operation_id = runtime
-            .lock()
-            .unwrap()
-            .worker(&worker.worker_ref)
-            .unwrap()
-            .pending_restore
-            .unwrap()
-            .operation_id;
-        // Model a retained resource snapshot before the peer authorization callback completes.
-        preparation
-            .repository_access
-            .push(runtime_api::WorkingDirectoryRepositoryAccessRequest {
-                working_directory_id: "peer-workdir".into(),
-                materialization: runtime_api::RepositoryMaterializationContext {
-                    workspace_id: "workspace-a".into(),
-                    runtime_id: "peer-runtime".into(),
-                    operation_id: format!("worker-restore:{}", request.request_id),
-                    config_projection_digest: "snapshot".into(),
-                    ssh: None,
+    fn retained_restore_resource_snapshot_does_not_require_workspace_completion() {
+        for completed in [false, true] {
+            let (runtime, backend) = runtime_and_backend();
+            let scope = scope("workspace-a", "server-a");
+            let worker = runtime
+                .create_worker_scoped(
+                    &scope,
+                    scoped_task_request("resource coordination", "workspace-a"),
+                )
+                .unwrap();
+            runtime.stop_worker(&worker.worker_ref, None).unwrap();
+            let mut request = runtime.test_restore_request(&worker.worker_ref);
+            let mut preparation = runtime_api::WorkerRestorePreparation::default();
+            preparation.repository_access_workdirs.push(
+                runtime_api::RepositoryAccessWorkdirReference {
+                    runtime_id: "offline-peer".into(),
+                    working_directory_id: "peer-workdir".into(),
                 },
-            });
-        coordination.preparation = Some(preparation.clone());
-        assert!(
-            runtime
-                .coordinate_worker_restore_operation(&worker.worker_ref, coordination, Some(&scope))
-                .unwrap()
-                .1
-                .is_some()
-        );
-        request.preparation = None;
-        assert_eq!(
-            runtime
+            );
+            if completed {
+                preparation.repository_access.push(
+                    runtime_api::WorkingDirectoryRepositoryAccessRequest {
+                        working_directory_id: "peer-workdir".into(),
+                        materialization: runtime_api::RepositoryMaterializationContext {
+                            workspace_id: "workspace-a".into(),
+                            runtime_id: "offline-peer".into(),
+                            operation_id: format!("worker-restore:{}", request.request_id),
+                            config_projection_digest: "snapshot".into(),
+                            ssh: None,
+                        },
+                    },
+                );
+            }
+            for include_resources in [false, completed] {
+                let mut admitted = preparation.clone();
+                if !include_resources {
+                    admitted.repository_access.clear();
+                }
+                runtime
+                    .coordinate_worker_restore_operation(
+                        &worker.worker_ref,
+                        runtime_api::WorkerRestoreCoordinationRequest {
+                            expected_observation_token: request.expected_observation_token.clone(),
+                            request_id: request.request_id.clone(),
+                            preparation: Some(admitted),
+                        },
+                        Some(&scope),
+                    )
+                    .unwrap();
+            }
+            request.preparation = None;
+            let restored = runtime
                 .restore_worker_operation_scoped(&scope, &worker.worker_ref, request.clone())
-                .unwrap()
-                .state,
-            WorkerRestoreState::ReconciliationRequired
-        );
-        assert_eq!(
-            runtime
-                .restore_admitted_worker_under_lock(&worker.worker_ref, WorkerRestoreMode::Explicit)
-                .unwrap()
-                .state,
-            WorkerRestoreState::ReconciliationRequired
-        );
-        assert_eq!(*backend.restore_count.lock().unwrap(), 0);
-        request.preparation = Some(runtime_api::WorkerRestorePreparation::default());
-        assert_eq!(
-            runtime
-                .restore_worker_operation_scoped(&scope, &worker.worker_ref, request.clone())
-                .unwrap()
-                .state,
-            WorkerRestoreState::ReconciliationRequired
-        );
-        assert_eq!(*backend.restore_count.lock().unwrap(), 0);
-        request.preparation = Some(preparation);
-        assert_eq!(
-            runtime
+                .unwrap();
+            assert_eq!(restored.state, WorkerRestoreState::Accepted);
+            let replay = runtime
                 .restore_worker_operation_scoped(&scope, &worker.worker_ref, request)
-                .unwrap()
-                .state,
-            WorkerRestoreState::Accepted
-        );
-        assert_eq!(
-            backend.restore_operation_ids.lock().unwrap().as_slice(),
-            &[operation_id]
-        );
+                .unwrap();
+            assert_eq!(replay, restored);
+            assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+            assert!(backend.repository_accesses.lock().unwrap().is_empty());
+        }
     }
 
     #[test]
@@ -12434,6 +12180,7 @@ mod tests {
                 session: protocol::SessionSnapshot {
                     pending_submissions: protocol::PendingSubmissionsSnapshot::default(),
                     entries: vec![protocol::SessionSnapshotEntry {
+                        use_policy: None,
                         entry_id: "restored-log-entry".to_owned(),
                         timestamp: 1,
                         provenance: protocol::SessionEntryProvenance::LegacyUnknown,
@@ -14007,6 +13754,129 @@ mod tests {
         let _ = std::fs::remove_dir_all(root);
     }
 
+    #[test]
+    fn restore_reconciliation_exposes_bounded_redacted_backend_reason() {
+        for has_state in [false, true] {
+            let (runtime, backend) = runtime_and_backend();
+            let worker = runtime
+                .create_worker(task_request("restore diagnosis"))
+                .unwrap();
+            let worker_state = if has_state {
+                runtime
+                    .worker_detail(&worker.worker_ref)
+                    .unwrap()
+                    .worker_state
+            } else {
+                None
+            };
+            runtime.stop_worker(&worker.worker_ref, None).unwrap();
+            *backend.restore_result.lock().unwrap() = Some(
+                WorkerExecutionSpawnResult::ReconciliationRequired {
+                    result: WorkerExecutionResult::busy(
+                        WorkerExecutionOperation::Restore,
+                        format!(
+                            "controller acknowledgement timed out at /private/runtime/worker.sock token=restore-token password hidden-value {}",
+                            "x".repeat(512),
+                        ),
+                    ),
+                    worker_state,
+                    workdir_attachments: Vec::new(),
+                },
+            );
+            let result = runtime
+                .restore_worker_operation(
+                    &worker.worker_ref,
+                    runtime.test_restore_request(&worker.worker_ref),
+                )
+                .unwrap();
+            assert_eq!(result.state, WorkerRestoreState::ReconciliationRequired);
+            assert_eq!(
+                result.reason_code.as_deref(),
+                Some("worker_restore_reconciliation_required")
+            );
+            let message = result.message.unwrap();
+            assert!(
+                message.contains("controller acknowledgement timed out"),
+                "{message}"
+            );
+            assert!(message.contains("Busy"), "{message}");
+            assert!(!message.contains("/private"), "{message}");
+            assert!(!message.contains("restore-token"), "{message}");
+            assert!(!message.contains("hidden-value"), "{message}");
+            assert!(message.chars().count() <= WORKER_CREATE_FAILURE_MESSAGE_LIMIT + 1);
+            assert!(
+                runtime
+                    .lock()
+                    .unwrap()
+                    .worker(&worker.worker_ref)
+                    .unwrap()
+                    .pending_restore
+                    .is_some()
+            );
+            assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+        }
+    }
+
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn startup_restore_persists_redacted_reconciliation_reason() {
+        // Real Runtime startup and durable diagnostics; only execution is substituted.
+        let dir = tempfile::tempdir().unwrap();
+        let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
+            .with_runtime_id("restore-diagnostics");
+        let initial = Runtime::with_fs_store_and_execution_backend(
+            options.clone(),
+            Arc::new(TestExecutionBackend::default()),
+        )
+        .unwrap();
+        initial.store_config_bundle(test_bundle()).unwrap();
+        let worker = initial
+            .create_worker(task_request("startup diagnosis"))
+            .unwrap();
+        drop(initial);
+        let backend = Arc::new(TestExecutionBackend::default());
+        *backend.restore_result.lock().unwrap() =
+            Some(WorkerExecutionSpawnResult::ReconciliationRequired {
+                result: WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Restore,
+                    "controller connection timed out at /private/worker.sock token=hidden-token",
+                ),
+                worker_state: None,
+                workdir_attachments: Vec::new(),
+            });
+        let restored =
+            Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
+        assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+        assert!(
+            restored
+                .lock()
+                .unwrap()
+                .worker(&worker.worker_ref)
+                .unwrap()
+                .pending_restore
+                .is_some()
+        );
+        drop(restored);
+        let reopened = Runtime::with_fs_store(options).unwrap();
+        let state = reopened.lock().unwrap();
+        let diagnostic = state
+            .diagnostics
+            .iter()
+            .find(|diagnostic| {
+                diagnostic.code == "worker_execution_restore_failed"
+                    && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
+            })
+            .unwrap();
+        assert!(
+            diagnostic
+                .message
+                .contains("controller connection timed out")
+        );
+        assert!(diagnostic.message.contains("Errored"));
+        assert!(!diagnostic.message.contains("/private"));
+        assert!(!diagnostic.message.contains("hidden-token"));
+    }
+
     #[test]
     fn rolled_back_restore_settles_the_pending_operation() {
         let (runtime, backend) = runtime_and_backend();

crates/worker-runtime/src/session_public.rs

@@ -398,6 +398,7 @@ mod tests {
                 content: vec![LoggedContentPart::Text { text: text.into() }],
             },
             metadata: LoggedSessionHistoryMetadata {
+                input_metadata: None,
                 entry_id: LoggedSessionHistoryEntryId(id.to_string()),
                 origin: LoggedSessionHistoryOrigin::HumanInput {
                     account_id: "account-1".to_string(),

crates/worker-runtime/src/worker_backend/deferred_workdir.rs

@@ -0,0 +1,322 @@
+//! Restore retains attachment identity without opening a provider. Failed opens
+//! are retryable at operation time; closing an unopened session never opens it.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use tokio::sync::{Mutex, broadcast};
+use workdir::*;
+
+type OpenSession = dyn Fn() -> Result<WorkdirSessionHandle, String> + Send + Sync;
+
+pub(super) struct DeferredWorkdirSession {
+    workdir: Workdir,
+    capabilities: WorkdirSessionCapabilities,
+    open: Arc<OpenSession>,
+    state: Mutex<State>,
+    events: broadcast::Sender<CommandEvent>,
+}
+
+#[derive(Default)]
+struct State {
+    session: Option<WorkdirSessionHandle>,
+    event_bridge: Option<tokio::task::JoinHandle<()>>,
+    closed: bool,
+}
+
+impl Drop for State {
+    fn drop(&mut self) {
+        if let Some(task) = &self.event_bridge {
+            task.abort();
+        }
+    }
+}
+
+impl std::fmt::Debug for DeferredWorkdirSession {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DeferredWorkdirSession")
+            .field("workdir", &self.workdir)
+            .finish_non_exhaustive()
+    }
+}
+
+impl DeferredWorkdirSession {
+    pub(super) fn handle(
+        workdir_id: &str,
+        capabilities: WorkdirSessionCapabilities,
+        open: impl Fn() -> Result<WorkdirSessionHandle, String> + Send + Sync + 'static,
+    ) -> WorkdirSessionHandle {
+        Arc::new(Self {
+            workdir: Workdir::new(workdir_id),
+            capabilities,
+            open: Arc::new(open),
+            state: Mutex::new(State::default()),
+            events: broadcast::channel(1024).0,
+        })
+    }
+
+    async fn session(&self) -> Result<WorkdirSessionHandle, WorkdirError> {
+        let mut state = self.state.lock().await;
+        if state.closed {
+            return Err(WorkdirError::SessionClosed);
+        }
+        if let Some(session) = &state.session {
+            return Ok(session.clone());
+        }
+        let open = self.open.clone();
+        let session = tokio::task::spawn_blocking(move || open())
+            .await
+            .map_err(|_| WorkdirError::Unavailable("Workdir session opener failed".into()))?
+            .map_err(WorkdirError::Unavailable)?;
+        if session.workdir().id() != self.workdir.id() {
+            session.close().await?;
+            return Err(WorkdirError::Unavailable(
+                "Workdir session identity mismatch".into(),
+            ));
+        }
+        if let Some(mut source) = session.subscribe_command_events() {
+            let events = self.events.clone();
+            state.event_bridge = Some(tokio::spawn(async move {
+                loop {
+                    match source.recv().await {
+                        Ok(event) => {
+                            let _ = events.send(event);
+                        }
+                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Closed) => break,
+                    }
+                }
+            }));
+        }
+        state.session = Some(session.clone());
+        Ok(session)
+    }
+}
+
+macro_rules! impl_deferred_session {
+    ($( $method:ident($request:ident: $request_type:ty) -> $result:ty; )*) => {
+        #[async_trait]
+        impl WorkdirSession for DeferredWorkdirSession {
+            fn workdir(&self) -> &Workdir {
+                &self.workdir
+            }
+
+            fn capabilities(&self) -> WorkdirSessionCapabilities {
+                self.capabilities
+            }
+
+            $(
+                async fn $method(&self, $request: $request_type) -> Result<$result, WorkdirError> {
+                    self.session().await?.$method($request).await
+                }
+            )*
+
+            fn subscribe_command_events(&self) -> Option<broadcast::Receiver<CommandEvent>> {
+                Some(self.events.subscribe())
+            }
+
+            fn command_snapshot(&self) -> Vec<CommandSnapshot> {
+                self.state.try_lock().ok().and_then(|state| {
+                    state.session.as_ref().map(|session| session.command_snapshot())
+                }).unwrap_or_default()
+            }
+
+            async fn close(&self) -> Result<(), WorkdirError> {
+                let mut state = self.state.lock().await;
+                state.closed = true;
+                if let Some(session) = &state.session {
+                    session.close().await?;
+                }
+                if let Some(task) = state.event_bridge.take() {
+                    task.abort();
+                    let _ = task.await;
+                }
+                Ok(())
+            }
+        }
+    };
+}
+
+impl_deferred_session! {
+    authorize_scope_path(request: WorkdirScopeAuthorizationRequest) -> ();
+    scope_rules_overlap(request: WorkdirScopeOverlapRequest) -> bool;
+    checkout_search(request: CheckoutSearchRequest) -> CheckoutSearchResult;
+    checkout_observe(path: WorkdirPath) -> CheckoutObservation;
+    checkout_execute(request: CheckoutRequest) -> CheckoutResult;
+    stat(request: StatRequest) -> StatResult;
+    read(request: ReadRequest) -> ReadResult;
+    read_bytes(request: ReadBytesRequest) -> ReadBytesResult;
+    write(request: WriteRequest) -> WriteResult;
+    edit(request: EditRequest) -> EditResult;
+    list(request: ListRequest) -> ListResult;
+    glob(request: GlobRequest) -> GlobResult;
+    grep(request: GrepRequest) -> GrepResult;
+    start_command(request: CommandRequest) -> CommandHandle;
+    command_status(handle: CommandHandle) -> CommandStatus;
+    command_output(request: CommandOutputRequest) -> CommandOutput;
+    cancel_command(handle: CommandHandle) -> ();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    fn short_tempdir() -> tempfile::TempDir {
+        let target = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .unwrap()
+            .parent()
+            .unwrap()
+            .join("target");
+        std::fs::create_dir_all(&target).unwrap();
+        tempfile::Builder::new()
+            .prefix("d")
+            .rand_bytes(3)
+            .tempdir_in(target)
+            .unwrap()
+    }
+
+    fn read_request() -> ReadRequest {
+        ReadRequest {
+            path: WorkdirPath::new("README.md").unwrap(),
+            offset: 0,
+            limit: 10,
+            max_bytes: 1024,
+        }
+    }
+
+    #[tokio::test]
+    async fn unopened_session_close_does_not_connect_and_is_terminal() {
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let observed = attempts.clone();
+        let session = DeferredWorkdirSession::handle(
+            "offline",
+            WorkdirSessionCapabilities::READ_ONLY,
+            move || {
+                observed.fetch_add(1, Ordering::SeqCst);
+                Err("offline".into())
+            },
+        );
+        assert_eq!(session.workdir().id().as_str(), "offline");
+        assert!(session.subscribe_command_events().is_some());
+        session.close().await.unwrap();
+        session.close().await.unwrap();
+        assert!(matches!(
+            session.read(read_request()).await,
+            Err(WorkdirError::SessionClosed)
+        ));
+        assert_eq!(attempts.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn router_cleanup_of_unopened_session_never_connects_or_follows_reused_alias() {
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let observed = attempts.clone();
+        let session = DeferredWorkdirSession::handle(
+            "old",
+            WorkdirSessionCapabilities::READ_ONLY,
+            move || {
+                observed.fetch_add(1, Ordering::SeqCst);
+                Err("offline".into())
+            },
+        );
+        let router = WorkdirSessionRouter::new();
+        let alias = WorkdirAttachmentAlias::new("w").unwrap();
+        let routed = router.attach(alias.clone(), session.clone()).unwrap();
+        assert!(routed.subscribe_command_events().is_some());
+        assert!(routed.command_snapshot().is_empty());
+        router.close_all().await.unwrap();
+        router.close_all().await.unwrap();
+        assert!(router.aliases().is_empty());
+        assert!(matches!(
+            session.read(read_request()).await,
+            Err(WorkdirError::SessionClosed)
+        ));
+        assert_eq!(attempts.load(Ordering::SeqCst), 0);
+
+        let new_attempts = Arc::new(AtomicUsize::new(0));
+        let new_observed = new_attempts.clone();
+        let replacement = DeferredWorkdirSession::handle(
+            "new",
+            WorkdirSessionCapabilities::READ_ONLY,
+            move || {
+                new_observed.fetch_add(1, Ordering::SeqCst);
+                Err("offline".into())
+            },
+        );
+        let new_routed = router.attach(alias, replacement).unwrap();
+        // A retained old connection cannot open the replacement provider.
+        assert!(routed.read(read_request()).await.is_err());
+        assert_eq!(new_attempts.load(Ordering::SeqCst), 0);
+        assert!(matches!(
+            new_routed.read(read_request()).await,
+            Err(WorkdirError::Unavailable(_))
+        ));
+        assert_eq!(new_attempts.load(Ordering::SeqCst), 1);
+        // Even after a failed open, cleanup must not try connecting again.
+        router.close_all().await.unwrap();
+        assert_eq!(new_attempts.load(Ordering::SeqCst), 1);
+        assert_eq!(attempts.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn failed_open_retries_on_use_and_keeps_read_only_capabilities() {
+        let temp = short_tempdir();
+        std::fs::write(temp.path().join("README.md"), "restored").unwrap();
+        let root = temp.path().to_path_buf();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let observed = attempts.clone();
+        let session = DeferredWorkdirSession::handle(
+            "saved",
+            WorkdirSessionCapabilities::READ_ONLY,
+            move || {
+                if observed.fetch_add(1, Ordering::SeqCst) == 0 {
+                    return Err("offline".into());
+                }
+                let scope = super::super::runtime_attachment_scope(
+                    &root,
+                    WorkdirSessionCapabilities::READ_ONLY,
+                )?;
+                Ok(super::super::runtime_local_workdir_session(
+                    "saved",
+                    &root,
+                    &root,
+                    scope,
+                    WorkdirSessionCapabilities::READ_ONLY,
+                    Default::default(),
+                    Vec::new(),
+                ))
+            },
+        );
+        assert_eq!(attempts.load(Ordering::SeqCst), 0);
+        assert!(matches!(
+            session.read(read_request()).await,
+            Err(WorkdirError::Unavailable(_))
+        ));
+        assert_eq!(
+            session.read(read_request()).await.unwrap().bytes,
+            b"restored"
+        );
+        assert_eq!(
+            session.read(read_request()).await.unwrap().bytes,
+            b"restored"
+        );
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert!(
+            session
+                .write(WriteRequest {
+                    path: WorkdirPath::new("README.md").unwrap(),
+                    content: b"forbidden".to_vec(),
+                    expected_hash: None
+                })
+                .await
+                .is_err()
+        );
+        assert_eq!(
+            std::fs::read(temp.path().join("README.md")).unwrap(),
+            b"restored"
+        );
+        session.close().await.unwrap();
+    }
+}

crates/worker-runtime/src/worker_backend_restore_failure_tests.rs

@@ -2,33 +2,137 @@
 // Worker/bootstrap, local Git Workdir provider, and allocation table. Model I/O
 // is not invoked; this is a deterministic ordinary suite, not product E2E.
 
+fn restore_failure_tempdir() -> tempfile::TempDir {
+    // Explicitly stay inside this checkout, even when TMPDIR points elsewhere.
+    // Keep fixture components short; no symlink/alias to evade socket limits.
+    let target = Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap()
+        .parent()
+        .unwrap()
+        .join("target");
+    fs::create_dir_all(&target).unwrap();
+    tempfile::Builder::new()
+        .prefix("r")
+        .rand_bytes(3)
+        .tempdir_in(target)
+        .unwrap()
+}
+
+struct RestoreOpenCountingMaterializer {
+    inner: RuntimeGitMaterializer,
+    opens: Arc<AtomicUsize>,
+}
+
+impl WorkingDirectoryMaterializer for RestoreOpenCountingMaterializer {
+    fn materialize(
+        &self,
+        worker_ref: &WorkerRef,
+        request: &WorkingDirectoryRequest,
+    ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
+        self.inner.materialize(worker_ref, request)
+    }
+
+    fn create(
+        &self,
+        request: &WorkingDirectoryRequest,
+    ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
+        self.inner.create(request)
+    }
+
+    fn authorize_repository_access(
+        &self,
+        request: &WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), WorkingDirectoryDiagnostic> {
+        self.inner.authorize_repository_access(request)
+    }
+
+    fn observe_repository_ref(
+        &self,
+        request: &RepositoryRefObservationRequest,
+    ) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
+        self.inner.observe_repository_ref(request)
+    }
+
+    fn bind_working_directory(
+        &self,
+        id: &str,
+        relative_cwd: Option<&str>,
+    ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
+        self.opens.fetch_add(1, Ordering::SeqCst);
+        self.inner.bind_working_directory(id, relative_cwd)
+    }
+
+    fn preflight_bind_working_directory(
+        &self,
+        id: &str,
+        relative_cwd: Option<&str>,
+    ) -> Result<(), WorkingDirectoryDiagnostic> {
+        self.inner
+            .preflight_bind_working_directory(id, relative_cwd)
+    }
+
+    fn list_working_directories(
+        &self,
+    ) -> Result<Vec<WorkingDirectoryStatus>, WorkingDirectoryDiagnostic> {
+        self.inner.list_working_directories()
+    }
+
+    fn working_directory_status(
+        &self,
+        id: &str,
+    ) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
+        self.inner.working_directory_status(id)
+    }
+
+    fn cleanup_working_directory(
+        &self,
+        id: &str,
+    ) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
+        self.inner.cleanup_working_directory(id)
+    }
+
+    fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> {
+        self.inner.cleanup(binding)
+    }
+}
+
 struct ProfileFailureFixture {
     runtime: EmbeddedRuntime,
     backend: Arc<WorkerRuntimeExecutionBackend>,
     worker: crate::catalog::WorkerDetail,
-    _repo: tempfile::TempDir,
+    provider_opens: Arc<AtomicUsize>,
     root: tempfile::TempDir,
 }
 
 impl ProfileFailureFixture {
     fn new(timeout: Duration) -> Self {
-        let root = tempfile::tempdir().unwrap();
-        let repo = create_clean_repo();
-        let store = root.path().join("private-restore-test/runtime");
+        let root = restore_failure_tempdir();
+        let repo = root.path().join("g");
+        fs::create_dir(&repo).unwrap();
+        git(&repo, &["init"]);
+        git(&repo, &["config", "user.email", "test@example.invalid"]);
+        git(&repo, &["config", "user.name", "Yoi Test"]);
+        fs::write(repo.join("README.md"), "clean\n").unwrap();
+        git(&repo, &["add", "README.md"]);
+        git(&repo, &["commit", "-m", "init"]);
+        let store = root.path().join("s");
+        let provider_opens = Arc::new(AtomicUsize::new(0));
         let backend = Arc::new(
             WorkerRuntimeExecutionBackend::new(
                 ProfileRuntimeWorkerFactory::new(root.path()).with_runtime_store_dir(&store),
             )
             .unwrap()
             .with_spawn_restore_timeout(timeout)
-            .with_working_directory_materializer(RuntimeGitMaterializer::new(
-                root.path().join("workdirs"),
-            )),
+            .with_working_directory_materializer(RestoreOpenCountingMaterializer {
+                inner: RuntimeGitMaterializer::new(root.path().join("w")),
+                opens: provider_opens.clone(),
+            }),
         );
         let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
             crate::fs_store::FsRuntimeStoreOptions {
                 root: store,
-                runtime_id: "restore-failure-test".into(),
+                runtime_id: "r".into(),
                 display_name: None,
             },
             backend.clone(),
@@ -39,13 +143,15 @@ impl ProfileFailureFixture {
         request.profile = ProfileSelector::Builtin("default".into());
         request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest {
             alias: WorkdirAttachmentAlias::new("checkout").unwrap(),
-            working_directory: working_directory_request(repo.path()),
+            working_directory: working_directory_request(&repo),
         }];
         let worker = runtime.create_worker(request).unwrap();
         runtime.stop_worker(&worker.worker_ref, None).unwrap();
+        // Count restore provider access, not the initial create/stop setup.
+        provider_opens.store(0, Ordering::SeqCst);
         Self {
             root,
-            _repo: repo,
+            provider_opens,
             runtime,
             backend,
             worker,
@@ -57,6 +163,7 @@ impl ProfileFailureFixture {
     }
 
     fn assert_released(&self) {
+        assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
         assert!(self.backend.pending_restores.lock().unwrap().is_empty());
         assert!(self.backend.workers.lock().unwrap().is_empty());
         assert!(
@@ -176,16 +283,22 @@ impl ProfileFailureFixture {
                 .load(Ordering::SeqCst),
             calls + 1
         );
+        // Restore/cleanup and old-request replay must not provider-open the
+        // retained attachment, including after a new execution was accepted.
+        assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
         // Only tear down after the fresh explicit Restore has already succeeded.
         self.runtime
             .stop_worker(&self.worker.worker_ref, None)
             .unwrap();
+        assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
     }
 }
 
 #[test]
 #[serial_test::serial(worker_allocation)]
 fn profile_restore_multistage_failure_cleans_then_fresh_restore_without_stop() {
+    // Binding means runtime Worker/attachment-router binding, not eager provider
+    // materialization. Provider access stays deferred through every boundary.
     for stage in [
         RuntimeRestoreStage::Reconstruction,
         RuntimeRestoreStage::Binding,
@@ -224,8 +337,7 @@ fn profile_restore_multistage_failure_cleans_then_fresh_restore_without_stop() {
 
 #[test]
 #[serial_test::serial(worker_allocation)]
-fn profile_restore_cleanup_failure_retains_owner_and_original_cause_until_same_operation_recovers()
-{
+fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_deferred_provider() {
     let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
     fixture.fail(RuntimeRestoreStage::Feature);
     fixture
@@ -265,6 +377,20 @@ fn profile_restore_cleanup_failure_retains_owner_and_original_cause_until_same_o
         "{message}"
     );
     assert!(message.contains("Workdir"), "{message}");
+    {
+        let retained = pending.result.lock().unwrap();
+        let failure = retained.as_ref().unwrap().as_ref().err().unwrap();
+        assert_eq!(failure.stage, RuntimeRestoreStage::Feature);
+        assert_eq!(failure.resources, RuntimeRestoreResources::WorkerStartup);
+        assert_eq!(
+            failure.cleanup,
+            RuntimeRestoreCleanup::Failed(RuntimeRestoreCleanupFailure::Workdir)
+        );
+        assert_eq!(
+            failure.cleanup_failure,
+            Some(RuntimeRestoreCleanupFailure::Workdir)
+        );
+    }
     let owned = fixture
         .backend
         .factory
@@ -274,6 +400,9 @@ fn profile_restore_cleanup_failure_retains_owner_and_original_cause_until_same_o
         .get(&fixture.worker.worker_ref)
         .cloned()
         .unwrap();
+    assert_eq!(fixture.provider_opens.load(Ordering::SeqCst), 0);
+    assert!(pending.request.workdir_attachments.is_empty());
+    assert!(!pending.request.deferred_workdir_sessions.is_empty());
     assert_eq!(owned.operation_id, pending.request.operation_id);
     assert!(owned.allocation.is_some());
     assert!(
@@ -307,6 +436,7 @@ fn profile_restore_cleanup_failure_retains_owner_and_original_cause_until_same_o
             .load(Ordering::SeqCst),
         1
     );
+    assert_eq!(fixture.provider_opens.load(Ordering::SeqCst), 0);
     fixture
         .backend
         .factory
@@ -334,6 +464,27 @@ fn profile_restore_cleanup_failure_retains_owner_and_original_cause_until_same_o
             .all(|sessions| sessions.aliases().is_empty())
     );
     fixture.assert_released();
+    let unopened = pending.request.deferred_workdir_sessions.clone();
+    fixture
+        .backend
+        .run_on_adapter_runtime(async move {
+            for session in unopened.values() {
+                assert!(matches!(
+                    session
+                        .read(workdir::ReadRequest {
+                            path: workdir::WorkdirPath::new("README.md").unwrap(),
+                            offset: 0,
+                            limit: 10,
+                            max_bytes: 1024,
+                        })
+                        .await,
+                    Err(workdir::WorkdirError::SessionClosed)
+                ));
+            }
+            Ok(())
+        })
+        .unwrap();
+    assert_eq!(fixture.provider_opens.load(Ordering::SeqCst), 0);
     eprintln!(
         "restore-fault cleanup=Workdir owner-retained=true retry=same-operation outcome=RolledBack"
     );
@@ -466,7 +617,19 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
         "{diagnostic}"
     );
     assert!(diagnostic.contains("InvalidInput"), "{diagnostic}");
-    assert!(!diagnostic.contains("private-restore-test"));
+    // This case deliberately exercises the real Unix socket path-limit error;
+    // positive restore cases retain the factory's normal InProcess transport.
+    // Short fixture roots must not accidentally turn it into a successful bind.
+    let operation_id =
+        fixture.backend.completed_restore_failures.lock().unwrap()[&fixture.worker.worker_ref].0;
+    let socket = fixture
+        .backend
+        .factory
+        .worker_restore_run_dir(&fixture.worker.worker_ref, operation_id)
+        .unwrap()
+        .join("worker.sock");
+    assert!(socket.as_os_str().as_encoded_bytes().len() >= 108);
+    assert!(!diagnostic.contains(&fixture.root.path().to_string_lossy().to_string()));
     fixture.assert_released();
     fixture
         .backend
@@ -844,102 +1007,150 @@ fn profile_restore_failure_settlement_write_retry_preserves_completed_cause_with
     }
 }
 
-struct RacedBindingFactory {
-    changed_checkout: PathBuf,
-    preflights: AtomicUsize,
-}
-
-#[async_trait]
-impl RuntimeWorkerFactory for RacedBindingFactory {
-    async fn spawn_controller(
-        &self,
-        _: WorkerExecutionSpawnRequest,
-    ) -> Result<RuntimeWorkerController, String> {
-        panic!("binding-only failure must not create a controller")
-    }
-    async fn preflight_restore(&self, _: &WorkerExecutionRestoreRequest) -> Result<(), String> {
-        // Inject another owner's checkout change AFTER the real materializer's
-        // read-only checks, but before its mutable bind. Recovery stays real.
-        if self.preflights.fetch_add(1, Ordering::SeqCst) == 0 {
-            fs::remove_dir_all(&self.changed_checkout).unwrap();
-        }
-        Ok(())
-    }
-    async fn restore_controller(
-        &self,
-        _: WorkerExecutionRestoreRequest,
-    ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
-        panic!("factory execution must not run after partial binding failed")
-    }
-}
-
+// Preflight accepts the real retained Profile source, but reconstruction
+// discovers that its committed segment has no system prompt. No fault checkpoint
+// supplies this Err.
 #[test]
-fn partial_restore_binding_failure_keeps_exact_operation_without_factory_cleanup_claim() {
-    let root = tempfile::tempdir().unwrap();
-    let repo = create_clean_repo();
-    let materializer = RuntimeGitMaterializer::new(root.path());
-    let first = materializer
-        .create(&working_directory_request(repo.path()))
-        .unwrap();
-    let second = materializer
-        .create(&working_directory_request(repo.path()))
-        .unwrap();
-    let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
-    request.previous_workdir_attachments = vec![
-        WorkingDirectoryAttachmentStatus {
-            alias: WorkdirAttachmentAlias::new("first").unwrap(),
-            working_directory: first.status(),
-        },
-        WorkingDirectoryAttachmentStatus {
-            alias: WorkdirAttachmentAlias::new("second").unwrap(),
-            working_directory: second.status(),
-        },
-    ];
-    let backend = WorkerRuntimeExecutionBackend::new(RacedBindingFactory {
-        changed_checkout: second.root().to_path_buf(),
-        preflights: AtomicUsize::new(0),
-    })
-    .unwrap()
-    .with_working_directory_materializer(materializer);
-    assert!(matches!(
-        backend.restore_worker(request.clone()),
-        WorkerExecutionSpawnResult::ReconciliationRequired { .. }
-    ));
-    let pending = backend
-        .pending_restore(&request.worker_ref)
-        .unwrap()
-        .unwrap();
-    assert_eq!(pending.request.operation_id, request.operation_id);
-    assert_eq!(pending.request.workdir_attachments.len(), 1);
-    assert_eq!(
-        pending
-            .request
-            .workdir_attachments
-            .values()
-            .next()
-            .unwrap()
-            .root(),
-        first.root()
+#[serial_test::serial(worker_allocation)]
+fn profile_restore_missing_committed_system_prompt_rolls_back_and_replays_classified_cause() {
+    let root = restore_failure_tempdir();
+    let runtime_store = root.path().join("s");
+    let worker_ref = WorkerRef::new(WorkerId::now_v7());
+    let (session_id, segment_id, _) = write_retained_profile_worker(&runtime_store, &worker_ref);
+    let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref);
+    let aggregate_dir = runtime_store
+        .join("workers")
+        .join(worker_ref.worker_id.to_string());
+    let metadata_store = WorkerAggregateStore::new(&aggregate_dir, &worker_name).unwrap();
+    let mut metadata = metadata_store.read_by_name(&worker_name).unwrap().unwrap();
+    let manifest = WorkerManifest::from_toml(
+        r#"
+        [worker]
+        name = "r"
+        [scope]
+        allow = []
+        [model]
+        scheme = "anthropic"
+        model_id = "test"
+        auth = { kind = "none" }
+        [engine]
+        instruction = "default"
+        max_tokens = 100
+        [feature.memory.profile]
+        enabled = false
+        "#,
+    )
+    .unwrap();
+    metadata.resolved_manifest_snapshot =
+        Some(manifest::write_persisted_worker_manifest_snapshot(&manifest).unwrap());
+    metadata_store.write(&metadata).unwrap();
+    let session_store = WorkerSessionStore::new(aggregate_dir.join("session")).unwrap();
+    let entries_before = session_store.read_all(session_id, segment_id).unwrap();
+    assert!(
+        session_store::collect_state(&entries_before)
+            .system_prompt
+            .is_none()
     );
-    for _ in 0..2 {
-        assert!(matches!(
-            backend.reconcile_restore(request.clone()),
-            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
-        ));
+
+    let backend = WorkerRuntimeExecutionBackend::new(
+        ProfileRuntimeWorkerFactory::new(root.path().join("p"))
+            .with_runtime_store_dir(&runtime_store),
+    )
+    .unwrap();
+    let request = direct_restore_request(worker_ref.clone());
+    backend.preflight_restore(&request).unwrap();
+    let first = match backend.restore_worker(request.clone()) {
+        WorkerExecutionSpawnResult::RolledBack(result) => result,
+        other => panic!("missing committed prompt must finish rollback: {other:?}"),
+    };
+    let diagnostic = first.message.as_deref().unwrap();
+    assert!(
+        diagnostic.contains("saved segment system prompt missing"),
+        "{diagnostic}"
+    );
+    assert!(diagnostic.contains("Reconstruction"), "{diagnostic}");
+    assert!(diagnostic.contains("Confirmed"), "{diagnostic}");
+    assert!(diagnostic.len() < 512, "{diagnostic}");
+    for private in [
+        root.path().to_string_lossy().into_owned(),
+        session_id.to_string(),
+        segment_id.to_string(),
+        "retained response".into(),
+    ] {
+        assert!(!diagnostic.contains(&private), "{diagnostic}");
     }
-    assert_eq!(backend.factory.preflights.load(Ordering::SeqCst), 1);
-    assert!(!backend.stop_worker(&request.worker_ref).is_accepted());
-    assert!(Arc::ptr_eq(
-        &pending,
-        &backend
-            .pending_restore(&request.worker_ref)
-            .unwrap()
-            .unwrap()
-    ));
+    // A completed Err is not evidence of zero resources. Require the owner to
+    // attest cleanup, and independently check its allocation/artifact authority.
+    {
+        let completed = backend.completed_restore_failures.lock().unwrap();
+        let (operation_id, failure) = completed.get(&worker_ref).unwrap();
+        assert_eq!(*operation_id, request.operation_id);
+        assert_eq!(failure.stage, RuntimeRestoreStage::Reconstruction);
+        assert_eq!(failure.resources, RuntimeRestoreResources::WorkerStartup);
+        assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed);
+        assert_eq!(failure.cleanup_failure, None);
+    }
+    assert!(backend.pending_restore(&worker_ref).unwrap().is_none());
     assert!(backend.workers.lock().unwrap().is_empty());
-    eprintln!(
-        "restore-fault stage=mutable-binding outcome=ReconciliationRequired partial-bindings=retained factory-executions=0"
+    assert!(
+        backend
+            .factory
+            .failed_restore_resources
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    assert!(
+        backend
+            .factory
+            .observation_hub
+            .workers
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    assert!(
+        !backend
+            .factory
+            .worker_restore_run_dir(&worker_ref, request.operation_id)
+            .unwrap()
+            .exists()
+    );
+    {
+        let table = worker::runtime::worker_allocation::LockFileGuard::open(
+            &worker::runtime::worker_allocation::default_allocation_path().unwrap(),
+        )
+        .unwrap();
+        assert!(table.data().find(&worker_name).is_none());
+        assert!(table.data().find("r").is_none());
+    }
+    for replay in [
+        backend.reconcile_restore(request.clone()),
+        backend.restore_worker(request),
+    ] {
+        let WorkerExecutionSpawnResult::RolledBack(result) = replay else {
+            panic!("same request must replay confirmed failure: {replay:?}");
+        };
+        assert_eq!(
+            serde_json::to_value(result).unwrap(),
+            serde_json::to_value(&first).unwrap()
+        );
+    }
+    assert_eq!(
+        backend
+            .factory
+            .restore_faults
+            .executions
+            .load(Ordering::SeqCst),
+        1
+    );
+    assert!(backend.pending_restore(&worker_ref).unwrap().is_none());
+    assert_eq!(
+        metadata_store.read_by_name(&worker_name).unwrap().unwrap(),
+        metadata
+    );
+    assert_eq!(
+        serde_json::to_value(session_store.read_all(session_id, segment_id).unwrap()).unwrap(),
+        serde_json::to_value(entries_before).unwrap()
     );
-    // Fixture teardown drops the retained passive local binding. This is not
-    // evidence that credential/policy materialization has been rolled back.
 }