diff --git a/crates/cli/src/config.rs b/crates/cli/src/config.rs index 8f6fcdabc..afa6b3eca 100644 --- a/crates/cli/src/config.rs +++ b/crates/cli/src/config.rs @@ -169,6 +169,9 @@ pub(crate) struct ServerArgs { /// OpenInference-compatible OTLP HTTP endpoint for streaming spans (Phoenix, Arize, etc.) #[arg(long, env = "NEMO_FLOW_OPENINFERENCE_ENDPOINT")] pub(crate) openinference_endpoint: Option, + /// Generic plugin configuration JSON for process-level gateway plugin activation. + #[arg(long, env = "NEMO_FLOW_PLUGIN_CONFIG")] + pub(crate) plugin_config: Option, } impl ServerArgs { @@ -184,6 +187,7 @@ impl ServerArgs { || self.atif_dir.is_some() || self.atof_dir.is_some() || self.openinference_endpoint.is_some() + || self.plugin_config.is_some() || self.config.is_some() } } @@ -409,9 +413,8 @@ impl Default for CursorAgentConfig { } // TOML file shape grouped by user intent. Sections map 1:1 onto fields already present on -// `GatewayConfig` / `AgentConfigs`; this is a rename pass — no new runtime knobs land in this -// pass. `[plugins]` is reserved as a forward-compatible block so users editing config today -// need no rewrite once the plugin runtime lands. +// `GatewayConfig` / `AgentConfigs`; plugin config is passed through to the runtime's generic +// `PluginConfig` activation path. #[derive(Debug, Clone, Default, Deserialize)] struct FileConfig { upstream: Option, @@ -472,8 +475,7 @@ struct FileOpenInferenceConfig { #[derive(Debug, Clone, Default, Deserialize)] struct FilePluginsConfig { - // Reserved for the plugin runtime. Stored on `GatewayConfig.plugin_config` for now; - // nothing in-process consumes it until the plugin runtime lands. + // Generic plugin initialization shape. The gateway activates this process-wide at startup. config: Option, } @@ -524,7 +526,7 @@ impl Default for GatewayConfig { /// server-facing command-line layer so launcher-only settings cannot leak into daemon mode. pub(crate) fn resolve_server_config(args: &ServerArgs) -> Result { let mut resolved = load_shared_config(args.config.as_ref())?; - apply_server_overrides(&mut resolved.gateway, args); + apply_server_overrides(&mut resolved.gateway, args)?; Ok(resolved) } @@ -543,7 +545,16 @@ pub(crate) fn resolve_run_config( .or_else(|| inherited.and_then(|args| args.config.as_ref())); let mut resolved = load_shared_config(config)?; if let Some(args) = inherited { - apply_server_overrides(&mut resolved.gateway, args); + // Run-subcommand plugin config has higher precedence than inherited top-level plugin + // config. Skip only that inherited field so file/plugin.toml conflicts are still caught + // when the run-level override is applied below. + if command.plugin_config.is_some() && args.plugin_config.is_some() { + let mut inherited = args.clone(); + inherited.plugin_config = None; + apply_server_overrides(&mut resolved.gateway, &inherited)?; + } else { + apply_server_overrides(&mut resolved.gateway, args)?; + } } apply_run_overrides(&mut resolved.gateway, command)?; resolved.gateway.bind = "127.0.0.1:0" @@ -590,14 +601,14 @@ fn apply_run_json_overrides( config.metadata = Some(parse_json_option("session metadata", value)?); } if let Some(value) = &command.plugin_config { - config.plugin_config = Some(parse_json_option("plugin config", value)?); + apply_cli_plugin_config(config, value)?; } Ok(()) } // Applies direct server flags on top of already-merged configuration. Only present options mutate // the config so lower-priority file values survive when a flag was omitted. -fn apply_server_overrides(config: &mut GatewayConfig, args: &ServerArgs) { +fn apply_server_overrides(config: &mut GatewayConfig, args: &ServerArgs) -> Result<(), CliError> { if let Some(value) = args.bind { config.bind = value; } @@ -616,13 +627,19 @@ fn apply_server_overrides(config: &mut GatewayConfig, args: &ServerArgs) { if let Some(value) = &args.openinference_endpoint { config.exporters.openinference.endpoint = Some(value.clone()); } + if let Some(value) = &args.plugin_config { + apply_cli_plugin_config(config, value)?; + } + Ok(()) } // Loads config from the ordered shared locations, deep-merges TOML tables, maps the typed file -// shape onto runtime structs, then lets environment variables override file values. Invalid TOML -// or typed shapes fail closed because they indicate an operator configuration error. +// shape onto runtime structs, applies a sibling/discovered plugin.toml when present, then lets +// environment variables override file values. Invalid TOML or typed shapes fail closed because +// they indicate an operator configuration error. fn load_shared_config(explicit: Option<&PathBuf>) -> Result { let mut merged = toml::Value::Table(toml::map::Map::new()); + let mut config_toml_plugin_sources = Vec::new(); for path in config_paths(explicit) { if path.exists() { let raw = std::fs::read_to_string(&path)?; @@ -632,14 +649,30 @@ fn load_shared_config(explicit: Option<&PathBuf>) -> Result 1 { + return Err(CliError::Config(format!( + "plugin config is defined in multiple config.toml files: {}; move it to one \ + [plugins].config block or use plugin.toml", + format_paths(&config_toml_plugin_sources) + ))); + } + let plugin_toml = load_plugin_toml_config(explicit)?; let mut resolved = ResolvedConfig { gateway: GatewayConfig::default(), ..ResolvedConfig::default() }; apply_file_config(&mut resolved, merged)?; + apply_plugin_toml_config( + &mut resolved.gateway, + config_toml_plugin_sources.first(), + plugin_toml, + )?; apply_env_config(&mut resolved.gateway); Ok(resolved) } @@ -669,6 +702,37 @@ fn config_paths(explicit: Option<&PathBuf>) -> Vec { paths } +// Returns the plugin config search path. An explicit gateway config path scopes plugin.toml to the +// same directory so `--config path/to/config.toml` can be extended by `path/to/plugin.toml` without +// reading unrelated implicit project/user/global plugin files. +fn plugin_config_paths(explicit: Option<&PathBuf>) -> Vec { + if let Some(path) = explicit { + return path + .parent() + .map(|parent| vec![parent.join("plugin.toml")]) + .unwrap_or_default(); + } + implicit_plugin_config_paths(std::env::current_dir().ok().as_deref(), user_config_dir()) +} + +fn implicit_plugin_config_paths( + cwd: Option<&std::path::Path>, + user_config_dir: Option, +) -> Vec { + // Ordered from lowest to highest precedence. User-level plugin config intentionally loads last + // so an operator can override project-local plugin defaults without editing the checkout. + let mut paths = vec![PathBuf::from("/etc/nemo-flow/plugin.toml")]; + if let Some(cwd) = cwd + && let Some(project) = find_project_plugin_config(cwd) + { + paths.push(project); + } + if let Some(user) = user_config_dir { + paths.push(user.join("plugin.toml")); + } + paths +} + // Walks upward from the current directory and returns the nearest project-local gateway config. // The first hit wins so nested projects can override parent workspace defaults. fn find_project_config(start: &std::path::Path) -> Option { @@ -681,6 +745,17 @@ fn find_project_config(start: &std::path::Path) -> Option { None } +// Walks upward from the current directory and returns the nearest project-local plugin config. +fn find_project_plugin_config(start: &std::path::Path) -> Option { + for ancestor in start.ancestors() { + let path = ancestor.join(".nemo-flow/plugin.toml"); + if path.exists() { + return Some(path); + } + } + None +} + // Resolves the user config using XDG first and HOME/USERPROFILE second. Returning `None` keeps // config loading portable in minimal environments where no home directory is visible. fn user_config_path() -> Option { @@ -807,8 +882,8 @@ fn apply_file_exporters_config( Ok(()) } -// Applies plugin config. Reserved for the plugin runtime — stored on `GatewayConfig.plugin_config` -// and forwarded through hook headers, but no in-process consumer until the runtime lands. +// Applies plugin config. The gateway activates process-level plugin config at startup; hook headers +// still carry the value as session metadata until scoped plugin activation exists. fn apply_file_plugins_config(gateway: &mut GatewayConfig, plugins: Option) { let Some(plugins) = plugins else { return; @@ -818,6 +893,77 @@ fn apply_file_plugins_config(gateway: &mut GatewayConfig, plugins: Option, +} + +fn load_plugin_toml_config( + explicit: Option<&PathBuf>, +) -> Result, CliError> { + load_plugin_toml_config_from_paths(plugin_config_paths(explicit)) +} + +fn load_plugin_toml_config_from_paths(paths: I) -> Result, CliError> +where + I: IntoIterator, +{ + let mut merged = toml::Value::Table(toml::map::Map::new()); + let mut sources = Vec::new(); + for path in paths { + if path.exists() { + let raw = std::fs::read_to_string(&path)?; + let parsed = raw + .parse::() + .map(toml::Value::Table) + .map_err(|error| { + CliError::Config(format!( + "invalid plugin TOML in {}: {error}", + path.display() + )) + })?; + merge_plugin_toml(&mut merged, parsed); + sources.push(path); + } + } + if sources.is_empty() { + return Ok(None); + } + let value = serde_json::to_value(merged) + .map_err(|error| CliError::Config(format!("invalid plugin TOML shape: {error}")))?; + Ok(Some(PluginTomlConfig { value, sources })) +} + +fn apply_plugin_toml_config( + gateway: &mut GatewayConfig, + config_toml_plugin_source: Option<&PathBuf>, + plugin_toml: Option, +) -> Result<(), CliError> { + let Some(plugin_toml) = plugin_toml else { + return Ok(()); + }; + if let Some(config_source) = config_toml_plugin_source { + return Err(CliError::Config(format!( + "plugin config is defined in both {} and {}; choose one source", + config_source.display(), + format_paths(&plugin_toml.sources) + ))); + } + gateway.plugin_config = Some(plugin_toml.value); + Ok(()) +} + +fn apply_cli_plugin_config(config: &mut GatewayConfig, value: &str) -> Result<(), CliError> { + if config.plugin_config.is_some() { + return Err(CliError::Config( + "plugin config is defined by both --plugin-config and file configuration; choose one source".into(), + )); + } + config.plugin_config = Some(parse_json_option("plugin config", value)?); + Ok(()) +} + // Applies configured agent commands and Cursor's temporary-hook behavior. Cursor's // `patch_restore_hooks` flag is intentionally tri-state in file config so omitted values preserve // the safe default while explicit `false` disables temporary hook mutation. @@ -886,6 +1032,74 @@ fn merge_toml(left: &mut toml::Value, right: toml::Value) { } } +// Plugin TOML uses normal recursive TOML merging except for the top-level components array. Each +// component is keyed by `kind`, so project/user plugin.toml files can add distinct plugin kinds or +// override one plugin kind without restating every other component. +fn merge_plugin_toml(left: &mut toml::Value, right: toml::Value) { + match (left, right) { + (toml::Value::Table(left), toml::Value::Table(right)) => { + for (key, value) in right { + match (key.as_str(), left.get_mut(&key)) { + ("components", Some(existing)) => merge_plugin_components(existing, value), + (_, Some(existing)) => merge_toml(existing, value), + _ => { + left.insert(key, value); + } + } + } + } + (left, right) => *left = right, + } +} + +fn merge_plugin_components(left: &mut toml::Value, right: toml::Value) { + let toml::Value::Array(left_components) = left else { + *left = right; + return; + }; + let toml::Value::Array(right_components) = right else { + *left = right; + return; + }; + + for component in right_components { + let Some(kind) = component_kind(&component).map(str::to_owned) else { + left_components.push(component); + continue; + }; + if let Some(existing) = left_components + .iter_mut() + .find(|candidate| component_kind(candidate) == Some(kind.as_str())) + { + merge_toml(existing, component); + } else { + left_components.push(component); + } + } +} + +fn component_kind(component: &toml::Value) -> Option<&str> { + component + .as_table() + .and_then(|table| table.get("kind")) + .and_then(toml::Value::as_str) +} + +fn has_config_toml_plugin_config(value: &toml::Value) -> bool { + value + .get("plugins") + .and_then(|plugins| plugins.get("config")) + .is_some() +} + +fn format_paths(paths: &[PathBuf]) -> String { + paths + .iter() + .map(|path| path.display().to_string()) + .collect::>() + .join(", ") +} + // Parses JSON-valued CLI options into runtime metadata/config values and labels errors with the // user-facing option name so callers can report which structured argument was malformed. fn parse_json_option(name: &str, value: &str) -> Result { diff --git a/crates/cli/src/server.rs b/crates/cli/src/server.rs index b6de3c547..5e90bb4e4 100644 --- a/crates/cli/src/server.rs +++ b/crates/cli/src/server.rs @@ -7,6 +7,7 @@ use axum::extract::State; use axum::http::HeaderMap; use axum::routing::{get, post}; use axum::{Json, Router}; +use nemo_flow::plugin::{PluginConfig, clear_plugin_configuration, initialize_plugins}; use reqwest::Client; use serde_json::Value; use tokio::net::TcpListener; @@ -64,20 +65,26 @@ pub(crate) async fn serve_listener( config: GatewayConfig, shutdown: Option>, ) -> Result<(), CliError> { + let plugin_activation = PluginActivation::initialize(config.plugin_config.clone()).await?; let app = router(config); - match shutdown { + let serve_result = match shutdown { Some(receiver) => { axum::serve(listener, app) .with_graceful_shutdown(async { let _ = receiver.await; }) - .await?; + .await } - None => { - axum::serve(listener, app).await?; + None => axum::serve(listener, app).await, + }; + let clear_result = plugin_activation.clear(); + if let Err(serve_error) = serve_result { + if let Err(clear_error) = clear_result { + eprintln!("plugin teardown failed after server error: {clear_error}"); } + return Err(serve_error.into()); } - Ok(()) + clear_result } /// Builds the gateway HTTP router and shared state. @@ -118,6 +125,42 @@ async fn healthz() -> Json { Json(serde_json::json!({ "status": "ok" })) } +struct PluginActivation { + active: bool, +} + +impl PluginActivation { + async fn initialize(config: Option) -> Result { + let Some(config) = config else { + return Ok(Self { active: false }); + }; + let plugin_config: PluginConfig = serde_json::from_value(config) + .map_err(|error| CliError::Config(format!("invalid plugin config: {error}")))?; + initialize_plugins(plugin_config) + .await + .map_err(|error| CliError::Config(format!("plugin activation failed: {error}")))?; + Ok(Self { active: true }) + } + + fn clear(mut self) -> Result<(), CliError> { + if self.active { + self.active = false; + clear_plugin_configuration() + .map_err(|error| CliError::Config(format!("plugin teardown failed: {error}")))?; + } + Ok(()) + } +} + +impl Drop for PluginActivation { + fn drop(&mut self) { + if self.active { + let _ = clear_plugin_configuration(); + self.active = false; + } + } +} + // Normalizes a Codex hook payload, applies all resulting events before responding, and returns the // adapter's pass-through response body so hook delivery stays causally ordered with observability. async fn codex_hook( diff --git a/crates/cli/tests/coverage/config_tests.rs b/crates/cli/tests/coverage/config_tests.rs index e3b37daa5..9c0d39b89 100644 --- a/crates/cli/tests/coverage/config_tests.rs +++ b/crates/cli/tests/coverage/config_tests.rs @@ -201,6 +201,10 @@ command = "hermes --yolo chat" Some("http://otel") ); assert_eq!(resolved.gateway.metadata, Some(json!({ "team": "obs" }))); + assert_eq!( + resolved.gateway.plugin_config, + Some(json!({ "components": [] })) + ); assert_eq!( resolved.agents.codex.command.as_deref(), Some("codex --approval-mode never") @@ -212,6 +216,258 @@ command = "hermes --yolo chat" assert!(!resolved.agents.cursor.patch_restore_hooks); } +#[test] +fn explicit_plugin_toml_maps_root_plugin_config() { + let temp = tempfile::tempdir().unwrap(); + let config_path = temp.path().join("config.toml"); + std::fs::write( + &config_path, + r#" +[upstream] +openai_base_url = "http://openai" +"#, + ) + .unwrap(); + std::fs::write( + temp.path().join("plugin.toml"), + r#" +version = 1 + +[[components]] +kind = "observability" +enabled = true + +[components.config] +version = 1 + +[components.config.atof] +enabled = true +output_directory = "atof" +filename = "events.jsonl" +mode = "overwrite" +"#, + ) + .unwrap(); + let command = RunCommand { + agent: Some(CodingAgent::Codex), + config: Some(config_path), + openai_base_url: None, + anthropic_base_url: None, + atif_dir: None, + atof_dir: None, + openinference_endpoint: None, + session_metadata: None, + plugin_config: None, + dry_run: false, + print: false, + command: vec!["codex".into()], + }; + + let resolved = resolve_run_config(&command, None).unwrap(); + + assert_eq!( + resolved.gateway.plugin_config, + Some(json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "output_directory": "atof", + "filename": "events.jsonl", + "mode": "overwrite" + } + } + } + ] + })) + ); +} + +#[test] +fn plugin_toml_path_resolution_tracks_config_scope() { + let temp = tempfile::tempdir().unwrap(); + let explicit = temp.path().join("custom-config.toml"); + assert_eq!( + plugin_config_paths(Some(&explicit)), + vec![temp.path().join("plugin.toml")] + ); + + let project = temp.path().join("workspace"); + let nested = project.join("a/b/c"); + std::fs::create_dir_all(project.join(".nemo-flow")).unwrap(); + std::fs::create_dir_all(&nested).unwrap(); + let plugin_path = project.join(".nemo-flow/plugin.toml"); + std::fs::write(&plugin_path, "version = 1").unwrap(); + let user_config = temp.path().join("xdg/nemo-flow"); + + assert_eq!(find_project_plugin_config(&nested), Some(plugin_path)); + assert_eq!( + implicit_plugin_config_paths(Some(&nested), Some(user_config.clone())), + vec![ + PathBuf::from("/etc/nemo-flow/plugin.toml"), + project.join(".nemo-flow/plugin.toml"), + user_config.join("plugin.toml"), + ] + ); +} + +#[test] +fn discovered_plugin_toml_upserts_components_by_kind() { + let temp = tempfile::tempdir().unwrap(); + let project_plugin = temp.path().join("project-plugin.toml"); + let user_plugin = temp.path().join("user-plugin.toml"); + std::fs::write( + &project_plugin, + r#" +version = 1 + +[[components]] +kind = "observability" +enabled = true + +[components.config] +version = 1 + +[components.config.atof] +enabled = true +filename = "project.jsonl" + +[[components]] +kind = "adaptive" +enabled = true + +[components.config] +mode = "project-only" +"#, + ) + .unwrap(); + std::fs::write( + &user_plugin, + r#" +version = 1 + +[[components]] +kind = "observability" +enabled = true + +[components.config] +version = 1 + +[components.config.atof] +enabled = true + +[components.config.atif] +enabled = true +filename_template = "user-{session_id}.json" + +[[components]] +kind = "custom" +enabled = true + +[components.config] +source = "user" +"#, + ) + .unwrap(); + + let resolved = load_plugin_toml_config_from_paths(vec![project_plugin, user_plugin]).unwrap(); + + assert_eq!( + resolved.map(|config| config.value), + Some(json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "filename": "project.jsonl" + }, + "atif": { + "enabled": true, + "filename_template": "user-{session_id}.json" + } + } + }, + { + "kind": "adaptive", + "enabled": true, + "config": { + "mode": "project-only" + } + }, + { + "kind": "custom", + "enabled": true, + "config": { + "source": "user" + } + } + ] + })) + ); +} + +#[test] +fn plugin_toml_conflicts_with_config_toml_plugins_config() { + let temp = tempfile::tempdir().unwrap(); + let config_path = temp.path().join("config.toml"); + std::fs::write( + &config_path, + r#" +[plugins] +config = { version = 1, components = [] } +"#, + ) + .unwrap(); + std::fs::write(temp.path().join("plugin.toml"), "version = 1\n").unwrap(); + let args = ServerArgs { + config: Some(config_path), + ..ServerArgs::default() + }; + + let error = resolve_server_config(&args).unwrap_err().to_string(); + + assert!(error.contains("plugin config is defined in both")); + assert!(error.contains("config.toml")); + assert!(error.contains("plugin.toml")); +} + +#[test] +fn cli_plugin_config_conflicts_with_file_plugin_config() { + let temp = tempfile::tempdir().unwrap(); + let config_path = temp.path().join("config.toml"); + std::fs::write(&config_path, "").unwrap(); + std::fs::write(temp.path().join("plugin.toml"), "version = 1\n").unwrap(); + let command = RunCommand { + agent: Some(CodingAgent::Codex), + config: Some(config_path), + openai_base_url: None, + anthropic_base_url: None, + atif_dir: None, + atof_dir: None, + openinference_endpoint: None, + session_metadata: None, + plugin_config: Some(r#"{"version":1,"components":[]}"#.into()), + dry_run: false, + print: false, + command: vec!["codex".into()], + }; + + let error = resolve_run_config(&command, None).unwrap_err().to_string(); + + assert!(error.contains("--plugin-config")); + assert!(error.contains("file configuration")); +} + #[test] fn cli_run_overrides_config_values() { let temp = tempfile::tempdir().unwrap(); @@ -292,6 +548,35 @@ openai_base_url = "http://file-openai" assert_eq!(resolved.gateway.openai_base_url, "http://top-level-openai"); } +#[test] +fn run_plugin_config_overrides_inherited_top_level_plugin_config() { + let server = ServerArgs { + plugin_config: Some(r#"{"components":["top-level"]}"#.into()), + ..ServerArgs::default() + }; + let command = RunCommand { + agent: Some(CodingAgent::Codex), + config: None, + openai_base_url: None, + anthropic_base_url: None, + atif_dir: None, + atof_dir: None, + openinference_endpoint: None, + session_metadata: None, + plugin_config: Some(r#"{"components":["run"]}"#.into()), + dry_run: false, + print: false, + command: vec!["codex".into()], + }; + + let resolved = resolve_run_config(&command, Some(&server)).unwrap(); + + assert_eq!( + resolved.gateway.plugin_config, + Some(json!({ "components": ["run"] })) + ); +} + #[test] fn server_resolution_applies_all_server_overrides() { let args = ServerArgs { @@ -302,6 +587,7 @@ fn server_resolution_applies_all_server_overrides() { atif_dir: Some(PathBuf::from("cli-atif")), atof_dir: None, openinference_endpoint: Some("http://cli-otel".into()), + plugin_config: Some(r#"{"version":1,"components":[]}"#.into()), }; let resolved = resolve_server_config(&args).unwrap(); @@ -317,6 +603,11 @@ fn server_resolution_applies_all_server_overrides() { resolved.gateway.exporters.openinference.endpoint.as_deref(), Some("http://cli-otel") ); + assert_eq!( + resolved.gateway.plugin_config, + Some(json!({ "version": 1, "components": [] })) + ); + assert!(args.requested_daemon_mode()); } #[test] @@ -379,6 +670,18 @@ fn malformed_shared_config_reports_context() { let error = resolve_server_config(&args).unwrap_err().to_string(); assert!(error.contains("invalid gateway configuration shape")); + + let plugin_config = temp.path().join("config-with-invalid-plugin.toml"); + std::fs::write(&plugin_config, "").unwrap(); + std::fs::write(temp.path().join("plugin.toml"), "version = [").unwrap(); + let args = ServerArgs { + config: Some(plugin_config), + ..ServerArgs::default() + }; + + let error = resolve_server_config(&args).unwrap_err().to_string(); + + assert!(error.contains("invalid plugin TOML")); } #[test] diff --git a/crates/cli/tests/coverage/server_tests.rs b/crates/cli/tests/coverage/server_tests.rs index 6ce34c826..99070b99d 100644 --- a/crates/cli/tests/coverage/server_tests.rs +++ b/crates/cli/tests/coverage/server_tests.rs @@ -1,14 +1,24 @@ // SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + use axum::body::Body; use axum::http::{Request, StatusCode, header}; use axum::response::IntoResponse; use bytes::Bytes; use futures_util::stream; use http_body_util::BodyExt; -use serde_json::{Value, json}; +use nemo_flow::plugin::{ + ConfigDiagnostic, Plugin, PluginRegistration, PluginRegistrationContext, deregister_plugin, + register_plugin, +}; +use serde_json::{Map, Value, json}; use tokio::net::TcpListener; +use tokio::sync::oneshot; use tokio::task::JoinHandle; use tower::ServiceExt; @@ -16,6 +26,42 @@ use super::*; use crate::config::ExportersConfig; use crate::error::CliError; +static PLUGIN_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +const GENERIC_TEST_PLUGIN_KIND: &str = "cli-test-generic-plugin"; +static GENERIC_TEST_PLUGIN_REGISTRATIONS: AtomicUsize = AtomicUsize::new(0); +static GENERIC_TEST_PLUGIN_DEREGISTRATIONS: AtomicUsize = AtomicUsize::new(0); + +struct GenericTestPlugin; + +impl Plugin for GenericTestPlugin { + fn plugin_kind(&self) -> &str { + GENERIC_TEST_PLUGIN_KIND + } + + fn validate(&self, _plugin_config: &Map) -> Vec { + vec![] + } + + fn register<'a>( + &'a self, + _plugin_config: &Map, + ctx: &'a mut PluginRegistrationContext, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + GENERIC_TEST_PLUGIN_REGISTRATIONS.fetch_add(1, Ordering::SeqCst); + ctx.add_registration(PluginRegistration::new( + "plugin", + GENERIC_TEST_PLUGIN_KIND, + Box::new(|| { + GENERIC_TEST_PLUGIN_DEREGISTRATIONS.fetch_add(1, Ordering::SeqCst); + Ok(()) + }), + )); + Ok(()) + }) + } +} + struct TestServer { url: String, handle: JoinHandle<()>, @@ -91,6 +137,256 @@ async fn healthz_returns_ok() { assert_eq!(body, json!({ "status": "ok" })); } +#[tokio::test] +async fn serve_listener_activates_plugin_config_and_clears_on_shutdown() { + let _guard = PLUGIN_TEST_LOCK.lock().await; + let _ = nemo_flow::plugin::clear_plugin_configuration(); + + let temp = tempfile::tempdir().unwrap(); + let atof_dir = temp.path().join("atof"); + let atif_dir = temp.path().join("atif"); + std::fs::create_dir_all(&atof_dir).unwrap(); + std::fs::create_dir_all(&atif_dir).unwrap(); + let mut config = test_config(); + config.plugin_config = Some(json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "output_directory": atof_dir, + "filename": "events.jsonl", + "mode": "overwrite" + }, + "atif": { + "enabled": true, + "output_directory": atif_dir, + "filename_template": "trajectory-{session_id}.json" + } + } + } + ] + })); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let url = format!("http://{address}"); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let handle = + tokio::spawn(async move { serve_listener(listener, config, Some(shutdown_rx)).await }); + + wait_for_gateway(&url).await; + assert!(nemo_flow::plugin::active_plugin_report().is_some()); + + let client = reqwest::Client::new(); + for hook_event_name in ["on_session_start", "on_session_finalize"] { + let response = client + .post(format!("{url}/hooks/hermes")) + .json(&json!({ + "session_id": "plugin-bridge-session", + "hook_event_name": hook_event_name + })) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + } + + shutdown_tx.send(()).unwrap(); + handle.await.unwrap().unwrap(); + assert!(nemo_flow::plugin::active_plugin_report().is_none()); + + let events = std::fs::read_to_string(temp.path().join("atof/events.jsonl")).unwrap(); + assert!( + events.lines().count() >= 2, + "expected ATOF lifecycle events, got {events:?}" + ); + let atif_files = std::fs::read_dir(temp.path().join("atif")) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(atif_files.len(), 1); + let trajectory: Value = + serde_json::from_slice(&std::fs::read(atif_files[0].path()).unwrap()).unwrap(); + assert!( + trajectory["extra"]["observed_events"] + .as_array() + .is_some_and(|events| events.len() >= 2) + ); +} + +#[tokio::test] +async fn serve_listener_observability_plugin_records_non_hermes_hooks() { + let _guard = PLUGIN_TEST_LOCK.lock().await; + let _ = nemo_flow::plugin::clear_plugin_configuration(); + + let temp = tempfile::tempdir().unwrap(); + let atof_dir = temp.path().join("atof"); + std::fs::create_dir_all(&atof_dir).unwrap(); + let mut config = test_config(); + config.plugin_config = Some(json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "output_directory": atof_dir, + "filename": "events.jsonl", + "mode": "overwrite" + } + } + } + ] + })); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let url = format!("http://{address}"); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let handle = + tokio::spawn(async move { serve_listener(listener, config, Some(shutdown_rx)).await }); + + wait_for_gateway(&url).await; + let client = reqwest::Client::new(); + for (path, session_id, start_event, end_event) in [ + ( + "/hooks/codex", + "codex-plugin-session", + "sessionStart", + "sessionEnd", + ), + ( + "/hooks/claude-code", + "claude-plugin-session", + "SessionStart", + "SessionEnd", + ), + ] { + for hook_event_name in [start_event, end_event] { + let response = client + .post(format!("{url}{path}")) + .json(&json!({ + "session_id": session_id, + "hook_event_name": hook_event_name + })) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + } + } + + shutdown_tx.send(()).unwrap(); + handle.await.unwrap().unwrap(); + assert!(nemo_flow::plugin::active_plugin_report().is_none()); + + let events = std::fs::read_to_string(temp.path().join("atof/events.jsonl")).unwrap(); + let agent_starts = events + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|event| { + event["kind"] == "scope" + && event["scope_category"] == "start" + && event["category"] == "agent" + }) + .filter_map(|event| event["name"].as_str().map(ToOwned::to_owned)) + .collect::>(); + assert!(agent_starts.contains(&"codex".to_string())); + assert!(agent_starts.contains(&"claude-code".to_string())); +} + +#[tokio::test] +async fn serve_listener_activates_any_registered_plugin_kind() { + let _guard = PLUGIN_TEST_LOCK.lock().await; + let _ = nemo_flow::plugin::clear_plugin_configuration(); + let _ = deregister_plugin(GENERIC_TEST_PLUGIN_KIND); + GENERIC_TEST_PLUGIN_REGISTRATIONS.store(0, Ordering::SeqCst); + GENERIC_TEST_PLUGIN_DEREGISTRATIONS.store(0, Ordering::SeqCst); + register_plugin(Arc::new(GenericTestPlugin)).unwrap(); + + let mut config = test_config(); + config.plugin_config = Some(json!({ + "version": 1, + "components": [ + { + "kind": GENERIC_TEST_PLUGIN_KIND, + "enabled": true, + "config": {} + } + ] + })); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let url = format!("http://{address}"); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let handle = + tokio::spawn(async move { serve_listener(listener, config, Some(shutdown_rx)).await }); + + wait_for_gateway(&url).await; + assert_eq!(GENERIC_TEST_PLUGIN_REGISTRATIONS.load(Ordering::SeqCst), 1); + + let response = reqwest::Client::new() + .post(format!("{url}/hooks/codex")) + .json(&json!({ + "session_id": "generic-plugin-session", + "hook_event_name": "sessionStart" + })) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + shutdown_tx.send(()).unwrap(); + handle.await.unwrap().unwrap(); + assert_eq!( + GENERIC_TEST_PLUGIN_DEREGISTRATIONS.load(Ordering::SeqCst), + 1 + ); + assert!(nemo_flow::plugin::active_plugin_report().is_none()); + let _ = deregister_plugin(GENERIC_TEST_PLUGIN_KIND); +} + +#[tokio::test] +async fn serve_listener_rejects_invalid_plugin_config() { + let _guard = PLUGIN_TEST_LOCK.lock().await; + let _ = nemo_flow::plugin::clear_plugin_configuration(); + + let mut config = test_config(); + config.plugin_config = Some(json!({ + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": true, + "config": { + "version": 1, + "atof": { + "enabled": true, + "mode": "invalid" + } + } + } + ] + })); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (_shutdown_tx, shutdown_rx) = oneshot::channel(); + let error = serve_listener(listener, config, Some(shutdown_rx)) + .await + .unwrap_err(); + + assert!(error.to_string().contains("ATOF mode")); + assert!(nemo_flow::plugin::active_plugin_report().is_none()); +} + #[tokio::test] async fn gateway_errors_render_structured_json_responses() { let response = CliError::InvalidPayload("bad input".into()).into_response(); @@ -399,6 +695,19 @@ async fn models_route_forwards_get_requests() { assert_eq!(body["authorization"], json!("Bearer test")); } +async fn wait_for_gateway(url: &str) { + let client = reqwest::Client::new(); + for _ in 0..50 { + if let Ok(response) = client.get(format!("{url}/healthz")).send().await + && response.status().is_success() + { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("gateway did not become healthy at {url}"); +} + async fn spawn_upstream(streaming: bool) -> TestServer { async fn chat(headers: HeaderMap, body: Bytes) -> impl IntoResponse { let payload: Value = serde_json::from_slice(&body).unwrap(); diff --git a/docs/export-observability-data/observability-plugin.md b/docs/export-observability-data/observability-plugin.md index bf8d72529..fcaccca7d 100644 --- a/docs/export-observability-data/observability-plugin.md +++ b/docs/export-observability-data/observability-plugin.md @@ -59,6 +59,109 @@ names from the plugin namespace: The active runtime names include the component namespace prefix used by the plugin system. +## CLI Gateway `plugin.toml` + +The `nemo-flow` CLI gateway can activate one process-level plugin config at +startup. Define it with one of these sources: + +- `--plugin-config` JSON on the command line. +- `[plugins].config` in `config.toml`. +- `plugin.toml` next to the resolved `config.toml`, or in the same discovered + system, project, and user scopes as `config.toml`. + +When multiple discovered `plugin.toml` files are present, the gateway loads +them from lowest to highest precedence: + +1. System: `/etc/nemo-flow/plugin.toml` +2. Project: `.nemo-flow/plugin.toml` +3. User: `$XDG_CONFIG_HOME/nemo-flow/plugin.toml`, or + `~/.config/nemo-flow/plugin.toml` + +Later files override earlier files. TOML tables merge recursively, so a +higher-precedence file can override one nested key while preserving sibling +keys from lower-precedence files. + +The top-level `[[components]]` array is merged by component `kind`. A +higher-precedence component with the same `kind` is merged into the lower +precedence component, and higher-precedence values win on conflicts. Components +with different `kind` values compose, so a project `observability` component and +a user `adaptive` component are both active in the effective config. + +Use only one source for plugin config. The gateway reports an error when +`plugin.toml`, `[plugins].config`, or `--plugin-config` are used together. + +`plugin.toml` uses the generic plugin config shape at the file root. The +example below shows every observability section; include only the sections you +want to configure. Missing sections behave like disabled sections. + +`version = 1` is recommended for clarity but not required. The root plugin +config version and observability component config version both default to `1` +when omitted; unsupported non-`1` versions fail validation by default. + +```toml +version = 1 + +[[components]] +kind = "observability" +enabled = true + +[components.config] +version = 1 + +[components.config.atof] +enabled = true +output_directory = "logs" +filename = "events.jsonl" +mode = "overwrite" + +[components.config.atif] +enabled = true +output_directory = "logs" +filename_template = "trajectory-{session_id}.json" + +[components.config.opentelemetry] +enabled = true +transport = "http_binary" +endpoint = "http://localhost:4318/v1/traces" +service_name = "nemo-flow" +service_namespace = "agent" +service_version = "0.2.0" +instrumentation_scope = "nemo-flow-observability" +timeout_millis = 3000 + +[components.config.opentelemetry.headers] +authorization = "Bearer " + +[components.config.opentelemetry.resource_attributes] +"deployment.environment" = "dev" +"service.instance.id" = "local" + +[components.config.openinference] +enabled = true +transport = "http_binary" +endpoint = "http://localhost:6006/v1/traces" +service_name = "nemo-flow" +service_namespace = "agent" +service_version = "0.2.0" +instrumentation_scope = "nemo-flow-openinference" +timeout_millis = 3000 + +[components.config.openinference.headers] +authorization = "Bearer " + +[components.config.openinference.resource_attributes] +"deployment.environment" = "dev" +"service.instance.id" = "local" + +[components.config.policy] +unknown_component = "warn" +unknown_field = "warn" +unsupported_value = "error" +``` + +The file format is generic. Other plugin kinds can use the same `components` +array when their plugin implementation is registered in the gateway process. + ## ATOF Section Use ATOF when you want the raw ATOF `0.1` event stream as JSONL.