refactor: version plugin component effective names (#889)

#### Overview

Relay creates effective names for dynamically registered plugin components so runtime discovery and middleware gates can unambiguously identify their owner. This change replaces the ambiguous dunder naming with a versioned, parsed format that carries the plugin kind, an always-present one-based component ordinal, and the local registration name. It makes singleton and multi-component registrations follow the same contract while retaining structured discovery as the stable client-facing identity.

- [x] I confirm this contribution is my own work, or I have the right to submit it under this project license.
- [x] I searched existing issues and open pull requests, and this does not duplicate existing work.

#### Details

- Replace the ambiguous dunder key with the versioned v1 grammar: `nemo-relay-plugin.v1.{percent-encoded-kind}:{one-based-ordinal}:{percent-encoded-local-name}`.
- Percent-encode UTF-8 component kinds and local names using RFC 3986 unreserved-character rules, and strictly decode only canonical v1 names. Invalid, legacy, and ordinary global names retain the global-API fallback.
- Preserve the existing discovery, worker, FFI, Python, Node, and Go owner transport. Relay-created plugin-component discovery now consistently reports `component_ordinal: Some(1)` for a singleton.
- Preserve encoding for PII child contexts and native runtime-created conditional-middleware gate names.
- Update exact cross-language expectations and the conditional-middleware guide.

Breaking behavior: effective names for Relay-created plugin components change from the unmerged dunder format to the v1 format. The structured discovery identity remains the public contract; clients must discover effective names and must not construct or persist them.

#### Where should the reviewer start?

Start with `crates/core/src/plugin.rs` for the namespace encoder/decoder, then `crates/core/src/api/registry.rs` for strict ownership discovery and `crates/core/tests/unit/plugin_tests.rs` for encoded-field and global-fallback coverage.

#### Related Issues: (use one of the action keywords Closes / Fixes / Resolves / Relates to)

- Relates to: #886


## Summary by CodeRabbit

* **New Features**
  * Introduced versioned, encoded registration names for plugin components.
  * Plugin registrations now provide clearer ownership metadata, including component kind and one-based ordinal.
  * Improved namespace handling for nested plugin components and conditional middleware guardrails.
  * Preserved reliable registration behavior for profile-scoped integrations and singleton plugins.

* **Documentation**
  * Documented plugin component ownership, ordinals, and registration naming rules.

* **Tests**
  * Expanded coverage for encoded namespaces, component discovery, ownership metadata, disabled components, and singleton plugins.

Authors:
  - Bryan Bednarski (https://github.com/bbednarski9)

Approvers:
  - Will Killian (https://github.com/willkill07)

URL: #889

Author: bbednarski9

crates/core/src/api/registry.rs

@@ -139,8 +139,9 @@ fn registration_identity(
     kind: RuntimeRegistrationKind,
     effective_name: &str,
 ) -> RuntimeRegistrationIdentity {
-    const PREFIX: &str = "__nemo_relay_plugin__";
-    let Some(rest) = effective_name.strip_prefix(PREFIX) else {
+    let Some((plugin_kind, component_ordinal, local_name)) =
+        crate::plugin::decode_plugin_component_effective_name(effective_name)
+    else {
         return RuntimeRegistrationIdentity {
             kind,
             local_name: effective_name.to_string(),
@@ -152,25 +153,14 @@ fn registration_identity(
             },
         };
     };
-    let mut pieces = rest.split("__");
-    let plugin_kind = pieces.next().unwrap_or_default().to_string();
-    let second = pieces.next().unwrap_or_default();
-    let (component_ordinal, local_name) = match second.parse::<u32>() {
-        Ok(ordinal) => (Some(ordinal), pieces.collect::<Vec<_>>().join("__")),
-        Err(_) => {
-            let mut local = vec![second];
-            local.extend(pieces);
-            (None, local.join("__"))
-        }
-    };
     RuntimeRegistrationIdentity {
         kind,
         local_name,
         effective_name: effective_name.to_string(),
         owner: RuntimeRegistrationOwner {
             kind: RuntimeRegistrationOwnerKind::Plugin,
             plugin_kind: Some(plugin_kind),
-            component_ordinal,
+            component_ordinal: Some(component_ordinal),
         },
     }
 }

crates/core/src/plugin.rs

@@ -407,7 +407,12 @@ impl PluginRegistration {
 #[derive(Default)]
 pub struct PluginRegistrationContext {
     registrations: Vec<PluginRegistration>,
-    namespace: Option<String>,
+    namespace: Option<PluginRegistrationNamespace>,
+}
+
+enum PluginRegistrationNamespace {
+    Plain(String),
+    PluginComponent(String),
 }
 
 impl PluginRegistrationContext {
@@ -420,7 +425,38 @@ impl PluginRegistrationContext {
     pub fn with_namespace(namespace: impl Into<String>) -> Self {
         Self {
             registrations: vec![],
-            namespace: Some(namespace.into()),
+            namespace: Some(PluginRegistrationNamespace::Plain(namespace.into())),
+        }
+    }
+
+    fn with_plugin_component_namespace(namespace: String) -> Self {
+        Self {
+            registrations: vec![],
+            namespace: Some(PluginRegistrationNamespace::PluginComponent(namespace)),
+        }
+    }
+
+    /// Creates a child context that extends this context's namespace.
+    ///
+    /// Child contexts preserve Relay-created plugin component qualification so
+    /// their local namespace segments and registration names remain encoded as
+    /// one effective component name.
+    pub fn with_child_namespace(&self, local_namespace: &str) -> Self {
+        let namespace = match &self.namespace {
+            Some(PluginRegistrationNamespace::Plain(namespace)) => {
+                PluginRegistrationNamespace::Plain(format!("{namespace}{local_namespace}"))
+            }
+            Some(PluginRegistrationNamespace::PluginComponent(namespace)) => {
+                PluginRegistrationNamespace::PluginComponent(format!(
+                    "{namespace}{}",
+                    encode_plugin_component_field(local_namespace)
+                ))
+            }
+            None => PluginRegistrationNamespace::Plain(local_namespace.to_string()),
+        };
+        Self {
+            registrations: vec![],
+            namespace: Some(namespace),
         }
     }
 
@@ -431,11 +467,21 @@ impl PluginRegistrationContext {
     /// not have to provide component instance ids.
     pub fn qualify_name(&self, name: &str) -> String {
         match &self.namespace {
-            Some(namespace) => format!("{namespace}{name}"),
+            Some(PluginRegistrationNamespace::Plain(namespace)) => format!("{namespace}{name}"),
+            Some(PluginRegistrationNamespace::PluginComponent(namespace)) => {
+                format!("{namespace}{}", encode_plugin_component_field(name))
+            }
             None => name.to_string(),
         }
     }
 
+    pub(crate) fn uses_plugin_component_namespace(&self) -> bool {
+        matches!(
+            &self.namespace,
+            Some(PluginRegistrationNamespace::PluginComponent(_))
+        )
+    }
+
     /// Registers an event subscriber and records its rollback closure.
     pub fn register_subscriber(&mut self, name: &str, callback: EventSubscriberFn) -> Result<()> {
         let qualified_name = self.qualify_name(name);
@@ -2653,7 +2699,6 @@ async fn initialize_plugin_components(
     rollback_failures: Option<Arc<Mutex<Vec<String>>>>,
 ) -> Result<Vec<PluginRegistration>> {
     ensure_builtin_plugins_registered()?;
-    let totals = plugin_component_totals(config);
     let mut ordinals: HashMap<&str, usize> = HashMap::new();
     let mut registrations = PendingPluginRegistrations::new(rollback_failures.clone());
 
@@ -2673,11 +2718,7 @@ async fn initialize_plugin_components(
             .entry(component.kind.as_str())
             .and_modify(|value| *value += 1)
             .or_insert(1);
-        let namespace = component_namespace(
-            &component.kind,
-            *ordinal,
-            totals.get(component.kind.as_str()).copied().unwrap_or(1),
-        );
+        let namespace = component_namespace(&component.kind, *ordinal);
 
         let mut pending =
             PendingPluginRegistrationContext::new(namespace, rollback_failures.clone());
@@ -2729,7 +2770,7 @@ struct PendingPluginRegistrationContext {
 impl PendingPluginRegistrationContext {
     fn new(namespace: String, rollback_failures: Option<Arc<Mutex<Vec<String>>>>) -> Self {
         Self {
-            context: PluginRegistrationContext::with_namespace(namespace),
+            context: PluginRegistrationContext::with_plugin_component_namespace(namespace),
             rollback_failures,
         }
     }
@@ -2804,11 +2845,85 @@ fn plugin_component_totals(config: &PluginConfig) -> HashMap<&str, usize> {
     totals
 }
 
-fn component_namespace(kind: &str, ordinal: usize, total: usize) -> String {
-    if total > 1 {
-        format!("__nemo_relay_plugin__{kind}__{ordinal}__")
-    } else {
-        format!("__nemo_relay_plugin__{kind}__")
+fn component_namespace(kind: &str, ordinal: usize) -> String {
+    assert!(ordinal > 0, "plugin component ordinals are one-based");
+    format!(
+        "nemo-relay-plugin.v1.{}:{ordinal}:",
+        encode_plugin_component_field(kind)
+    )
+}
+
+pub(crate) fn encode_plugin_component_field(value: &str) -> String {
+    const HEX: &[u8; 16] = b"0123456789ABCDEF";
+
+    let mut encoded = String::with_capacity(value.len());
+    for byte in value.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
+            encoded.push(char::from(byte));
+        } else {
+            encoded.push('%');
+            encoded.push(char::from(HEX[(byte >> 4) as usize]));
+            encoded.push(char::from(HEX[(byte & 0x0f) as usize]));
+        }
+    }
+    encoded
+}
+
+pub(crate) fn decode_plugin_component_effective_name(
+    effective_name: &str,
+) -> Option<(String, u32, String)> {
+    const PREFIX: &str = "nemo-relay-plugin.v1.";
+
+    let rest = effective_name.strip_prefix(PREFIX)?;
+    let (encoded_kind, rest) = rest.split_once(':')?;
+    let (ordinal_text, encoded_local_name) = rest.split_once(':')?;
+    let ordinal = ordinal_text
+        .parse::<u32>()
+        .ok()
+        .filter(|ordinal| *ordinal > 0)?;
+    if ordinal.to_string() != ordinal_text {
+        return None;
+    }
+    let plugin_kind = decode_plugin_component_field(encoded_kind)?;
+    let local_name = decode_plugin_component_field(encoded_local_name)?;
+    (!plugin_kind.is_empty() && !local_name.is_empty()).then_some((
+        plugin_kind,
+        ordinal,
+        local_name,
+    ))
+}
+
+fn decode_plugin_component_field(encoded: &str) -> Option<String> {
+    let mut bytes = Vec::with_capacity(encoded.len());
+    let encoded = encoded.as_bytes();
+    let mut index = 0;
+
+    while index < encoded.len() {
+        let byte = encoded[index];
+        if byte == b'%' {
+            let high = *encoded.get(index + 1)?;
+            let low = *encoded.get(index + 2)?;
+            bytes.push((hex_value(high)? << 4) | hex_value(low)?);
+            index += 3;
+        } else if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
+            bytes.push(byte);
+            index += 1;
+        } else {
+            return None;
+        }
+    }
+
+    let decoded = String::from_utf8(bytes).ok()?;
+    (encode_plugin_component_field(&decoded) == std::str::from_utf8(encoded).ok()?)
+        .then_some(decoded)
+}
+
+fn hex_value(byte: u8) -> Option<u8> {
+    match byte {
+        b'0'..=b'9' => Some(byte - b'0'),
+        b'A'..=b'F' => Some(byte - b'A' + 10),
+        b'a'..=b'f' => Some(byte - b'a' + 10),
+        _ => None,
     }
 }
 

crates/core/src/plugin/dynamic/native.rs

@@ -529,6 +529,7 @@ struct NativeHostPluginContext {
 
 struct NativeHostPluginRuntime {
     namespace: String,
+    encode_local_names: bool,
     active: AtomicBool,
     gates: Mutex<HashMap<String, NativeOwnedGate>>,
 }
@@ -3935,9 +3936,11 @@ unsafe extern "C" fn native_plugin_context_runtime(
         Ok(ctx) => ctx,
         Err(status) => return status,
     };
-    let namespace = unsafe { &*host_ctx.ctx }.qualify_name("");
+    let context = unsafe { &*host_ctx.ctx };
+    let namespace = context.qualify_name("");
     let runtime = Arc::new(NativeHostPluginRuntime {
         namespace,
+        encode_local_names: context.uses_plugin_component_namespace(),
         active: AtomicBool::new(true),
         gates: Mutex::new(HashMap::new()),
     });
@@ -4060,7 +4063,15 @@ unsafe extern "C" fn native_plugin_runtime_register_conditional_middleware_guard
         return NemoRelayStatus::AlreadyExists;
     }
     let handle = format!("gate-{}", Uuid::now_v7());
-    let qualified_name = format!("{}{}", runtime.namespace, local_name);
+    let qualified_name = if runtime.encode_local_names {
+        format!(
+            "{}{}",
+            runtime.namespace,
+            crate::plugin::encode_plugin_component_field(&local_name)
+        )
+    } else {
+        format!("{}{}", runtime.namespace, local_name)
+    };
     if let Err(error) = register_conditional_middleware_guardrail(
         &qualified_name,
         kinds,

crates/core/tests/integration/middleware_tests.rs

@@ -6736,7 +6736,7 @@ async fn test_runtime_registration_discovery_and_cross_owner_subscriber_gate() {
     setup_isolated_thread();
 
     let mut context =
-        PluginRegistrationContext::with_namespace("__nemo_relay_plugin__observability__");
+        PluginRegistrationContext::with_namespace("nemo-relay-plugin.v1.observability:1:");
     let deliveries = Arc::new(AtomicU32::new(0));
     let captured_deliveries = Arc::clone(&deliveries);
     context
@@ -6757,7 +6757,7 @@ async fn test_runtime_registration_discovery_and_cross_owner_subscriber_gate() {
     assert_eq!(registration.kind, RuntimeRegistrationKind::Subscriber);
     assert_eq!(
         registration.effective_name,
-        "__nemo_relay_plugin__observability__opentelemetry"
+        "nemo-relay-plugin.v1.observability:1:opentelemetry"
     );
     assert_eq!(
         registration.owner.kind,
@@ -6767,6 +6767,7 @@ async fn test_runtime_registration_discovery_and_cross_owner_subscriber_gate() {
         registration.owner.plugin_kind.as_deref(),
         Some("observability")
     );
+    assert_eq!(registration.owner.component_ordinal, Some(1));
 
     register_conditional_middleware_guardrail(
         "external_observability_gate",

crates/core/tests/unit/observability/plugin_component_tests.rs

@@ -1511,7 +1511,7 @@ fn opentelemetry_endpoint_header_env_rejects_missing_and_duplicate_headers() {
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
     clear_plugin_configuration().unwrap();
 }
@@ -1547,7 +1547,7 @@ fn invalid_log_endpoint_keeps_valid_signal_subscriber() {
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
     clear_plugin_configuration().unwrap();
 }
@@ -1583,7 +1583,7 @@ fn invalid_metric_endpoint_keeps_valid_signal_subscriber() {
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
     clear_plugin_configuration().unwrap();
 }
@@ -1635,7 +1635,7 @@ fn malformed_derived_signal_endpoint_keeps_valid_peers() {
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
     clear_plugin_configuration().unwrap();
 }
@@ -2576,7 +2576,7 @@ fn atof_enabled_writes_jsonl_and_teardown_flushes() {
             .keys()
             .cloned()
             .collect::<Vec<_>>();
-        assert_eq!(names, vec!["__nemo_relay_plugin__observability__atof"]);
+        assert_eq!(names, vec!["nemo-relay-plugin.v1.observability:1:atof"]);
     }
 
     let agent = push_agent("atof-agent");
@@ -3703,11 +3703,11 @@ fn otlp_sections_register_inferred_subscribers_with_full_config() {
         .keys()
         .cloned()
         .collect::<Vec<_>>();
-    assert!(names.contains(&"__nemo_relay_plugin__observability__opentelemetry".to_string()));
+    assert!(names.contains(&"nemo-relay-plugin.v1.observability:1:opentelemetry".to_string()));
     assert_eq!(
         names
             .iter()
-            .filter(|name| *name == "__nemo_relay_plugin__observability__opentelemetry")
+            .filter(|name| *name == "nemo-relay-plugin.v1.observability:1:opentelemetry")
             .count(),
         1
     );
@@ -3918,7 +3918,7 @@ fn opentelemetry_rejects_canonical_collision_during_validation_and_activation()
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
 }
 
@@ -3994,7 +3994,7 @@ fn invalid_later_opentelemetry_endpoint_keeps_fanout_registration() {
             .read()
             .unwrap()
             .event_subscribers
-            .contains_key("__nemo_relay_plugin__observability__opentelemetry")
+            .contains_key("nemo-relay-plugin.v1.observability:1:opentelemetry")
     );
     clear_plugin_configuration().unwrap();
 }