diff --git a/README.md b/README.md index ee2bddae5..dc298bc58 100644 --- a/README.md +++ b/README.md @@ -49,8 +49,8 @@ consistent across frameworks and languages. ecosystem components, third-party agent frameworks, provider adapters, or direct application code. - ⚙️ **Install reusable runtime behavior**: Plugins configure middleware, - subscribers, adaptive components, and custom runtime behavior from one shared - system. + subscribers, adaptive components, observability exporters, and custom runtime + behavior from one shared system. ## What You Get @@ -64,6 +64,9 @@ consistent across frameworks and languages. - ✅ **Observability-ready events**: Preserve model metadata, tool call IDs, inputs, outputs, scope relationships, and lifecycle timing for downstream analysis. +- ✅ **Built-in observability plugin**: Configure Agent Trajectory Observability + Format (ATOF), ATIF, OpenTelemetry, and OpenInference exporters without + registering subscribers by hand. - ✅ **Extension points for framework authors**: Wrap stable tool and provider callbacks while preserving framework-owned scheduling, retries, memory, and result handling. diff --git a/crates/core/README.md b/crates/core/README.md index 3b819d111..02db9192d 100644 --- a/crates/core/README.md +++ b/crates/core/README.md @@ -44,6 +44,9 @@ Node.js bindings mirror the semantics exposed by this crate. scope that owns them and clean them up when that scope closes. - ✅ **Plugin primitives**: Register reusable runtime behavior configured from one shared plugin system. +- ✅ **Built-in observability plugin**: Configure first-party Agent Trajectory + Observability Format (ATOF), ATIF, OpenTelemetry, and OpenInference exporters + from the core crate. - ✅ **Codec and typed helpers**: Normalize provider requests and responses for framework integrations. - ✅ **Binding source of truth**: Use the runtime semantics mirrored by the diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 075d0a1c5..d7caf2f32 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -17,3 +17,4 @@ pub mod atof; pub mod openinference; #[cfg(feature = "otel")] pub mod otel; +pub mod plugin_component; diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs new file mode 100644 index 000000000..2aa3fbd2e --- /dev/null +++ b/crates/core/src/observability/plugin_component.rs @@ -0,0 +1,1179 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Built-in observability plugin component. +//! +//! This module packages NeMo Flow's first-party observability exporters behind +//! the shared plugin configuration system. Each exporter section is opt-in: +//! omitted sections and sections with `enabled = false` validate but do not +//! register subscribers or construct exporters. +//! +//! The plugin intentionally infers subscriber names from the component namespace +//! so configuration remains portable across bindings. ATOF, OpenTelemetry, and +//! OpenInference each register one global subscriber when enabled. ATIF uses a +//! global dispatcher that detects direct child agent scopes and creates one +//! scope-local exporter for each top-level agent run. + +use std::collections::HashMap; +use std::future::Future; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value as Json}; +use uuid::Uuid; + +use crate::api::event::{Event, ScopeCategory}; +use crate::api::runtime::{EventSubscriberFn, current_scope_stack}; +use crate::api::scope::ScopeType; +use crate::api::subscriber::{scope_deregister_subscriber, scope_register_subscriber}; +use crate::observability::atif::{AtifAgentInfo, AtifExporter}; +use crate::observability::atof::{ + AtofExporter, AtofExporterConfig as CoreAtofExporterConfig, AtofExporterMode, +}; +#[cfg(feature = "openinference")] +use crate::observability::openinference::{ + OpenInferenceConfig as CoreOpenInferenceConfig, OpenInferenceSubscriber, + OtlpTransport as OpenInferenceTransport, +}; +#[cfg(feature = "otel")] +use crate::observability::otel::{ + OpenTelemetryConfig as CoreOpenTelemetryConfig, OpenTelemetrySubscriber, +}; +use crate::plugin::{ + ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, PluginComponentSpec, PluginError, + PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, + deregister_plugin, register_plugin, +}; + +/// The plugin kind registered by the core crate. +pub const OBSERVABILITY_PLUGIN_KIND: &str = "observability"; + +/// Top-level observability component wrapper. +/// +/// Use this wrapper when constructing a [`PluginComponentSpec`] from Rust +/// instead of hand-writing the generic plugin component shape. The component +/// kind is always [`OBSERVABILITY_PLUGIN_KIND`]. +#[derive(Debug, Clone)] +pub struct ComponentSpec { + /// Whether the observability component should be activated. + pub enabled: bool, + /// Observability config for this top-level component. + pub config: ObservabilityConfig, +} + +impl ComponentSpec { + /// Creates an enabled observability component spec. + /// + /// The returned component can be converted into the generic plugin config + /// entry with `PluginComponentSpec::from(...)`. + pub fn new(config: ObservabilityConfig) -> Self { + Self { + enabled: true, + config, + } + } +} + +impl From for PluginComponentSpec { + fn from(value: ComponentSpec) -> Self { + let Json::Object(config) = serde_json::to_value(value.config) + .expect("observability config should serialize to object") + else { + unreachable!("observability config must serialize to object"); + }; + + PluginComponentSpec { + kind: OBSERVABILITY_PLUGIN_KIND.to_string(), + enabled: value.enabled, + config, + } + } +} + +/// Canonical config document for the observability plugin component. +/// +/// Every section is optional. A missing section has the same activation +/// behavior as a section with `enabled = false`: it contributes no runtime +/// subscribers and performs no export work. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ObservabilityConfig { + /// Observability config schema version. + #[serde(default = "default_observability_config_version")] + pub version: u32, + /// Filesystem-backed raw ATOF JSONL export. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub atof: Option, + /// Per-top-level-agent ATIF trajectory export. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub atif: Option, + /// OpenTelemetry trace export. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub opentelemetry: Option, + /// OpenInference trace export. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub openinference: Option, + /// Observability-local unsupported-config policy. + #[serde(default)] + pub policy: ConfigPolicy, +} + +impl Default for ObservabilityConfig { + fn default() -> Self { + Self { + version: default_observability_config_version(), + atof: None, + atif: None, + opentelemetry: None, + openinference: None, + policy: ConfigPolicy::default(), + } + } +} + +/// Filesystem-backed ATOF JSONL exporter config. +/// +/// When enabled, this section wraps +/// [`crate::observability::atof::AtofExporter`] and writes the raw ATOF event +/// stream as JSONL. The exporter uses the current working directory and a +/// timestamped filename when no explicit path settings are supplied. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AtofSectionConfig { + /// Whether ATOF JSONL export is active. + #[serde(default)] + pub enabled: bool, + /// Directory containing the JSONL output file. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub output_directory: Option, + /// Output filename. Defaults to the underlying ATOF exporter timestamped filename. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub filename: Option, + /// File open mode: `append` or `overwrite`. + #[serde(default = "default_atof_mode")] + pub mode: String, +} + +impl Default for AtofSectionConfig { + fn default() -> Self { + Self { + enabled: false, + output_directory: None, + filename: None, + mode: default_atof_mode(), + } + } +} + +/// Per-agent ATIF trajectory exporter config. +/// +/// When enabled, this section creates a dispatcher that opens a separate +/// [`crate::observability::atif::AtifExporter`] for each top-level agent scope. The `{session_id}` +/// placeholder in [`AtifSectionConfig::filename_template`] is required so +/// concurrent sibling agents cannot overwrite each other's trajectory files. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AtifSectionConfig { + /// Whether ATIF export is active. + #[serde(default)] + pub enabled: bool, + /// Human-readable agent name. + #[serde(default = "default_agent_name")] + pub agent_name: String, + /// Agent version string. + #[serde(default = "default_agent_version")] + pub agent_version: String, + /// Default model name. + #[serde(default = "default_model_name")] + pub model_name: String, + /// Tool definitions available to the agent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_definitions: Option>, + /// Extra ATIF agent metadata. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub extra: Option, + /// Directory containing trajectory JSON files. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub output_directory: Option, + /// Filename template. `{session_id}` is replaced with the top-level agent scope UUID. + #[serde(default = "default_atif_filename_template")] + pub filename_template: String, +} + +impl Default for AtifSectionConfig { + fn default() -> Self { + Self { + enabled: false, + agent_name: default_agent_name(), + agent_version: default_agent_version(), + model_name: default_model_name(), + tool_definitions: None, + extra: None, + output_directory: None, + filename_template: default_atif_filename_template(), + } + } +} + +/// Shared OTLP exporter config for OpenTelemetry and OpenInference. +/// +/// The `opentelemetry` and `openinference` sections share the same shape but +/// construct different subscriber implementations. Both sections are disabled +/// by default and use `http_binary` transport unless configured otherwise. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct OtlpSectionConfig { + /// Whether the subscriber is active. + #[serde(default)] + pub enabled: bool, + /// OTLP transport: `http_binary` or `grpc`. + #[serde(default = "default_otlp_transport")] + pub transport: String, + /// OTLP endpoint. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub endpoint: Option, + /// Extra exporter headers or metadata. + #[serde(default)] + pub headers: HashMap, + /// Extra resource attributes. + #[serde(default)] + pub resource_attributes: HashMap, + /// `service.name` resource attribute. + #[serde(default = "default_service_name")] + pub service_name: String, + /// Optional `service.namespace` resource attribute. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub service_namespace: Option, + /// Optional `service.version` resource attribute. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub service_version: Option, + /// Instrumentation scope name. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub instrumentation_scope: Option, + /// Export timeout in milliseconds. + #[serde(default = "default_timeout_millis")] + pub timeout_millis: u64, +} + +impl Default for OtlpSectionConfig { + fn default() -> Self { + Self { + enabled: false, + transport: default_otlp_transport(), + endpoint: None, + headers: HashMap::new(), + resource_attributes: HashMap::new(), + service_name: default_service_name(), + service_namespace: None, + service_version: None, + instrumentation_scope: None, + timeout_millis: default_timeout_millis(), + } + } +} + +struct ObservabilityPlugin; + +impl Plugin for ObservabilityPlugin { + fn plugin_kind(&self) -> &str { + OBSERVABILITY_PLUGIN_KIND + } + + fn allows_multiple_components(&self) -> bool { + false + } + + fn validate(&self, plugin_config: &Map) -> Vec { + validate_observability_plugin_config(plugin_config) + } + + fn register<'a>( + &'a self, + plugin_config: &Map, + ctx: &'a mut PluginRegistrationContext, + ) -> Pin> + Send + 'a>> { + let plugin_config = plugin_config.clone(); + Box::pin(async move { + let config = parse_observability_config(&plugin_config)?; + register_observability(config, ctx) + }) + } +} + +/// Registers the observability component kind in the core plugin registry. +/// +/// Calling this function more than once is safe. The core plugin APIs call it +/// automatically before listing, looking up, validating, or initializing plugin +/// components, so applications normally do not need to invoke it directly. +pub fn register_observability_component() -> PluginResult<()> { + match register_plugin(Arc::new(ObservabilityPlugin)) { + Ok(()) => Ok(()), + Err(PluginError::RegistrationFailed(message)) if message.contains("already registered") => { + Ok(()) + } + Err(err) => Err(err), + } +} + +/// Deregisters the observability component kind from the core plugin registry. +/// +/// This helper exists primarily for tests and specialized embedding scenarios. +/// It removes the plugin kind from future registry lookups but does not clear an +/// already active plugin configuration. +pub fn deregister_observability_component() -> bool { + deregister_plugin(OBSERVABILITY_PLUGIN_KIND) +} + +fn register_observability( + config: ObservabilityConfig, + ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + if let Some(atof) = config.atof.filter(|section| section.enabled) { + register_atof_exporter(atof, ctx)?; + } + if let Some(atif) = config.atif.filter(|section| section.enabled) { + register_atif_dispatcher(atif, ctx)?; + } + if let Some(otel) = config.opentelemetry.filter(|section| section.enabled) { + register_opentelemetry(otel, ctx)?; + } + if let Some(openinference) = config.openinference.filter(|section| section.enabled) { + register_openinference(openinference, ctx)?; + } + Ok(()) +} + +fn register_atof_exporter( + section: AtofSectionConfig, + ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + let mode = AtofExporterMode::parse(§ion.mode).ok_or_else(|| { + PluginError::InvalidConfig("ATOF mode must be 'append' or 'overwrite'".to_string()) + })?; + let mut config = CoreAtofExporterConfig::new().with_mode(mode); + if let Some(output_directory) = section.output_directory { + config = config.with_output_directory(output_directory); + } + if let Some(filename) = section.filename { + config = config.with_filename(filename); + } + + let exporter = Arc::new(AtofExporter::new(config).map_err(observability_registration_error)?); + ctx.register_subscriber("atof", exporter.subscriber())?; + ctx.add_registration(PluginRegistration::new( + "observability", + ctx.qualify_name("atof.shutdown"), + Box::new(move || { + exporter + .shutdown() + .map_err(observability_registration_error) + }), + )); + Ok(()) +} + +fn register_atif_dispatcher( + section: AtifSectionConfig, + ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + if !section.filename_template.contains("{session_id}") { + return Err(PluginError::InvalidConfig( + "ATIF filename_template must contain '{session_id}'".to_string(), + )); + } + + let manager = Arc::new(Mutex::new(AtifDispatcher::new(section))); + let dispatcher = atif_dispatcher_subscriber(Arc::clone(&manager), ctx.qualify_name("atif-")); + ctx.register_subscriber("atif", dispatcher)?; + ctx.add_registration(PluginRegistration::new( + "observability", + ctx.qualify_name("atif.shutdown"), + Box::new(move || { + let work = { + let mut guard = manager.lock().map_err(|err| { + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) + })?; + guard + .flush_open_agents() + .map_err(observability_registration_error)? + }; + for (scope_uuid, name) in work.scope_subscribers { + let _ = scope_deregister_subscriber(&scope_uuid, &name); + } + for write in work.writes { + let agent_uuid = write.agent_uuid; + let result = write_atif_file(&write); + let mut guard = manager.lock().map_err(|err| { + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) + })?; + guard + .finish_agent_write(agent_uuid, result) + .map_err(observability_registration_error)?; + } + let guard = manager.lock().map_err(|err| { + PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}")) + })?; + guard + .last_error_result() + .map_err(observability_registration_error) + }), + )); + Ok(()) +} + +#[cfg(feature = "otel")] +fn register_opentelemetry( + section: OtlpSectionConfig, + ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + let subscriber = Arc::new( + OpenTelemetrySubscriber::new(build_otel_config(section)?) + .map_err(observability_registration_error)?, + ); + ctx.register_subscriber("opentelemetry", subscriber.subscriber())?; + ctx.add_registration(PluginRegistration::new( + "observability", + ctx.qualify_name("opentelemetry.shutdown"), + Box::new(move || { + subscriber + .shutdown() + .map_err(observability_registration_error) + }), + )); + Ok(()) +} + +#[cfg(not(feature = "otel"))] +fn register_opentelemetry( + _section: OtlpSectionConfig, + _ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + Err(PluginError::InvalidConfig( + "OpenTelemetry support is not enabled in this build".to_string(), + )) +} + +#[cfg(feature = "openinference")] +fn register_openinference( + section: OtlpSectionConfig, + ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + let subscriber = Arc::new( + OpenInferenceSubscriber::new(build_openinference_config(section)?) + .map_err(observability_registration_error)?, + ); + ctx.register_subscriber("openinference", subscriber.subscriber())?; + ctx.add_registration(PluginRegistration::new( + "observability", + ctx.qualify_name("openinference.shutdown"), + Box::new(move || { + subscriber + .shutdown() + .map_err(observability_registration_error) + }), + )); + Ok(()) +} + +#[cfg(not(feature = "openinference"))] +fn register_openinference( + _section: OtlpSectionConfig, + _ctx: &mut PluginRegistrationContext, +) -> PluginResult<()> { + Err(PluginError::InvalidConfig( + "OpenInference support is not enabled in this build".to_string(), + )) +} + +struct AtifDispatcher { + config: AtifSectionConfig, + agents: HashMap, + scope_subscribers: HashMap, + last_error: Option, +} + +struct ManagedAtifExporter { + exporter: AtifExporter, + path: PathBuf, + observed_events: Vec, + written: bool, +} + +struct PendingAtifWrite { + agent_uuid: Uuid, + path: PathBuf, + payload: Vec, +} + +struct AtifFlushWork { + writes: Vec, + scope_subscribers: Vec<(Uuid, String)>, +} + +impl AtifDispatcher { + fn new(config: AtifSectionConfig) -> Self { + Self { + config, + agents: HashMap::new(), + scope_subscribers: HashMap::new(), + last_error: None, + } + } + + fn observe_global(&mut self, event: &Event, subscriber_prefix: &str, state: Arc>) { + if self.last_error.is_some() || !is_top_level_agent_start(event) { + return; + } + + // The top-level agent scope UUID is the ATIF session ID. The global + // dispatcher records the start event itself because the scope-local + // subscriber is attached after that start event has already been + // emitted. + let session_id = event.uuid().to_string(); + let exporter = AtifExporter::new(session_id.clone(), self.agent_info()); + (exporter.subscriber())(event); + let path = self.output_path(&session_id); + self.agents.insert( + event.uuid(), + ManagedAtifExporter { + exporter, + path, + observed_events: vec![event.clone()], + written: false, + }, + ); + + let agent_uuid = event.uuid(); + let name = format!("{subscriber_prefix}{agent_uuid}"); + let callback = atif_scope_subscriber(state, agent_uuid); + // Attach the per-agent subscriber to the agent scope rather than the + // global registry so sibling top-level agents never share events. + if let Err(err) = scope_register_subscriber(&agent_uuid, &name, callback) { + self.last_error = Some(format!("failed to register ATIF scope subscriber: {err}")); + } else { + self.scope_subscribers.insert(agent_uuid, name); + } + } + + fn observe_scope(&mut self, event: &Event, agent_uuid: Uuid) -> Option { + if self.last_error.is_some() { + return None; + } + let should_finalize = + event.uuid() == agent_uuid && event.scope_category() == Some(ScopeCategory::End); + let agent = self.agents.get_mut(&agent_uuid)?; + (agent.exporter.subscriber())(event); + agent.observed_events.push(event.clone()); + if !should_finalize || agent.written { + return None; + } + match prepare_atif_file(agent_uuid, agent) { + Ok(write) => Some(write), + Err(err) => { + self.last_error = Some(err.to_string()); + None + } + } + } + + fn complete_scope_write( + &mut self, + agent_uuid: Uuid, + result: std::io::Result<()>, + ) -> Option<(Uuid, String)> { + if self.finish_agent_write(agent_uuid, result).is_err() { + return None; + } + self.agents.remove(&agent_uuid); + self.scope_subscribers + .remove(&agent_uuid) + .map(|name| (agent_uuid, name)) + } + + fn flush_open_agents(&mut self) -> std::io::Result { + // Plugin teardown may run before an agent scope closes. Remove dynamic + // scope-local subscribers first so the later scope end event cannot + // trigger a second write after the dispatcher has flushed. + let scope_subscribers = std::mem::take(&mut self.scope_subscribers) + .into_iter() + .collect(); + let agent_uuids = self + .agents + .iter() + .filter_map(|(agent_uuid, agent)| (!agent.written).then_some(*agent_uuid)) + .collect::>(); + let mut writes = Vec::with_capacity(agent_uuids.len()); + for agent_uuid in agent_uuids { + if let Some(agent) = self.agents.get_mut(&agent_uuid) { + writes.push(prepare_atif_file(agent_uuid, agent)?); + } + } + Ok(AtifFlushWork { + writes, + scope_subscribers, + }) + } + + fn finish_agent_write( + &mut self, + agent_uuid: Uuid, + result: std::io::Result<()>, + ) -> std::io::Result<()> { + match result { + Ok(()) => { + if let Some(agent) = self.agents.get_mut(&agent_uuid) { + agent.observed_events.clear(); + } + Ok(()) + } + Err(err) => { + if let Some(agent) = self.agents.get_mut(&agent_uuid) { + agent.written = false; + } + self.last_error = Some(err.to_string()); + Err(err) + } + } + } + + fn last_error_result(&self) -> std::io::Result<()> { + if let Some(message) = &self.last_error { + return Err(std::io::Error::other(message.clone())); + } + Ok(()) + } + + fn agent_info(&self) -> AtifAgentInfo { + AtifAgentInfo { + name: self.config.agent_name.clone(), + version: self.config.agent_version.clone(), + model_name: Some(self.config.model_name.clone()), + tool_definitions: self.config.tool_definitions.clone(), + extra: self.config.extra.clone(), + } + } + + fn output_path(&self, session_id: &str) -> PathBuf { + let directory = self + .config + .output_directory + .clone() + .unwrap_or_else(default_output_directory); + let filename = self + .config + .filename_template + .replace("{session_id}", session_id); + directory.join(filename) + } +} + +fn atif_dispatcher_subscriber( + manager: Arc>, + subscriber_prefix: String, +) -> EventSubscriberFn { + Arc::new(move |event: &Event| { + let Ok(mut guard) = manager.lock() else { + return; + }; + guard.observe_global(event, &subscriber_prefix, Arc::clone(&manager)); + }) +} + +fn atif_scope_subscriber( + manager: Arc>, + agent_uuid: Uuid, +) -> EventSubscriberFn { + Arc::new(move |event: &Event| { + let pending_write = { + let Ok(mut guard) = manager.lock() else { + return; + }; + guard.observe_scope(event, agent_uuid) + }; + let Some(write) = pending_write else { + return; + }; + let result = write_atif_file(&write); + let scope_subscriber = { + let Ok(mut guard) = manager.lock() else { + return; + }; + guard.complete_scope_write(write.agent_uuid, result) + }; + if let Some((scope_uuid, name)) = scope_subscriber { + let _ = scope_deregister_subscriber(&scope_uuid, &name); + } + }) +} + +fn prepare_atif_file( + agent_uuid: Uuid, + agent: &mut ManagedAtifExporter, +) -> std::io::Result { + let trajectory = agent.exporter.export(); + let mut value = serde_json::to_value(trajectory)?; + if let Some(object) = value.as_object_mut() { + object.insert( + "extra".to_string(), + serde_json::json!({ + "observed_events": agent.observed_events, + }), + ); + } + let payload = serde_json::to_vec_pretty(&value)?; + agent.written = true; + Ok(PendingAtifWrite { + agent_uuid, + path: agent.path.clone(), + payload, + }) +} + +fn write_atif_file(write: &PendingAtifWrite) -> std::io::Result<()> { + if let Some(parent) = write.path.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::write(&write.path, &write.payload)?; + Ok(()) +} + +fn is_top_level_agent_start(event: &Event) -> bool { + if event.scope_category() != Some(ScopeCategory::Start) + || event.scope_type() != Some(ScopeType::Agent) + { + return false; + } + let Some(parent_uuid) = event.parent_uuid() else { + return false; + }; + current_scope_stack() + .read() + .map(|stack| stack.root_uuid() == parent_uuid) + .unwrap_or(false) +} + +#[cfg(feature = "otel")] +fn build_otel_config(section: OtlpSectionConfig) -> PluginResult { + let mut config = match section.transport.as_str() { + "http_binary" => CoreOpenTelemetryConfig::http_binary(section.service_name), + "grpc" => CoreOpenTelemetryConfig::grpc(section.service_name), + other => { + return Err(PluginError::InvalidConfig(format!( + "OpenTelemetry transport must be 'http_binary' or 'grpc', got {other:?}" + ))); + } + } + .with_timeout(Duration::from_millis(section.timeout_millis)); + + if let Some(endpoint) = section.endpoint { + config = config.with_endpoint(endpoint); + } + if let Some(namespace) = section.service_namespace { + config = config.with_service_namespace(namespace); + } + if let Some(version) = section.service_version { + config = config.with_service_version(version); + } + if let Some(scope) = section.instrumentation_scope { + config = config.with_instrumentation_scope(scope); + } + for (key, value) in section.headers { + config = config.with_header(key, value); + } + for (key, value) in section.resource_attributes { + config = config.with_resource_attribute(key, value); + } + Ok(config) +} + +#[cfg(feature = "openinference")] +fn build_openinference_config(section: OtlpSectionConfig) -> PluginResult { + let transport = match section.transport.as_str() { + "http_binary" => OpenInferenceTransport::HttpBinary, + "grpc" => OpenInferenceTransport::Grpc, + other => { + return Err(PluginError::InvalidConfig(format!( + "OpenInference transport must be 'http_binary' or 'grpc', got {other:?}" + ))); + } + }; + let mut config = CoreOpenInferenceConfig::new() + .with_transport(transport) + .with_service_name(section.service_name) + .with_timeout(Duration::from_millis(section.timeout_millis)); + + if let Some(endpoint) = section.endpoint { + config = config.with_endpoint(endpoint); + } + if let Some(namespace) = section.service_namespace { + config = config.with_service_namespace(namespace); + } + if let Some(version) = section.service_version { + config = config.with_service_version(version); + } + if let Some(scope) = section.instrumentation_scope { + config = config.with_instrumentation_scope(scope); + } + for (key, value) in section.headers { + config = config.with_header(key, value); + } + for (key, value) in section.resource_attributes { + config = config.with_resource_attribute(key, value); + } + Ok(config) +} + +fn parse_observability_config( + plugin_config: &Map, +) -> PluginResult { + serde_json::from_value(Json::Object(plugin_config.clone())).map_err(|err| { + PluginError::InvalidConfig(format!("invalid observability plugin config: {err}")) + }) +} + +fn validate_observability_plugin_config( + plugin_config: &Map, +) -> Vec { + let config = match parse_observability_config(plugin_config) { + Ok(config) => config, + Err(err) => { + return vec![ConfigDiagnostic { + level: DiagnosticLevel::Error, + code: "observability.invalid_plugin_config".to_string(), + component: Some(OBSERVABILITY_PLUGIN_KIND.to_string()), + field: None, + message: err.to_string(), + }]; + } + }; + + let mut diagnostics = vec![]; + validate_unknown_fields( + &mut diagnostics, + &config.policy, + Some(OBSERVABILITY_PLUGIN_KIND.to_string()), + plugin_config, + &[ + "version", + "atof", + "atif", + "opentelemetry", + "openinference", + "policy", + ], + ); + + validate_version(&mut diagnostics, &config.policy, config.version); + validate_policy_fields(&mut diagnostics, &config.policy, plugin_config); + validate_section_fields( + &mut diagnostics, + &config.policy, + plugin_config, + "atof", + &["enabled", "output_directory", "filename", "mode"], + ); + validate_section_fields( + &mut diagnostics, + &config.policy, + plugin_config, + "atif", + &[ + "enabled", + "agent_name", + "agent_version", + "model_name", + "tool_definitions", + "extra", + "output_directory", + "filename_template", + ], + ); + validate_section_fields( + &mut diagnostics, + &config.policy, + plugin_config, + "opentelemetry", + &[ + "enabled", + "transport", + "endpoint", + "headers", + "resource_attributes", + "service_name", + "service_namespace", + "service_version", + "instrumentation_scope", + "timeout_millis", + ], + ); + validate_section_fields( + &mut diagnostics, + &config.policy, + plugin_config, + "openinference", + &[ + "enabled", + "transport", + "endpoint", + "headers", + "resource_attributes", + "service_name", + "service_namespace", + "service_version", + "instrumentation_scope", + "timeout_millis", + ], + ); + + if let Some(section) = &config.atof { + validate_atof_values(&mut diagnostics, &config.policy, section); + #[cfg(target_arch = "wasm32")] + if section.enabled { + push_policy_diag( + &mut diagnostics, + config.policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some("enabled".to_string()), + "ATOF file export is not supported on WebAssembly".to_string(), + ); + } + } + if let Some(section) = &config.atif { + validate_atif_values(&mut diagnostics, &config.policy, section); + #[cfg(target_arch = "wasm32")] + if section.enabled { + push_policy_diag( + &mut diagnostics, + config.policy.unsupported_value, + "observability.unsupported_value", + Some("atif".to_string()), + Some("enabled".to_string()), + "ATIF file export is not supported on WebAssembly".to_string(), + ); + } + } + if let Some(section) = &config.opentelemetry { + validate_otlp_values(&mut diagnostics, &config.policy, "opentelemetry", section); + #[cfg(not(feature = "otel"))] + if section.enabled { + push_policy_diag( + &mut diagnostics, + config.policy.unsupported_value, + "observability.feature_disabled", + Some("opentelemetry".to_string()), + Some("enabled".to_string()), + "OpenTelemetry support is not enabled in this build".to_string(), + ); + } + } + if let Some(section) = &config.openinference { + validate_otlp_values(&mut diagnostics, &config.policy, "openinference", section); + #[cfg(not(feature = "openinference"))] + if section.enabled { + push_policy_diag( + &mut diagnostics, + config.policy.unsupported_value, + "observability.feature_disabled", + Some("openinference".to_string()), + Some("enabled".to_string()), + "OpenInference support is not enabled in this build".to_string(), + ); + } + } + + diagnostics +} + +fn validate_version(diagnostics: &mut Vec, policy: &ConfigPolicy, version: u32) { + if version != 1 { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_config_version", + Some(OBSERVABILITY_PLUGIN_KIND.to_string()), + Some("version".to_string()), + format!("observability config version {version} is unsupported"), + ); + } +} + +fn validate_policy_fields( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + plugin_config: &Map, +) { + if let Some(policy_json) = plugin_config.get("policy").and_then(Json::as_object) { + validate_unknown_fields( + diagnostics, + policy, + Some("policy".to_string()), + policy_json, + &["unknown_component", "unknown_field", "unsupported_value"], + ); + } +} + +fn validate_section_fields( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + plugin_config: &Map, + section: &str, + known_fields: &[&str], +) { + if let Some(section_json) = plugin_config.get(section).and_then(Json::as_object) { + validate_unknown_fields( + diagnostics, + policy, + Some(section.to_string()), + section_json, + known_fields, + ); + } +} + +fn validate_atof_values( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtofSectionConfig, +) { + if AtofExporterMode::parse(§ion.mode).is_none() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some("mode".to_string()), + "ATOF mode must be 'append' or 'overwrite'".to_string(), + ); + } +} + +fn validate_atif_values( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtifSectionConfig, +) { + if !section.filename_template.contains("{session_id}") { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atif".to_string()), + Some("filename_template".to_string()), + "ATIF filename_template must contain '{session_id}'".to_string(), + ); + } +} + +fn validate_otlp_values( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section_name: &str, + section: &OtlpSectionConfig, +) { + if !matches!(section.transport.as_str(), "http_binary" | "grpc") { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some(section_name.to_string()), + Some("transport".to_string()), + format!("{section_name} transport must be 'http_binary' or 'grpc'"), + ); + } +} + +fn validate_unknown_fields( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + component: Option, + config: &Map, + known_fields: &[&str], +) { + for field in config.keys() { + if !known_fields.contains(&field.as_str()) { + push_policy_diag( + diagnostics, + policy.unknown_field, + "observability.unknown_field", + component.clone(), + Some(field.clone()), + format!( + "field '{}' is not recognized for '{}'", + field, + component.as_deref().unwrap_or("unknown") + ), + ); + } + } +} + +fn push_policy_diag( + diagnostics: &mut Vec, + behavior: UnsupportedBehavior, + code: &str, + component: Option, + field: Option, + message: String, +) { + let level = match behavior { + UnsupportedBehavior::Ignore => return, + UnsupportedBehavior::Warn => DiagnosticLevel::Warning, + UnsupportedBehavior::Error => DiagnosticLevel::Error, + }; + diagnostics.push(ConfigDiagnostic { + level, + code: code.to_string(), + component, + field, + message, + }); +} + +fn observability_registration_error(error: impl std::fmt::Display) -> PluginError { + PluginError::RegistrationFailed(error.to_string()) +} + +fn default_observability_config_version() -> u32 { + 1 +} + +fn default_atof_mode() -> String { + "append".to_string() +} + +fn default_agent_name() -> String { + "NeMo Flow".to_string() +} + +fn default_agent_version() -> String { + env!("CARGO_PKG_VERSION").to_string() +} + +fn default_model_name() -> String { + "unknown".to_string() +} + +fn default_atif_filename_template() -> String { + "nemo-flow-atif-{session_id}.json".to_string() +} + +fn default_otlp_transport() -> String { + "http_binary".to_string() +} + +fn default_service_name() -> String { + "nemo-flow".to_string() +} + +fn default_timeout_millis() -> u64 { + 3_000 +} + +fn default_output_directory() -> PathBuf { + std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")) +} + +#[cfg(test)] +#[path = "../../tests/unit/observability/plugin_component_tests.rs"] +mod tests; diff --git a/crates/core/src/plugin.rs b/crates/core/src/plugin.rs index e119de7d8..3f8187f1a 100644 --- a/crates/core/src/plugin.rs +++ b/crates/core/src/plugin.rs @@ -13,7 +13,7 @@ use std::collections::{HashMap, HashSet}; use std::fmt; use std::future::Future; use std::pin::Pin; -use std::sync::{Arc, LazyLock, Mutex, RwLock}; +use std::sync::{Arc, LazyLock, Mutex, OnceLock, RwLock}; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value as Json}; @@ -44,6 +44,7 @@ type PluginMap = HashMap>; static PLUGIN_HANDLERS: LazyLock> = LazyLock::new(|| RwLock::new(HashMap::new())); static ACTIVE_PLUGIN_CONFIGURATION: LazyLock>> = LazyLock::new(|| Mutex::new(None)); +static BUILTIN_PLUGIN_REGISTRATION: OnceLock> = OnceLock::new(); /// Error type for generic plugin operations. #[derive(Debug, Error)] @@ -729,6 +730,31 @@ pub fn register_plugin(plugin: Arc) -> Result<()> { Ok(()) } +/// Registers core-provided plugin kinds. +/// +/// Built-in plugins are available to validation and initialization without a +/// binding or application-specific registration call. +pub fn ensure_builtin_plugins_registered() -> Result<()> { + match BUILTIN_PLUGIN_REGISTRATION + .get_or_init(crate::observability::plugin_component::register_observability_component) + { + Ok(()) => Ok(()), + Err(err) => Err(clone_cached_plugin_error(err)), + } +} + +fn clone_cached_plugin_error(err: &PluginError) -> PluginError { + match err { + PluginError::InvalidConfig(message) => PluginError::InvalidConfig(message.clone()), + PluginError::NotFound(message) => PluginError::NotFound(message.clone()), + PluginError::Serialization(err) => PluginError::Internal(err.to_string()), + PluginError::Internal(message) => PluginError::Internal(message.clone()), + PluginError::RegistrationFailed(message) => { + PluginError::RegistrationFailed(message.clone()) + } + } +} + /// Removes a previously registered plugin. /// /// This affects future validation and initialization only. Active runtime @@ -764,6 +790,7 @@ pub fn deregister_plugin(plugin_kind: &str) -> bool { /// Disabled or inactive components still appear here when their plugin kind is /// registered. pub fn list_plugin_kinds() -> Vec { + let _ = ensure_builtin_plugins_registered(); let mut kinds = PLUGIN_HANDLERS .read() .map(|guard| guard.keys().cloned().collect::>()) @@ -784,6 +811,7 @@ pub fn list_plugin_kinds() -> Vec { /// # Notes /// The returned plugin is shared by [`Arc`], so callers receive a cheap clone. pub fn lookup_plugin(plugin_kind: &str) -> Option> { + let _ = ensure_builtin_plugins_registered(); PLUGIN_HANDLERS .read() .ok() @@ -806,6 +834,7 @@ pub fn lookup_plugin(plugin_kind: &str) -> Option> { /// Validation checks host policy, plugin multiplicity rules, unknown component /// kinds, and plugin-provided validation hooks. pub fn validate_plugin_config(config: &PluginConfig) -> ConfigReport { + let _ = ensure_builtin_plugins_registered(); let mut report = ConfigReport::default(); if config.version != 1 { @@ -967,6 +996,7 @@ struct ActivePluginConfiguration { } async fn initialize_plugin_components(config: &PluginConfig) -> Result> { + ensure_builtin_plugins_registered()?; let totals = plugin_component_totals(config); let mut ordinals: HashMap<&str, usize> = HashMap::new(); let mut registrations = vec![]; diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs new file mode 100644 index 000000000..97b83c34d --- /dev/null +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -0,0 +1,605 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for the built-in observability plugin component. + +use super::*; +use crate::api::event::{BaseEvent, EventCategory, ScopeEvent}; +use crate::api::runtime::NemoFlowContextState; +use crate::api::runtime::global_context; +use crate::api::scope::{PopScopeParams, PushScopeParams}; +use crate::plugin::{ + PluginComponentSpec, PluginConfig, clear_plugin_configuration, initialize_plugins, + list_plugin_kinds, lookup_plugin, validate_plugin_config, +}; +use serde_json::json; +use std::fs; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn temp_dir(prefix: &str) -> PathBuf { + let id = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!("nemo-flow-{prefix}-{id}")); + fs::create_dir_all(&path).unwrap(); + path +} + +fn reset_runtime() { + let _ = clear_plugin_configuration(); + crate::shared_runtime::reset_runtime_owner_for_tests(); + let context = global_context(); + *context.write().unwrap() = NemoFlowContextState::new(); +} + +fn component(config: Json) -> PluginComponentSpec { + let Json::Object(config) = config else { + panic!("component config must be an object"); + }; + PluginComponentSpec { + kind: OBSERVABILITY_PLUGIN_KIND.to_string(), + enabled: true, + config, + } +} + +fn plugin_config(config: Json) -> PluginConfig { + PluginConfig { + version: 1, + components: vec![component(config)], + policy: Default::default(), + } +} + +fn push_agent(name: &str) -> crate::api::scope::ScopeHandle { + crate::api::scope::push_scope( + PushScopeParams::builder() + .name(name) + .scope_type(ScopeType::Agent) + .input(json!({"agent": name})) + .build(), + ) + .unwrap() +} + +fn push_function(name: &str) -> crate::api::scope::ScopeHandle { + crate::api::scope::push_scope( + PushScopeParams::builder() + .name(name) + .scope_type(ScopeType::Function) + .input(json!({"function": name})) + .build(), + ) + .unwrap() +} + +fn pop(handle: &crate::api::scope::ScopeHandle) { + crate::api::scope::pop_scope( + PopScopeParams::builder() + .handle_uuid(&handle.uuid) + .output(json!({"done": handle.name})) + .build(), + ) + .unwrap(); +} + +#[test] +fn default_config_and_component_conversion_cover_public_shape() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let defaults = ObservabilityConfig::default(); + assert_eq!(defaults.version, 1); + assert!(defaults.atof.is_none()); + assert!(defaults.atif.is_none()); + assert!(defaults.opentelemetry.is_none()); + assert!(defaults.openinference.is_none()); + + let atof = AtofSectionConfig::default(); + assert!(!atof.enabled); + assert_eq!(atof.mode, "append"); + assert!(atof.output_directory.is_none()); + assert!(atof.filename.is_none()); + + let atif = AtifSectionConfig::default(); + assert!(!atif.enabled); + assert_eq!(atif.agent_name, "NeMo Flow"); + assert_eq!(atif.agent_version, env!("CARGO_PKG_VERSION")); + assert_eq!(atif.model_name, "unknown"); + assert_eq!(atif.filename_template, "nemo-flow-atif-{session_id}.json"); + + let otlp = OtlpSectionConfig::default(); + assert!(!otlp.enabled); + assert_eq!(otlp.transport, "http_binary"); + assert_eq!(otlp.service_name, "nemo-flow"); + assert_eq!(otlp.timeout_millis, 3_000); + + let generic: PluginComponentSpec = ComponentSpec::new(ObservabilityConfig { + atof: Some(atof), + atif: Some(atif), + opentelemetry: Some(otlp.clone()), + openinference: Some(otlp), + ..ObservabilityConfig::default() + }) + .into(); + assert_eq!(generic.kind, OBSERVABILITY_PLUGIN_KIND); + assert!(generic.enabled); + assert_eq!(generic.config["version"], json!(1)); + assert_eq!(generic.config["atif"]["agent_name"], json!("NeMo Flow")); +} + +#[test] +fn built_in_registration_is_automatic() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + assert!(list_plugin_kinds().contains(&OBSERVABILITY_PLUGIN_KIND.to_string())); + assert!(lookup_plugin(OBSERVABILITY_PLUGIN_KIND).is_some()); + + let config = plugin_config(json!({})); + assert!(!validate_plugin_config(&config).has_errors()); +} + +#[test] +fn empty_and_disabled_config_register_nothing() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let config = plugin_config(json!({ + "atof": {"enabled": false, "mode": "overwrite"}, + "atif": {"enabled": false}, + "opentelemetry": {"enabled": false, "transport": "grpc"}, + "openinference": {"enabled": false, "transport": "grpc"} + })); + assert!(!validate_plugin_config(&config).has_errors()); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let state = global_context(); + assert!(state.read().unwrap().event_subscribers.is_empty()); +} + +#[test] +fn disabled_file_sections_do_not_create_files() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-disabled-files"); + + let config = plugin_config(json!({ + "atof": { + "enabled": false, + "output_directory": dir, + "filename": "events.jsonl" + }, + "atif": { + "enabled": false, + "output_directory": dir, + "filename_template": "trajectory-{session_id}.json" + } + })); + assert!(!validate_plugin_config(&config).has_errors()); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let agent = push_agent("disabled-agent"); + pop(&agent); + clear_plugin_configuration().unwrap(); + + assert!(!dir.join("events.jsonl").exists()); + assert!(!dir.join(format!("trajectory-{}.json", agent.uuid)).exists()); +} + +#[test] +fn duplicate_component_is_rejected_as_singleton() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let config = PluginConfig { + version: 1, + components: vec![component(json!({})), component(json!({}))], + policy: Default::default(), + }; + let report = validate_plugin_config(&config); + assert!(report.has_errors()); + assert!( + report + .diagnostics + .iter() + .any(|diag| diag.code == "plugin.duplicate_component") + ); +} + +#[test] +fn unknown_fields_and_bad_values_follow_policy() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let warn_report = validate_plugin_config(&plugin_config(json!({ + "atof": {"bogus": true, "mode": "invalid"}, + "atif": {"filename_template": "missing-session"} + }))); + assert!(warn_report.has_errors()); + assert!( + warn_report + .diagnostics + .iter() + .any(|diag| diag.code == "observability.unknown_field") + ); + assert!( + warn_report + .diagnostics + .iter() + .any(|diag| diag.field.as_deref() == Some("mode")) + ); + assert!( + warn_report + .diagnostics + .iter() + .any(|diag| diag.field.as_deref() == Some("filename_template")) + ); + + let ignore_report = validate_plugin_config(&plugin_config(json!({ + "policy": {"unknown_field": "ignore", "unsupported_value": "ignore"}, + "atof": {"bogus": true, "mode": "invalid"}, + "atif": {"filename_template": "missing-session"} + }))); + assert!(!ignore_report.has_errors()); + assert!(ignore_report.diagnostics.is_empty()); +} + +#[test] +fn invalid_shapes_and_strict_policy_are_reported() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let invalid_shape = validate_plugin_config(&plugin_config(json!({ + "version": "one", + }))); + assert!(invalid_shape.has_errors()); + assert!( + invalid_shape + .diagnostics + .iter() + .any(|diag| diag.code == "observability.invalid_plugin_config") + ); + + let strict_unknown = validate_plugin_config(&plugin_config(json!({ + "policy": {"unknown_field": "error"}, + "opentelemetry": {"unexpected": true} + }))); + assert!(strict_unknown.has_errors()); + assert!( + strict_unknown + .diagnostics + .iter() + .any(|diag| diag.code == "observability.unknown_field" + && diag.component.as_deref() == Some("opentelemetry") + && diag.field.as_deref() == Some("unexpected")) + ); + + let strict_bad_transport = validate_plugin_config(&plugin_config(json!({ + "openinference": {"enabled": true, "transport": "udp"} + }))); + assert!(strict_bad_transport.has_errors()); + assert!( + strict_bad_transport + .diagnostics + .iter() + .any(|diag| diag.field.as_deref() == Some("transport")) + ); +} + +#[test] +fn initialization_fails_for_invalid_enabled_file_exporters() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-invalid-exporters"); + let not_a_directory = dir.join("not-a-directory"); + fs::write(¬_a_directory, "file").unwrap(); + + let invalid_atof = plugin_config(json!({ + "policy": {"unsupported_value": "ignore"}, + "atof": { + "enabled": true, + "mode": "invalid", + "output_directory": dir, + "filename": "events.jsonl" + } + })); + let error = futures::executor::block_on(initialize_plugins(invalid_atof)).unwrap_err(); + assert!(error.to_string().contains("ATOF mode")); + + let invalid_path = plugin_config(json!({ + "atof": { + "enabled": true, + "output_directory": not_a_directory, + "filename": "events.jsonl" + } + })); + let error = futures::executor::block_on(initialize_plugins(invalid_path)).unwrap_err(); + assert!(error.to_string().contains("registration failed")); +} + +#[test] +fn atof_enabled_writes_jsonl_and_teardown_flushes() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atof"); + + let config = plugin_config(json!({ + "atof": { + "enabled": true, + "output_directory": dir, + "filename": "events.jsonl", + "mode": "overwrite" + } + })); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + { + let state = global_context(); + let names = state + .read() + .unwrap() + .event_subscribers + .keys() + .cloned() + .collect::>(); + assert_eq!(names, vec!["__nemo_flow_plugin__observability__atof"]); + } + + let agent = push_agent("atof-agent"); + crate::api::scope::event( + crate::api::scope::EmitMarkEventParams::builder() + .name("checkpoint") + .parent(&agent) + .data(json!({"step": 1})) + .build(), + ) + .unwrap(); + pop(&agent); + clear_plugin_configuration().unwrap(); + + let content = fs::read_to_string(dir.join("events.jsonl")).unwrap(); + let lines = content.lines().collect::>(); + assert_eq!(lines.len(), 3); + assert!(lines[0].contains("\"kind\":\"scope\"")); + assert!(lines[1].contains("\"kind\":\"mark\"")); + assert!(lines[2].contains("\"scope_category\":\"end\"")); +} + +#[test] +fn atif_defaults_create_one_file_per_top_level_agent() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-defaults"); + + let config = plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir + } + })); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let first = push_agent("first-agent"); + let nested = push_agent("nested-agent"); + pop(&nested); + pop(&first); + + let second = push_agent("second-agent"); + pop(&second); + clear_plugin_configuration().unwrap(); + + let first_path = dir.join(format!("nemo-flow-atif-{}.json", first.uuid)); + let second_path = dir.join(format!("nemo-flow-atif-{}.json", second.uuid)); + assert!(first_path.exists()); + assert!(second_path.exists()); + + let first_json: Json = serde_json::from_str(&fs::read_to_string(first_path).unwrap()).unwrap(); + let second_json: Json = + serde_json::from_str(&fs::read_to_string(second_path).unwrap()).unwrap(); + + assert_eq!(first_json["session_id"], first.uuid.to_string()); + assert_eq!(first_json["agent"]["name"], "NeMo Flow"); + assert_eq!(first_json["agent"]["version"], env!("CARGO_PKG_VERSION")); + assert_eq!(first_json["agent"]["model_name"], "unknown"); + let first_serialized = first_json.to_string(); + assert!(first_serialized.contains("first-agent")); + assert!(first_serialized.contains("nested-agent")); + assert!(!first_serialized.contains("second-agent")); + + let second_serialized = second_json.to_string(); + assert!(second_serialized.contains("second-agent")); + assert!(!second_serialized.contains("first-agent")); +} + +#[test] +fn atif_completed_top_level_agent_is_evicted_after_write() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-evict"); + let root_uuid = crate::api::runtime::current_scope_stack() + .read() + .unwrap() + .root_uuid(); + let agent = push_agent("evicted-agent"); + let manager = Arc::new(Mutex::new(AtifDispatcher::new(AtifSectionConfig { + enabled: true, + output_directory: Some(dir.clone()), + ..AtifSectionConfig::default() + }))); + + let start_event = Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .uuid(agent.uuid) + .parent_uuid(root_uuid) + .name("evicted-agent") + .build(), + ScopeCategory::Start, + vec![], + EventCategory::agent(), + None, + )); + manager + .lock() + .unwrap() + .observe_global(&start_event, "__test__", Arc::clone(&manager)); + { + let dispatcher = manager.lock().unwrap(); + assert!(dispatcher.agents.contains_key(&agent.uuid)); + assert!(dispatcher.scope_subscribers.contains_key(&agent.uuid)); + } + + let end_event = Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .uuid(agent.uuid) + .parent_uuid(root_uuid) + .name("evicted-agent") + .build(), + ScopeCategory::End, + vec![], + EventCategory::agent(), + None, + )); + let pending_write = manager + .lock() + .unwrap() + .observe_scope(&end_event, agent.uuid) + .unwrap(); + let path = dir.join(format!("nemo-flow-atif-{}.json", agent.uuid)); + assert!(!path.exists()); + write_atif_file(&pending_write).unwrap(); + let scope_subscriber = manager + .lock() + .unwrap() + .complete_scope_write(agent.uuid, Ok(())); + if let Some((scope_uuid, name)) = scope_subscriber { + let _ = scope_deregister_subscriber(&scope_uuid, &name); + } + + let dispatcher = manager.lock().unwrap(); + assert!(dispatcher.last_error.is_none()); + assert!(!dispatcher.agents.contains_key(&agent.uuid)); + assert!(!dispatcher.scope_subscribers.contains_key(&agent.uuid)); + assert!(path.exists()); + drop(dispatcher); + pop(&agent); +} + +#[test] +fn atif_explicit_options_and_open_agent_teardown_are_written() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-explicit"); + + let config = plugin_config(json!({ + "atif": { + "enabled": true, + "agent_name": "custom-agent", + "agent_version": "9.9.9", + "model_name": "demo-model", + "tool_definitions": [{"name": "search"}], + "extra": {"team": "runtime"}, + "output_directory": dir, + "filename_template": "custom-{session_id}.atif.json" + } + })); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let ignored = push_function("not-an-agent"); + pop(&ignored); + let agent = push_agent("open-agent"); + clear_plugin_configuration().unwrap(); + + let path = dir.join(format!("custom-{}.atif.json", agent.uuid)); + assert!(path.exists()); + let value: Json = serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap(); + assert_eq!(value["agent"]["name"], "custom-agent"); + assert_eq!(value["agent"]["version"], "9.9.9"); + assert_eq!(value["agent"]["model_name"], "demo-model"); + assert_eq!(value["agent"]["tool_definitions"][0]["name"], "search"); + assert_eq!(value["agent"]["extra"]["team"], "runtime"); + assert!(fs::read_dir(dir).unwrap().count() == 1); + pop(&agent); +} + +#[test] +fn atif_rejects_unsafe_template_and_ignores_non_top_level_agents() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-errors"); + + let invalid_template = plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir, + "filename_template": "single-file.json" + } + })); + assert!(validate_plugin_config(&invalid_template).has_errors()); + assert!(futures::executor::block_on(initialize_plugins(invalid_template)).is_err()); + + let config = plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir, + "filename_template": "trajectory-{session_id}.json" + } + })); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let function = push_function("top-level-function"); + let nested_agent = push_agent("nested-under-function"); + pop(&nested_agent); + pop(&function); + clear_plugin_configuration().unwrap(); + + assert_eq!(fs::read_dir(dir).unwrap().count(), 0); +} + +#[test] +fn otlp_sections_register_inferred_subscribers_with_full_config() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let config = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "transport": "http_binary", + "endpoint": "http://127.0.0.1:4318/v1/traces", + "headers": {"authorization": "token"}, + "resource_attributes": {"deployment.environment": "test"}, + "service_name": "otel-service", + "service_namespace": "agents", + "service_version": "1.2.3", + "instrumentation_scope": "test-otel", + "timeout_millis": 1 + }, + "openinference": { + "enabled": true, + "transport": "http_binary", + "endpoint": "http://127.0.0.1:4318/v1/traces", + "headers": {"authorization": "token"}, + "resource_attributes": {"deployment.environment": "test"}, + "service_name": "oi-service", + "service_namespace": "agents", + "service_version": "1.2.3", + "instrumentation_scope": "test-openinference", + "timeout_millis": 1 + } + })); + assert!(!validate_plugin_config(&config).has_errors()); + futures::executor::block_on(initialize_plugins(config)).unwrap(); + + let state = global_context(); + let names = state + .read() + .unwrap() + .event_subscribers + .keys() + .cloned() + .collect::>(); + assert!(names.contains(&"__nemo_flow_plugin__observability__opentelemetry".to_string())); + assert!(names.contains(&"__nemo_flow_plugin__observability__openinference".to_string())); + clear_plugin_configuration().unwrap(); +} diff --git a/crates/ffi/nemo_flow.h b/crates/ffi/nemo_flow.h index 6251e253b..c76af4c1f 100644 --- a/crates/ffi/nemo_flow.h +++ b/crates/ffi/nemo_flow.h @@ -841,6 +841,35 @@ NemoFlowStatus nemo_flow_register_subscriber(const char *name, */ NemoFlowStatus nemo_flow_deregister_subscriber(const char *name); +/** + * Return the built-in observability plugin kind. + * + * The caller owns the returned string and must free it with `nemo_flow_string_free`. + */ +char *nemo_flow_observability_plugin_kind(void); + +/** + * Return the default observability plugin config as JSON. + * + * # Safety + * `out_json` must be a valid, non-null pointer. + */ +NemoFlowStatus nemo_flow_observability_default_config_json(char **out_json); + +/** + * Wrap an observability config JSON object as a top-level plugin component. + * + * Pass null for `config_json` to use the default observability config. The + * returned JSON can be inserted into `PluginConfig.components`. + * + * # Safety + * `config_json`, when non-null, must be a valid C string. `out_json` must be a + * valid, non-null pointer. + */ +NemoFlowStatus nemo_flow_observability_component_spec_json(const char *config_json, + bool enabled, + char **out_json); + /** * Creates a new ATIF exporter. * diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs index f0b58806b..971026b84 100644 --- a/crates/ffi/src/api/observability.rs +++ b/crates/ffi/src/api/observability.rs @@ -3,8 +3,9 @@ use super::{ Duration, FfiAtifExporter, FfiAtofExporter, FfiOpenInferenceSubscriber, - FfiOpenTelemetrySubscriber, NemoFlowStatus, c_char, c_str_to_string, clear_last_error, - core_subscriber_api, set_last_error, status_from_error, str_to_c_string, tokio_runtime, + FfiOpenTelemetrySubscriber, NemoFlowStatus, c_char, c_str_to_json, c_str_to_string, + clear_last_error, core_subscriber_api, json_to_c_string, set_last_error, status_from_error, + str_to_c_string, tokio_runtime, }; type AtofExporter = nemo_flow::observability::atof::AtofExporter; @@ -15,6 +16,8 @@ type OpenTelemetryConfig = nemo_flow::observability::otel::OpenTelemetryConfig; type OpenTelemetrySubscriber = nemo_flow::observability::otel::OpenTelemetrySubscriber; type OpenInferenceConfig = nemo_flow::observability::openinference::OpenInferenceConfig; type OpenInferenceSubscriber = nemo_flow::observability::openinference::OpenInferenceSubscriber; +type ObservabilityComponentSpec = nemo_flow::observability::plugin_component::ComponentSpec; +type ObservabilityConfig = nemo_flow::observability::plugin_component::ObservabilityConfig; fn status_from_atof_error(error: &AtofExporterError) -> NemoFlowStatus { set_last_error(&error.to_string()); @@ -24,6 +27,88 @@ fn status_from_atof_error(error: &AtofExporterError) -> NemoFlowStatus { } } +// --------------------------------------------------------------------------- +// Observability plugin component helpers +// --------------------------------------------------------------------------- + +/// Return the built-in observability plugin kind. +/// +/// The caller owns the returned string and must free it with `nemo_flow_string_free`. +#[unsafe(no_mangle)] +pub extern "C" fn nemo_flow_observability_plugin_kind() -> *mut c_char { + str_to_c_string(nemo_flow::observability::plugin_component::OBSERVABILITY_PLUGIN_KIND) +} + +/// Return the default observability plugin config as JSON. +/// +/// # Safety +/// `out_json` must be a valid, non-null pointer. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_flow_observability_default_config_json( + out_json: *mut *mut c_char, +) -> NemoFlowStatus { + clear_last_error(); + if out_json.is_null() { + set_last_error("out_json pointer is null"); + return NemoFlowStatus::NullPointer; + } + let config_json = match serde_json::to_value(ObservabilityConfig::default()) { + Ok(value) => value, + Err(error) => { + set_last_error(&error.to_string()); + return NemoFlowStatus::Internal; + } + }; + unsafe { *out_json = json_to_c_string(&config_json) }; + NemoFlowStatus::Ok +} + +/// Wrap an observability config JSON object as a top-level plugin component. +/// +/// Pass null for `config_json` to use the default observability config. The +/// returned JSON can be inserted into `PluginConfig.components`. +/// +/// # Safety +/// `config_json`, when non-null, must be a valid C string. `out_json` must be a +/// valid, non-null pointer. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_flow_observability_component_spec_json( + config_json: *const c_char, + enabled: bool, + out_json: *mut *mut c_char, +) -> NemoFlowStatus { + clear_last_error(); + if out_json.is_null() { + set_last_error("out_json pointer is null"); + return NemoFlowStatus::NullPointer; + } + let config = if config_json.is_null() { + ObservabilityConfig::default() + } else { + let Some(config_value) = c_str_to_json(config_json) else { + return NemoFlowStatus::InvalidJson; + }; + match serde_json::from_value::(config_value) { + Ok(config) => config, + Err(error) => { + set_last_error(&error.to_string()); + return NemoFlowStatus::InvalidJson; + } + } + }; + let component: nemo_flow::plugin::PluginComponentSpec = + ObservabilityComponentSpec { enabled, config }.into(); + let component_json = match serde_json::to_value(component) { + Ok(value) => value, + Err(error) => { + set_last_error(&error.to_string()); + return NemoFlowStatus::Internal; + } + }; + unsafe { *out_json = json_to_c_string(&component_json) }; + NemoFlowStatus::Ok +} + // --------------------------------------------------------------------------- // ATIF exporter // --------------------------------------------------------------------------- diff --git a/crates/ffi/tests/unit/api/core_tests.rs b/crates/ffi/tests/unit/api/core_tests.rs index f827bbed7..22f4f9de0 100644 --- a/crates/ffi/tests/unit/api/core_tests.rs +++ b/crates/ffi/tests/unit/api/core_tests.rs @@ -57,6 +57,11 @@ fn test_ffi_plugin_config_validate_initialize_and_clear() { .as_array() .is_some_and(|values| values.iter().any(|value| value == "adaptive")) ); + assert!( + kinds + .as_array() + .is_some_and(|values| values.iter().any(|value| value == "observability")) + ); let mut configured_json = ptr::null_mut(); assert_eq!( @@ -84,6 +89,288 @@ fn test_ffi_plugin_config_validate_initialize_and_clear() { assert_eq!(unsafe { returned_json(cleared_json) }, Json::Null); } +#[test] +fn test_ffi_observability_plugin_file_sinks() { + let _guard = TEST_MUTEX.lock().unwrap(); + reset_globals(); + let _ = nemo_flow_clear_plugin_configuration(); + let dir = std::env::temp_dir().join(unique_name("ffi_observability_plugin")); + std::fs::create_dir_all(&dir).unwrap(); + let dir_text = dir.to_string_lossy().into_owned(); + + let config = cstring( + &json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "output_directory": dir_text, + "filename": "events.jsonl", + "mode": "overwrite" + }, + "atif": { + "enabled": true, + "agent_name": "ffi-agent", + "agent_version": "1.2.3", + "model_name": "ffi-model", + "tool_definitions": [{"name": "search"}], + "extra": {"binding": "ffi"}, + "output_directory": dir_text, + "filename_template": "trajectory-{session_id}.json" + } + } + } + ] + }) + .to_string(), + ); + + unsafe { + assert_eq!( + take_string(nemo_flow_observability_plugin_kind()).unwrap(), + "observability" + ); + let mut default_config_json = ptr::null_mut(); + assert_eq!( + nemo_flow_observability_default_config_json(&mut default_config_json), + NemoFlowStatus::Ok + ); + assert_eq!(returned_json(default_config_json)["version"], json!(1)); + let mut component_json = ptr::null_mut(); + assert_eq!( + nemo_flow_observability_component_spec_json(ptr::null(), true, &mut component_json), + NemoFlowStatus::Ok + ); + let component = returned_json(component_json); + assert_eq!(component["kind"], "observability"); + assert_eq!(component["enabled"], true); + + let mut report_json = ptr::null_mut(); + assert_eq!( + nemo_flow_validate_plugin_config(config.as_ptr(), &mut report_json), + NemoFlowStatus::Ok + ); + assert_eq!(returned_json(report_json)["diagnostics"], json!([])); + + let mut initialized_json = ptr::null_mut(); + assert_eq!( + nemo_flow_initialize_plugins(config.as_ptr(), &mut initialized_json), + NemoFlowStatus::Ok + ); + assert_eq!(returned_json(initialized_json)["diagnostics"], json!([])); + + let stack = fresh_scope_stack(); + let scope_name = cstring("ffi-observability-agent"); + let input = cstring(r#"{"agent":true}"#); + let mut scope = ptr::null_mut(); + assert_eq!( + nemo_flow_push_scope( + scope_name.as_ptr(), + NemoFlowScopeType::Agent, + ptr::null(), + 0, + ptr::null(), + ptr::null(), + input.as_ptr(), + &mut scope, + ), + NemoFlowStatus::Ok + ); + let scope_uuid = take_string(nemo_flow_scope_handle_uuid(scope)).unwrap(); + + let mark_name = cstring("ffi-observability-mark"); + let mark_data = cstring(r#"{"step":1}"#); + assert_eq!( + nemo_flow_event(mark_name.as_ptr(), scope, mark_data.as_ptr(), ptr::null()), + NemoFlowStatus::Ok + ); + assert_eq!(nemo_flow_pop_scope(scope, ptr::null()), NemoFlowStatus::Ok); + nemo_flow_scope_handle_free(scope); + nemo_flow_scope_stack_free(stack); + assert_eq!(nemo_flow_clear_plugin_configuration(), NemoFlowStatus::Ok); + + let jsonl = std::fs::read_to_string(dir.join("events.jsonl")).unwrap(); + assert_eq!(jsonl.trim().lines().count(), 3); + + let trajectory_path = dir.join(format!("trajectory-{scope_uuid}.json")); + let trajectory: Json = + serde_json::from_str(&std::fs::read_to_string(trajectory_path).unwrap()).unwrap(); + assert_eq!(trajectory["agent"]["name"], "ffi-agent"); + assert_eq!(trajectory["agent"]["version"], "1.2.3"); + assert_eq!(trajectory["agent"]["model_name"], "ffi-model"); + assert!( + trajectory["extra"] + .to_string() + .contains("ffi-observability-agent") + ); + } +} + +#[test] +fn test_ffi_observability_plugin_atif_splits_multiple_top_level_agents() { + let _guard = TEST_MUTEX.lock().unwrap(); + reset_globals(); + let _ = nemo_flow_clear_plugin_configuration(); + let dir = std::env::temp_dir().join(unique_name("ffi_observability_plugin_multi_agent")); + std::fs::create_dir_all(&dir).unwrap(); + let dir_text = dir.to_string_lossy().into_owned(); + + let config = cstring( + &json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atif": { + "enabled": true, + "output_directory": dir_text, + "filename_template": "trajectory-{session_id}.json" + } + } + } + ] + }) + .to_string(), + ); + + unsafe { + let mut initialized_json = ptr::null_mut(); + assert_eq!( + nemo_flow_initialize_plugins(config.as_ptr(), &mut initialized_json), + NemoFlowStatus::Ok + ); + assert_eq!(returned_json(initialized_json)["diagnostics"], json!([])); + + let stack = fresh_scope_stack(); + + let first_name = cstring("ffi-first-agent"); + let first_input = cstring(r#"{"agent":"first"}"#); + let mut first = ptr::null_mut(); + assert_eq!( + nemo_flow_push_scope( + first_name.as_ptr(), + NemoFlowScopeType::Agent, + ptr::null(), + 0, + ptr::null(), + ptr::null(), + first_input.as_ptr(), + &mut first, + ), + NemoFlowStatus::Ok + ); + let first_uuid = take_string(nemo_flow_scope_handle_uuid(first)).unwrap(); + + let first_mark = cstring("ffi-first-mark"); + let first_mark_data = cstring(r#"{"agent":"first"}"#); + assert_eq!( + nemo_flow_event( + first_mark.as_ptr(), + first, + first_mark_data.as_ptr(), + ptr::null() + ), + NemoFlowStatus::Ok + ); + + let nested_name = cstring("ffi-nested-agent"); + let nested_input = cstring(r#"{"agent":"nested"}"#); + let mut nested = ptr::null_mut(); + assert_eq!( + nemo_flow_push_scope( + nested_name.as_ptr(), + NemoFlowScopeType::Agent, + ptr::null(), + 0, + ptr::null(), + ptr::null(), + nested_input.as_ptr(), + &mut nested, + ), + NemoFlowStatus::Ok + ); + let nested_mark = cstring("ffi-nested-mark"); + let nested_mark_data = cstring(r#"{"agent":"nested"}"#); + assert_eq!( + nemo_flow_event( + nested_mark.as_ptr(), + nested, + nested_mark_data.as_ptr(), + ptr::null() + ), + NemoFlowStatus::Ok + ); + assert_eq!(nemo_flow_pop_scope(nested, ptr::null()), NemoFlowStatus::Ok); + nemo_flow_scope_handle_free(nested); + assert_eq!(nemo_flow_pop_scope(first, ptr::null()), NemoFlowStatus::Ok); + nemo_flow_scope_handle_free(first); + + let second_name = cstring("ffi-second-agent"); + let second_input = cstring(r#"{"agent":"second"}"#); + let mut second = ptr::null_mut(); + assert_eq!( + nemo_flow_push_scope( + second_name.as_ptr(), + NemoFlowScopeType::Agent, + ptr::null(), + 0, + ptr::null(), + ptr::null(), + second_input.as_ptr(), + &mut second, + ), + NemoFlowStatus::Ok + ); + let second_uuid = take_string(nemo_flow_scope_handle_uuid(second)).unwrap(); + let second_mark = cstring("ffi-second-mark"); + let second_mark_data = cstring(r#"{"agent":"second"}"#); + assert_eq!( + nemo_flow_event( + second_mark.as_ptr(), + second, + second_mark_data.as_ptr(), + ptr::null() + ), + NemoFlowStatus::Ok + ); + assert_eq!(nemo_flow_pop_scope(second, ptr::null()), NemoFlowStatus::Ok); + nemo_flow_scope_handle_free(second); + nemo_flow_scope_stack_free(stack); + assert_eq!(nemo_flow_clear_plugin_configuration(), NemoFlowStatus::Ok); + + let files = std::fs::read_dir(&dir) + .unwrap() + .filter(|entry| { + entry + .as_ref() + .ok() + .and_then(|entry| entry.file_name().into_string().ok()) + .is_some_and(|name| name.starts_with("trajectory-")) + }) + .count(); + assert_eq!(files, 2); + + let first_payload = + std::fs::read_to_string(dir.join(format!("trajectory-{first_uuid}.json"))).unwrap(); + let second_payload = + std::fs::read_to_string(dir.join(format!("trajectory-{second_uuid}.json"))).unwrap(); + assert!(first_payload.contains("ffi-first-agent")); + assert!(first_payload.contains("ffi-nested-agent")); + assert!(!first_payload.contains("ffi-second-agent")); + assert!(second_payload.contains("ffi-second-agent")); + assert!(!second_payload.contains("ffi-first-agent")); + assert!(!second_payload.contains("ffi-nested-agent")); + } +} + #[test] fn test_ffi_plugin_top_level_null_and_invalid_paths() { let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/node/README.md b/crates/node/README.md index 5f3896320..1f53cb1f1 100644 --- a/crates/node/README.md +++ b/crates/node/README.md @@ -32,7 +32,7 @@ should install it from npm rather than depend on the Rust crate directly. - 📡 **Emit one lifecycle stream**: Send runtime events to in-process subscribers, ATIF, OpenTelemetry, or OpenInference workflows. - 🧩 **Use package entry points by need**: Import the main runtime surface plus - typed, plugin, and adaptive helpers from npm. + typed, plugin, adaptive, and observability helpers from npm. ## What You Get @@ -44,7 +44,8 @@ should install it from npm rather than depend on the Rust crate directly. - ✅ **Observability exporters**: Subscriber and exporter support for common runtime telemetry flows. - ✅ **Additional entry points**: `nemo-flow-node/typed`, - `nemo-flow-node/plugin`, and `nemo-flow-node/adaptive`. + `nemo-flow-node/plugin`, `nemo-flow-node/adaptive`, and + `nemo-flow-node/observability`. ## Installation @@ -86,8 +87,8 @@ main().catch((error) => { ``` The main runtime API is exported from `nemo-flow-node`. Additional entry points -are available at `nemo-flow-node/typed`, `nemo-flow-node/plugin`, and -`nemo-flow-node/adaptive`. +are available at `nemo-flow-node/typed`, `nemo-flow-node/plugin`, +`nemo-flow-node/adaptive`, and `nemo-flow-node/observability`. ## Documentation diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts new file mode 100644 index 000000000..811bce129 --- /dev/null +++ b/crates/node/observability.d.ts @@ -0,0 +1,71 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import type { Json } from './index'; +import type { ConfigPolicy, ConfigDiagnostic, ConfigReport } from './plugin'; + +export { ConfigPolicy, ConfigDiagnostic, ConfigReport }; + +export interface AtofConfig { + enabled?: boolean; + output_directory?: string; + filename?: string; + mode?: 'append' | 'overwrite' | string; +} + +export interface AtifConfig { + enabled?: boolean; + agent_name?: string; + agent_version?: string; + model_name?: string; + tool_definitions?: Record[]; + extra?: Record; + output_directory?: string; + filename_template?: string; +} + +export interface OtlpConfig { + enabled?: boolean; + transport?: 'http_binary' | 'grpc' | string; + endpoint?: string; + headers?: Record; + resource_attributes?: Record; + service_name?: string; + service_namespace?: string; + service_version?: string; + instrumentation_scope?: string; + timeout_millis?: number; +} + +export interface Config { + version?: number; + atof?: AtofConfig; + atif?: AtifConfig; + opentelemetry?: OtlpConfig; + openinference?: OtlpConfig; + policy?: ConfigPolicy; +} + +export interface ComponentSpec { + kind: 'observability'; + enabled?: boolean; + config: Config; +} + +/** Top-level plugin kind used by the built-in observability component. */ +export declare const OBSERVABILITY_PLUGIN_KIND: 'observability'; +/** Create a default observability component config. */ +export declare function defaultConfig(): Config; +/** Create filesystem-backed ATOF JSONL settings with defaults applied. */ +export declare function atofConfig(config?: AtofConfig): AtofConfig; +/** Create per-agent ATIF trajectory settings with defaults applied. */ +export declare function atifConfig(config?: AtifConfig): AtifConfig; +/** Create OTLP exporter settings for OpenTelemetry or OpenInference. */ +export declare function otlpConfig(config?: OtlpConfig): OtlpConfig; +/** Wrap observability config as a top-level plugin component. */ +export declare function ComponentSpec( + config: Config, + options?: { + enabled?: boolean; + }, +): ComponentSpec; diff --git a/crates/node/observability.js b/crates/node/observability.js new file mode 100644 index 000000000..06a53811e --- /dev/null +++ b/crates/node/observability.js @@ -0,0 +1,89 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +'use strict'; + +const plugin = require('./plugin.js'); + +const OBSERVABILITY_PLUGIN_KIND = 'observability'; + +/** + * Create a default observability component config. + * + * @returns {object} The minimal observability config with schema version 1. + */ +function defaultConfig() { + return { + version: 1, + }; +} + +/** + * Create filesystem-backed ATOF JSONL settings with defaults applied. + * + * @param {object} [config={}] - Partial ATOF settings to override. + * @returns {object} A normalized ATOF config object. + */ +function atofConfig(config = {}) { + return { + enabled: false, + mode: 'append', + ...config, + }; +} + +/** + * Create per-agent ATIF trajectory settings with defaults applied. + * + * @param {object} [config={}] - Partial ATIF settings to override. + * @returns {object} A normalized ATIF config object. + */ +function atifConfig(config = {}) { + return { + enabled: false, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + ...config, + }; +} + +/** + * Create OTLP exporter settings for OpenTelemetry or OpenInference. + * + * @param {object} [config={}] - Partial OTLP settings to override. + * @returns {object} A normalized OTLP config object. + */ +function otlpConfig(config = {}) { + return { + enabled: false, + transport: 'http_binary', + headers: {}, + resource_attributes: {}, + service_name: 'nemo-flow', + timeout_millis: 3000, + ...config, + }; +} + +/** + * Wrap observability config as a top-level plugin component. + * + * @param {object} config - Observability component configuration document. + * @param {{ enabled?: boolean }} [options={}] - Optional component-level flags. + * @returns {object} A plugin component spec for the observability plugin. + */ +function ComponentSpec(config, { enabled = true } = {}) { + return plugin.ComponentSpec(OBSERVABILITY_PLUGIN_KIND, config, { + enabled, + }); +} + +module.exports = { + OBSERVABILITY_PLUGIN_KIND, + defaultConfig, + atofConfig, + atifConfig, + otlpConfig, + ComponentSpec, +}; diff --git a/crates/node/package.json b/crates/node/package.json index 49efafa12..9d71017c8 100644 --- a/crates/node/package.json +++ b/crates/node/package.json @@ -40,6 +40,10 @@ "./adaptive": { "types": "./adaptive.d.ts", "default": "./adaptive.js" + }, + "./observability": { + "types": "./observability.d.ts", + "default": "./observability.js" } }, "engines": { diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs new file mode 100644 index 000000000..7c6625fbd --- /dev/null +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -0,0 +1,178 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdtempSync, readdirSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +const require = createRequire(import.meta.url); +const plugin = require('../plugin.js'); +const observability = require('../observability.js'); +const { ScopeType, pushScope, popScope, event } = require('../index.js'); + +function tempDir(prefix) { + return mkdtempSync(join(tmpdir(), `nemo-flow-${prefix}-`)); +} + +describe('observability plugin helpers', () => { + it('builds defaults and plugin component shape', () => { + assert.deepEqual(observability.defaultConfig(), { version: 1 }); + assert.deepEqual(observability.atofConfig(), { enabled: false, mode: 'append' }); + assert.deepEqual(observability.atifConfig(), { + enabled: false, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + }); + assert.deepEqual(observability.otlpConfig(), { + enabled: false, + transport: 'http_binary', + headers: {}, + resource_attributes: {}, + service_name: 'nemo-flow', + timeout_millis: 3000, + }); + + const component = observability.ComponentSpec({ version: 1, atof: observability.atofConfig() }); + assert.equal(component.kind, observability.OBSERVABILITY_PLUGIN_KIND); + assert.equal(component.enabled, true); + }); + + it('lists builtin observability kind and validates bad values', () => { + assert.equal(plugin.listKinds().includes(observability.OBSERVABILITY_PLUGIN_KIND), true); + const report = plugin.validate({ + version: 1, + components: [ + observability.ComponentSpec({ + version: 1, + atof: observability.atofConfig({ mode: 'bad' }), + atif: observability.atifConfig({ filename_template: 'missing-placeholder.json' }), + }), + ], + }); + assert.deepEqual( + report.diagnostics.map((diagnostic) => diagnostic.field).sort(), + ['filename_template', 'mode'], + ); + }); + + it('activates ATOF and ATIF file sinks', async () => { + const outputDirectory = tempDir('node-observability-plugin'); + const config = { + version: 1, + atof: observability.atofConfig({ + enabled: true, + output_directory: outputDirectory, + filename: 'events.jsonl', + mode: 'overwrite', + }), + atif: observability.atifConfig({ + enabled: true, + agent_name: 'node-agent', + agent_version: '1.2.3', + model_name: 'node-model', + tool_definitions: [{ name: 'search' }], + extra: { binding: 'node' }, + output_directory: outputDirectory, + filename_template: 'trajectory-{session_id}.json', + }), + }; + + await plugin.initialize({ + version: 1, + components: [observability.ComponentSpec(config)], + }); + let scope = null; + try { + scope = pushScope('node-observability-agent', ScopeType.Agent, null, null, null, null, { agent: true }); + event('node-mark', scope, { step: 1 }, null); + popScope(scope, { done: true }); + scope = null; + } finally { + plugin.clear(); + if (scope) { + popScope(scope, { done: true }); + } + } + + const records = readFileSync(join(outputDirectory, 'events.jsonl'), 'utf8').trim().split('\n').map(JSON.parse); + assert.deepEqual(records.map((record) => record.kind), ['scope', 'mark', 'scope']); + + const trajectory = JSON.parse(readFileSync(join(outputDirectory, `trajectory-${records[0].uuid}.json`), 'utf8')); + assert.equal(trajectory.agent.name, 'node-agent'); + assert.equal(trajectory.agent.version, '1.2.3'); + assert.equal(trajectory.agent.model_name, 'node-model'); + assert.equal(trajectory.agent.tool_definitions[0].name, 'search'); + assert.equal(trajectory.agent.extra.binding, 'node'); + assert.match(JSON.stringify(trajectory.extra), /node-observability-agent/); + }); + + it('splits ATIF files for multiple top-level agent scopes', async () => { + const outputDirectory = tempDir('node-observability-plugin-multi-agent'); + const config = { + version: 1, + atif: observability.atifConfig({ + enabled: true, + output_directory: outputDirectory, + filename_template: 'trajectory-{session_id}.json', + }), + }; + + await plugin.initialize({ + version: 1, + components: [observability.ComponentSpec(config)], + }); + + let first = null; + let nested = null; + let second = null; + let firstUuid = null; + let secondUuid = null; + try { + first = pushScope('node-first-agent', ScopeType.Agent, null, null, null, null, { agent: 'first' }); + firstUuid = first.uuid; + event('node-first-mark', first, { agent: 'first' }, null); + nested = pushScope('node-nested-agent', ScopeType.Agent, null, null, null, null, { agent: 'nested' }); + event('node-nested-mark', nested, { agent: 'nested' }, null); + popScope(nested, { done: true }); + nested = null; + popScope(first, { done: true }); + first = null; + + second = pushScope('node-second-agent', ScopeType.Agent, null, null, null, null, { agent: 'second' }); + secondUuid = second.uuid; + event('node-second-mark', second, { agent: 'second' }, null); + popScope(second, { done: true }); + second = null; + } finally { + plugin.clear(); + if (nested) { + popScope(nested, { done: true }); + } + if (first) { + popScope(first, { done: true }); + } + if (second) { + popScope(second, { done: true }); + } + } + + const files = readdirSync(outputDirectory).filter((name) => name.startsWith('trajectory-')); + assert.equal(files.length, 2); + + const firstTrajectory = JSON.parse(readFileSync(join(outputDirectory, `trajectory-${firstUuid}.json`), 'utf8')); + const secondTrajectory = JSON.parse(readFileSync(join(outputDirectory, `trajectory-${secondUuid}.json`), 'utf8')); + const firstPayload = JSON.stringify(firstTrajectory.extra); + const secondPayload = JSON.stringify(secondTrajectory.extra); + + assert.match(firstPayload, /node-first-agent/); + assert.match(firstPayload, /node-nested-agent/); + assert.doesNotMatch(firstPayload, /node-second-agent/); + assert.match(secondPayload, /node-second-agent/); + assert.doesNotMatch(secondPayload, /node-first-agent/); + assert.doesNotMatch(secondPayload, /node-nested-agent/); + }); +}); diff --git a/crates/wasm/scripts/prepare_pkg.mjs b/crates/wasm/scripts/prepare_pkg.mjs index a3124aff2..9ceed9086 100644 --- a/crates/wasm/scripts/prepare_pkg.mjs +++ b/crates/wasm/scripts/prepare_pkg.mjs @@ -12,8 +12,8 @@ const nodeJsWrapperDir = path.join(crateDir, 'wrappers', 'nodejs'); const pkgDir = process.argv[2] ? path.resolve(process.argv[2]) : path.join(crateDir, 'pkg'); const rootJsFiles = ['index.js']; -const jsWrapperFiles = ['typed.js', 'plugin.js', 'adaptive.js']; -const typeWrapperFiles = ['typed.d.ts', 'plugin.d.ts', 'adaptive.d.ts']; +const jsWrapperFiles = ['typed.js', 'plugin.js', 'adaptive.js', 'observability.js']; +const typeWrapperFiles = ['typed.d.ts', 'plugin.d.ts', 'adaptive.d.ts', 'observability.d.ts']; const wrapperFiles = [...rootJsFiles, ...jsWrapperFiles, ...typeWrapperFiles]; const packageMetadata = { description: 'WebAssembly bindings for the NeMo Flow agent runtime.', @@ -99,6 +99,10 @@ function updatePackageManifest(manifest) { types: './adaptive.d.ts', default: './adaptive.js', }, + './observability': { + types: './observability.d.ts', + default: './observability.js', + }, './typed.js': { types: './typed.d.ts', default: './typed.js', @@ -111,6 +115,10 @@ function updatePackageManifest(manifest) { types: './adaptive.d.ts', default: './adaptive.js', }, + './observability.js': { + types: './observability.d.ts', + default: './observability.js', + }, }; fs.writeFileSync(manifestPath, `${JSON.stringify(manifest, null, 2)}\n`); diff --git a/crates/wasm/tests-js/observability_tests.mjs b/crates/wasm/tests-js/observability_tests.mjs new file mode 100644 index 000000000..94a9351c9 --- /dev/null +++ b/crates/wasm/tests-js/observability_tests.mjs @@ -0,0 +1,72 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import * as observability from '../pkg/observability.js'; +import * as plugin from '../pkg/plugin.js'; + +test('WebAssembly observability wrappers expose helper defaults', () => { + assert.deepEqual(observability.defaultConfig(), { + version: 1, + }); + assert.deepEqual(observability.atofConfig(), { + enabled: false, + mode: 'append', + }); + assert.deepEqual(observability.atifConfig(), { + enabled: false, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + }); + assert.deepEqual(observability.otlpConfig(), { + enabled: false, + transport: 'http_binary', + headers: {}, + resource_attributes: {}, + service_name: 'nemo-flow', + timeout_millis: 3000, + }); +}); + +test('WebAssembly observability wrappers build component specs and validate file sinks', () => { + assert.equal(plugin.listKinds().includes(observability.OBSERVABILITY_PLUGIN_KIND), true); + + const component = observability.ComponentSpec({ + version: 1, + atof: observability.atofConfig({ enabled: true }), + atif: observability.atifConfig({ enabled: true }), + }); + + assert.deepEqual(component, { + kind: 'observability', + enabled: true, + config: { + version: 1, + atof: { + enabled: true, + mode: 'append', + }, + atif: { + enabled: true, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + }, + }, + }); + + const report = plugin.validate({ + version: 1, + components: [component], + }); + assert.deepEqual( + report.diagnostics.map((diagnostic) => [diagnostic.component, diagnostic.field]).sort(), + [ + ['atif', 'enabled'], + ['atof', 'enabled'], + ], + ); +}); diff --git a/crates/wasm/wrappers/esm/observability.d.ts b/crates/wasm/wrappers/esm/observability.d.ts new file mode 100644 index 000000000..6f45069c2 --- /dev/null +++ b/crates/wasm/wrappers/esm/observability.d.ts @@ -0,0 +1,71 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import type { ConfigPolicy, ConfigDiagnostic, ConfigReport } from './plugin.js'; +import type { JsonObject } from './typed.js'; + +export { ConfigPolicy, ConfigDiagnostic, ConfigReport }; + +export interface AtofConfig { + enabled?: boolean; + output_directory?: string; + filename?: string; + mode?: 'append' | 'overwrite' | string; +} + +export interface AtifConfig { + enabled?: boolean; + agent_name?: string; + agent_version?: string; + model_name?: string; + tool_definitions?: JsonObject[]; + extra?: JsonObject; + output_directory?: string; + filename_template?: string; +} + +export interface OtlpConfig { + enabled?: boolean; + transport?: 'http_binary' | 'grpc' | string; + endpoint?: string; + headers?: Record; + resource_attributes?: Record; + service_name?: string; + service_namespace?: string; + service_version?: string; + instrumentation_scope?: string; + timeout_millis?: number; +} + +export interface Config { + version?: number; + atof?: AtofConfig; + atif?: AtifConfig; + opentelemetry?: OtlpConfig; + openinference?: OtlpConfig; + policy?: ConfigPolicy; +} + +export interface ComponentSpec { + kind: 'observability'; + enabled?: boolean; + config: Config; +} + +/** Top-level plugin kind used by the built-in observability component. */ +export declare const OBSERVABILITY_PLUGIN_KIND: 'observability'; +/** Create a default observability component config. */ +export declare function defaultConfig(): Config; +/** Create filesystem-backed ATOF JSONL settings with defaults applied. */ +export declare function atofConfig(config?: AtofConfig): AtofConfig; +/** Create per-agent ATIF trajectory settings with defaults applied. */ +export declare function atifConfig(config?: AtifConfig): AtifConfig; +/** Create OTLP exporter settings for OpenTelemetry or OpenInference. */ +export declare function otlpConfig(config?: OtlpConfig): OtlpConfig; +/** Wrap observability config as a top-level plugin component. */ +export declare function ComponentSpec( + config: Config, + options?: { + enabled?: boolean; + }, +): import('./plugin.js').ComponentSpec; diff --git a/crates/wasm/wrappers/esm/observability.js b/crates/wasm/wrappers/esm/observability.js new file mode 100644 index 000000000..e1b56cd4c --- /dev/null +++ b/crates/wasm/wrappers/esm/observability.js @@ -0,0 +1,78 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import * as plugin from './plugin.js'; + +export const OBSERVABILITY_PLUGIN_KIND = 'observability'; + +/** + * Create a default observability component config. + * + * @returns {object} The minimal observability config with schema version 1. + */ +export function defaultConfig() { + return { + version: 1, + }; +} + +/** + * Create filesystem-backed ATOF JSONL settings with defaults applied. + * + * @param {object} [config={}] - Partial ATOF settings to override. + * @returns {object} A normalized ATOF config object. + */ +export function atofConfig(config = {}) { + return { + enabled: false, + mode: 'append', + ...config, + }; +} + +/** + * Create per-agent ATIF trajectory settings with defaults applied. + * + * @param {object} [config={}] - Partial ATIF settings to override. + * @returns {object} A normalized ATIF config object. + */ +export function atifConfig(config = {}) { + return { + enabled: false, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + ...config, + }; +} + +/** + * Create OTLP exporter settings for OpenTelemetry or OpenInference. + * + * @param {object} [config={}] - Partial OTLP settings to override. + * @returns {object} A normalized OTLP config object. + */ +export function otlpConfig(config = {}) { + return { + enabled: false, + transport: 'http_binary', + headers: {}, + resource_attributes: {}, + service_name: 'nemo-flow', + timeout_millis: 3000, + ...config, + }; +} + +/** + * Wrap observability config as a top-level plugin component. + * + * @param {object} config - Observability component configuration document. + * @param {{ enabled?: boolean }} [options={}] - Optional component-level flags. + * @returns {object} A plugin component spec for the observability plugin. + */ +export function ComponentSpec(config, { enabled = true } = {}) { + return plugin.ComponentSpec(OBSERVABILITY_PLUGIN_KIND, config, { + enabled, + }); +} diff --git a/crates/wasm/wrappers/nodejs/observability.js b/crates/wasm/wrappers/nodejs/observability.js new file mode 100644 index 000000000..06a53811e --- /dev/null +++ b/crates/wasm/wrappers/nodejs/observability.js @@ -0,0 +1,89 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +'use strict'; + +const plugin = require('./plugin.js'); + +const OBSERVABILITY_PLUGIN_KIND = 'observability'; + +/** + * Create a default observability component config. + * + * @returns {object} The minimal observability config with schema version 1. + */ +function defaultConfig() { + return { + version: 1, + }; +} + +/** + * Create filesystem-backed ATOF JSONL settings with defaults applied. + * + * @param {object} [config={}] - Partial ATOF settings to override. + * @returns {object} A normalized ATOF config object. + */ +function atofConfig(config = {}) { + return { + enabled: false, + mode: 'append', + ...config, + }; +} + +/** + * Create per-agent ATIF trajectory settings with defaults applied. + * + * @param {object} [config={}] - Partial ATIF settings to override. + * @returns {object} A normalized ATIF config object. + */ +function atifConfig(config = {}) { + return { + enabled: false, + agent_name: 'NeMo Flow', + model_name: 'unknown', + filename_template: 'nemo-flow-atif-{session_id}.json', + ...config, + }; +} + +/** + * Create OTLP exporter settings for OpenTelemetry or OpenInference. + * + * @param {object} [config={}] - Partial OTLP settings to override. + * @returns {object} A normalized OTLP config object. + */ +function otlpConfig(config = {}) { + return { + enabled: false, + transport: 'http_binary', + headers: {}, + resource_attributes: {}, + service_name: 'nemo-flow', + timeout_millis: 3000, + ...config, + }; +} + +/** + * Wrap observability config as a top-level plugin component. + * + * @param {object} config - Observability component configuration document. + * @param {{ enabled?: boolean }} [options={}] - Optional component-level flags. + * @returns {object} A plugin component spec for the observability plugin. + */ +function ComponentSpec(config, { enabled = true } = {}) { + return plugin.ComponentSpec(OBSERVABILITY_PLUGIN_KIND, config, { + enabled, + }); +} + +module.exports = { + OBSERVABILITY_PLUGIN_KIND, + defaultConfig, + atofConfig, + atifConfig, + otlpConfig, + ComponentSpec, +}; diff --git a/docs/about/concepts/plugins.md b/docs/about/concepts/plugins.md index aa3203e4a..a0c002a7f 100644 --- a/docs/about/concepts/plugins.md +++ b/docs/about/concepts/plugins.md @@ -60,7 +60,7 @@ what did not. flowchart TB subgraph Config[Plugin Configuration] Document[Plugin Config
version + components + policy] - Components[Components
custom or adaptive] + Components[Components
custom, adaptive, or observability] Document --> Components end @@ -126,7 +126,13 @@ that should activate once for the running process rather than once per request. Scope-local behavior still matters after plugin installation, but the plugin system itself is a global activation layer. -## Built-In Plugin Example: Adaptive +## Built-In Plugin Examples + +Core plugin APIs register built-in components before lookup, validation, and +initialization. Applications can still register custom plugins, but first-party +components are available by kind without an explicit registration call. + +### Adaptive Adaptive is implemented as a built-in plugin component. It is not a separate runtime model. It uses the same plugin system as custom components. @@ -140,6 +146,15 @@ through the same component lifecycle as other plugins: Detailed adaptive configuration belongs in [Configure Adaptive Optimization](../../use-adaptive-optimization/configure.md), [Adaptive Code Examples](../../use-adaptive-optimization/code-examples.md), and [Advanced Guide: Configure Adaptive Components](../../use-adaptive-optimization/adaptive-components.md). +### Observability + +The core crate ships a built-in `observability` plugin component for ATOF, +ATIF, OpenTelemetry, and OpenInference exporters. Each exporter section is +disabled unless its section sets `enabled: true`, and subscriber names are +inferred from the plugin namespace instead of exposed in public config. + +Detailed observability plugin configuration belongs in [Configure the Observability Plugin](../../export-observability-data/observability-plugin.md). + ## Practical Guidance Use these practices when applying the concept in application or integration code. diff --git a/docs/about/concepts/subscribers.md b/docs/about/concepts/subscribers.md index 1fa33ca62..5bb48d8be 100644 --- a/docs/about/concepts/subscribers.md +++ b/docs/about/concepts/subscribers.md @@ -89,9 +89,7 @@ offline analysis, replay, or debugging. ### ATOF JSONL Exporter The ATOF JSONL exporter writes the canonical event stream to a native -filesystem path as one raw ATOF event per line. It is available for native -Rust, Python, Node.js, Go, and C FFI use. It is not exposed in WebAssembly -because arbitrary filesystem writes are not portable there. +filesystem path as one raw ATOF event per line. ### OpenTelemetry Subscriber @@ -106,6 +104,10 @@ OpenInference semantics for model-centric observability. Detailed setup, configuration, and API shape for these subscribers belongs in [Export Observability Data](../../export-observability-data/basic-guide.md) and [Observability Code Examples](../../export-observability-data/code-examples.md). +For configuration-driven setup, use the built-in +[`observability` plugin](../../export-observability-data/observability-plugin.md) +to install ATOF, ATIF, OpenTelemetry, and OpenInference subscribers from one +plugin component. ## Practical Guidance diff --git a/docs/export-observability-data/about.md b/docs/export-observability-data/about.md index ad8d49e20..4822b8f4f 100644 --- a/docs/export-observability-data/about.md +++ b/docs/export-observability-data/about.md @@ -14,6 +14,10 @@ event stream. Subscribers consume that stream inside the process, and exporter-oriented subscribers write raw ATOF JSONL or translate it into formats such as ATIF, OpenTelemetry, and OpenInference. +For standard exporters, use the built-in `observability` plugin to configure +ATOF, per-agent ATIF, OpenTelemetry, and OpenInference from one plugin document. +Each section is disabled unless it explicitly sets `enabled: true`. + Use these guides to confirm what ran, where it belonged, which model or tool was involved, and what sanitized payload was observed across Rust, Python, and Node.js. @@ -35,6 +39,7 @@ If you have not instrumented any scopes, tools, or LLM calls yet, start with [In The following guides describe available tutorials and exporters: - [Basic Guide: Register a Subscriber](basic-guide.md) shows a simple subscriber lifecycle and validation workflow. +- [Basic Guide: Configure the Observability Plugin](observability-plugin.md) shows the built-in exporter plugin and its config schema. - [Code Examples](code-examples.md#atof-jsonl-export) shows how to write raw ATOF events as JSONL. - [Advanced Guide: Export OpenTelemetry Data](opentelemetry.md) shows how to export generic OTLP spans. - [Advanced Guide: Export OpenInference Data](advanced-guide.md) shows how to configure and operate the OpenInference exporter. @@ -50,7 +55,6 @@ request, use stable service identity fields, keep credentials outside source code, flush during graceful shutdown, and filter by `root_uuid` when analyzing concurrent agent runs. -The filesystem-backed ATOF JSONL exporter is available on native Rust, Python, -Node.js, Go, and C FFI surfaces. It is not exposed in WebAssembly because -arbitrary filesystem writes are not portable across browser and hosted WASM -environments. +The filesystem-backed ATOF JSONL exporter and ATIF plugin file sink require +native filesystem access. Use explicit plugin teardown or exporter shutdown to +flush files during graceful shutdown. diff --git a/docs/export-observability-data/advanced-guide.md b/docs/export-observability-data/advanced-guide.md index 264f114a3..56404d4cb 100644 --- a/docs/export-observability-data/advanced-guide.md +++ b/docs/export-observability-data/advanced-guide.md @@ -26,8 +26,6 @@ Complete these steps: 3. Start an OTLP/HTTP collector or tracing backend. 4. Redact sensitive event payloads with sanitize guardrails before production export. -Use `http_binary` transport for the current OpenInference path. The configuration surfaces expose `grpc`, but the current core OpenInference subscriber returns an unsupported-transport error for gRPC. - ## Configure the Exporter Set these fields first: diff --git a/docs/export-observability-data/atif.md b/docs/export-observability-data/atif.md index b0a7e24ca..1dd640bcc 100644 --- a/docs/export-observability-data/atif.md +++ b/docs/export-observability-data/atif.md @@ -13,6 +13,11 @@ You will create an ATIF exporter, register it as a subscriber, run instrumented Unlike OpenTelemetry and OpenInference export, ATIF export is in-process and buffered. The exporter collects events until you call `export`, `export_json`, or `clear`. +For automatic per-agent file export, use the built-in +[Observability plugin](observability-plugin.md). Its `atif` section creates one +scope-local exporter for each top-level agent scope and writes each trajectory +when that agent scope ends. + ## Before You Start Complete these steps: @@ -130,6 +135,7 @@ Choose one of these patterns: | One exporter per run | Each agent run should produce one trajectory. | | Long-lived exporter with `clear` | A test or local tool exports multiple trajectories in one process. | | Filtered analysis by root scope | Concurrent runs share one process but can be separated later. | +| Observability plugin ATIF section | Each direct child agent scope should write its own trajectory file automatically. | For production services, prefer bounded collection windows. Long-lived unbounded exporters can accumulate more event data than expected. diff --git a/docs/export-observability-data/basic-guide.md b/docs/export-observability-data/basic-guide.md index 8c4ae6d2c..9278ad988 100644 --- a/docs/export-observability-data/basic-guide.md +++ b/docs/export-observability-data/basic-guide.md @@ -136,6 +136,7 @@ The table below compares subscriber and exporter options for common observabilit | ATIF exporter | Collect events and export ATIF v1.6 trajectories. | | OpenTelemetry subscriber | Export lifecycle events as OTLP spans. | | OpenInference subscriber | Export lifecycle events as OTLP spans with OpenInference-oriented semantics. | +| Observability plugin | Configure ATOF, per-agent ATIF, OpenTelemetry, and OpenInference from one built-in plugin component. | ## Validate the Subscriber @@ -164,5 +165,6 @@ Use these links to continue from this workflow into the next related task. - Export generic OTLP spans with [Advanced Guide: Export OpenTelemetry Data](opentelemetry.md). - Export traces with [Advanced Guide: Export OpenInference Data](advanced-guide.md). - Export trajectory artifacts with [Advanced Guide: Export ATIF](atif.md). +- Configure standard exporters with [Basic Guide: Configure the Observability Plugin](observability-plugin.md). - Use [Code Examples](code-examples.md) for event shape, scope-local subscribers, ATIF, and OpenTelemetry snippets. - Add redaction with [Advanced Guide: Add Middleware](../instrument-applications/advanced-guide.md). diff --git a/docs/export-observability-data/code-examples.md b/docs/export-observability-data/code-examples.md index 1d0ddf125..99872ba3b 100644 --- a/docs/export-observability-data/code-examples.md +++ b/docs/export-observability-data/code-examples.md @@ -167,12 +167,121 @@ scope_register_subscriber(&scope.uuid, "scoped-logger", Arc::new(|event| { :::: +## Observability Plugin Configuration + +Use the built-in plugin when one config document should own standard exporter +setup and teardown. Each exporter section stays disabled unless it explicitly +sets `enabled: true`. + +::::{tab-set} +:sync-group: language + +:::{tab-item} Python +:sync: python + +```python +from nemo_flow import plugin +from nemo_flow.observability import ( + AtifConfig, + AtofConfig, + ComponentSpec, + ObservabilityConfig, +) + +await plugin.initialize( + plugin.PluginConfig( + components=[ + ComponentSpec( + ObservabilityConfig( + atof=AtofConfig( + enabled=True, + output_directory="logs", + filename="events.jsonl", + mode="overwrite", + ), + atif=AtifConfig( + enabled=True, + output_directory="logs", + filename_template="trajectory-{session_id}.json", + ), + ) + ) + ] + ) +) +``` +::: + +:::{tab-item} Node.js +:sync: node + +```ts +import * as plugin from 'nemo-flow-node/plugin'; +import * as observability from 'nemo-flow-node/observability'; + +await plugin.initialize({ + version: 1, + components: [ + observability.ComponentSpec({ + version: 1, + atof: observability.atofConfig({ + enabled: true, + output_directory: 'logs', + filename: 'events.jsonl', + mode: 'overwrite', + }), + atif: observability.atifConfig({ + enabled: true, + output_directory: 'logs', + filename_template: 'trajectory-{session_id}.json', + }), + }), + ], +}); +``` +::: + +:::{tab-item} Rust +:sync: rust + +```rust +use nemo_flow::observability::plugin_component::{ + AtifSectionConfig, AtofSectionConfig, ComponentSpec, ObservabilityConfig, +}; +use nemo_flow::plugin::{initialize_plugins, PluginConfig}; + +let component = ComponentSpec::new(ObservabilityConfig { + atof: Some(AtofSectionConfig { + enabled: true, + output_directory: Some("logs".into()), + filename: Some("events.jsonl".into()), + mode: "overwrite".into(), + }), + atif: Some(AtifSectionConfig { + enabled: true, + output_directory: Some("logs".into()), + filename_template: "trajectory-{session_id}.json".into(), + ..AtifSectionConfig::default() + }), + ..ObservabilityConfig::default() +}); + +initialize_plugins(PluginConfig { + version: 1, + components: vec![component.into()], + policy: Default::default(), +}) +.await?; +``` +::: + +:::: + ## ATOF JSONL Export Use the ATOF JSONL exporter when you want the raw canonical event stream on disk. The exporter writes one ATOF event JSON object per line, opens files in -append mode by default, and flushes after every event. WebAssembly does not -expose this filesystem-backed exporter. +append mode by default, and flushes after every event. ::::{tab-set} :sync-group: language @@ -243,31 +352,6 @@ exporter.shutdown()?; ``` ::: -:::{tab-item} Go -:sync: go - -```go -exporter, err := nemo_flow.NewAtofExporter(nemo_flow.AtofExporterConfig{ - OutputDirectory: "logs", - Mode: nemo_flow.AtofExporterModeAppend, - Filename: "nemo-flow-events.jsonl", -}) -if err != nil { - return err -} -defer exporter.Close() - -if err := exporter.Register("atof-jsonl"); err != nil { - return err -} - -// Run instrumented application work here. - -_ = exporter.Deregister("atof-jsonl") -return exporter.Shutdown() -``` -::: - :::: ## ATIF Export diff --git a/docs/export-observability-data/observability-plugin.md b/docs/export-observability-data/observability-plugin.md new file mode 100644 index 000000000..bf8d72529 --- /dev/null +++ b/docs/export-observability-data/observability-plugin.md @@ -0,0 +1,252 @@ + + +# Basic Guide: Configure the Observability Plugin + +Use the built-in Observability plugin when an application should install +standard exporters from one plugin configuration document instead of manually +registering each subscriber. + +The plugin kind is `observability`. It is registered by the core runtime, so +applications do not need to register a plugin implementation before validation +or initialization. + +## What It Installs + +The component accepts four optional sections: + +| Section | Runtime behavior | +|---|---| +| `atof` | Registers a global ATOF JSONL exporter for raw lifecycle events. | +| `atif` | Registers one ATIF dispatcher that writes one trajectory file for each top-level agent scope. | +| `opentelemetry` | Registers a global OpenTelemetry OTLP subscriber. | +| `openinference` | Registers a global OpenInference OTLP subscriber. | + +Every section defaults to disabled. A section is active only when it includes +`enabled: true`. + +## Top-Level Shape + +The generic plugin config wraps the observability component: + +```json +{ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { "enabled": true, "filename": "events.jsonl" } + } + } + ] +} +``` + +`subscriber_name` is not part of this config. The runtime infers subscriber +names from the plugin namespace: + +- ATOF: `atof` +- ATIF dispatcher: `atif` +- Per-agent ATIF scope subscriber: `atif-{agent_scope_uuid}` +- OpenTelemetry: `opentelemetry` +- OpenInference: `openinference` + +The active runtime names include the component namespace prefix used by the +plugin system. + +## ATOF Section + +Use ATOF when you want the raw ATOF `0.1` event stream as JSONL. + +| Field | Default | Notes | +|---|---|---| +| `enabled` | `false` | Must be `true` to write events. | +| `output_directory` | Current working directory | Directory containing the JSONL file. | +| `filename` | Timestamped `nemo-flow-events-*.jsonl` | Explicit output filename. | +| `mode` | `append` | `append` or `overwrite`. | + +## ATIF Section + +Use ATIF when you want one trajectory artifact per top-level agent run. + +| Field | Default | Notes | +|---|---|---| +| `enabled` | `false` | Must be `true` to write trajectories. | +| `agent_name` | `NeMo Flow` | Agent metadata written into the trajectory. | +| `agent_version` | NeMo Flow crate version | Agent version metadata. | +| `model_name` | `unknown` | Default model metadata when no call-level model is present. | +| `tool_definitions` | Omitted | Optional ATIF tool metadata. | +| `extra` | Omitted | Optional ATIF agent metadata. | +| `output_directory` | Current working directory | Directory containing trajectory files. | +| `filename_template` | `nemo-flow-atif-{session_id}.json` | Must contain `{session_id}`. | + +A top-level agent is a scope start event with category `agent` whose parent is +the implicit root scope. The ATIF plugin creates a separate exporter for each +direct child agent scope, records that start event, attaches a scope-local +subscriber to the agent scope, and writes the file when the agent scope ends. +If the plugin is cleared while an agent is still open, teardown flushes the +partial trajectory. + +Nested agent scopes under a top-level agent remain in the parent trajectory. +Direct child scopes that are not `agent` scopes do not create ATIF files. + +## OpenTelemetry and OpenInference Sections + +OpenTelemetry and OpenInference use the same section shape: + +| Field | Default | Notes | +|---|---|---| +| `enabled` | `false` | Must be `true` to construct and register the subscriber. | +| `transport` | `http_binary` | `http_binary` or `grpc`. | +| `endpoint` | Exporter default | OTLP endpoint. | +| `headers` | `{}` | String-to-string exporter headers. | +| `resource_attributes` | `{}` | String-to-string OTLP resource attributes. | +| `service_name` | `nemo-flow` | `service.name` resource attribute. | +| `service_namespace` | Omitted | Optional `service.namespace`. | +| `service_version` | Omitted | Optional `service.version`. | +| `instrumentation_scope` | Omitted | Optional instrumentation scope name. | +| `timeout_millis` | `3000` | Export timeout. | + +Disabled OTLP sections do not construct exporters and do not contact endpoints. + +## Configure + +:::::{tab-set} +:sync-group: language + +::::{tab-item} Python +:sync: python + +```python +from nemo_flow import plugin, scope, ScopeType +from nemo_flow.observability import ( + AtifConfig, + AtofConfig, + ComponentSpec, + ObservabilityConfig, +) + +config = plugin.PluginConfig( + components=[ + ComponentSpec( + ObservabilityConfig( + atof=AtofConfig( + enabled=True, + output_directory="logs", + filename="events.jsonl", + mode="overwrite", + ), + atif=AtifConfig( + enabled=True, + output_directory="logs", + filename_template="trajectory-{session_id}.json", + ), + ) + ) + ] +) + +report = plugin.validate(config) +if report["diagnostics"]: + raise RuntimeError(report["diagnostics"]) + +await plugin.initialize(config) +try: + with scope.scope("agent", ScopeType.Agent): + pass +finally: + plugin.clear() +``` + +:::: + +::::{tab-item} Node.js +:sync: node + +```js +const plugin = require('nemo-flow-node/plugin'); +const observability = require('nemo-flow-node/observability'); + +await plugin.initialize({ + version: 1, + components: [ + observability.ComponentSpec({ + version: 1, + atof: observability.atofConfig({ + enabled: true, + output_directory: 'logs', + filename: 'events.jsonl', + mode: 'overwrite', + }), + atif: observability.atifConfig({ + enabled: true, + output_directory: 'logs', + filename_template: 'trajectory-{session_id}.json', + }), + }), + ], +}); + +try { + // Run instrumented application work here. +} finally { + plugin.clear(); +} +``` + +:::: + +::::{tab-item} Rust +:sync: rust + +```rust +use nemo_flow::observability::plugin_component::{ + AtifSectionConfig, AtofSectionConfig, ComponentSpec, ObservabilityConfig, +}; +use nemo_flow::plugin::{PluginConfig, initialize_plugins}; + +let component = ComponentSpec::new(ObservabilityConfig { + atof: Some(AtofSectionConfig { + enabled: true, + output_directory: Some("logs".into()), + filename: Some("events.jsonl".into()), + mode: "overwrite".into(), + }), + atif: Some(AtifSectionConfig { + enabled: true, + output_directory: Some("logs".into()), + filename_template: "trajectory-{session_id}.json".into(), + ..AtifSectionConfig::default() + }), + ..ObservabilityConfig::default() +}); + +let config = PluginConfig { + version: 1, + components: vec![component.into()], + policy: Default::default(), +}; + +let report = initialize_plugins(config).await?; +assert!(!report.has_errors()); +``` + +:::: + +::::: + +## Validation and Teardown + +Validate plugin configuration before activating it. The plugin reports +unsupported transports, unsupported ATOF modes, unsafe ATIF filename templates, +unknown fields according to policy, and enabled exporters that are unavailable +in the current build or target. + +Call `plugin.clear()` or `clear_plugin_configuration()` during teardown. Clearing +the plugin config deregisters inferred subscribers, flushes file exporters, and +shuts down owned OTLP subscribers. diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index cad6b5d66..7f62ce0c2 100644 --- a/docs/getting-started/configuration.md +++ b/docs/getting-started/configuration.md @@ -30,10 +30,13 @@ Start with [Basic Guide: Define a Plugin](../build-plugins/basic-guide.md) when ## Observability Setup -ATIF exporters, OpenTelemetry subscribers, and OpenInference subscribers are -configured through their binding-native config objects. See -[Export Observability Data](../export-observability-data/code-examples.md) for -the supported export paths. +ATOF exporters, ATIF exporters, OpenTelemetry subscribers, and OpenInference +subscribers can be configured directly through binding-native config objects. +Use the built-in `observability` plugin when you want one plugin component to +own standard exporter setup and teardown. See +[Configure the Observability Plugin](../export-observability-data/observability-plugin.md) +and [Export Observability Data](../export-observability-data/code-examples.md) +for the supported export paths. NeMo Flow does not require application-level environment variables for normal runtime use. Configure most behavior through API objects, registration calls, or diff --git a/docs/index.md b/docs/index.md index 9b2a0a90c..8f182797a 100644 --- a/docs/index.md +++ b/docs/index.md @@ -195,6 +195,7 @@ Code Examples About Basic Guide: Register a Subscriber +Basic Guide: Configure the Observability Plugin Advanced Guide: Export OpenTelemetry Data Advanced Guide: Export OpenInference Data Advanced Guide: Export ATIF diff --git a/docs/reference/api/nodejs/index.md b/docs/reference/api/nodejs/index.md index 25dc1b259..970219aa6 100644 --- a/docs/reference/api/nodejs/index.md +++ b/docs/reference/api/nodejs/index.md @@ -17,7 +17,7 @@ This summary lists the package identity and support status for the binding. The Node.js binding is built with `napi-rs`. The package root exports the core runtime lifecycle APIs, and the package also publishes focused subpath exports -for typed helpers, plugin helpers, and adaptive helpers. +for typed helpers, plugin helpers, adaptive helpers, and observability helpers. ## Main Binding Surfaces @@ -27,6 +27,7 @@ These entry points are the primary APIs to use from this binding. - `nemo-flow-node/typed`: typed wrappers and codec-aware execution helpers - `nemo-flow-node/plugin`: plugin-facing helpers and configuration types - `nemo-flow-node/adaptive`: adaptive helpers layered on top of the runtime +- `nemo-flow-node/observability`: built-in observability plugin helpers ## How To Read The Generated Pages @@ -36,6 +37,7 @@ The generated pages are organized around the package export map: - `Typed Helpers`: the `./typed` export - `Plugins`: the `./plugin` export - `Adaptive`: the `./adaptive` export +- `Observability`: the `./observability` export Use the generated Node.js pages for symbol-level documentation: @@ -43,6 +45,7 @@ Use the generated Node.js pages for symbol-level documentation: - {doc}`Typed Helpers <_generated/typed>` - {doc}`Plugins <_generated/plugin>` - {doc}`Adaptive <_generated/adaptive>` +- {doc}`Observability <_generated/observability>` ```{toctree} :maxdepth: 1 @@ -51,6 +54,7 @@ runtime <_generated/runtime> typed <_generated/typed> plugin <_generated/plugin> adaptive <_generated/adaptive> +observability <_generated/observability> ``` ## Related Guides @@ -64,6 +68,7 @@ Use these links to continue from the API reference into task-focused guides. - [Subscribers](../../../about/concepts/subscribers.md) - [Plugins](../../../about/concepts/plugins.md) - [Adaptive Optimization](../../../use-adaptive-optimization/about.md) +- [Configure the Observability Plugin](../../../export-observability-data/observability-plugin.md) - [Instrument a Tool Call](../../../instrument-applications/instrument-tool-call.md) - [Typed Wrappers and Codecs](../../../integrate-frameworks/using-codecs.md) - [Framework Integration Surfaces](../../../integrate-frameworks/about.md) diff --git a/docs/reference/api/python/index.md b/docs/reference/api/python/index.md index 198aee51f..932cbe81b 100644 --- a/docs/reference/api/python/index.md +++ b/docs/reference/api/python/index.md @@ -29,7 +29,7 @@ These entry points are the primary APIs to use from this binding. - `nemo_flow.guardrails` and `nemo_flow.intercepts`: register global middleware - `nemo_flow.scope_local`: register middleware against a specific scope hierarchy - `nemo_flow.subscribers`: observe emitted runtime lifecycle events -- `nemo_flow.plugin` and `nemo_flow.adaptive`: configure plugin-backed and adaptive behavior +- `nemo_flow.plugin`, `nemo_flow.adaptive`, and `nemo_flow.observability`: configure plugin-backed, adaptive, and exporter behavior - `nemo_flow.typed` and `nemo_flow.codecs`: use typed wrappers and request/response codecs ## How To Read The Generated Pages @@ -46,6 +46,7 @@ will find submodule pages for the public binding surface, including: - `subscribers` - `plugin` - `adaptive` +- `observability` - `typed` - `codecs` @@ -69,5 +70,6 @@ Use these links to continue from the API reference into task-focused guides. - [Subscribers](../../../about/concepts/subscribers.md) - [Plugins](../../../about/concepts/plugins.md) - [Adaptive Optimization](../../../use-adaptive-optimization/about.md) +- [Configure the Observability Plugin](../../../export-observability-data/observability-plugin.md) - [Typed Wrappers and Codecs](../../../integrate-frameworks/using-codecs.md) - [Framework Integration Surfaces](../../../integrate-frameworks/about.md) diff --git a/docs/reference/api/rust/index.md b/docs/reference/api/rust/index.md index 66f9d8adc..18078dd0e 100644 --- a/docs/reference/api/rust/index.md +++ b/docs/reference/api/rust/index.md @@ -25,14 +25,16 @@ module tree. These entry points are the primary APIs to use from this binding. -- `nemo-flow`: core runtime APIs for scopes, tools, LLMs, registries, subscribers, codecs, streams, and observability +- `nemo-flow`: core runtime APIs for scopes, tools, LLMs, registries, subscribers, codecs, streams, observability exporters, and the built-in observability plugin - `nemo-flow-adaptive`: adaptive runtime helpers, learner implementations, storage backends, and adaptive configuration - `nemo-flow-cli`: binary gateway for coding-agent hooks and passthrough LLM observability - `nemo-flow-ffi`: raw C ABI used by downstream native bindings Within `nemo-flow`, most integrations start in `api`, especially the `scope`, `tool`, `llm`, `registry`, and `subscriber` modules. Other important public -modules include `codec`, `observability`, `stream`, `error`, and `json`. +modules include `codec`, `observability`, `stream`, `error`, and `json`. The +`observability::plugin_component` module contains the built-in `observability` +plugin config types. Within `nemo-flow-adaptive`, the main surfaces include adaptive configuration, plugin components, storage abstractions, learners, trie-backed data @@ -74,6 +76,7 @@ Use these links to continue from the API reference into task-focused guides. - [Subscribers](../../../about/concepts/subscribers.md) - [Plugins](../../../about/concepts/plugins.md) - [Adaptive Optimization](../../../use-adaptive-optimization/about.md) +- [Configure the Observability Plugin](../../../export-observability-data/observability-plugin.md) - [Typed Wrappers and Codecs](../../../integrate-frameworks/using-codecs.md) - [Framework Integration Surfaces](../../../integrate-frameworks/about.md) - [Coding-Agent Gateway](../../../integrate-frameworks/coding-agent-gateway.md) diff --git a/go/nemo_flow/observability_plugin.go b/go/nemo_flow/observability_plugin.go new file mode 100644 index 000000000..fcc1e8b45 --- /dev/null +++ b/go/nemo_flow/observability_plugin.go @@ -0,0 +1,111 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package nemo_flow + +// ObservabilityPluginKind is the top-level plugin kind used by the core observability component. +const ObservabilityPluginKind = "observability" + +// ObservabilityConfig is the canonical Go shape for the observability plugin config document. +type ObservabilityConfig struct { + Version uint32 `json:"version,omitempty"` + Atof *ObservabilityAtofConfig `json:"atof,omitempty"` + Atif *ObservabilityAtifConfig `json:"atif,omitempty"` + OpenTelemetry *ObservabilityOtlpConfig `json:"opentelemetry,omitempty"` + OpenInference *ObservabilityOtlpConfig `json:"openinference,omitempty"` + Policy *ConfigPolicy `json:"policy,omitempty"` +} + +// ObservabilityAtofConfig configures filesystem-backed raw ATOF JSONL export. +type ObservabilityAtofConfig struct { + Enabled bool `json:"enabled,omitempty"` + OutputDirectory string `json:"output_directory,omitempty"` + Filename string `json:"filename,omitempty"` + Mode string `json:"mode,omitempty"` +} + +// ObservabilityAtifConfig configures per-top-level-agent ATIF file export. +type ObservabilityAtifConfig struct { + Enabled bool `json:"enabled,omitempty"` + AgentName string `json:"agent_name,omitempty"` + AgentVersion string `json:"agent_version,omitempty"` + ModelName string `json:"model_name,omitempty"` + ToolDefinitions []map[string]any `json:"tool_definitions,omitempty"` + Extra map[string]any `json:"extra,omitempty"` + OutputDirectory string `json:"output_directory,omitempty"` + FilenameTemplate string `json:"filename_template,omitempty"` +} + +// ObservabilityOtlpConfig configures OpenTelemetry or OpenInference OTLP export. +type ObservabilityOtlpConfig struct { + Enabled bool `json:"enabled,omitempty"` + Transport string `json:"transport,omitempty"` + Endpoint string `json:"endpoint,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + ResourceAttributes map[string]string `json:"resource_attributes,omitempty"` + ServiceName string `json:"service_name,omitempty"` + ServiceNamespace string `json:"service_namespace,omitempty"` + ServiceVersion string `json:"service_version,omitempty"` + InstrumentationScope string `json:"instrumentation_scope,omitempty"` + TimeoutMillis uint64 `json:"timeout_millis,omitempty"` +} + +// ObservabilityComponentSpec wraps one observability config as a top-level plugin component. +type ObservabilityComponentSpec struct { + Enabled bool `json:"enabled,omitempty"` + Config ObservabilityConfig `json:"config"` +} + +// NewObservabilityConfig returns a default observability config with version 1. +func NewObservabilityConfig() ObservabilityConfig { + return ObservabilityConfig{Version: 1} +} + +// NewObservabilityAtofConfig returns disabled ATOF JSONL settings with native defaults. +func NewObservabilityAtofConfig() ObservabilityAtofConfig { + return ObservabilityAtofConfig{ + Mode: "append", + } +} + +// NewObservabilityAtifConfig returns disabled ATIF settings with core defaults. +func NewObservabilityAtifConfig() ObservabilityAtifConfig { + return ObservabilityAtifConfig{ + AgentName: "NeMo Flow", + ModelName: "unknown", + FilenameTemplate: "nemo-flow-atif-{session_id}.json", + } +} + +// NewObservabilityOtlpConfig returns disabled OTLP settings with core defaults. +func NewObservabilityOtlpConfig() ObservabilityOtlpConfig { + return ObservabilityOtlpConfig{ + Transport: "http_binary", + Headers: map[string]string{}, + ResourceAttributes: map[string]string{}, + ServiceName: "nemo-flow", + TimeoutMillis: 3000, + } +} + +// NewObservabilityComponentSpec wraps observability config as an enabled top-level component. +func NewObservabilityComponentSpec(config ObservabilityConfig) ObservabilityComponentSpec { + return ObservabilityComponentSpec{ + Enabled: true, + Config: config, + } +} + +// PluginComponent converts the observability component wrapper into the shared plugin shape. +func (spec ObservabilityComponentSpec) PluginComponent() PluginComponentSpec { + return PluginComponentSpec{ + Kind: ObservabilityPluginKind, + Enabled: spec.Enabled, + Config: mustConfigMap(spec.Config), + } +} + +// ObservabilityComponent converts observability config directly into a shared plugin component. +func ObservabilityComponent(config ObservabilityConfig) PluginComponentSpec { + return NewObservabilityComponentSpec(config).PluginComponent() +} diff --git a/go/nemo_flow/observability_plugin_test.go b/go/nemo_flow/observability_plugin_test.go new file mode 100644 index 000000000..6e28aae47 --- /dev/null +++ b/go/nemo_flow/observability_plugin_test.go @@ -0,0 +1,240 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package nemo_flow + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestObservabilityConfigHelpers(t *testing.T) { + config := NewObservabilityConfig() + if config.Version != 1 { + t.Fatalf("expected version 1, got %d", config.Version) + } + atof := NewObservabilityAtofConfig() + if atof.Enabled || atof.Mode != "append" { + t.Fatalf("unexpected ATOF defaults: %#v", atof) + } + atif := NewObservabilityAtifConfig() + if atif.Enabled || atif.AgentName != "NeMo Flow" || atif.ModelName != "unknown" || atif.FilenameTemplate != "nemo-flow-atif-{session_id}.json" { + t.Fatalf("unexpected ATIF defaults: %#v", atif) + } + otlp := NewObservabilityOtlpConfig() + if otlp.Enabled || otlp.Transport != "http_binary" || otlp.ServiceName != "nemo-flow" || otlp.TimeoutMillis != 3000 { + t.Fatalf("unexpected OTLP defaults: %#v", otlp) + } + + config.Atof = &atof + wrapped := ObservabilityComponent(config) + if wrapped.Kind != ObservabilityPluginKind || !wrapped.Enabled { + t.Fatalf("unexpected component wrapper: %#v", wrapped) + } + if _, ok := wrapped.Config["atof"].(map[string]any); !ok { + t.Fatalf("expected serialized ATOF config object, got %#v", wrapped.Config) + } +} + +func TestObservabilityPluginAtofAndAtifFiles(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + dir := t.TempDir() + config := NewObservabilityConfig() + atof := NewObservabilityAtofConfig() + atof.Enabled = true + atof.OutputDirectory = dir + atof.Filename = "events.jsonl" + atof.Mode = "overwrite" + config.Atof = &atof + atif := NewObservabilityAtifConfig() + atif.Enabled = true + atif.AgentName = "go-agent" + atif.AgentVersion = "1.2.3" + atif.ModelName = "go-model" + atif.ToolDefinitions = []map[string]any{{"name": "search"}} + atif.Extra = map[string]any{"binding": "go"} + atif.OutputDirectory = dir + atif.FilenameTemplate = "trajectory-{session_id}.json" + config.Atif = &atif + + if report, err := ValidatePluginConfig(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { + t.Fatalf("ValidatePluginConfig failed: %v", err) + } else if len(report.Diagnostics) != 0 { + t.Fatalf("unexpected diagnostics: %#v", report.Diagnostics) + } + if _, err := InitializePlugins(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { + t.Fatalf("InitializePlugins failed: %v", err) + } + + handle, err := PushScope("go-observability-agent", ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":true}`))) + if err != nil { + t.Fatalf("PushScope failed: %v", err) + } + if err := EmitEvent("go-mark", WithEventParent(handle), WithEventData(json.RawMessage(`{"step":1}`))); err != nil { + t.Fatalf("EmitEvent failed: %v", err) + } + if err := PopScope(handle, WithOutput(json.RawMessage(`{"done":true}`))); err != nil { + t.Fatalf("PopScope failed: %v", err) + } + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + + jsonl := string(mustReadFile(t, filepath.Join(dir, "events.jsonl"))) + if got := strings.Count(strings.TrimSpace(jsonl), "\n") + 1; got != 3 { + t.Fatalf("expected 3 JSONL records, got %d: %s", got, jsonl) + } + + trajectoryPath := filepath.Join(dir, "trajectory-"+handle.UUID()+".json") + var trajectory map[string]any + if err := json.Unmarshal(mustReadFile(t, trajectoryPath), &trajectory); err != nil { + t.Fatalf("failed to read trajectory: %v", err) + } + agent := trajectory["agent"].(map[string]any) + if agent["name"] != "go-agent" || agent["version"] != "1.2.3" || agent["model_name"] != "go-model" { + t.Fatalf("unexpected ATIF agent metadata: %#v", agent) + } + if !strings.Contains(string(mustReadFile(t, trajectoryPath)), "go-observability-agent") { + t.Fatalf("expected top-level agent event in ATIF file") + } +} + +func TestObservabilityPluginAtifSplitsMultipleTopLevelAgents(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + dir := t.TempDir() + config := NewObservabilityConfig() + atif := NewObservabilityAtifConfig() + atif.Enabled = true + atif.OutputDirectory = dir + atif.FilenameTemplate = "trajectory-{session_id}.json" + config.Atif = &atif + + if _, err := InitializePlugins(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { + t.Fatalf("InitializePlugins failed: %v", err) + } + + first, err := PushScope("go-first-agent", ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":"first"}`))) + if err != nil { + t.Fatalf("PushScope first failed: %v", err) + } + if err := EmitEvent("go-first-mark", WithEventParent(first), WithEventData(json.RawMessage(`{"agent":"first"}`))); err != nil { + t.Fatalf("EmitEvent first failed: %v", err) + } + nested, err := PushScope("go-nested-agent", ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":"nested"}`))) + if err != nil { + t.Fatalf("PushScope nested failed: %v", err) + } + if err := EmitEvent("go-nested-mark", WithEventParent(nested), WithEventData(json.RawMessage(`{"agent":"nested"}`))); err != nil { + t.Fatalf("EmitEvent nested failed: %v", err) + } + if err := PopScope(nested, WithOutput(json.RawMessage(`{"done":true}`))); err != nil { + t.Fatalf("PopScope nested failed: %v", err) + } + if err := PopScope(first, WithOutput(json.RawMessage(`{"done":true}`))); err != nil { + t.Fatalf("PopScope first failed: %v", err) + } + + second, err := PushScope("go-second-agent", ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":"second"}`))) + if err != nil { + t.Fatalf("PushScope second failed: %v", err) + } + if err := EmitEvent("go-second-mark", WithEventParent(second), WithEventData(json.RawMessage(`{"agent":"second"}`))); err != nil { + t.Fatalf("EmitEvent second failed: %v", err) + } + if err := PopScope(second, WithOutput(json.RawMessage(`{"done":true}`))); err != nil { + t.Fatalf("PopScope second failed: %v", err) + } + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + + files, err := filepath.Glob(filepath.Join(dir, "trajectory-*.json")) + if err != nil { + t.Fatalf("Glob failed: %v", err) + } + if len(files) != 2 { + t.Fatalf("expected 2 ATIF trajectory files, got %d: %#v", len(files), files) + } + + firstPayload := string(mustReadFile(t, filepath.Join(dir, "trajectory-"+first.UUID()+".json"))) + secondPayload := string(mustReadFile(t, filepath.Join(dir, "trajectory-"+second.UUID()+".json"))) + if !strings.Contains(firstPayload, "go-first-agent") || !strings.Contains(firstPayload, "go-nested-agent") { + t.Fatalf("expected first trajectory to include first and nested agents: %s", firstPayload) + } + if strings.Contains(firstPayload, "go-second-agent") { + t.Fatalf("first trajectory leaked second agent events: %s", firstPayload) + } + if !strings.Contains(secondPayload, "go-second-agent") { + t.Fatalf("expected second trajectory to include second agent: %s", secondPayload) + } + if strings.Contains(secondPayload, "go-first-agent") || strings.Contains(secondPayload, "go-nested-agent") { + t.Fatalf("second trajectory leaked first trajectory events: %s", secondPayload) + } +} + +func TestObservabilityPluginValidationRejectsBadValues(t *testing.T) { + config := NewObservabilityConfig() + atof := NewObservabilityAtofConfig() + atof.Mode = "bad" + config.Atof = &atof + atif := NewObservabilityAtifConfig() + atif.FilenameTemplate = "missing-placeholder.json" + config.Atif = &atif + + report, err := ValidatePluginConfig(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}) + if err != nil { + t.Fatalf("ValidatePluginConfig failed: %v", err) + } + if len(report.Diagnostics) < 2 { + t.Fatalf("expected validation diagnostics, got %#v", report.Diagnostics) + } +} + +func TestObservabilityPluginListKindIsAutomatic(t *testing.T) { + kinds, err := ListPluginKinds() + if err != nil { + t.Fatalf("ListPluginKinds failed: %v", err) + } + for _, kind := range kinds { + if kind == ObservabilityPluginKind { + return + } + } + t.Fatalf("expected %q in registered kinds: %#v", ObservabilityPluginKind, kinds) +} + +func TestObservabilityAtifOpenAgentFlushesOnClear(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + dir := t.TempDir() + config := NewObservabilityConfig() + atif := NewObservabilityAtifConfig() + atif.Enabled = true + atif.OutputDirectory = dir + config.Atif = &atif + if _, err := InitializePlugins(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { + t.Fatalf("InitializePlugins failed: %v", err) + } + handle, err := PushScope("go-open-agent", ScopeTypeAgent) + if err != nil { + t.Fatalf("PushScope failed: %v", err) + } + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration failed: %v", err) + } + path := filepath.Join(dir, "nemo-flow-atif-"+handle.UUID()+".json") + if _, err := os.Stat(path); err != nil { + t.Fatalf("expected open-agent ATIF file at %s: %v", path, err) + } + if err := PopScope(handle); err != nil { + t.Fatalf("PopScope failed: %v", err) + } +} diff --git a/python/nemo_flow/README.md b/python/nemo_flow/README.md index 3c2b19435..20a82194d 100644 --- a/python/nemo_flow/README.md +++ b/python/nemo_flow/README.md @@ -44,7 +44,7 @@ runtime semantics as the Rust and Node.js surfaces. - ✅ **Subscribers and exporters**: Event consumers for observability and diagnostics. - ✅ **Plugin and typed helpers**: Public modules for plugins, codecs, typed - wrappers, and adaptive runtime behavior. + wrappers, adaptive runtime behavior, and observability plugin configuration. - ✅ **Shared Rust runtime semantics**: Python behavior aligned with the Rust and Node.js surfaces. @@ -94,6 +94,7 @@ The public package modules are: - `nemo_flow.subscribers` - `nemo_flow.plugin` - `nemo_flow.adaptive` +- `nemo_flow.observability` - `nemo_flow.typed` - `nemo_flow.codecs` diff --git a/python/nemo_flow/__init__.py b/python/nemo_flow/__init__.py index c29f6bedf..85dce458e 100644 --- a/python/nemo_flow/__init__.py +++ b/python/nemo_flow/__init__.py @@ -15,6 +15,7 @@ - ``nemo_flow.typed`` for codec-based typed wrappers - ``nemo_flow.plugin`` for global plugin configuration and custom plugin registration - ``nemo_flow.adaptive`` for adaptive component configuration helpers +- ``nemo_flow.observability`` for observability component configuration helpers Top-level exports also include: @@ -181,6 +182,7 @@ async def main(): guardrails, intercepts, llm, + observability, plugin, scope, scope_local, @@ -422,6 +424,7 @@ def worker() -> None: "typed", "plugin", "adaptive", + "observability", # Scope stack isolation "ScopeStack", "create_scope_stack", diff --git a/python/nemo_flow/__init__.pyi b/python/nemo_flow/__init__.pyi index ed3187860..c506b2d26 100644 --- a/python/nemo_flow/__init__.pyi +++ b/python/nemo_flow/__init__.pyi @@ -30,6 +30,7 @@ from nemo_flow import codecs as codecs from nemo_flow import guardrails as guardrails from nemo_flow import intercepts as intercepts from nemo_flow import llm as llm +from nemo_flow import observability as observability from nemo_flow import plugin as plugin from nemo_flow import scope as scope from nemo_flow import scope_local as scope_local diff --git a/python/nemo_flow/observability.py b/python/nemo_flow/observability.py new file mode 100644 index 000000000..306667f59 --- /dev/null +++ b/python/nemo_flow/observability.py @@ -0,0 +1,191 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Observability plugin configuration helpers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field, fields, is_dataclass +from typing import Literal, Protocol, cast + +from nemo_flow import Json, JsonObject, UnsupportedBehavior + + +class _SupportsToDict(Protocol): + def to_dict(self) -> JsonObject: ... + + +def _normalize(value: object) -> Json: + if hasattr(value, "to_dict"): + return cast(_SupportsToDict, value).to_dict() + if is_dataclass(value) and not isinstance(value, type): + return { + field_info.name: _normalize(field_value) + for field_info in fields(value) + if (field_value := getattr(value, field_info.name)) is not None + } + if isinstance(value, list): + return [_normalize(item) for item in value] + if isinstance(value, dict): + return {cast(str, key): _normalize(val) for key, val in value.items() if val is not None} + return cast(Json, value) + + +def _normalize_object(value: object) -> JsonObject: + return cast(JsonObject, _normalize(value)) + + +@dataclass(slots=True) +class ConfigPolicy: + """Policy for unsupported observability configuration.""" + + unknown_component: UnsupportedBehavior = "warn" + unknown_field: UnsupportedBehavior = "warn" + unsupported_value: UnsupportedBehavior = "error" + + def to_dict(self) -> JsonObject: + """Serialize this policy to the canonical JSON object shape.""" + return { + "unknown_component": self.unknown_component, + "unknown_field": self.unknown_field, + "unsupported_value": self.unsupported_value, + } + + +@dataclass(slots=True) +class AtofConfig: + """Filesystem-backed raw ATOF JSONL export settings.""" + + enabled: bool = False + output_directory: str | None = None + filename: str | None = None + mode: Literal["append", "overwrite"] = "append" + + def to_dict(self) -> JsonObject: + """Serialize this ATOF config to the canonical JSON object shape.""" + return _normalize_object( + { + "enabled": self.enabled, + "output_directory": self.output_directory, + "filename": self.filename, + "mode": self.mode, + } + ) + + +@dataclass(slots=True) +class AtifConfig: + """Per-top-level-agent ATIF file export settings.""" + + enabled: bool = False + agent_name: str = "NeMo Flow" + agent_version: str | None = None + model_name: str = "unknown" + tool_definitions: list[JsonObject] | None = None + extra: JsonObject | None = None + output_directory: str | None = None + filename_template: str = "nemo-flow-atif-{session_id}.json" + + def to_dict(self) -> JsonObject: + """Serialize this ATIF config to the canonical JSON object shape.""" + value = { + "enabled": self.enabled, + "agent_name": self.agent_name, + "agent_version": self.agent_version, + "model_name": self.model_name, + "tool_definitions": self.tool_definitions, + "extra": self.extra, + "output_directory": self.output_directory, + "filename_template": self.filename_template, + } + if value["agent_version"] is None: + value.pop("agent_version") + return _normalize_object(value) + + +@dataclass(slots=True) +class OtlpConfig: + """Shared OpenTelemetry/OpenInference OTLP export settings.""" + + enabled: bool = False + transport: Literal["http_binary", "grpc"] = "http_binary" + endpoint: str | None = None + headers: dict[str, str] = field(default_factory=dict) + resource_attributes: dict[str, str] = field(default_factory=dict) + service_name: str = "nemo-flow" + service_namespace: str | None = None + service_version: str | None = None + instrumentation_scope: str | None = None + timeout_millis: int = 3000 + + def to_dict(self) -> JsonObject: + """Serialize this OTLP config to the canonical JSON object shape.""" + return _normalize_object( + { + "enabled": self.enabled, + "transport": self.transport, + "endpoint": self.endpoint, + "headers": self.headers, + "resource_attributes": self.resource_attributes, + "service_name": self.service_name, + "service_namespace": self.service_namespace, + "service_version": self.service_version, + "instrumentation_scope": self.instrumentation_scope, + "timeout_millis": self.timeout_millis, + } + ) + + +@dataclass(slots=True) +class ObservabilityConfig: + """Canonical config document for the top-level observability component.""" + + version: int = 1 + atof: AtofConfig | None = None + atif: AtifConfig | None = None + opentelemetry: OtlpConfig | None = None + openinference: OtlpConfig | None = None + policy: ConfigPolicy = field(default_factory=ConfigPolicy) + + def to_dict(self) -> JsonObject: + """Serialize this observability config to the canonical JSON object shape.""" + return _normalize_object( + { + "version": self.version, + "atof": self.atof, + "atif": self.atif, + "opentelemetry": self.opentelemetry, + "openinference": self.openinference, + "policy": self.policy, + } + ) + + +OBSERVABILITY_PLUGIN_KIND = "observability" + + +@dataclass(slots=True) +class ComponentSpec: + """Top-level observability component wrapper.""" + + config: ObservabilityConfig | JsonObject + enabled: bool = True + + def to_dict(self) -> JsonObject: + """Serialize this component to the canonical plugin shape.""" + return { + "kind": OBSERVABILITY_PLUGIN_KIND, + "enabled": self.enabled, + "config": _normalize_object(self.config), + } + + +__all__ = [ + "ConfigPolicy", + "AtofConfig", + "AtifConfig", + "OtlpConfig", + "ObservabilityConfig", + "OBSERVABILITY_PLUGIN_KIND", + "ComponentSpec", +] diff --git a/python/nemo_flow/observability.pyi b/python/nemo_flow/observability.pyi new file mode 100644 index 000000000..595a812ca --- /dev/null +++ b/python/nemo_flow/observability.pyi @@ -0,0 +1,70 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Type stubs for ``nemo_flow.observability``.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Literal + +from nemo_flow import JsonObject, UnsupportedBehavior + +@dataclass(slots=True) +class ConfigPolicy: + unknown_component: UnsupportedBehavior = ... + unknown_field: UnsupportedBehavior = ... + unsupported_value: UnsupportedBehavior = ... + def to_dict(self) -> JsonObject: ... + +@dataclass(slots=True) +class AtofConfig: + enabled: bool = ... + output_directory: str | None = ... + filename: str | None = ... + mode: Literal["append", "overwrite"] = ... + def to_dict(self) -> JsonObject: ... + +@dataclass(slots=True) +class AtifConfig: + enabled: bool = ... + agent_name: str = ... + agent_version: str | None = ... + model_name: str = ... + tool_definitions: list[JsonObject] | None = ... + extra: JsonObject | None = ... + output_directory: str | None = ... + filename_template: str = ... + def to_dict(self) -> JsonObject: ... + +@dataclass(slots=True) +class OtlpConfig: + enabled: bool = ... + transport: Literal["http_binary", "grpc"] = ... + endpoint: str | None = ... + headers: dict[str, str] = field(default_factory=dict) + resource_attributes: dict[str, str] = field(default_factory=dict) + service_name: str = ... + service_namespace: str | None = ... + service_version: str | None = ... + instrumentation_scope: str | None = ... + timeout_millis: int = ... + def to_dict(self) -> JsonObject: ... + +@dataclass(slots=True) +class ObservabilityConfig: + version: int = ... + atof: AtofConfig | None = ... + atif: AtifConfig | None = ... + opentelemetry: OtlpConfig | None = ... + openinference: OtlpConfig | None = ... + policy: ConfigPolicy = field(default_factory=ConfigPolicy) + def to_dict(self) -> JsonObject: ... + +OBSERVABILITY_PLUGIN_KIND: Literal["observability"] + +@dataclass(slots=True) +class ComponentSpec: + config: ObservabilityConfig | JsonObject + enabled: bool = ... + def to_dict(self) -> JsonObject: ... diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py new file mode 100644 index 000000000..e3f00323d --- /dev/null +++ b/python/tests/test_observability_plugin.py @@ -0,0 +1,160 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the built-in observability plugin config helpers.""" + +from __future__ import annotations + +import json + +from nemo_flow import ScopeType, plugin, scope +from nemo_flow.observability import ( + OBSERVABILITY_PLUGIN_KIND, + AtifConfig, + AtofConfig, + ComponentSpec, + ObservabilityConfig, + OtlpConfig, +) + + +class TestObservabilityConfigHelpers: + def test_defaults_and_component_wrapper(self): + assert AtofConfig().to_dict() == {"enabled": False, "mode": "append"} + assert AtifConfig().to_dict() == { + "enabled": False, + "agent_name": "NeMo Flow", + "model_name": "unknown", + "filename_template": "nemo-flow-atif-{session_id}.json", + } + assert OtlpConfig().to_dict() == { + "enabled": False, + "transport": "http_binary", + "headers": {}, + "resource_attributes": {}, + "service_name": "nemo-flow", + "timeout_millis": 3000, + } + + wrapped = ComponentSpec(ObservabilityConfig(atof=AtofConfig())).to_dict() + assert wrapped["kind"] == OBSERVABILITY_PLUGIN_KIND + assert wrapped["enabled"] is True + wrapped_config = wrapped["config"] + assert isinstance(wrapped_config, dict) + assert wrapped_config["version"] == 1 + + def test_validation_rejects_bad_values(self): + report = plugin.validate( + plugin.PluginConfig( + components=[ + ComponentSpec( + { + "version": 1, + "atof": {"mode": "bad"}, + "atif": {"filename_template": "missing-placeholder"}, + } + ) + ] + ) + ) + fields = {diag.get("field") for diag in report["diagnostics"]} + assert {"mode", "filename_template"} <= fields + + def test_list_kinds_includes_builtin_observability(self): + assert OBSERVABILITY_PLUGIN_KIND in plugin.list_kinds() + + async def test_atof_and_atif_file_outputs(self, tmp_path): + config = ObservabilityConfig( + atof=AtofConfig( + enabled=True, + output_directory=str(tmp_path), + filename="events.jsonl", + mode="overwrite", + ), + atif=AtifConfig( + enabled=True, + agent_name="python-agent", + agent_version="1.2.3", + model_name="python-model", + tool_definitions=[{"name": "search"}], + extra={"binding": "python"}, + output_directory=str(tmp_path), + filename_template="trajectory-{session_id}.json", + ), + ) + + await plugin.initialize(plugin.PluginConfig(components=[ComponentSpec(config)])) + try: + with scope.scope("python-observability-agent", ScopeType.Agent) as handle: + scope.event("python-mark", handle=handle, data={"step": 1}) + finally: + plugin.clear() + + lines = (tmp_path / "events.jsonl").read_text().strip().splitlines() + assert len(lines) == 3 + assert json.loads(lines[1])["name"] == "python-mark" + + trajectory = json.loads((tmp_path / f"trajectory-{handle.uuid}.json").read_text()) + assert trajectory["agent"]["name"] == "python-agent" + assert trajectory["agent"]["version"] == "1.2.3" + assert trajectory["agent"]["model_name"] == "python-model" + assert trajectory["agent"]["tool_definitions"][0]["name"] == "search" + assert trajectory["agent"]["extra"]["binding"] == "python" + assert "python-observability-agent" in json.dumps(trajectory["extra"]) + + async def test_atif_flushes_open_agent_on_clear(self, tmp_path): + await plugin.initialize( + plugin.PluginConfig( + components=[ + ComponentSpec(ObservabilityConfig(atif=AtifConfig(enabled=True, output_directory=str(tmp_path)))) + ] + ) + ) + handle = scope.push("python-open-agent", ScopeType.Agent) + try: + plugin.clear() + assert (tmp_path / f"nemo-flow-atif-{handle.uuid}.json").exists() + finally: + scope.pop(handle) + + async def test_atif_splits_multiple_top_level_agent_scopes(self, tmp_path): + await plugin.initialize( + plugin.PluginConfig( + components=[ + ComponentSpec( + ObservabilityConfig( + atif=AtifConfig( + enabled=True, + output_directory=str(tmp_path), + filename_template="trajectory-{session_id}.json", + ) + ) + ) + ] + ) + ) + try: + with scope.scope("python-first-agent", ScopeType.Agent) as first: + scope.event("python-first-mark", handle=first, data={"agent": "first"}) + with scope.scope("python-nested-agent", ScopeType.Agent) as nested: + scope.event("python-nested-mark", handle=nested, data={"agent": "nested"}) + + with scope.scope("python-second-agent", ScopeType.Agent) as second: + scope.event("python-second-mark", handle=second, data={"agent": "second"}) + finally: + plugin.clear() + + files = sorted(tmp_path.glob("trajectory-*.json")) + assert len(files) == 2 + + first_trajectory = json.loads((tmp_path / f"trajectory-{first.uuid}.json").read_text()) + second_trajectory = json.loads((tmp_path / f"trajectory-{second.uuid}.json").read_text()) + first_payload = json.dumps(first_trajectory["extra"]) + second_payload = json.dumps(second_trajectory["extra"]) + + assert "python-first-agent" in first_payload + assert "python-nested-agent" in first_payload + assert "python-second-agent" not in first_payload + assert "python-second-agent" in second_payload + assert "python-first-agent" not in second_payload + assert "python-nested-agent" not in second_payload diff --git a/scripts/docs/build_node_docs_artifacts.mjs b/scripts/docs/build_node_docs_artifacts.mjs index 39c182d5a..01ff1c65a 100644 --- a/scripts/docs/build_node_docs_artifacts.mjs +++ b/scripts/docs/build_node_docs_artifacts.mjs @@ -48,6 +48,13 @@ const MODULES = [ pageName: 'adaptive', title: 'Adaptive', }, + { + declaration: 'observability.d.ts', + deppath: './declarations/observability.d', + entryTarget: './declarations/observability', + pageName: 'observability', + title: 'Observability', + }, ]; // These rewrites are a docs-only compatibility layer for TypeDoc/sphinx-js @@ -108,6 +115,31 @@ const DECLARATION_REWRITES = new Map([ }, ], ], + [ + 'observability.d.ts', + [ + { + original: "import type { Json } from './index';", + replacement: 'type Json = import("./index").Json;', + }, + { + original: "import type { ConfigPolicy, ConfigDiagnostic, ConfigReport } from './plugin';\n\nexport { ConfigPolicy, ConfigDiagnostic, ConfigReport };", + replacement: [ + 'export type ConfigPolicy = import("./plugin").ConfigPolicy;', + 'export type ConfigDiagnostic = import("./plugin").ConfigDiagnostic;', + 'export type ConfigReport = import("./plugin").ConfigReport;', + ].join('\n'), + }, + { + original: 'export interface ComponentSpec {', + replacement: 'interface ComponentSpecShape {', + }, + { + original: '): ComponentSpec;', + replacement: '): ComponentSpecShape;', + }, + ], + ], ]); const PUBLIC_NAME_REWRITES = new Map([['ComponentSpecShape', 'ComponentSpec']]);