[PATCH datacenter-manager v4 09/10] subscription: add Adopt Key action for foreign live subscriptions

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Add a dedicated endpoint plus CLI / UI wiring for importing a
remote node's live subscription into the pool as a bound entry,
without touching the remote. The action covers the case where a
key was already installed on a node before PDM took over its pool
management; bringing it under the registry is required for any
subsequent pool action to operate on it.

Three sub-cases for the live key:

 - Not in the pool: insert with source=Adopted, bound to (remote, node).
 - In the pool but unbound: rebind, leaving the source field as-is
   so a key originally added by hand keeps its Manual label.
 - In the pool but bound elsewhere: refused, the operator has to
   reconcile the binding first.

The endpoint pre-fetches the pool digest before the live network
read and refuses with CONFLICT on mismatch, so a parallel
set_assignment landing during the .await cannot silently rebind
the key. Per-remote PRIV_RESOURCE_MODIFY is enforced inside the
handler so operators with only global system access cannot pull
subscriptions off remotes they have no other authority on.

The Node Subscription Status tree marks adoptable rows (live key
set, no pool binding yet) with a download hint icon so the action
is discoverable without consulting the docs. The pool grid gets a
new Source column exposing the Manual vs Adopted origin, hidden by
default; available via the column picker.

Tested-by: Lukas Wagner <l.wagner@proxmox.com>
Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---

Changes v3 -> 4:
* Bundles v3-0009 (Adopt Key) and v3-0010 (Adopt All); they share the
  dropdown filter and Source column.
* Adopt All preview Key column drops the `FontStyle::LabelMedium` span
  wrapper, same 12px-vs-14px fix as v4-0005 (Lukas).

 cli/client/src/subscriptions.rs               |  60 ++++
 docs/subscription-registry.rst                |  15 +
 lib/pdm-api-types/src/subscription.rs         |  20 ++
 lib/pdm-api-types/tests/test_import.rs        |  29 ++
 lib/pdm-client/src/lib.rs                     |  61 +++-
 server/src/api/subscriptions/mod.rs           | 333 +++++++++++++++++-
 ui/src/configuration/subscription_keys.rs     |  16 +-
 ui/src/configuration/subscription_registry.rs | 262 +++++++++++++-
 8 files changed, 788 insertions(+), 8 deletions(-)

diff --git a/cli/client/src/subscriptions.rs b/cli/client/src/subscriptions.rs
index 02dfa0f2..79810841 100644
--- a/cli/client/src/subscriptions.rs
+++ b/cli/client/src/subscriptions.rs
@@ -48,6 +48,11 @@ pub fn cli() -> CommandLineInterface {
             "revert-clear",
             CliCommand::new(&API_METHOD_REVERT_CLEAR).arg_param(&["remote", "node"]),
         )
+        .insert(
+            "adopt-key",
+            CliCommand::new(&API_METHOD_ADOPT_KEY).arg_param(&["remote", "node"]),
+        )
+        .insert("adopt-all", CliCommand::new(&API_METHOD_ADOPT_ALL))
         .into()
 }
 
@@ -308,6 +313,61 @@ async fn auto_assign(apply: bool) -> Result<(), Error> {
     Ok(())
 }
 
+#[api(
+    input: {
+        properties: {
+            remote: { schema: REMOTE_ID_SCHEMA },
+            node: { schema: NODE_SCHEMA },
+            digest: {
+                schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+                optional: true,
+            },
+        },
+    },
+)]
+/// Adopt the live subscription on a remote node into the pool.
+///
+/// Brings a foreign subscription under PDM management without touching the remote: the live
+/// current key on `remote`/`node` is imported as a pool entry bound to that node. Refuses if
+/// the (remote, node) target already has a pool-managed binding.
+async fn adopt_key(remote: String, node: String, digest: Option<String>) -> Result<(), Error> {
+    let digest = digest.map(ConfigDigest::from);
+    client()?
+        .subscription_adopt_key(&remote, &node, digest)
+        .await?;
+    println!("Adopted live subscription on {remote}/{node} into the pool.");
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            digest: {
+                schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+                optional: true,
+            },
+        },
+    },
+)]
+/// Adopt every foreign live subscription into the pool in one transaction.
+///
+/// Walks all remotes the caller can audit, imports any (remote, node) with a live current key
+/// and no pool binding. Candidates the caller has no modify privilege on, or whose key is
+/// already bound elsewhere in the pool, are silently skipped.
+async fn adopt_all(digest: Option<String>) -> Result<(), Error> {
+    let digest = digest.map(ConfigDigest::from);
+    let adopted = client()?.subscription_adopt_all(digest).await?;
+    if adopted.is_empty() {
+        println!("No foreign live subscriptions to adopt.");
+        return Ok(());
+    }
+    println!("Adopted {} live subscription(s):", adopted.len());
+    for e in &adopted {
+        println!("  {}/{} -> {}", e.remote, e.node, e.key);
+    }
+    Ok(())
+}
+
 #[api(
     input: {
         properties: {
diff --git a/docs/subscription-registry.rst b/docs/subscription-registry.rst
index 68b879be..6d599fe2 100644
--- a/docs/subscription-registry.rst
+++ b/docs/subscription-registry.rst
@@ -44,6 +44,21 @@ issues the removal on the remote and releases the pool binding so the key become
 for reassignment. Discard Pending drops the queued clear without touching the remote; the
 binding stays intact and the operator can retry.
 
+The Adopt Key action imports the live subscription on a remote node into the pool as a
+bound entry, without touching the remote. Use it to bring a pre-existing subscription -- one
+installed on a node before PDM took over its pool management -- under the registry so that
+pool actions such as Clear Key and Auto-Assign can act on it. Nodes that are eligible for
+adoption are highlighted with a download hint icon in the Node Subscription Status tree;
+the pool grid carries a hidden-by-default Source column distinguishing manually-added from
+adopted entries, which can be enabled via the column picker if the distinction matters.
+
+The Adopt All action runs the same import across every remote the operator can audit in one
+transaction. Use it after first connecting an existing fleet of nodes to PDM so the pool
+catches up with the live subscriptions already deployed, without having to click through
+Adopt Key for each node. Candidates the operator has no modify privilege on, whose key is
+already bound elsewhere in the pool, whose (remote, node) target is already bound by another
+pool entry, or whose key or node name fails schema validation are skipped silently.
+
 The proposed plan can be inspected before it is applied. Apply Pending walks the queue in
 order; if any push or clear fails the remaining queue is kept intact for retry. Discard Pending
 drops the plan without touching any remote.
diff --git a/lib/pdm-api-types/src/subscription.rs b/lib/pdm-api-types/src/subscription.rs
index 7dd16c62..22408a90 100644
--- a/lib/pdm-api-types/src/subscription.rs
+++ b/lib/pdm-api-types/src/subscription.rs
@@ -307,6 +307,9 @@ pub enum SubscriptionKeySource {
     /// UI or CLI, and as the `serde(default)` for entries that predate this field.
     #[default]
     Manual,
+    /// Imported from a remote node's live subscription via the Adopt Key action, that is, a key
+    /// that was already installed on a remote before PDM took over its pool management.
+    Adopted,
 }
 
 #[api(
@@ -563,6 +566,23 @@ pub struct ClearPendingResult {
     pub cleared: u32,
 }
 
+#[api(
+    properties: {
+        "key": { schema: SUBSCRIPTION_KEY_SCHEMA },
+    },
+)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+/// One entry imported by the bulk Adopt-All endpoint.
+pub struct AdoptedEntry {
+    /// Remote the live subscription was running on.
+    pub remote: String,
+    /// Node within the remote.
+    pub node: String,
+    /// The adopted subscription key.
+    pub key: String,
+}
+
 #[api]
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "kebab-case")]
diff --git a/lib/pdm-api-types/tests/test_import.rs b/lib/pdm-api-types/tests/test_import.rs
index 33601620..72177460 100644
--- a/lib/pdm-api-types/tests/test_import.rs
+++ b/lib/pdm-api-types/tests/test_import.rs
@@ -40,6 +40,35 @@ fn entry_roundtrip() {
     assert_eq!(back.next_due_date.as_deref(), Some("2027-06-01"));
 }
 
+#[test]
+fn adopted_entry_roundtrip() {
+    // Ensure SubscriptionKeySource::Adopted serializes to its kebab-case form `adopted` and
+    // parses back to the same variant, so an in-place upgrade does not silently rewrite
+    // adopted pool entries to Manual on the next save.
+    let mut config = SectionConfigData::<SubscriptionKeyEntry>::default();
+    config.insert(
+        "pbsc-1122334455".to_string(),
+        SubscriptionKeyEntry {
+            key: "pbsc-1122334455".to_string(),
+            product_type: ProductType::Pbs,
+            source: SubscriptionKeySource::Adopted,
+            remote: Some("backup-cluster".to_string()),
+            node: Some("pbs-1".to_string()),
+            ..Default::default()
+        },
+    );
+
+    let raw = SubscriptionKeyEntry::write_section_config("test", &config).expect("write failed");
+    assert!(
+        raw.contains("\tsource adopted"),
+        "expected kebab-case `adopted` in serialised form, got:\n{raw}",
+    );
+    let parsed = SubscriptionKeyEntry::parse_section_config("test", &raw).expect("parse failed");
+    let back = parsed.get("pbsc-1122334455").expect("key not found");
+    assert_eq!(back.source, SubscriptionKeySource::Adopted);
+    assert_eq!(back.remote.as_deref(), Some("backup-cluster"));
+}
+
 #[test]
 fn shadow_roundtrip() {
     let mut shadow = SectionConfigData::<SubscriptionKeyShadow>::default();
diff --git a/lib/pdm-client/src/lib.rs b/lib/pdm-client/src/lib.rs
index c6630e03..a547860f 100644
--- a/lib/pdm-client/src/lib.rs
+++ b/lib/pdm-client/src/lib.rs
@@ -1280,9 +1280,66 @@ impl<T: HttpApiClient> PdmClient<T> {
             .data)
     }
 
+    /// Adopt the live subscription on `remote`/`node` into the pool: imports the live key as a
+    /// new pool entry bound to (remote, node) without touching the remote. Refuses if (remote,
+    /// node) already has a pool entry bound to it. See the server endpoint docs for the full
+    /// per-sub-case semantics (existing-unbound, existing-bound-elsewhere, not-in-pool).
+    pub async fn subscription_adopt_key(
+        &self,
+        remote: &str,
+        node: &str,
+        digest: Option<ConfigDigest>,
+    ) -> Result<(), Error> {
+        #[derive(Serialize)]
+        struct AdoptArgs<'a> {
+            remote: &'a str,
+            node: &'a str,
+            #[serde(skip_serializing_if = "Option::is_none")]
+            digest: Option<ConfigDigest>,
+        }
+        self.0
+            .post(
+                "/api2/extjs/subscriptions/adopt-key",
+                &AdoptArgs {
+                    remote,
+                    node,
+                    digest,
+                },
+            )
+            .await?
+            .nodata()
+    }
+
+    /// Adopt every foreign live subscription that the caller can modify, in one transaction.
+    /// Returns the list of `(remote, node, key)` tuples that were imported into the pool;
+    /// candidates the caller has no `PRIV_RESOURCE_MODIFY` on (or that fail validation, or that
+    /// are already bound elsewhere in the pool) are silently skipped. See the server endpoint
+    /// docs for the full skip rules.
+    pub async fn subscription_adopt_all(
+        &self,
+        digest: Option<ConfigDigest>,
+    ) -> Result<Vec<pdm_api_types::subscription::AdoptedEntry>, Error> {
+        #[derive(Serialize)]
+        struct AdoptAllArgs {
+            #[serde(skip_serializing_if = "Option::is_none")]
+            digest: Option<ConfigDigest>,
+        }
+        Ok(self
+            .0
+            .post(
+                "/api2/extjs/subscriptions/adopt-all",
+                &AdoptAllArgs { digest },
+            )
+            .await?
+            .expect_json()?
+            .data)
+    }
+
     /// Queue a clear for the subscription on `remote`/`node`. Apply Pending later removes the
-    /// subscription from the node so the key can be reassigned elsewhere; Discard Pending undoes
-    /// the queueing without touching the remote.
+    /// subscription from the node so the key can be reassigned elsewhere; Discard Pending
+    /// undoes the queueing without touching the remote. Returns `BAD_REQUEST` if no pool entry
+    /// is bound to (remote, node); callers must run Adopt Key first to import a foreign
+    /// subscription.
     pub async fn subscription_queue_clear(
         &self,
         remote: &str,
diff --git a/server/src/api/subscriptions/mod.rs b/server/src/api/subscriptions/mod.rs
index 706636c9..7eb758cc 100644
--- a/server/src/api/subscriptions/mod.rs
+++ b/server/src/api/subscriptions/mod.rs
@@ -21,9 +21,9 @@ use proxmox_sortable_macro::sortable;
 
 use pdm_api_types::remotes::{Remote, REMOTE_ID_SCHEMA};
 use pdm_api_types::subscription::{
-    pick_best_pve_socket_key, socket_count_from_key, AutoAssignProposal, ClearPendingResult,
-    ProductType, ProposedAssignment, RemoteNodeStatus, SubscriptionKeyEntry, SubscriptionKeySource,
-    SubscriptionLevel, SUBSCRIPTION_KEY_SCHEMA,
+    pick_best_pve_socket_key, socket_count_from_key, AdoptedEntry, AutoAssignProposal,
+    ClearPendingResult, ProductType, ProposedAssignment, RemoteNodeStatus, SubscriptionKeyEntry,
+    SubscriptionKeySource, SubscriptionLevel, SUBSCRIPTION_KEY_SCHEMA,
 };
 use pdm_api_types::{
     Authid, NODE_SCHEMA, PRIV_RESOURCE_AUDIT, PRIV_RESOURCE_MODIFY, PRIV_SYS_AUDIT, PRIV_SYS_MODIFY,
@@ -39,6 +39,8 @@ pub const ROUTER: Router = Router::new()
 
 #[sortable]
 const SUBDIRS: SubdirMap = &sorted!([
+    ("adopt-all", &Router::new().post(&API_METHOD_ADOPT_ALL)),
+    ("adopt-key", &Router::new().post(&API_METHOD_ADOPT_KEY)),
     (
         "apply-pending",
         &Router::new().post(&API_METHOD_APPLY_PENDING)
@@ -87,6 +89,11 @@ const PANEL_NODE_STATUS_MAX_AGE: u64 = 5 * 60;
 /// Keeps the product prefix and the first/last hex characters of the secret so an operator can
 /// still tell two keys apart in a tail of `journalctl`, but the full key never lands in a log
 /// file readable by anyone other than the priv user.
+///
+/// Uses `chars()` rather than byte slicing so a hostile remote returning a non-ASCII subscription
+/// key cannot trigger a slice-on-non-char-boundary panic; schema-validated pool keys are pure
+/// ASCII per `PRODUCT_KEY_REGEX`, but `redact_key` is also reached by the adoption path on a
+/// live key the remote owned, which can be any string.
 fn redact_key(key: &str) -> String {
     let Some((prefix, secret)) = key.split_once('-') else {
         return "<malformed-key>".to_string();
@@ -956,6 +963,317 @@ async fn revert_pending_clear(
     Ok(())
 }
 
+#[api(
+    input: {
+        properties: {
+            remote: { schema: REMOTE_ID_SCHEMA },
+            // NODE_SCHEMA rejects path-traversal input before it ends up interpolated into the
+            // remote URL `/api2/extjs/nodes/{node}/subscription`.
+            node: { schema: NODE_SCHEMA },
+            digest: {
+                type: ConfigDigest,
+                optional: true,
+            },
+        },
+    },
+    access: {
+        permission: &Permission::Privilege(&["system"], PRIV_SYS_MODIFY, false),
+    },
+)]
+/// Adopt the live subscription on a remote node into the pool.
+///
+/// Reads the live current key from `remote`/`node` and brings the pool under management of it
+/// without touching the remote (no DELETE / push). Three sub-cases for the live key:
+///
+/// - Not in the pool: a fresh `Adopted` entry is inserted, bound to (remote, node).
+/// - In the pool, unbound: rebound to (remote, node); the source is left untouched so a key
+///   that was originally added manually keeps its `Manual` label even after a remote re-import.
+/// - In the pool, bound elsewhere: refused; the operator has to reconcile the binding first.
+///
+/// Refuses if a pool entry is already bound to (remote, node): adopting a node that is already
+/// pool-managed would either be a no-op or a footgun (rebinding the same node to a different
+/// key in the pool), so the caller has to pick the right Assign/Clear path explicitly.
+///
+/// Per-remote `PRIV_RESOURCE_MODIFY` is enforced inside the handler so an operator with global
+/// system access alone cannot pull subscriptions off remotes they have no other authority on
+/// (an adopted key bound to (remote, node) is itself an audit-side surface against that node).
+async fn adopt_key(
+    remote: String,
+    node: String,
+    digest: Option<ConfigDigest>,
+    rpcenv: &mut dyn RpcEnvironment,
+) -> Result<(), Error> {
+    let auth_id: Authid = rpcenv
+        .get_auth_id()
+        .context("no authid available")?
+        .parse()?;
+    let user_info = CachedUserInfo::new()?;
+    user_info.check_privs(
+        &auth_id,
+        &["resource", &remote],
+        PRIV_RESOURCE_MODIFY,
+        false,
+    )?;
+
+    // Pre-fetch digest to catch a parallel set_assignment during the live read below.
+    let (_pre_config, pre_digest) = pdm_config::subscriptions::config()?;
+
+    // Fetch live state before grabbing the config lock so the network call does not pin the
+    // lock for the duration of a remote query.
+    let (remotes_config, _) = pdm_config::remotes::config()?;
+    let remote_entry = remotes_config
+        .get(&remote)
+        .ok_or_else(|| http_err!(NOT_FOUND, "remote '{remote}' not found"))?;
+    let live = get_subscription_info_for_remote(remote_entry, FRESH_NODE_STATUS_MAX_AGE)
+        .await
+        .map_err(|err| {
+            http_err!(
+                BAD_REQUEST,
+                "could not read subscription on {remote}/{node}: {err}"
+            )
+        })?;
+    let live_current_key: String = live
+        .get(&node)
+        .and_then(|info| info.as_ref())
+        .and_then(|info| info.key.clone())
+        .ok_or_else(|| {
+            http_err!(
+                NOT_FOUND,
+                "no live subscription on {remote}/{node} to adopt"
+            )
+        })?;
+
+    // The lock + sync IO runs on a blocking thread so the async runtime stays free for other
+    // work even when /etc/proxmox-datacenter-manager/subscriptions is on slow storage.
+    let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
+        let _lock = pdm_config::subscriptions::lock_config()?;
+        let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+        config_digest.detect_modification(digest.as_ref())?;
+        if config_digest != pre_digest {
+            http_bail!(
+                CONFLICT,
+                "pool config changed during live fetch; refresh and retry adopt of \
+                 {remote}/{node}"
+            );
+        }
+
+        let target_bound = config.iter().any(|(_, e)| {
+            e.remote.as_deref() == Some(remote.as_str()) && e.node.as_deref() == Some(node.as_str())
+        });
+        if target_bound {
+            http_bail!(
+                BAD_REQUEST,
+                "{remote}/{node} is already pool-managed; adopt only applies to foreign \
+                 subscriptions"
+            );
+        }
+
+        if let Some(existing) = config.get_mut(&live_current_key) {
+            if existing.remote.is_some() || existing.node.is_some() {
+                http_bail!(
+                    CONFLICT,
+                    "key '{}' is in the pool but bound elsewhere; resolve manually first",
+                    redact_key(&live_current_key),
+                );
+            }
+            existing.remote = Some(remote.clone());
+            existing.node = Some(node.clone());
+        } else {
+            // Schema-validate the live key before letting it touch the on-disk pool. The
+            // remote claimed it via /nodes/{node}/subscription, but that surface is not a
+            // strict-schema gate (older PVE versions accept whatever the operator typed at
+            // setup time), so re-validate here against the same schema that manual entry
+            // uses.
+            SUBSCRIPTION_KEY_SCHEMA
+                .parse_simple_value(&live_current_key)
+                .map_err(|err| {
+                    http_err!(
+                        BAD_REQUEST,
+                        "key '{}' rejected: {err}",
+                        redact_key(&live_current_key),
+                    )
+                })?;
+            let product_type = ProductType::from_key(&live_current_key).ok_or_else(|| {
+                http_err!(
+                    BAD_REQUEST,
+                    "unrecognised key prefix: {}",
+                    redact_key(&live_current_key),
+                )
+            })?;
+            let entry = SubscriptionKeyEntry {
+                key: live_current_key.clone(),
+                product_type,
+                level: SubscriptionLevel::from_key(Some(&live_current_key)),
+                source: SubscriptionKeySource::Adopted,
+                remote: Some(remote.clone()),
+                node: Some(node.clone()),
+                ..Default::default()
+            };
+            config.insert(live_current_key, entry);
+        }
+
+        pdm_config::subscriptions::save_config(&config)
+    })
+    .await??;
+    rpcenv["digest"] = new_digest.to_hex().into();
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            digest: {
+                type: ConfigDigest,
+                optional: true,
+            },
+        },
+    },
+    returns: {
+        type: Array,
+        description: "List of (remote, node, key) tuples that were adopted into the pool.",
+        items: { type: AdoptedEntry },
+    },
+    access: {
+        permission: &Permission::Privilege(&["system"], PRIV_SYS_MODIFY, false),
+    },
+)]
+/// Adopt every foreign live subscription in one transaction.
+///
+/// Walks the node-status view (so only remotes the caller can audit are considered), collects
+/// every (remote, node) that has a live current key but no pool entry bound to it, and imports
+/// each one into the pool with source = `Adopted`. Candidates are skipped (not adopted, not an
+/// error) when:
+///
+/// - The caller has no `PRIV_RESOURCE_MODIFY` on the candidate's remote: an audit-only operator
+///   should not be able to materialise pool state for a remote they cannot manage.
+/// - The live key is already in the pool but bound elsewhere: leaving the rebind as a manual
+///   step keeps the bulk action from silently competing with a deliberate prior assignment.
+/// - The live key fails schema validation or its prefix is unknown: a buggy or malicious
+///   remote should not be able to inject garbage into the pool through a bulk shortcut.
+///
+/// Successfully-adopted entries are returned so the caller (CLI / UI) can summarise the outcome
+/// without needing a separate refresh round-trip.
+async fn adopt_all(
+    digest: Option<ConfigDigest>,
+    rpcenv: &mut dyn RpcEnvironment,
+) -> Result<Vec<AdoptedEntry>, Error> {
+    let auth_id: Authid = rpcenv
+        .get_auth_id()
+        .context("no authid available")?
+        .parse()?;
+
+    // Use a fresh node-status snapshot: a cached entry from minutes ago could miss a live
+    // subscription that was just installed on a remote, or vice-versa, claim a subscription
+    // that has since been removed. Adopting bogus or already-cleared keys would be a footgun.
+    let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+
+    // Lock + sync IO under spawn_blocking. The closure re-resolves the candidate set under the
+    // lock: a parallel admin's Assign / Adopt between the network read above and the lock
+    // acquisition here would otherwise let us race-import a key that has just been bound by
+    // them.
+    let (adopted, new_digest_opt) = tokio::task::spawn_blocking(
+        move || -> Result<(Vec<AdoptedEntry>, Option<ConfigDigest>), Error> {
+            let user_info = CachedUserInfo::new()?;
+            let _lock = pdm_config::subscriptions::lock_config()?;
+            let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+            config_digest.detect_modification(digest.as_ref())?;
+
+            let mut adopted: Vec<AdoptedEntry> = Vec::new();
+            for n in &node_statuses {
+                let Some(current_key) = n.current_key.as_deref() else {
+                    continue;
+                };
+                if n.assigned_key.is_some() {
+                    continue;
+                }
+                if user_info.lookup_privs(&auth_id, &["resource", &n.remote]) & PRIV_RESOURCE_MODIFY
+                    == 0
+                {
+                    continue;
+                }
+                // Re-validate foreign node name: later interpolated into remote URL.
+                if NODE_SCHEMA.parse_simple_value(&n.node).is_err() {
+                    warn!(
+                        "skipping adopt-all candidate on {}/{}: node name fails schema",
+                        n.remote, n.node,
+                    );
+                    continue;
+                }
+                // Re-check binding state under the lock - between the network read and here a
+                // parallel Adopt / Assign on the same target could have created a pool entry
+                // bound to (remote, node) that the cached node-status snapshot did not see.
+                let target_bound = config.iter().any(|(_, e)| {
+                    e.remote.as_deref() == Some(n.remote.as_str())
+                        && e.node.as_deref() == Some(n.node.as_str())
+                });
+                if target_bound {
+                    continue;
+                }
+
+                if let Some(existing) = config.get_mut(current_key) {
+                    if existing.remote.is_some() || existing.node.is_some() {
+                        // Bound elsewhere: leave the rebind as an explicit operator decision.
+                        continue;
+                    }
+                    existing.remote = Some(n.remote.clone());
+                    existing.node = Some(n.node.clone());
+                } else {
+                    if SUBSCRIPTION_KEY_SCHEMA
+                        .parse_simple_value(current_key)
+                        .is_err()
+                    {
+                        warn!(
+                            "skipping adopt-all candidate on {}/{}: key '{}' fails schema",
+                            n.remote,
+                            n.node,
+                            redact_key(current_key),
+                        );
+                        continue;
+                    }
+                    let Some(product_type) = ProductType::from_key(current_key) else {
+                        warn!(
+                            "skipping adopt-all candidate on {}/{}: unrecognised key prefix \
+                             '{}'",
+                            n.remote,
+                            n.node,
+                            redact_key(current_key),
+                        );
+                        continue;
+                    };
+                    let entry = SubscriptionKeyEntry {
+                        key: current_key.to_string(),
+                        product_type,
+                        level: SubscriptionLevel::from_key(Some(current_key)),
+                        source: SubscriptionKeySource::Adopted,
+                        remote: Some(n.remote.clone()),
+                        node: Some(n.node.clone()),
+                        ..Default::default()
+                    };
+                    config.insert(current_key.to_string(), entry);
+                }
+                adopted.push(AdoptedEntry {
+                    remote: n.remote.clone(),
+                    node: n.node.clone(),
+                    key: current_key.to_string(),
+                });
+            }
+
+            let new_digest = if adopted.is_empty() {
+                None
+            } else {
+                Some(pdm_config::subscriptions::save_config(&config)?)
+            };
+            Ok((adopted, new_digest))
+        },
+    )
+    .await??;
+
+    if let Some(new_digest) = new_digest_opt {
+        rpcenv["digest"] = new_digest.to_hex().into();
+    }
+    Ok(adopted)
+}
+
 #[api(
     input: {
         properties: {
@@ -1735,6 +2053,15 @@ mod tests {
         assert_eq!(redact_key("pbsc-abcdef0123"), "pbsc-a...3");
     }
 
+    #[test]
+    fn redact_key_safe_on_non_ascii_secret() {
+        // Slicing by byte index on a UTF-8 boundary would panic; chars()-based redaction must
+        // tolerate hostile / buggy remote inputs in the foreign-key adoption path.
+        let key = "pve4b-1\u{1F600}";
+        let redacted = redact_key(key);
+        assert!(redacted.starts_with("pve4b-1..."));
+    }
+
     #[test]
     fn redact_key_safe_on_single_char_secret() {
         assert_eq!(redact_key("pve4b-x"), "pve4b-x...");
diff --git a/ui/src/configuration/subscription_keys.rs b/ui/src/configuration/subscription_keys.rs
index 10255b5a..a0ddcbe3 100644
--- a/ui/src/configuration/subscription_keys.rs
+++ b/ui/src/configuration/subscription_keys.rs
@@ -6,7 +6,7 @@ use anyhow::Error;
 
 use pdm_api_types::remotes::RemoteType;
 use pdm_api_types::subscription::{
-    ProductType, RemoteNodeStatus, SubscriptionKeyEntry,
+    ProductType, RemoteNodeStatus, SubscriptionKeyEntry, SubscriptionKeySource,
 };
 use yew::virtual_dom::{Key, VComp, VNode};
 
@@ -123,8 +123,9 @@ impl SubscriptionKeyGridComp {
         Rc::new(vec![
             DataTableColumn::new(tr!("Key"))
                 .flex(2)
-                .get_property(|entry: &SubscriptionKeyEntry| entry.key.as_str())
+                .sorter(|a: &SubscriptionKeyEntry, b: &SubscriptionKeyEntry| a.key.cmp(&b.key))
                 .sort_order(true)
+                .render(|entry: &SubscriptionKeyEntry| entry.key.as_str().into())
                 .into(),
             DataTableColumn::new(tr!("Product"))
                 .width("80px")
@@ -140,6 +141,17 @@ impl SubscriptionKeyGridComp {
                 .sorter(|a: &SubscriptionKeyEntry, b: &SubscriptionKeyEntry| a.level.cmp(&b.level))
                 .render(|entry: &SubscriptionKeyEntry| entry.level.to_string().into())
                 .into(),
+            DataTableColumn::new(tr!("Source"))
+                .width("90px")
+                .hidden(true)
+                .sorter(|a: &SubscriptionKeyEntry, b: &SubscriptionKeyEntry| {
+                    (a.source as u8).cmp(&(b.source as u8))
+                })
+                .render(|entry: &SubscriptionKeyEntry| match entry.source {
+                    SubscriptionKeySource::Manual => tr!("Manual").into(),
+                    SubscriptionKeySource::Adopted => tr!("Adopted").into(),
+                })
+                .into(),
             DataTableColumn::new(tr!("Assignment"))
                 .flex(2)
                 .sorter(|a: &SubscriptionKeyEntry, b: &SubscriptionKeyEntry| {
diff --git a/ui/src/configuration/subscription_registry.rs b/ui/src/configuration/subscription_registry.rs
index e89dc217..0f7250ba 100644
--- a/ui/src/configuration/subscription_registry.rs
+++ b/ui/src/configuration/subscription_registry.rs
@@ -99,6 +99,14 @@ fn pending_badge(push_count: u32, clear_count: u32) -> Row {
     row
 }
 
+/// Row shape for the Adopt All preview table.
+#[derive(Clone, PartialEq)]
+struct AdoptCandidate {
+    remote: String,
+    node: String,
+    key: String,
+}
+
 #[derive(Clone, Debug, PartialEq)]
 enum NodeTreeEntry {
     Root,
@@ -235,6 +243,11 @@ pub enum Msg {
     QueueClearForSelectedNode,
     /// Open the Assign Key dialog for the currently-selected node.
     AssignKeyToSelectedNode,
+    /// Open the confirmation dialog for adopting the live subscription on the selected node
+    /// into the pool.
+    AdoptKeyForSelectedNode,
+    /// Open the confirmation dialog for adopting every foreign live subscription into the pool.
+    AdoptAllPreview,
 }
 
 #[derive(PartialEq)]
@@ -249,6 +262,19 @@ pub enum ViewState {
         node: String,
         current_key: Option<String>,
     },
+    /// Pending confirmation to adopt the live subscription on `(remote, node)` into the pool.
+    /// The live key is captured here so the dialog body can show what will be imported.
+    ConfirmAdoptKey {
+        remote: String,
+        node: String,
+        current_key: String,
+    },
+    /// Pending confirmation to bulk-adopt every foreign live subscription. The candidate list
+    /// is captured at view-open time so the dialog body can show the operator exactly what
+    /// will be imported; the server re-computes the set under the lock at commit time.
+    ConfirmAdoptAll {
+        candidates: Vec<(String, String, String)>,
+    },
     /// Assign a pool key to the given node. Opens from the right panel's Assign Key button.
     AssignKeyToNode {
         remote: String,
@@ -264,6 +290,7 @@ pub struct SubscriptionRegistryComp {
     tree_store: TreeStore<NodeTreeEntry>,
     tree_columns: Rc<Vec<DataTableHeader<NodeTreeEntry>>>,
     proposal_columns: Rc<Vec<DataTableHeader<ProposedAssignment>>>,
+    adopt_columns: Rc<Vec<DataTableHeader<AdoptCandidate>>>,
     node_selection: Selection,
     last_node_data: Vec<RemoteNodeStatus>,
     /// Canonical pool snapshot. Passed down to the key grid (display) and shared with the
@@ -443,6 +470,19 @@ impl SubscriptionRegistryComp {
                 .into(),
         ])
     }
+
+    fn adopt_columns() -> Rc<Vec<DataTableHeader<AdoptCandidate>>> {
+        Rc::new(vec![
+            DataTableColumn::new(tr!("Remote / Node"))
+                .flex(2)
+                .render(|c: &AdoptCandidate| format!("{} / {}", c.remote, c.node).into())
+                .into(),
+            DataTableColumn::new(tr!("Key"))
+                .flex(2)
+                .render(|c: &AdoptCandidate| c.key.clone().into())
+                .into(),
+        ])
+    }
 }
 
 fn key_cell(n: &RemoteNodeStatus) -> Html {
@@ -489,6 +529,18 @@ fn key_cell(n: &RemoteNodeStatus) -> Html {
                     .with_child(Fa::new("clock-o").class(FontColor::Warning))
                     .with_child(text)
                     .into()
+            } else if assigned.is_none() && current.is_some() {
+                Tooltip::new(
+                    Row::new()
+                        .class(AlignItems::Baseline)
+                        .gap(2)
+                        .with_child(Fa::new("download").class(FontColor::Primary))
+                        .with_child(text),
+                )
+                .tip(tr!(
+                    "Not in pool - Adopt Key imports this live subscription."
+                ))
+                .into()
             } else {
                 text.into()
             }
@@ -515,6 +567,7 @@ impl LoadableComponent for SubscriptionRegistryComp {
             tree_store: store.clone(),
             tree_columns: Self::tree_columns(store),
             proposal_columns: Self::proposal_columns(),
+            adopt_columns: Self::adopt_columns(),
             node_selection,
             last_node_data: Vec::new(),
             pool_keys: Rc::new(Vec::new()),
@@ -655,6 +708,24 @@ impl LoadableComponent for SubscriptionRegistryComp {
                     node_sockets,
                 }));
             }
+            Msg::AdoptKeyForSelectedNode => {
+                let Some((remote, node, current_key)) = self.selected_node_for_adopt() else {
+                    return false;
+                };
+                ctx.link().change_view(Some(ViewState::ConfirmAdoptKey {
+                    remote,
+                    node,
+                    current_key,
+                }));
+            }
+            Msg::AdoptAllPreview => {
+                let candidates = self.adopt_all_candidates();
+                if candidates.is_empty() {
+                    return false;
+                }
+                ctx.link()
+                    .change_view(Some(ViewState::ConfirmAdoptAll { candidates }));
+            }
         }
         true
     }
@@ -662,6 +733,7 @@ impl LoadableComponent for SubscriptionRegistryComp {
     fn toolbar(&self, ctx: &LoadableComponentContext<Self>) -> Option<Html> {
         let link = ctx.link();
         let (push_count, clear_count) = self.pending_counts();
+        let adopt_all_count = self.adopt_all_candidates().len();
         let mut toolbar = Toolbar::new()
             .border_bottom(true)
             .with_child(
@@ -675,6 +747,18 @@ impl LoadableComponent for SubscriptionRegistryComp {
                      subscription, then queue it pending Apply."
                 )),
             )
+            .with_child(
+                Tooltip::new(
+                    Button::new(tr!("Adopt All"))
+                        .icon_class("fa fa-download")
+                        .disabled(adopt_all_count == 0)
+                        .on_activate(link.callback(|_| Msg::AdoptAllPreview)),
+                )
+                .tip(tr!(
+                    "Import every foreign live subscription that is not yet tracked by the \
+                     pool. The remote is not contacted; only the pool config is updated."
+                )),
+            )
             .with_spacer()
             .with_child(
                 Tooltip::new(
@@ -823,6 +907,58 @@ impl LoadableComponent for SubscriptionRegistryComp {
             ViewState::ConfirmAutoAssign(proposal) => {
                 Some(self.render_auto_assign_dialog(ctx, proposal))
             }
+            ViewState::ConfirmAdoptAll { candidates } => {
+                Some(self.render_adopt_all_dialog(ctx, candidates))
+            }
+            ViewState::ConfirmAdoptKey {
+                remote,
+                node,
+                current_key,
+            } => {
+                use pwt::widget::ConfirmDialog;
+                let question = tr!(
+                    "Adopt {key} from {remote}/{node} into the pool?",
+                    key = current_key.clone(),
+                    remote = remote.clone(),
+                    node = node.clone(),
+                );
+                let body = Column::new()
+                    .gap(2)
+                    .with_child(Container::from_tag("p").with_child(question))
+                    .with_child(Container::from_tag("p").with_child(tr!(
+                        "The live subscription is imported as a pool entry bound to this node; the remote is not contacted. After adoption the key participates in pool operations such as Clear Key and Auto-Assign."
+                    )));
+                let remote_for_cb = remote.clone();
+                let node_for_cb = node.clone();
+                let link = ctx.link().clone();
+                let close_link = ctx.link().clone();
+                let digest_for_cb = self.pool_digest.clone();
+                Some(
+                    ConfirmDialog::default()
+                        .title(tr!("Adopt Key"))
+                        .icon_class("fa fa-question-circle")
+                        .confirm_message(body)
+                        .on_confirm(move |_| {
+                            let link = link.clone();
+                            let remote = remote_for_cb.clone();
+                            let node = node_for_cb.clone();
+                            let digest = digest_for_cb.clone();
+                            link.clone().spawn(async move {
+                                let digest = digest.map(pdm_client::ConfigDigest::from);
+                                if let Err(err) = crate::pdm_client()
+                                    .subscription_adopt_key(&remote, &node, digest)
+                                    .await
+                                {
+                                    link.show_error(tr!("Adopt Key"), err.to_string(), true);
+                                }
+                                link.change_view(None);
+                                link.send_reload();
+                            });
+                        })
+                        .on_close(move |_| close_link.change_view(None))
+                        .into(),
+                )
+            }
             ViewState::ConfirmQueueClear {
                 remote,
                 node,
@@ -940,6 +1076,7 @@ impl SubscriptionRegistryComp {
         let can_assign_key = self.assign_target_for_selected_node().is_some();
         let can_revert = self.revert_target().is_some();
         let can_clear_key = self.selected_node_for_clear().is_some();
+        let can_adopt_key = self.selected_node_for_adopt().is_some();
         let assign_button = Tooltip::new(
             Button::new(tr!("Assign Key"))
                 .icon_class("fa fa-link")
@@ -968,7 +1105,16 @@ impl SubscriptionRegistryComp {
         .tip(tr!(
             "Queue the live subscription on the selected node for removal at next Apply \
              Pending, freeing the key for reassignment. Requires the node to be \
-             pool-managed."
+             pool-managed; for foreign subscriptions, run Adopt Key first."
+        ));
+        let adopt_key_button = Tooltip::new(
+            Button::new(tr!("Adopt Key"))
+                .icon_class("fa fa-download")
+                .disabled(!can_adopt_key)
+                .on_activate(ctx.link().callback(|_| Msg::AdoptKeyForSelectedNode)),
+        )
+        .tip(tr!(
+            "Import the live subscription on the selected node into the pool."
         ));
 
         Panel::new()
@@ -978,6 +1124,7 @@ impl SubscriptionRegistryComp {
             .min_width(400)
             .title(tr!("Node Subscription Status"))
             .with_tool(assign_button)
+            .with_tool(adopt_key_button)
             .with_tool(revert_button)
             .with_tool(clear_key_button)
             .with_child(table)
@@ -1065,6 +1212,38 @@ impl SubscriptionRegistryComp {
         Some((n.remote.clone(), n.node.clone(), n.current_key.clone()))
     }
 
+    /// Returns `(remote, node, current_key)` when the selected node has a foreign live
+    /// subscription eligible for Adopt Key: a current key is set on the node and no pool entry
+    /// is bound to (remote, node) yet. Mutually exclusive with `selected_node_for_clear` so the
+    /// toolbar can offer exactly one of Clear Key / Adopt Key for any given selection.
+    fn selected_node_for_adopt(&self) -> Option<(String, String, String)> {
+        let n = self.selected_node_status()?;
+        if n.assigned_key.is_some() {
+            return None;
+        }
+        let current_key = n.current_key.clone()?;
+        Some((n.remote.clone(), n.node.clone(), current_key))
+    }
+
+    /// Iterate the loaded node-status snapshot and return every `(remote, node, current_key)`
+    /// eligible for bulk Adopt-All (live key set, no pool binding). Used both for the toolbar
+    /// disabled gate and for the preview list in the confirm dialog; the authoritative set is
+    /// recomputed by the server under the lock at commit time, so this view is a hint, not a
+    /// contract.
+    fn adopt_all_candidates(&self) -> Vec<(String, String, String)> {
+        self.last_node_data
+            .iter()
+            .filter_map(|n| {
+                if n.assigned_key.is_some() {
+                    return None;
+                }
+                n.current_key
+                    .clone()
+                    .map(|k| (n.remote.clone(), n.node.clone(), k))
+            })
+            .collect()
+    }
+
     /// Returns `(remote, node, type, node_sockets)` for the right-panel Assign button:
     /// selected row is a node, no assigned key in the pool yet, and no live active subscription.
     /// Refusing earlier than the server keeps the button-disable affordance honest.
@@ -1140,4 +1319,85 @@ impl SubscriptionRegistryComp {
             .with_child(body)
             .into()
     }
+
+    fn render_adopt_all_dialog(
+        &self,
+        ctx: &LoadableComponentContext<Self>,
+        candidates: &[(String, String, String)],
+    ) -> Html {
+        use pwt::widget::Dialog;
+
+        let rows: Vec<AdoptCandidate> = candidates
+            .iter()
+            .map(|(r, n, k)| AdoptCandidate {
+                remote: r.clone(),
+                node: n.clone(),
+                key: k.clone(),
+            })
+            .collect();
+        let n = rows.len();
+        let store: Store<AdoptCandidate> = Store::with_extract_key(|c: &AdoptCandidate| {
+            format!("{}/{}", c.remote, c.node).into()
+        });
+        store.set_data(rows);
+
+        let link_close = ctx.link().clone();
+        let link_apply = ctx.link().clone();
+        let digest = self.pool_digest.clone();
+        let body = Column::new()
+            .class(Flex::Fill)
+            .class(Overflow::Hidden)
+            .min_height(0)
+            .padding(2)
+            .gap(2)
+            .min_width(600)
+            .with_child(Container::from_tag("p").with_child(tr!(
+                "The following {n} live subscription(s) will be imported into the pool; \
+                 the remote is not contacted.",
+                n = n,
+            )))
+            .with_child(
+                DataTable::new(self.adopt_columns.clone(), store)
+                    .striped(true)
+                    .class(FlexFit)
+                    .min_height(140),
+            )
+            .with_child(
+                Row::new()
+                    .class(JustifyContent::FlexEnd)
+                    .gap(2)
+                    .padding_top(2)
+                    .with_child(
+                        Button::new(tr!("Cancel"))
+                            .on_activate(move |_| link_close.change_view(None)),
+                    )
+                    .with_child(Button::new(tr!("Adopt")).on_activate(move |_| {
+                        let link = link_apply.clone();
+                        let digest = digest.clone();
+                        link.clone().spawn(async move {
+                            let digest = digest.map(pdm_client::ConfigDigest::from);
+                            if let Err(err) =
+                                crate::pdm_client().subscription_adopt_all(digest).await
+                            {
+                                link.show_error(tr!("Adopt All"), err.to_string(), true);
+                            }
+                            link.change_view(None);
+                            link.send_reload();
+                        });
+                    })),
+            );
+
+        Dialog::new(tr!("Adopt All"))
+            .resizable(true)
+            .width(700)
+            .min_width(500)
+            .min_height(300)
+            .max_height("80vh")
+            .on_close({
+                let link = ctx.link().clone();
+                move |_| link.change_view(None)
+            })
+            .with_child(body)
+            .into()
+    }
 }
-- 
2.47.3