feat(plugins): add typed marks and runtime diagnostics

crates/core/src/plugin.rs

@@ -10,7 +10,7 @@
 //! - rollback bookkeeping for registrations created during plugin setup
 
 use std::cell::Cell;
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::fmt;
 use std::future::Future;
 use std::panic::{AssertUnwindSafe, catch_unwind};
@@ -219,6 +219,16 @@ pub struct RuntimeDiagnostic {
     pub count: u64,
 }
 
+/// Read-only projection of runtime diagnostics exposed to dynamic plugins.
+#[derive(Debug, Clone, Serialize)]
+pub(crate) struct RuntimeDiagnosticsSnapshotEntry {
+    pub(crate) code: String,
+    pub(crate) message: String,
+    pub(crate) count: u64,
+}
+
+const MAX_DYNAMIC_PLUGIN_RUNTIME_DIAGNOSTICS: usize = 32;
+
 impl ConfigReport {
     /// Returns `true` when the report contains at least one error diagnostic.
     pub fn has_errors(&self) -> bool {
@@ -1684,6 +1694,7 @@ fn install_previous_configuration_for_teardown(
     *guard = Some(ActivePluginConfiguration {
         config: previous_state.config.clone(),
         report: previous_state.report.clone(),
+        runtime_diagnostics: previous_state.runtime_diagnostics.clone(),
         registrations: Vec::new(),
     });
     Ok(())
@@ -1752,9 +1763,10 @@ async fn restore_previous_plugin_configuration(
     .await
     {
         Ok(registrations) => {
-            store_active_plugin_configuration(
+            store_active_plugin_configuration_with_runtime_diagnostics(
                 previous_state.config,
                 previous_state.report,
+                previous_state.runtime_diagnostics,
                 registrations,
             )?;
             log::warn!(
@@ -2451,6 +2463,19 @@ pub fn record_active_plugin_runtime_diagnostic(diagnostic: RuntimeDiagnostic) {
     let Some(state) = guard.as_mut() else {
         return;
     };
+    if let Some(existing) = state.runtime_diagnostics.get_mut(&diagnostic.code) {
+        existing.message = diagnostic.message.clone();
+        existing.count = existing.count.saturating_add(diagnostic.count);
+    } else if state.runtime_diagnostics.len() < MAX_DYNAMIC_PLUGIN_RUNTIME_DIAGNOSTICS {
+        state.runtime_diagnostics.insert(
+            diagnostic.code.clone(),
+            RuntimeDiagnosticsSnapshotEntry {
+                code: diagnostic.code.clone(),
+                message: diagnostic.message.clone(),
+                count: diagnostic.count,
+            },
+        );
+    }
     if let Some(existing) = state
         .report
         .runtime_diagnostics
@@ -2469,6 +2494,19 @@ pub fn record_active_plugin_runtime_diagnostic(diagnostic: RuntimeDiagnostic) {
     }
 }
 
+/// Return a bounded, active-only runtime-diagnostics snapshot for dynamic plugins.
+pub(crate) fn active_runtime_diagnostics_snapshot() -> Vec<RuntimeDiagnosticsSnapshotEntry> {
+    ACTIVE_PLUGIN_CONFIGURATION
+        .lock()
+        .ok()
+        .and_then(|guard| {
+            guard
+                .as_ref()
+                .map(|state| state.runtime_diagnostics.values().cloned().collect())
+        })
+        .unwrap_or_default()
+}
+
 /// Rolls back registrations in reverse order, ignoring rollback failures.
 ///
 /// This is used internally during failed initialization and by
@@ -2537,6 +2575,7 @@ fn panic_payload_message(payload: Box<dyn std::any::Any + Send>) -> String {
 struct ActivePluginConfiguration {
     config: PluginConfig,
     report: ConfigReport,
+    runtime_diagnostics: BTreeMap<String, RuntimeDiagnosticsSnapshotEntry>,
     registrations: Vec<PluginRegistration>,
 }
 
@@ -2658,13 +2697,28 @@ fn store_active_plugin_configuration(
     config: PluginConfig,
     report: ConfigReport,
     registrations: Vec<PluginRegistration>,
+) -> Result<()> {
+    store_active_plugin_configuration_with_runtime_diagnostics(
+        config,
+        report,
+        BTreeMap::new(),
+        registrations,
+    )
+}
+
+fn store_active_plugin_configuration_with_runtime_diagnostics(
+    config: PluginConfig,
+    report: ConfigReport,
+    runtime_diagnostics: BTreeMap<String, RuntimeDiagnosticsSnapshotEntry>,
+    registrations: Vec<PluginRegistration>,
 ) -> Result<()> {
     let mut guard = ACTIVE_PLUGIN_CONFIGURATION.lock().map_err(|err| {
         PluginError::Internal(format!("active plugin configuration lock poisoned: {err}"))
     })?;
     *guard = Some(ActivePluginConfiguration {
         config,
         report,
+        runtime_diagnostics,
         registrations,
     });
     if let Ok(mut guard) = LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT.lock() {

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

@@ -21,7 +21,7 @@ use std::task::{Context, Poll};
 
 use futures_util::FutureExt;
 
-use crate::api::event::{Event, EventSanitizeFields};
+use crate::api::event::{DataSchema, Event, EventSanitizeFields, LogSeverity};
 use crate::api::llm::{LlmRequest, LlmRequestInterceptOutcome};
 use crate::api::runtime::{
     EventSanitizeFn, EventSubscriberFn, LlmCodecIdentity, LlmConditionalFn, LlmExecutionFn,
@@ -45,7 +45,8 @@ use crate::codec::traits::{LlmCodec, LlmResponseCodec};
 use crate::error::{FlowError, Result as FlowResult};
 use crate::plugin::{
     ConfigDiagnostic, DiagnosticLevel, Plugin, PluginError, PluginRegistrationContext,
-    deregister_plugin_registration_checked, register_plugin_tracked,
+    active_runtime_diagnostics_snapshot, deregister_plugin_registration_checked,
+    register_plugin_tracked,
 };
 use chrono::{DateTime, Utc};
 use libloading::{Library, Symbol};
@@ -399,9 +400,9 @@ fn load_one_native_plugin(
                 ))
             })?;
         let mut status = entry(native_host_api(), &mut plugin);
-        // Older SDKs reject newer tables. Negotiate through separately frozen
-        // v4, v3, and v2 tables so their struct sizes and function pointers do
-        // not change as the current ABI grows.
+        // Older SDKs reject newer tables. Negotiate from the current v4 table
+        // through separately frozen v3 and v2 tables so their struct sizes and
+        // function pointers do not change as the current ABI grows.
         if status == NemoRelayStatus::InvalidArg {
             drop_native_plugin_descriptor(&mut plugin);
             status = entry(native_host_api_v3(), &mut plugin);
@@ -919,6 +920,8 @@ fn build_native_host_api_v4() -> NemoRelayNativeHostApiV4 {
         async_llm_stream_release: native_async_llm_stream_release,
         async_completion_retain: native_async_completion_retain,
         async_stream_is_backpressured: native_async_stream_is_backpressured,
+        emit_mark_v2: native_emit_mark_v2,
+        get_runtime_diagnostics: native_get_runtime_diagnostics,
     }
 }
 
@@ -1004,6 +1007,38 @@ fn optional_json_from_native_string(
     })
 }
 
+fn optional_typed_json_from_native_string<T: serde::de::DeserializeOwned>(
+    value: *const NemoRelayNativeString,
+    field: &str,
+) -> Result<Option<T>, NemoRelayStatus> {
+    optional_json_from_native_string(value, field)?
+        .map(|value| {
+            serde_json::from_value(value).map_err(|err| {
+                set_native_last_error(format!("{field} has an invalid shape: {err}"));
+                NemoRelayStatus::InvalidArg
+            })
+        })
+        .transpose()
+}
+
+fn optional_severity_from_native_string(
+    value: *const NemoRelayNativeString,
+) -> Result<Option<LogSeverity>, NemoRelayStatus> {
+    if value.is_null() {
+        return Ok(None);
+    }
+    let value = read_native_string(value).map_err(|err| {
+        set_native_last_error(err.to_string());
+        NemoRelayStatus::InvalidUtf8
+    })?;
+    serde_json::from_value(Json::String(value))
+        .map(Some)
+        .map_err(|err| {
+            set_native_last_error(format!("mark severity is invalid: {err}"));
+            NemoRelayStatus::InvalidArg
+        })
+}
+
 fn optional_timestamp_from_native(
     timestamp_unix_micros: *const i64,
 ) -> Result<Option<DateTime<Utc>>, NemoRelayStatus> {
@@ -1199,6 +1234,78 @@ unsafe extern "C" fn native_emit_mark(
     }
 }
 
+#[allow(clippy::too_many_arguments)] // Mirrors the append-only native ABI function.
+unsafe extern "C" fn native_emit_mark_v2(
+    name: *const NemoRelayNativeString,
+    parent: *const NemoRelayNativeScopeHandle,
+    data_json: *const NemoRelayNativeString,
+    metadata_json: *const NemoRelayNativeString,
+    data_schema_json: *const NemoRelayNativeString,
+    severity: *const NemoRelayNativeString,
+    timestamp_unix_micros: *const i64,
+) -> NemoRelayStatus {
+    clear_native_last_error();
+    let name = match read_name(name) {
+        Ok(name) => name,
+        Err(status) => return status,
+    };
+    let data = match optional_json_from_native_string(data_json, "mark data") {
+        Ok(data) => data,
+        Err(status) => return status,
+    };
+    let metadata = match optional_json_from_native_string(metadata_json, "mark metadata") {
+        Ok(metadata) => metadata,
+        Err(status) => return status,
+    };
+    let data_schema = match optional_typed_json_from_native_string::<DataSchema>(
+        data_schema_json,
+        "mark data schema",
+    ) {
+        Ok(data_schema) => data_schema,
+        Err(status) => return status,
+    };
+    let severity = match optional_severity_from_native_string(severity) {
+        Ok(severity) => severity,
+        Err(status) => return status,
+    };
+    let timestamp = match optional_timestamp_from_native(timestamp_unix_micros) {
+        Ok(timestamp) => timestamp,
+        Err(status) => return status,
+    };
+    let parent_ref = native_scope_ref(parent);
+    match emit_scope_mark(
+        EmitMarkEventParams::builder()
+            .name(&name)
+            .parent_opt(parent_ref)
+            .data_opt(data)
+            .metadata_opt(metadata)
+            .data_schema_opt(data_schema)
+            .severity_opt(severity)
+            .timestamp_opt(timestamp)
+            .build(),
+    ) {
+        Ok(()) => NemoRelayStatus::Ok,
+        Err(err) => status_from_flow_error(err),
+    }
+}
+
+unsafe extern "C" fn native_get_runtime_diagnostics(
+    out_json: *mut *mut NemoRelayNativeString,
+) -> NemoRelayStatus {
+    clear_native_last_error();
+    let diagnostics = active_runtime_diagnostics_snapshot();
+    match serde_json::to_value(diagnostics) {
+        Ok(entries) => {
+            let value = Json::Object(Map::from_iter([("entries".into(), entries)]));
+            write_native_json(&value, out_json)
+        }
+        Err(error) => {
+            set_native_last_error(format!("failed to serialize runtime diagnostics: {error}"));
+            NemoRelayStatus::Internal
+        }
+    }
+}
+
 unsafe extern "C" fn native_scope_stack_create(
     out: *mut *mut NemoRelayNativeScopeStack,
 ) -> NemoRelayStatus {