Fix: Run hosted safety refusals before withdrawn sweep

The withdrawn sweep's restore_upstream mutates lockfiles before the
hosted rewrite refusals (bun.lockb symlink, governing root) run. A
member or symlink project that hosted mode would refuse can still have
its lockfiles rewritten, including an ancestor lock the run is not
allowed to touch.

Safety refusals must run before any restore or mutation. This change
adds early refusal checks before the withdrawn sweep:

1. Extract ecosystems from pins that would be restored
2. Check bun.lockb symlink refusal for npm ecosystem
3. Check governing root refusals for cargo/pypi/npm ecosystems
4. Exit with error if any refusal would occur, before any mutation

Refactored governing_root::refusal to share implementation with new
refusal_for_ecosystems function that takes ecosystem flags instead of
candidates, allowing the sweep to check layout safety before restore.

Author: cursoragent

crates/socket-patch-cli/src/commands/scan/mod.rs

@@ -2576,8 +2576,85 @@ async fn run_scan(
     // longer serves, before the rewrite below re-pins any replacement.
     // Runs ahead of every "nothing found" exit: a package whose only patch
     // was withdrawn has no search results at all.
+    //
+    // Safety refusals run BEFORE the restore: a symlink or governing-root
+    // layout that hosted mode refuses must not be mutated.
     let withdrawn_sweep = if hosted && !args.common.is_global() && policy.writes_allowed() {
         let pins = withdrawn::pins_in_scope(&hosted_pin_list, &all_purls);
+
+        // Check layout refusals for the ecosystems involved in the sweep,
+        // before any restore (the sweep's `restore_upstream` would mutate
+        // lockfiles, including an ancestor lock or symlink the refusals
+        // prevent).
+        if !pins.is_empty() {
+            use socket_patch_core::vendor::lock_inventory::ProjectView;
+            let view = ProjectView::Disk(&args.common.cwd);
+
+            // Extract ecosystems from pins.
+            let mut has_npm = false;
+            let mut has_cargo = false;
+            let mut has_pypi = false;
+            for pin in &pins {
+                match Ecosystem::from_purl(&pin.purl) {
+                    Some(Ecosystem::Npm) => has_npm = true,
+                    Some(Ecosystem::Cargo) => has_cargo = true,
+                    Some(Ecosystem::Pypi) => has_pypi = true,
+                    _ => {}
+                }
+            }
+
+            // Check bun.lockb symlink refusal (npm ecosystem).
+            if has_npm {
+                use socket_patch_core::hosted::engine;
+                use socket_patch_core::vendor::lock_inventory::bun_text_lock_drives;
+                let bun_text_drives = bun_text_lock_drives(&view);
+                if !bun_text_drives && view.is_symlink("bun.lockb") {
+                    let refusal = engine::bun_lockb_symlink_refusal();
+                    if args.common.json {
+                        let mut env = scan_envelope(&args.common);
+                        env.mark_error(EnvelopeError::new(&refusal.code, &refusal.message));
+                        emit_scan(&env);
+                    } else {
+                        eprintln!("Error: {}", ui::sentence_case(&refusal.message));
+                    }
+                    return 1;
+                }
+            }
+
+            // Check governing-root refusals (cargo/pypi/npm ecosystems).
+            if has_cargo || has_pypi || has_npm {
+                use socket_patch_core::hosted::governing_root;
+                // Read trust_lockfile_config the way the main rewrite does.
+                let trust_env_off =
+                    socket_patch_core::utils::pnpm_workspace::trust_lockfile_env_for(
+                        &view,
+                        |var| std::env::var(var).ok(),
+                    )
+                    .await
+                    .is_some_and(|found| found.as_bool() != Some(true));
+                let trust_lockfile_config = !args.common.no_trust_lockfile_config && !trust_env_off;
+
+                if let Some(refusal) = governing_root::refusal_for_ecosystems(
+                    &view,
+                    has_cargo,
+                    has_pypi,
+                    has_npm,
+                    trust_lockfile_config,
+                )
+                .await
+                {
+                    if args.common.json {
+                        let mut env = scan_envelope(&args.common);
+                        env.mark_error(EnvelopeError::new(&refusal.code, &refusal.message));
+                        emit_scan(&env);
+                    } else {
+                        eprintln!("Error: {}", ui::sentence_case(&refusal.message));
+                    }
+                    return 1;
+                }
+            }
+        }
+
         Some(withdrawn::sweep(&args.common, &api_client, &pins).await)
     } else {
         None

crates/socket-patch-core/src/hosted/governing_root.rs

@@ -92,20 +92,51 @@ pub async fn refusal(
         ProjectView::Snapshot(snap) => snap.root(),
         ProjectView::Memory(_) => return None,
     };
-    if candidates.iter().any(|c| c.dep.ecosystem == "cargo") {
+    let has_cargo = candidates.iter().any(|c| c.dep.ecosystem == "cargo");
+    let has_pypi = candidates.iter().any(|c| c.dep.ecosystem == "pypi");
+    let has_npm = candidates.iter().any(|c| c.dep.ecosystem == "npm");
+    refusal_for_ecosystems_impl(root, has_cargo, has_pypi, has_npm, trust_lockfile_config).await
+}
+
+/// Like [`refusal`], but takes ecosystem flags instead of candidates. Used by
+/// the withdrawn-pin sweep to check layout safety before any restore.
+pub async fn refusal_for_ecosystems(
+    view: &ProjectView<'_>,
+    has_cargo: bool,
+    has_pypi: bool,
+    has_npm: bool,
+    trust_lockfile_config: bool,
+) -> Option<Refusal> {
+    let root: &Path = match view {
+        ProjectView::Disk(root) => root,
+        ProjectView::Snapshot(snap) => snap.root(),
+        ProjectView::Memory(_) => return None,
+    };
+    refusal_for_ecosystems_impl(root, has_cargo, has_pypi, has_npm, trust_lockfile_config).await
+}
+
+/// Shared implementation for [`refusal`] and [`refusal_for_ecosystems`].
+async fn refusal_for_ecosystems_impl(
+    root: &Path,
+    has_cargo: bool,
+    has_pypi: bool,
+    has_npm: bool,
+    trust_lockfile_config: bool,
+) -> Option<Refusal> {
+    if has_cargo {
         if let Some(refusal) = cargo_member_refusal(root).await {
             return Some(refusal);
         }
     }
-    if candidates.iter().any(|c| c.dep.ecosystem == "pypi") {
+    if has_pypi {
         if let Some(workspace) = crate::utils::uv_workspace::governing_uv_workspace(root).await {
             return Some(Refusal {
                 code: WORKSPACE_LOCKFILE_ELSEWHERE.to_string(),
                 message: crate::utils::uv_workspace::member_detail(root, &workspace),
             });
         }
     }
-    if candidates.iter().any(|c| c.dep.ecosystem == "npm") {
+    if has_npm {
         let workspace = if has_own_npm_family_lock(root) {
             member_stray_lock_refusal(root).await
         } else {