fix: retain archive and worker restore failure diagnostics

crates/worker-runtime/src/retention.rs

@@ -1258,8 +1258,11 @@ fn for_each_live_session(
             let session_id = match identity {
                 Ok(Some(session_id)) => session_id,
                 Ok(None) => continue,
-                Err(_) if is_archive => {
+                Err(error) if is_archive => {
-                    warn_unavailable_archive(&aggregate);
+                    warn_unavailable_archive(
+                        &aggregate,
+                        ArchiveMigrationFailure::runtime("session_identity", &error),
+                    );
                     continue;
                 }
                 Err(error) => return Err(error),
@@ -1271,9 +1274,11 @@ fn for_each_live_session(
                 Err(session_store::StoreError::SessionErased(_)) => continue,
                 Err(error) => return Err(retention_error(error)),
             }
-            if is_archive && validate_archive_migration(&aggregate).is_err() {
+            if is_archive {
-                warn_unavailable_archive(&aggregate);
+                if let Err(failure) = validate_archive_migration(&aggregate) {
-                continue;
+                    warn_unavailable_archive(&aggregate, failure);
+                    continue;
+                }
             }
             // Publication failures still abort the outer staging transaction.
             visit(&aggregate, is_archive)?;
@@ -1282,7 +1287,75 @@ fn for_each_live_session(
     Ok(())
 }
 
-fn validate_archive_migration(aggregate: &Path) -> Result<(), RuntimeError> {
+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.
@@ -1292,16 +1365,21 @@ fn validate_archive_migration(aggregate: &Path) -> Result<(), RuntimeError> {
     let manifest: WorkerSessionArchiveManifest = read_json(
         &aggregate.join("manifest.json"),
         "read archive before Session migration",
-    )?;
+    )
-    validate_existing_archive(aggregate, &manifest, false)?;
+    .map_err(|error| ArchiveMigrationFailure::runtime("archive_manifest", &error))?;
-    session_store::WorkerSessionStore::validate_migration(&session).map_err(retention_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) {
+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"
     );
 }
@@ -2117,6 +2195,160 @@ mod tests {
         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/runtime.rs

@@ -2893,7 +2893,7 @@ impl Runtime {
                 Ok(receipt)
             }
             WorkerExecutionSpawnResult::ReconciliationRequired {
-                result: _result,
+                result,
                 worker_state,
                 workdir_attachments,
             } => {
@@ -2922,7 +2922,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) => {
@@ -13738,6 +13742,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();