feat(middleware): add dynamic conditional middleware guardrails (#831)

#### Overview

Add dynamically registered conditional middleware guardrails that can enable or disable global runtime registrations by registration kind and effective name. The surface is available to Rust, primary language bindings, native plugins, and gRPC worker plugins without introducing a new native ABI version beyond V4.

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

#### Details

- Add stable runtime-registration kinds, structured registration identity/discovery, and a process-global conditional guardrail registry.
- Centralize the shared runtime-registration discovery DTOs in `nemo-relay-types` while preserving the existing core, native plugin SDK, and worker SDK import paths through re-exports.
- Apply matching gates at runtime snapshot boundaries so registering or deregistering a gate changes behavior between turns without rebuilding the underlying registration registry.
- Preserve the agreed invariants: global registrations only, scope-local registrations remain ungated, every matching gate must allow a target, and gate failures fail open.
- Expose registration discovery and dynamic gate lifecycle APIs through Rust, Python, Node.js, experimental Go/C FFI, native plugin ABI V4, and the gRPC worker protocol/Python worker SDK.
- Document native and worker timer-driven control patterns and middleware semantics.

Validation:

- `cargo test -p nemo-relay-types` — passed, including stable kind serialization, DTO round trips, and trait coverage.
- `just test-rust` — passed.
- `just test-python` — passed (689 package tests and 19 Python plugin-example tests).
- `just test-node` — passed (396 package tests and 21 Node plugin-example tests).
- The new Go conditional-guardrail test passed. `just test-go` otherwise reaches the existing `TestObservabilityPluginActivatesDerivedLogsAndExplicitMetrics` timeout waiting for `/v1/logs`; the same failure reproduced twice from untouched `origin/release/0.8`, so it is not introduced by this branch.
- `cargo clippy --workspace --all-targets -- -D warnings` — passed.
- `uv run pre-commit run --all-files` — passed.

Real-provider end-to-end validation has not been run. Current executable coverage uses in-process/runtime fixtures and local test collectors.

Breaking changes: none expected. This extends the V4 ABI being introduced for the 0.8 release rather than adding V5. The shared DTO move preserves existing Rust import paths and JSON/protobuf/C ABI wire shapes.

#### Where should the reviewer start?

Start with `crates/types/src/api/registry.rs` for the canonical discovery model and `crates/core/src/api/registry.rs` for the gate registry and compatibility re-exports, then `crates/core/src/api/runtime/state.rs` for runtime filtering semantics. Cross-plane APIs are centered in `crates/plugin/src/lib.rs`, `crates/core/src/plugin/dynamic/worker.rs`, and `crates/worker-proto/proto/nemo/relay/worker/v1/plugin_worker.proto`. End-to-end core behavior is covered in `crates/core/tests/integration/middleware_tests.rs`.

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

- Relates to: N/A



## Summary by CodeRabbit

- **New Features**
  - Added conditional middleware guardrails that can temporarily disable matching subscribers, sanitizers, guardrails, metadata injectors, or intercepts.
  - Added runtime registration discovery with filtering and ownership metadata.
  - Added management APIs across Python, Node.js, Go, C, native plugins, and worker plugins.
  - Added lifecycle cleanup, callback safety, and fail-open behavior.
- **Bug Fixes**
  - Improved runtime consistency by evaluating middleware from stable snapshots.
- **Documentation**
  - Documented guardrail behavior, ordering, ownership, discovery, and usage.

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

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

URL: #831

Author: bbednarski9

crates/core/src/api/llm.rs

@@ -5,14 +5,273 @@
 //! intercepts, and subscribers.
 
 use crate::api::runtime::{
-    EventMetadataInjectorFn, EventSanitizeFn, LlmConditionalFn, LlmExecutionFn,
-    LlmRequestInterceptFn, LlmSanitizeRequestFn, LlmSanitizeResponseFn, LlmStreamExecutionFn,
-    ToolConditionalFn, ToolExecutionFn, ToolInterceptFn, ToolSanitizeFn,
+    ConditionalMiddlewareGuardrailFn, EventMetadataInjectorFn, EventSanitizeFn, LlmConditionalFn,
+    LlmExecutionFn, LlmRequestInterceptFn, LlmSanitizeRequestFn, LlmSanitizeResponseFn,
+    LlmStreamExecutionFn, ToolConditionalFn, ToolExecutionFn, ToolInterceptFn, ToolSanitizeFn,
 };
 use crate::api::runtime::{current_scope_stack, global_context};
 use crate::api::shared::ensure_runtime_owner;
 use crate::error::{FlowError, Result};
 use crate::registry::RegistryEntry;
+pub use nemo_relay_types::api::registry::{
+    RuntimeRegistrationIdentity, RuntimeRegistrationKind, RuntimeRegistrationOwner,
+    RuntimeRegistrationOwnerKind,
+};
+use std::collections::{BTreeMap, BTreeSet};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::sync::{OnceLock, RwLock};
+
+#[derive(Clone)]
+struct ConditionalMiddlewareGuardrail {
+    kinds: BTreeSet<RuntimeRegistrationKind>,
+    registration_name: String,
+    callback: ConditionalMiddlewareGuardrailFn,
+}
+
+static CONDITIONAL_MIDDLEWARE_GUARDRAILS: OnceLock<
+    RwLock<BTreeMap<String, ConditionalMiddlewareGuardrail>>,
+> = OnceLock::new();
+
+fn conditional_middleware_guardrails()
+-> &'static RwLock<BTreeMap<String, ConditionalMiddlewareGuardrail>> {
+    CONDITIONAL_MIDDLEWARE_GUARDRAILS.get_or_init(|| RwLock::new(BTreeMap::new()))
+}
+
+/// Register a global conditional middleware guardrail.
+pub fn register_conditional_middleware_guardrail(
+    name: &str,
+    kinds: BTreeSet<RuntimeRegistrationKind>,
+    registration_name: &str,
+    guardrail: ConditionalMiddlewareGuardrailFn,
+) -> Result<()> {
+    ensure_runtime_owner()?;
+    if kinds.is_empty() {
+        return Err(FlowError::InvalidArgument(
+            "conditional middleware guardrail kinds must not be empty".to_string(),
+        ));
+    }
+    let mut gates = conditional_middleware_guardrails()
+        .write()
+        .map_err(|error| FlowError::Internal(error.to_string()))?;
+    if gates.contains_key(name) {
+        return Err(FlowError::AlreadyExists(format!(
+            "{name} conditional middleware guardrail already exists"
+        )));
+    }
+    gates.insert(
+        name.to_string(),
+        ConditionalMiddlewareGuardrail {
+            kinds,
+            registration_name: registration_name.to_string(),
+            callback: guardrail,
+        },
+    );
+    Ok(())
+}
+
+/// Deregister a global conditional middleware guardrail.
+pub fn deregister_conditional_middleware_guardrail(name: &str) -> Result<bool> {
+    ensure_runtime_owner()?;
+    let mut gates = conditional_middleware_guardrails()
+        .write()
+        .map_err(|error| FlowError::Internal(error.to_string()))?;
+    Ok(gates.remove(name).is_some())
+}
+
+/// Return whether one global runtime registration is enabled by every
+/// matching conditional middleware guardrail.
+pub(crate) fn runtime_registration_is_enabled(
+    kind: RuntimeRegistrationKind,
+    effective_name: &str,
+) -> bool {
+    let Some(registry) = CONDITIONAL_MIDDLEWARE_GUARDRAILS.get() else {
+        return true;
+    };
+    let matching = match registry.read() {
+        Ok(gates) => gates
+            .iter()
+            .filter(|(_, gate)| {
+                gate.kinds.contains(&kind) && gate.registration_name == effective_name
+            })
+            .map(|(name, gate)| (name.clone(), gate.kinds.clone(), gate.callback.clone()))
+            .collect::<Vec<_>>(),
+        Err(error) => {
+            log::error!(
+                target: "nemo_relay.runtime",
+                event = "conditional_middleware_guardrail_registry_failed",
+                registration_kind = kind.as_str(),
+                registration_name = effective_name;
+                "Conditional middleware guardrail registry read failed; enabling target: {error}"
+            );
+            return true;
+        }
+    };
+
+    for (gate_name, kinds, callback) in matching {
+        match catch_unwind(AssertUnwindSafe(|| callback(&kinds, effective_name))) {
+            Ok(None) => {}
+            Ok(Some(reason)) => {
+                log::debug!(
+                    target: "nemo_relay.runtime",
+                    event = "runtime_registration_disabled",
+                    gate = gate_name.as_str(),
+                    registration_kind = kind.as_str(),
+                    registration_name = effective_name,
+                    reason = reason.as_str();
+                    "Conditional middleware guardrail disabled runtime registration"
+                );
+                return false;
+            }
+            Err(_) => log::error!(
+                target: "nemo_relay.runtime",
+                event = "conditional_middleware_guardrail_panicked",
+                gate = gate_name.as_str(),
+                registration_kind = kind.as_str(),
+                registration_name = effective_name;
+                "Conditional middleware guardrail panicked; enabling target"
+            ),
+        }
+    }
+    true
+}
+
+fn registration_identity(
+    kind: RuntimeRegistrationKind,
+    effective_name: &str,
+) -> RuntimeRegistrationIdentity {
+    const PREFIX: &str = "__nemo_relay_plugin__";
+    let Some(rest) = effective_name.strip_prefix(PREFIX) else {
+        return RuntimeRegistrationIdentity {
+            kind,
+            local_name: effective_name.to_string(),
+            effective_name: effective_name.to_string(),
+            owner: RuntimeRegistrationOwner {
+                kind: RuntimeRegistrationOwnerKind::GlobalApi,
+                plugin_kind: None,
+                component_ordinal: None,
+            },
+        };
+    };
+    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,
+        },
+    }
+}
+
+/// List a deterministic snapshot of global gateable runtime registrations.
+pub fn list_runtime_registrations(
+    kinds: Option<&BTreeSet<RuntimeRegistrationKind>>,
+) -> Result<Vec<RuntimeRegistrationIdentity>> {
+    ensure_runtime_owner()?;
+    let context = global_context();
+    let state = context
+        .read()
+        .map_err(|error| FlowError::Internal(error.to_string()))?;
+    let mut registrations = Vec::new();
+    macro_rules! collect_registry {
+        ($kind:expr, $field:ident) => {
+            if kinds.is_none_or(|selected| selected.contains(&$kind)) {
+                registrations.extend(
+                    state
+                        .$field
+                        .values()
+                        .map(|entry| registration_identity($kind, &entry.name)),
+                );
+            }
+        };
+    }
+    if kinds.is_none_or(|selected| selected.contains(&RuntimeRegistrationKind::Subscriber)) {
+        registrations.extend(
+            state
+                .event_subscribers
+                .keys()
+                .map(|name| registration_identity(RuntimeRegistrationKind::Subscriber, name)),
+        );
+    }
+    collect_registry!(
+        RuntimeRegistrationKind::EventMetadataInjector,
+        event_metadata_injectors
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::MarkSanitizeGuardrail,
+        mark_sanitize_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ScopeSanitizeStartGuardrail,
+        scope_sanitize_start_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ScopeSanitizeEndGuardrail,
+        scope_sanitize_end_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ToolSanitizeRequestGuardrail,
+        tool_sanitize_request_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ToolSanitizeResponseGuardrail,
+        tool_sanitize_response_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ToolConditionalExecutionGuardrail,
+        tool_conditional_execution_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ToolRequestIntercept,
+        tool_request_intercepts
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::ToolExecutionIntercept,
+        tool_execution_intercepts
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmSanitizeRequestGuardrail,
+        llm_sanitize_request_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmSanitizeResponseGuardrail,
+        llm_sanitize_response_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmConditionalExecutionGuardrail,
+        llm_conditional_execution_guardrails
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmRequestIntercept,
+        llm_request_intercepts
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmExecutionIntercept,
+        llm_execution_intercepts
+    );
+    collect_registry!(
+        RuntimeRegistrationKind::LlmStreamExecutionIntercept,
+        llm_stream_execution_intercepts
+    );
+    registrations.sort_by(|left, right| {
+        left.kind
+            .cmp(&right.kind)
+            .then_with(|| left.effective_name.cmp(&right.effective_name))
+    });
+    Ok(registrations)
+}
 
 /// A priority-ordered registration record.
 ///

crates/core/src/api/registry.rs

@@ -11,12 +11,12 @@ pub mod state;
 pub mod subscriber_dispatcher;
 
 pub use callbacks::{
-    BuiltinLlmCodec, EventMetadataInjectorFn, EventSanitizeFn, EventSubscriberFn, LlmCodecIdentity,
-    LlmCollectorFn, LlmConditionalFn, LlmExecutionFn, LlmExecutionNextFn, LlmFinalizerFn,
-    LlmJsonStream, LlmRequestInterceptFn, LlmSanitizeRequestContext, LlmSanitizeRequestFn,
-    LlmSanitizeResponseContext, LlmSanitizeResponseFn, LlmStreamExecutionFn,
-    LlmStreamExecutionNextFn, LlmStreamInner, ToolConditionalFn, ToolExecutionFn,
-    ToolExecutionNextFn, ToolInterceptFn, ToolSanitizeFn,
+    BuiltinLlmCodec, ConditionalMiddlewareGuardrailFn, EventMetadataInjectorFn, EventSanitizeFn,
+    EventSubscriberFn, LlmCodecIdentity, LlmCollectorFn, LlmConditionalFn, LlmExecutionFn,
+    LlmExecutionNextFn, LlmFinalizerFn, LlmJsonStream, LlmRequestInterceptFn,
+    LlmSanitizeRequestContext, LlmSanitizeRequestFn, LlmSanitizeResponseContext,
+    LlmSanitizeResponseFn, LlmStreamExecutionFn, LlmStreamExecutionNextFn, LlmStreamInner,
+    ToolConditionalFn, ToolExecutionFn, ToolExecutionNextFn, ToolInterceptFn, ToolSanitizeFn,
 };
 #[doc(hidden)]
 pub use continuation_context::MiddlewareContinuationContext;

crates/core/src/api/runtime.rs

@@ -8,7 +8,7 @@
 //! so the runtime can compose tool and LLM middleware consistently across
 //! bindings.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -18,13 +18,22 @@ use tokio_stream::Stream;
 
 use crate::api::event::{Event, EventSanitizeFields};
 use crate::api::llm::{LlmRequest, LlmRequestInterceptOutcome};
+use crate::api::registry::RuntimeRegistrationKind;
 use crate::api::tool::{ToolExecutionInterceptOutcome, ToolExecutionResult};
 use crate::codec::request::AnnotatedLlmRequest;
 use crate::codec::traits::{LlmCodec, LlmResponseCodec};
 use crate::error::Result;
 use crate::json::Json;
 pub use nemo_relay_types::codec::identity::{BuiltinLlmCodec, LlmCodecIdentity};
 
+/// Decide whether matching global runtime registrations remain eligible.
+///
+/// Relay invokes this callback with the gate's configured registration kinds
+/// and the target registration's effective name. Returning `None` enables the
+/// target. Returning a reason string disables it for the current snapshot.
+pub type ConditionalMiddlewareGuardrailFn =
+    Arc<dyn Fn(&BTreeSet<RuntimeRegistrationKind>, &str) -> Option<String> + Send + Sync>;
+
 /// Sanitize mutable observability fields on a fully constructed event.
 ///
 /// The callback receives the current event as immutable context and the fields

crates/core/src/api/runtime/callbacks.rs

@@ -369,6 +369,21 @@ impl ScopeStack {
             .collect()
     }
 
+    /// Clone one registry field from every active scope that owns it.
+    ///
+    /// Eligibility callbacks must not run while the scope-stack lock is held.
+    /// Resolution paths use these owned snapshots before consulting global
+    /// conditional middleware guardrails.
+    pub(crate) fn snapshot_scope_local_registries<T: RegistryEntry + Clone>(
+        &self,
+        field: impl Fn(&ScopeLocalRegistries) -> &SortedRegistry<T>,
+    ) -> Vec<SortedRegistry<T>> {
+        self.collect_scope_local_registries(field)
+            .into_iter()
+            .cloned()
+            .collect()
+    }
+
     /// Collect all scope-local subscribers visible from the active stack.
     ///
     /// # Returns