Merge origin/main into arch-fix/takeover-atomic

Resolve the hosted.rs conflict with #1045: this branch moved the
takeover out of hosted.rs into hosted/takeover.rs, so keep that layout
and port #1045's change there. The takeover's ledger drop now compares
PurlKeys and the local canonical_purl key helper is gone.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: mikolalysenko

crates/socket-patch-cli/src/commands/apply.rs

@@ -15,7 +15,8 @@ use socket_patch_core::patch::redirect::golang_local::{
 use socket_patch_core::patch::sidecars::{maven as maven_sidecars, SidecarAdvisoryCode};
 use socket_patch_core::telemetry::{track_patch_applied, track_patch_apply_failed};
 use socket_patch_core::utils::purl::parse_golang_purl;
-use socket_patch_core::utils::purl::{normalize_purl, purl_eq, strip_purl_qualifiers};
+use socket_patch_core::utils::purl::{normalize_purl, strip_purl_qualifiers};
+use socket_patch_core::utils::purl_key::PurlKey;
 use socket_patch_core::vendor::purl_keys_cover;
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::path::{Path, PathBuf};
@@ -103,7 +104,7 @@ async fn ensure_blobs_for_mismatches(
     args: &ApplyArgs,
     manifest: &PatchManifest,
     all_packages: &HashMap<String, Vec<PathBuf>>,
-    vendored_purls: &HashSet<String>,
+    vendored_purls: &HashSet<PurlKey>,
     staged: &mut StagedSources,
     client: &ApiClient,
 ) {
@@ -224,7 +225,7 @@ fn format_mismatch_fetch_result(downloaded: usize, needed: usize) -> String {
 async fn mismatch_blob_gaps(
     manifest: &PatchManifest,
     all_packages: &HashMap<String, Vec<PathBuf>>,
-    vendored_purls: &HashSet<String>,
+    vendored_purls: &HashSet<PurlKey>,
     blobs_path: &Path,
     force: bool,
 ) -> HashSet<String> {
@@ -247,10 +248,11 @@ async fn mismatch_blob_gaps(
         };
         let variant_eco = Ecosystem::from_purl(purl).is_some_and(|e| e.supports_release_variants());
         let stripped = strip_purl_qualifiers(purl);
+        let identity = PurlKey::new(purl);
         let records: Vec<(&String, &PatchRecord)> = manifest
             .patches
             .iter()
-            .filter(|(key, _)| *key == purl || strip_purl_qualifiers(key) == stripped)
+            .filter(|(key, _)| *key == purl || PurlKey::new(key) == identity)
             .collect();
         if purl_keys_cover(vendored_purls, purl)
             || records
@@ -1729,7 +1731,7 @@ async fn report_apply_failure(
 /// main-thread stack in debug builds.
 fn synthesize_vendor_owned_results(
     target_manifest_purls: &HashSet<String>,
-    vendored_purls: &HashSet<String>,
+    vendored_purls: &HashSet<PurlKey>,
 ) -> (Vec<ApplyResult>, HashSet<String>, HashSet<String>) {
     let is_vendored = |p: &str| purl_keys_cover(vendored_purls, p);
     let mut results: Vec<ApplyResult> = Vec::new();
@@ -3264,18 +3266,10 @@ async fn lockfile_resolved(common: &GlobalArgs, unmatched: &[String]) -> HashSet
     }
     let ctx = crate::commands::context::ProjectContext::new(common);
     let entries = &ctx.locks().await.entries;
-    let lock_purls: HashSet<String> = entries
-        .iter()
-        .map(|e| normalize_purl(strip_purl_qualifiers(&e.purl)).into_owned())
-        .collect();
+    let lock_purls: HashSet<PurlKey> = entries.iter().map(|e| PurlKey::new(&e.purl)).collect();
     unmatched
         .iter()
-        .filter(|p| {
-            let base = strip_purl_qualifiers(p);
-            lock_purls.contains(normalize_purl(base).as_ref())
-                || (base.starts_with("pkg:composer/")
-                    && entries.iter().any(|e| purl_eq(&e.purl, base)))
-        })
+        .filter(|p| lock_purls.contains(&PurlKey::new(p)))
         .cloned()
         .collect()
 }
@@ -4001,7 +3995,7 @@ mod tests {
             "without a vendor claim the drifted singleton must queue (fixture sanity)"
         );
 
-        let vendored = HashSet::from(["pkg:gem/foo@1.0.0".to_string()]);
+        let vendored = HashSet::from([PurlKey::new("pkg:gem/foo@1.0.0")]);
         let needed = mismatch_blob_gaps(&manifest, &all_packages, &vendored, &blobs, false).await;
         assert!(
             needed.is_empty(),

crates/socket-patch-cli/src/commands/get.rs

@@ -23,6 +23,7 @@ use socket_patch_core::utils::concurrent::{api_concurrency_for, ordered_concurre
 use socket_patch_core::utils::purl::{
     canonical_purl, is_purl, normalize_purl, strip_purl_qualifiers,
 };
+use socket_patch_core::utils::purl_key::PurlKey;
 use socket_patch_core::vendor::{load_state, lookup_entry, VendorEntry, VendorState};
 use std::collections::HashMap;
 use std::fmt;
@@ -1430,9 +1431,8 @@ struct InstalledNarrowing {
 /// runs — the coarse layer above [`filter_to_installed_releases`]'s
 /// per-release variant narrowing (which still runs later, unchanged).
 ///
-/// Presence evidence per result purl (compared on
-/// `normalize_purl(strip_purl_qualifiers(..))` — API purls are
-/// percent-encoded/qualified, crawler purls literal):
+/// Presence evidence per result purl (compared by [`PurlKey`] — API purls
+/// are percent-encoded/qualified/mixed-case, crawler purls literal):
 /// * installed on disk — `find_packages_for_rollback` over the deduped base
 ///   purls (the qualified-aware resolver; memory invariant);
 /// * already tracked in the manifest — the user opted this purl in earlier,
@@ -1465,8 +1465,6 @@ async fn filter_to_installed_purls(
     use socket_patch_core::vendor::lock_inventory;
     use std::collections::HashSet;
 
-    let canon = canonical_purl;
-
     // Deduped base purls, probed against the installed tree. The resolver
     // keys its result by the purls we pass, so canonicalize the found keys
     // the same way as the membership probes below.
@@ -1480,14 +1478,14 @@ async fn filter_to_installed_purls(
     };
     let partitioned = partition_purls(&bases, None);
     let found = find_packages_for_rollback(&partitioned, &common.crawler_options(), true).await;
-    let mut present: HashSet<String> = found.keys().map(|k| canon(k)).collect();
+    let mut present: HashSet<PurlKey> = found.keys().map(|k| PurlKey::new(k)).collect();
 
     let ctx = super::context::ProjectContext::rooted(common, common.cwd.clone());
     // Manifest membership counts as presence (read-only probe: a corrupt
     // manifest degrades to "no extension" here — the download path's
     // fail-closed read still guards every write).
     if let Some(manifest) = ctx.ledgers().await.manifest {
-        present.extend(manifest.patches.keys().map(|k| canon(k)));
+        present.extend(manifest.patches.keys().map(|k| PurlKey::new(k)));
     }
 
     // scan's lockfile + vendored-ledger discovery supplements (and their
@@ -1498,11 +1496,11 @@ async fn filter_to_installed_purls(
         let supplement = super::scan::project_lockfile_supplement(&ctx, &[], None).await;
         pnp_diags = supplement.unsupported;
         if mode != super::scan::ScanMode::Agent {
-            present.extend(supplement.entries.iter().map(|e| canon(&e.purl)));
+            present.extend(supplement.entries.iter().map(|e| PurlKey::new(&e.purl)));
             let vendored =
                 super::scan::project_vendored_supplement(common, &[], &ctx.loaded().await.vendor)
                     .await;
-            present.extend(vendored.packages.iter().map(|p| canon(&p.purl)));
+            present.extend(vendored.packages.iter().map(|p| PurlKey::new(&p.purl)));
         }
     }
 
@@ -1536,7 +1534,7 @@ async fn filter_to_installed_purls(
         warnings,
     };
     for result in accessible {
-        if present.contains(&canon(&result.purl)) {
+        if present.contains(&PurlKey::new(&result.purl)) {
             out.kept.push(result.clone());
             continue;
         }
@@ -1565,7 +1563,7 @@ async fn filter_to_installed_purls(
             // nothing — so it carries the same `package_not_installed` code
             // a non-PnP pnpm project would get; only an UNREADABLE lock
             // (no judgment possible) keeps the layout-refusal code.
-            let decoded = canon(&result.purl);
+            let decoded = canonical_purl(&result.purl);
             let coord = decoded.strip_prefix("pkg:npm/").unwrap_or(&decoded);
             if mode == super::scan::ScanMode::Hosted {
                 match (&pnpm_pnp_lock, coord.rsplit_once('@')) {
@@ -1780,7 +1778,8 @@ async fn lock_text_refusals_for(
         )
         .await,
     );
-    let claimed: Vec<String> = pins.iter().map(|pin| canonical_purl(&pin.purl)).collect();
+    let claimed: std::collections::HashSet<PurlKey> =
+        pins.iter().map(|pin| PurlKey::new(&pin.purl)).collect();
     let fetchable: Vec<&PatchSearchResult> = selected
         .iter()
         .filter(|sr| bun_refusal.filter(|r| r.applies_to(&sr.purl)).is_none())
@@ -1791,7 +1790,7 @@ async fn lock_text_refusals_for(
         .collect();
     let candidates: Vec<(&str, &str)> = fetchable
         .iter()
-        .filter(|sr| !claimed.contains(&canonical_purl(&sr.purl)))
+        .filter(|sr| !claimed.contains(&PurlKey::new(&sr.purl)))
         .map(|sr| (sr.purl.as_str(), sr.uuid.as_str()))
         .collect();
     let refused = socket_patch_core::vendor::lock_text_refusals(cwd, &candidates).await;
@@ -1815,7 +1814,7 @@ async fn lock_text_refusals_for(
             cwd,
             fetchable
                 .iter()
-                .filter(|sr| claimed.contains(&canonical_purl(&sr.purl)))
+                .filter(|sr| claimed.contains(&PurlKey::new(&sr.purl)))
                 .map(|sr| sr.purl.as_str()),
             &pins,
             false,
@@ -2389,8 +2388,8 @@ async fn run_nested_apply(
 /// qualified record (apply keys a release-variant base by its base purl).
 /// A qualified key never covers a sibling variant.
 fn apply_key_covers(key: &str, record: &str) -> bool {
-    let (key, record) = (normalize_purl(key), normalize_purl(record));
-    key == record || (!key.contains(['?', '#']) && record.split(['?', '#']).next() == Some(&*key))
+    PurlKey::qualified(key) == PurlKey::qualified(record)
+        || (!key.trim().contains(['?', '#']) && PurlKey::same(key, record))
 }
 
 /// Fold a failed nested apply into a `get` / `scan --mode agent` JSON
@@ -2433,7 +2432,8 @@ fn fold_apply_failures(
             }
         }
         let appended = patches[selected..].iter().any(|r| {
-            normalize_purl(r["purl"].as_str().unwrap_or_default()) == normalize_purl(&failure.purl)
+            PurlKey::qualified(r["purl"].as_str().unwrap_or_default())
+                == PurlKey::qualified(&failure.purl)
         });
         if !hit && !appended {
             let mut rec = serde_json::json!({

crates/socket-patch-cli/src/commands/list.rs

@@ -70,19 +70,14 @@ impl HostedListing {
     /// One listing per hosted pin in `pins`, detailed from `legacy` where
     /// it records the same purl and uuid.
     pub(crate) fn from_pins(pins: &[HostedPin], legacy: Option<&RedirectState>) -> Vec<Self> {
-        let canon = |p: &str| {
-            socket_patch_core::utils::purl::normalize_purl(
-                socket_patch_core::utils::purl::strip_purl_qualifiers(p),
-            )
-            .into_owned()
-        };
+        use socket_patch_core::utils::purl_key::PurlKey;
         pins.iter()
             .map(|pin| {
                 let record = legacy
                     .and_then(|l| {
                         l.records
                             .iter()
-                            .find(|(k, r)| canon(k) == canon(&pin.purl) && r.uuid == pin.uuid)
+                            .find(|(k, r)| PurlKey::same(k, &pin.purl) && r.uuid == pin.uuid)
                             .map(|(_, r)| r.clone())
                     })
                     .unwrap_or_else(|| PatchRecord {

crates/socket-patch-cli/src/commands/remove.rs

@@ -7,6 +7,7 @@ use socket_patch_core::manifest::schema::PatchManifest;
 use socket_patch_core::patch::redirect::upstream::HostedPin;
 use socket_patch_core::telemetry::{track_patch_remove_failed, track_patch_removed};
 use socket_patch_core::utils::purl::patch_matches;
+use socket_patch_core::utils::purl_key::PurlKey;
 use socket_patch_core::vendor::{
     load_state, RevertOpts, VendorEntry, VendorState, VENDOR_STATE_REL,
 };
@@ -562,7 +563,7 @@ pub async fn run(args: RemoveArgs) -> i32 {
     // Vendor-owned purls are excluded from the in-place restore (the
     // vendored leg below reverts them); an unreadable ledger degrades to
     // "nothing vendored" here and fails closed at that leg.
-    let vendored_keys: HashSet<String> = vendor_state_result
+    let vendored_keys: HashSet<PurlKey> = vendor_state_result
         .as_ref()
         .map(socket_patch_core::vendor::VendorState::purl_keys)
         .unwrap_or_default();

crates/socket-patch-cli/src/commands/rollback.rs

@@ -15,8 +15,8 @@ use socket_patch_core::patch::rollback::{
 };
 use socket_patch_core::telemetry::{track_patch_rollback_failed, track_patch_rolled_back};
 use socket_patch_core::utils::purl::{patch_matches, strip_purl_qualifiers};
+use socket_patch_core::utils::purl_key::PurlKey;
 use socket_patch_core::vendor::{purl_keys_cover, RevertOpts, VendorState};
-use socket_patch_core::vex::discover::{canonical_base_purl, same_release};
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
@@ -1168,7 +1168,7 @@ pub async fn run(args: RollbackArgs) -> i32 {
     // runs, and the ledger does not own the global copies, so the in-place
     // leg restores them.
     let loaded_vendor_state = socket_patch_core::vendor::load_state(&cwd).await;
-    let project_vendored_keys: HashSet<String> = loaded_vendor_state
+    let project_vendored_keys: HashSet<PurlKey> = loaded_vendor_state
         .as_ref()
         .map(VendorState::purl_keys)
         .unwrap_or_default();
@@ -1181,7 +1181,7 @@ pub async fn run(args: RollbackArgs) -> i32 {
     let vendor_corrupt = vendor_state_result.is_err();
     // An unreadable ledger degrades to "nothing vendored" for the in-place
     // leg (its own containment is the `vendor_state_unreadable` exit below).
-    let vendored_keys: HashSet<String> = vendor_state_result
+    let vendored_keys: HashSet<PurlKey> = vendor_state_result
         .as_ref()
         .map(VendorState::purl_keys)
         .unwrap_or_default();
@@ -2020,7 +2020,7 @@ pub(crate) async fn rollback_patches_inner(
     common: &GlobalArgs,
     socket_dir: &Path,
     manifest: &PatchManifest,
-    vendored_keys: &HashSet<String>,
+    vendored_keys: &HashSet<PurlKey>,
     selection: InnerSelection<'_>,
     // Manifest purl -> the hosted uuid a live lockfile pin superseded its
     // record with ([`superseded_by_hosted`]); empty when no hosted pin
@@ -2811,10 +2811,10 @@ pub(crate) fn superseded_by_hosted(
         .patches
         .iter()
         .filter_map(|(purl, record)| {
-            let pkg = canonical_base_purl(purl);
+            let pkg = PurlKey::new(purl);
             let same: Vec<&HostedPin> = pins
                 .iter()
-                .filter(|pin| same_release(&pin.purl, &pkg))
+                .filter(|pin| PurlKey::new(&pin.purl) == pkg)
                 .collect();
             if same.iter().any(|pin| pin.uuid == record.uuid) {
                 return None;