diff --git a/Cargo.lock b/Cargo.lock index b06faa5a..8760f602 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "anyhow" version = "1.0.102" @@ -49,6 +55,7 @@ version = "0.1.0" dependencies = [ "async-trait", "futures", + "hashbrown 0.15.5", "serde", "serde_json", "serde_yaml", @@ -197,6 +204,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash", ] diff --git a/Cargo.toml b/Cargo.toml index 03fcb104..8ee43bc0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,3 +29,4 @@ tracing = "0.1" uuid = { version = "1", features = ["v4"] } paste = "1" futures = "0.3" +hashbrown = "0.15" diff --git a/crates/cpex-core/Cargo.toml b/crates/cpex-core/Cargo.toml index 4e0d4006..1a6d3351 100644 --- a/crates/cpex-core/Cargo.toml +++ b/crates/cpex-core/Cargo.toml @@ -25,3 +25,4 @@ thiserror = { workspace = true } tracing = { workspace = true } uuid = { workspace = true } futures = { workspace = true } +hashbrown = { workspace = true } diff --git a/crates/cpex-core/examples/README.md b/crates/cpex-core/examples/README.md new file mode 100644 index 00000000..9be92c2d --- /dev/null +++ b/crates/cpex-core/examples/README.md @@ -0,0 +1,43 @@ +# CPEX Core Examples + +## plugin_demo + +A complete end-to-end example showing how to build plugins, load config, and invoke hooks with the CPEX runtime. + +### What it demonstrates + +- **Defining hook types and payloads** — `ToolPreInvoke` and `ToolPostInvoke` hooks with a shared `ToolInvokePayload` +- **Building plugins** — three plugins (`IdentityResolver`, `PiiGuard`, `AuditLogger`) implementing `Plugin` + `HookHandler` for different hook types +- **Multi-hook registration** — a single plugin instance (e.g., `IdentityResolver`) registered for multiple hooks (`tool_pre_invoke` and `tool_post_invoke`) via the factory pattern +- **Plugin factories** — `PluginFactory` implementations that create plugin instances and wire up typed handler adapters +- **YAML config loading** — `plugin_demo.yaml` declares plugins, policy groups, and routing rules +- **Policy groups and tag-based routing** — the `pii` policy group activates `PiiGuard` only for tools tagged with `pii` +- **Route resolution** — exact tool matches, wildcard catch-all, tag-driven plugin selection +- **PluginContext** — `global_state` used to pass PII clearance between hooks, `local_state` for per-plugin scratch data +- **BackgroundTasks** — fire-and-forget plugins (`AuditLogger`) spawn background tasks; `wait_for_background_tasks()` awaits them +- **PluginContextTable** — context table threaded from pre-invoke to post-invoke to preserve plugin state + +### Running + +From the workspace root: + +``` +cargo run --example plugin_demo +``` + +### Scenarios + +The demo runs five scenarios against three registered plugins: + +| Scenario | Tool | User | Outcome | +|----------|------|------|---------| +| 1 | get_compensation | alice (no clearance) | DENIED by pii-guard | +| 2 | get_compensation | alice (with clearance) | ALLOWED, then post-invoke fires | +| 3 | list_departments | bob | ALLOWED (no PII tag, pii-guard skipped) | +| 4 | some_other_tool | charlie | ALLOWED (wildcard route) | +| 5 | list_departments | (empty) | DENIED by identity-resolver | + +### Files + +- `plugin_demo.rs` — Rust source with plugins, factories, and main +- `plugin_demo.yaml` — YAML config with plugins, policy groups, and routes diff --git a/crates/cpex-core/examples/plugin_demo.rs b/crates/cpex-core/examples/plugin_demo.rs new file mode 100644 index 00000000..8e5fb602 --- /dev/null +++ b/crates/cpex-core/examples/plugin_demo.rs @@ -0,0 +1,394 @@ +// CPEX Plugin Demo +// +// Demonstrates how to: +// 1. Define hook types and payloads +// 2. Build plugins that implement HookHandler +// 3. Create plugin factories for config-driven loading +// 4. Load a YAML config with routing rules +// 5. Invoke hooks with MetaExtension for route resolution +// +// Run with: cargo run --example plugin_demo + +use std::sync::Arc; + +use async_trait::async_trait; +use cpex_core::context::PluginContext; +use cpex_core::error::{PluginError, PluginViolation}; +use cpex_core::executor::PipelineResult; +use cpex_core::factory::{PluginFactory, PluginInstance}; +use cpex_core::hooks::adapter::TypedHandlerAdapter; +use cpex_core::hooks::payload::{Extensions, FilteredExtensions, MetaExtension}; +use cpex_core::hooks::trait_def::{HookHandler, HookTypeDef, PluginResult}; +use cpex_core::manager::PluginManager; +use cpex_core::plugin::{Plugin, PluginConfig}; + +// --------------------------------------------------------------------------- +// Step 1: Define a payload and hook type +// --------------------------------------------------------------------------- + +/// The payload carried through the tool_pre_invoke hook. +#[derive(Debug, Clone)] +struct ToolInvokePayload { + tool_name: String, + user: String, + arguments: String, +} +cpex_core::impl_plugin_payload!(ToolInvokePayload); + +/// Hook type for tool_pre_invoke — runs before a tool executes. +struct ToolPreInvoke; +impl HookTypeDef for ToolPreInvoke { + type Payload = ToolInvokePayload; + type Result = PluginResult; + const NAME: &'static str = "tool_pre_invoke"; +} + +/// Hook type for tool_post_invoke — runs after a tool executes. +struct ToolPostInvoke; +impl HookTypeDef for ToolPostInvoke { + type Payload = ToolInvokePayload; + type Result = PluginResult; + const NAME: &'static str = "tool_post_invoke"; +} + +// --------------------------------------------------------------------------- +// Step 2: Build plugins +// --------------------------------------------------------------------------- + +/// Identity resolver — checks that a user is present. +struct IdentityResolver { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for IdentityResolver { + fn config(&self) -> &PluginConfig { &self.cfg } + async fn initialize(&self) -> Result<(), PluginError> { + println!(" [identity-resolver] initialized"); + Ok(()) + } + async fn shutdown(&self) -> Result<(), PluginError> { + println!(" [identity-resolver] shutdown"); + Ok(()) + } +} + +impl HookHandler for IdentityResolver { + fn handle( + &self, + payload: &ToolInvokePayload, + _extensions: &FilteredExtensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + if payload.user.is_empty() { + println!(" [identity-resolver] DENIED: no user identity"); + return PluginResult::deny( + PluginViolation::new("no_identity", "User identity is required"), + ); + } + println!(" [identity-resolver] OK: user '{}' identified", payload.user); + PluginResult::allow() + } +} + +impl HookHandler for IdentityResolver { + fn handle( + &self, + payload: &ToolInvokePayload, + _extensions: &FilteredExtensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + println!(" [identity-resolver] post-invoke: user '{}' completed '{}'", + payload.user, payload.tool_name); + PluginResult::allow() + } +} + +/// PII guard — blocks access to sensitive tools without clearance. +struct PiiGuard { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for PiiGuard { + fn config(&self) -> &PluginConfig { &self.cfg } + // initialize() and shutdown() use defaults — no setup needed +} + +impl HookHandler for PiiGuard { + fn handle( + &self, + payload: &ToolInvokePayload, + _extensions: &FilteredExtensions, + ctx: &mut PluginContext, + ) -> PluginResult { + // Check if the user has PII clearance (simulated via context) + let has_clearance = ctx + .get_global("pii_clearance") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + + if !has_clearance { + println!(" [pii-guard] DENIED: user '{}' lacks PII clearance for '{}'", + payload.user, payload.tool_name); + return PluginResult::deny( + PluginViolation::new("pii_access_denied", "PII clearance required"), + ); + } + + println!(" [pii-guard] OK: user '{}' has PII clearance", payload.user); + PluginResult::allow() + } +} + +/// Audit logger — logs all tool invocations (fire-and-forget). +struct AuditLogger { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for AuditLogger { + fn config(&self) -> &PluginConfig { &self.cfg } + // initialize() and shutdown() use defaults — no setup needed +} + +impl HookHandler for AuditLogger { + fn handle( + &self, + payload: &ToolInvokePayload, + _extensions: &FilteredExtensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + println!(" [audit-logger] LOG: user='{}' tool='{}' args='{}'", + payload.user, payload.tool_name, payload.arguments); + PluginResult::allow() + } +} + +impl HookHandler for AuditLogger { + fn handle( + &self, + payload: &ToolInvokePayload, + _extensions: &FilteredExtensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + println!(" [audit-logger] LOG: post-invoke user='{}' tool='{}'", + payload.user, payload.tool_name); + PluginResult::allow() + } +} + +// --------------------------------------------------------------------------- +// Step 3: Create plugin factories +// --------------------------------------------------------------------------- + +struct IdentityFactory; +impl PluginFactory for IdentityFactory { + fn create(&self, config: &PluginConfig) -> Result { + let plugin = Arc::new(IdentityResolver { cfg: config.clone() }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![ + ("tool_pre_invoke", Arc::new(TypedHandlerAdapter::::new(plugin.clone()))), + ("tool_post_invoke", Arc::new(TypedHandlerAdapter::::new(plugin))), + ], + }) + } +} + +struct PiiGuardFactory; +impl PluginFactory for PiiGuardFactory { + fn create(&self, config: &PluginConfig) -> Result { + let plugin = Arc::new(PiiGuard { cfg: config.clone() }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![ + ("tool_pre_invoke", Arc::new(TypedHandlerAdapter::::new(plugin))), + ], + }) + } +} + +struct AuditLoggerFactory; +impl PluginFactory for AuditLoggerFactory { + fn create(&self, config: &PluginConfig) -> Result { + let plugin = Arc::new(AuditLogger { cfg: config.clone() }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![ + ("tool_pre_invoke", Arc::new(TypedHandlerAdapter::::new(plugin.clone()))), + ("tool_post_invoke", Arc::new(TypedHandlerAdapter::::new(plugin))), + ], + }) + } +} + +// --------------------------------------------------------------------------- +// Step 4: Build extensions with MetaExtension for routing +// --------------------------------------------------------------------------- + +fn make_tool_extensions(tool_name: &str, tags: &[&str]) -> Extensions { + Extensions { + meta: Some(MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some(tool_name.into()), + tags: tags.iter().map(|s| s.to_string()).collect(), + ..Default::default() + }), + ..Default::default() + } +} + +// --------------------------------------------------------------------------- +// Helper to print results +// --------------------------------------------------------------------------- + +fn print_result(_label: &str, result: &PipelineResult) { + if result.continue_processing { + println!(" Result: ALLOWED"); + } else { + let violation = result.violation.as_ref().unwrap(); + println!(" Result: DENIED by '{}' — {} [{}]", + violation.plugin_name.as_deref().unwrap_or("unknown"), + violation.reason, + violation.code, + ); + } + println!(); +} + +// --------------------------------------------------------------------------- +// Step 5: Main — load config, invoke hooks, see results +// --------------------------------------------------------------------------- + +#[tokio::main] +async fn main() { + println!("=== CPEX Plugin Demo ===\n"); + + // --- Load config from YAML file --- + let config_path = "crates/cpex-core/examples/plugin_demo.yaml"; + println!("--- Loading config from {} ---\n", config_path); + let yaml = std::fs::read_to_string(config_path) + .unwrap_or_else(|e| panic!("Failed to read {}: {}", config_path, e)); + let cpex_config = cpex_core::config::parse_config(&yaml).unwrap(); + + let mut mgr = PluginManager::default(); + mgr.register_factory("builtin/identity", Box::new(IdentityFactory)); + mgr.register_factory("builtin/pii", Box::new(PiiGuardFactory)); + mgr.register_factory("builtin/audit", Box::new(AuditLoggerFactory)); + mgr.load_config(cpex_config).unwrap(); + + println!("\n--- Initializing plugins ---\n"); + mgr.initialize().await.unwrap(); + + println!("\nPlugins loaded: {}", mgr.plugin_count()); + println!("Hooks registered: tool_pre_invoke={}, tool_post_invoke={}\n", + mgr.has_hooks_for("tool_pre_invoke"), + mgr.has_hooks_for("tool_post_invoke"), + ); + + // --- Scenario 1: PII tool without clearance --- + println!("=== Scenario 1: get_compensation (PII tool, no clearance) ===\n"); + let payload = ToolInvokePayload { + tool_name: "get_compensation".into(), + user: "alice".into(), + arguments: "employee_id=42".into(), + }; + let ext = make_tool_extensions("get_compensation", &[]); + let (result, bg) = mgr.invoke::( + payload, ext, None, + ).await; + print_result("get_compensation (no clearance)", &result); + // Wait for any fire-and-forget tasks + bg.wait_for_background_tasks().await; + + // --- Scenario 2: PII tool with clearance --- + println!("=== Scenario 2: get_compensation (PII tool, with clearance) ===\n"); + let payload = ToolInvokePayload { + tool_name: "get_compensation".into(), + user: "alice".into(), + arguments: "employee_id=42".into(), + }; + let ext = make_tool_extensions("get_compensation", &[]); + // Simulate clearance by pre-populating global_state + // (In production, an earlier hook would set this from a token claim) + let mut global_state = std::collections::HashMap::new(); + global_state.insert( + "pii_clearance".into(), + serde_json::Value::Bool(true), + ); + // Pass global state via context table + let mut ctx_table = cpex_core::context::PluginContextTable::new(); + // We need to seed global_state — create a dummy entry + ctx_table.insert( + "__seed__".into(), + cpex_core::context::PluginContext::with_global_state(global_state), + ); + let (result, bg) = mgr.invoke::( + payload, ext, Some(ctx_table), + ).await; + print_result("get_compensation (with clearance)", &result); + bg.wait_for_background_tasks().await; + + // Now call post-invoke — threads the context table from pre-invoke + println!(" --- post-invoke for get_compensation ---\n"); + let payload = ToolInvokePayload { + tool_name: "get_compensation".into(), + user: "alice".into(), + arguments: "employee_id=42".into(), + }; + let ext = make_tool_extensions("get_compensation", &[]); + let (post_result, bg) = mgr.invoke::( + payload, ext, Some(result.context_table), + ).await; + print_result("get_compensation post-invoke", &post_result); + bg.wait_for_background_tasks().await; + + // --- Scenario 3: Non-PII tool --- + println!("=== Scenario 3: list_departments (non-PII tool) ===\n"); + let payload = ToolInvokePayload { + tool_name: "list_departments".into(), + user: "bob".into(), + arguments: "".into(), + }; + let ext = make_tool_extensions("list_departments", &[]); + let (result, bg) = mgr.invoke::( + payload, ext, None, + ).await; + print_result("list_departments", &result); + bg.wait_for_background_tasks().await; + + // --- Scenario 4: Unknown tool (wildcard route) --- + println!("=== Scenario 4: some_other_tool (wildcard route) ===\n"); + let payload = ToolInvokePayload { + tool_name: "some_other_tool".into(), + user: "charlie".into(), + arguments: "foo=bar".into(), + }; + let ext = make_tool_extensions("some_other_tool", &[]); + let (result, bg) = mgr.invoke::( + payload, ext, None, + ).await; + print_result("some_other_tool (wildcard)", &result); + bg.wait_for_background_tasks().await; + + // --- Scenario 5: No user identity --- + println!("=== Scenario 5: list_departments (no user identity) ===\n"); + let payload = ToolInvokePayload { + tool_name: "list_departments".into(), + user: "".into(), + arguments: "".into(), + }; + let ext = make_tool_extensions("list_departments", &[]); + let (result, bg) = mgr.invoke::( + payload, ext, None, + ).await; + print_result("list_departments (no user)", &result); + bg.wait_for_background_tasks().await; + + // --- Shutdown --- + println!("--- Shutting down ---\n"); + mgr.shutdown().await; + + println!("=== Demo complete ==="); +} diff --git a/crates/cpex-core/examples/plugin_demo.yaml b/crates/cpex-core/examples/plugin_demo.yaml new file mode 100644 index 00000000..9e3dd610 --- /dev/null +++ b/crates/cpex-core/examples/plugin_demo.yaml @@ -0,0 +1,59 @@ +# CPEX Plugin Demo Configuration +# +# Three plugins, policy groups with tag-based activation, +# and routes that map tools to different plugin combinations. + +plugin_settings: + routing_enabled: true + plugin_timeout: 30 + +global: + policies: + # "all" is reserved — these plugins fire on every invocation + all: + plugins: [identity-resolver] + # "pii" group — activated when a route has the "pii" tag + pii: + plugins: [pii-guard] + +plugins: + - name: identity-resolver + kind: builtin/identity + hooks: [tool_pre_invoke, tool_post_invoke] + mode: sequential + priority: 10 + on_error: fail + + - name: pii-guard + kind: builtin/pii + hooks: [tool_pre_invoke] + mode: sequential + priority: 20 + on_error: fail + config: + clearance_level: confidential + + - name: audit-logger + kind: builtin/audit + hooks: [tool_pre_invoke, tool_post_invoke] + mode: fire_and_forget + priority: 100 + on_error: ignore + +routes: + # HR compensation tool — contains PII, gets full security stack + - tool: get_compensation + meta: + tags: [pii, hr] + plugins: + - audit-logger + + # Public department listing — standard security only + - tool: list_departments + plugins: + - audit-logger + + # Wildcard — catch-all for unmatched tools + - tool: "*" + plugins: + - audit-logger diff --git a/crates/cpex-core/src/config.rs b/crates/cpex-core/src/config.rs index 02496747..375094e5 100644 --- a/crates/cpex-core/src/config.rs +++ b/crates/cpex-core/src/config.rs @@ -5,11 +5,1157 @@ // // Unified YAML configuration parsing. // -// Parses the unified config format that combines global settings, -// plugin declarations, named policy groups, and per-entity routes -// into a single YAML document. +// Parses the config format that combines global settings, plugin +// declarations, and per-entity routes into a single YAML document. // -// Mirrors the unified config proposal in -// apl-plugins/docs/unified-config-proposal.md. +// Supports two modes controlled by `plugin_settings.routing_enabled`: +// - false (default, backward compatible): plugins declare their +// own conditions for when they fire. +// - true: per-entity routing rules determine which plugins fire, +// with plugin selection via policy groups and meta.tags. +// +// The two modes are mutually exclusive. When routing is disabled, +// the routes and global sections are ignored. When routing is +// enabled, conditions on individual plugins are ignored. + +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use serde::{Deserialize, Serialize}; + +use crate::error::PluginError; +use crate::plugin::PluginConfig; + +// --------------------------------------------------------------------------- +// Top-Level Config +// --------------------------------------------------------------------------- + +/// Top-level CPEX configuration. +/// +/// Parsed from a single YAML file. Plugin scoping mode is controlled +/// by `plugin_settings.routing_enabled` — if absent or false, plugins +/// use their own `conditions:` field (backward compatible). If true, +/// the `routes:` and `global:` sections take over. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct CpexConfig { + /// Global configuration — policies, defaults. + /// Only used when `plugin_settings.routing_enabled` is true. + #[serde(default)] + pub global: GlobalConfig, + + /// Directories to scan for plugin modules. + #[serde(default)] + pub plugin_dirs: Vec, + + /// Plugin declarations. + #[serde(default)] + pub plugins: Vec, + + /// Per-entity routing rules. + /// Only used when `plugin_settings.routing_enabled` is true. + #[serde(default)] + pub routes: Vec, + + /// Global plugin settings (timeout, error behavior, routing mode). + #[serde(default)] + pub plugin_settings: PluginSettings, +} + +impl CpexConfig { + /// Whether route-based plugin selection is enabled. + pub fn routing_enabled(&self) -> bool { + self.plugin_settings.routing_enabled + } +} + +// --------------------------------------------------------------------------- +// Plugin Settings +// --------------------------------------------------------------------------- + +/// Global plugin settings. +/// +/// Controls executor behavior and routing mode. All fields have +/// sensible defaults — a missing `plugin_settings:` section is valid. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PluginSettings { + /// Enable route-based plugin selection. + /// When false (default), plugins use their own `conditions:` field. + /// When true, the `routes:` and `global:` sections determine which + /// plugins fire per entity. + #[serde(default)] + pub routing_enabled: bool, + + /// Default timeout per plugin in seconds. + #[serde(default = "default_timeout")] + pub plugin_timeout: u64, + + /// Whether to halt on first deny in concurrent mode. + #[serde(default = "default_true")] + pub short_circuit_on_deny: bool, + + /// Whether plugins can execute in parallel within a mode band. + #[serde(default)] + pub parallel_execution_within_band: bool, + + /// Whether to halt the pipeline on any plugin error. + #[serde(default)] + pub fail_on_plugin_error: bool, +} + +impl Default for PluginSettings { + fn default() -> Self { + Self { + routing_enabled: false, + plugin_timeout: 30, + short_circuit_on_deny: true, + parallel_execution_within_band: false, + fail_on_plugin_error: false, + } + } +} + +fn default_timeout() -> u64 { + 30 +} + +fn default_true() -> bool { + true +} + +// --------------------------------------------------------------------------- +// Global Config +// --------------------------------------------------------------------------- + +/// Global configuration — applies across all routes. +/// +/// Only used when routing is enabled. Contains named policy groups +/// (including the reserved `all` group) and per-entity-type defaults. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct GlobalConfig { + /// Named policy groups. The reserved name `all` is applied to + /// every request unconditionally. Other groups are inherited + /// by routes via `meta.tags`. + #[serde(default)] + pub policies: HashMap, + + /// Per-entity-type default policy groups. + /// Keys are `tool`, `resource`, `prompt`, `llm`. + #[serde(default)] + pub defaults: HashMap, +} + +// --------------------------------------------------------------------------- +// Policy Group +// --------------------------------------------------------------------------- + +/// A named policy group — plugins to activate and optional metadata. +/// +/// The `all` group is reserved and always applied. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct PolicyGroup { + /// Human-readable description. + #[serde(default)] + pub description: Option, + + /// Arbitrary metadata for tooling and audit. + #[serde(default)] + pub metadata: HashMap, + + /// Plugin references to activate when this group matches. + #[serde(default)] + pub plugins: Vec, +} + +// --------------------------------------------------------------------------- +// Plugin Ref (route/group plugin reference) +// --------------------------------------------------------------------------- + +/// A reference to a plugin in a route or policy group. +/// +/// ```yaml +/// plugins: +/// - rate_limiter # bare name +/// - pii_scanner: # name with config overrides +/// config: +/// sensitivity: high +/// ``` +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum PluginRef { + /// Just the name — activate the plugin with no config overrides. + Name(String), + /// Name with config overrides — single-key map. + WithOverrides(HashMap), +} + +impl PluginRef { + /// Extract the plugin name from this reference. + pub fn name(&self) -> &str { + match self { + Self::Name(name) => name, + Self::WithOverrides(map) => map.keys().next().map(|s| s.as_str()).unwrap_or(""), + } + } + + /// Extract config overrides, if any. + pub fn overrides(&self) -> Option<&serde_json::Value> { + match self { + Self::Name(_) => None, + Self::WithOverrides(map) => map.values().next(), + } + } +} + +// --------------------------------------------------------------------------- +// Route Entry +// --------------------------------------------------------------------------- + +/// A per-entity routing rule. +/// +/// Matches one entity type (tool, resource, prompt, or LLM) and +/// determines which plugins fire. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct RouteEntry { + /// Match a tool by exact name, list, or glob. + #[serde(default)] + pub tool: Option, + + /// Match a resource by exact URI, list, or glob. + #[serde(default)] + pub resource: Option, + + /// Match a prompt by exact name, list, or glob. + #[serde(default)] + pub prompt: Option, + + /// Match an LLM by exact model name, list, or glob. + #[serde(default)] + pub llm: Option, + + /// Operational metadata — tags, scope, properties. + #[serde(default)] + pub meta: Option, + + /// Conditional match expression — carried but not evaluated + /// during static resolution. Evaluated at runtime when payload + /// data is available (future: APL evaluator). + #[serde(default)] + pub when: Option, + + /// Plugin references to activate for this route. + #[serde(default)] + pub plugins: Vec, +} + +// --------------------------------------------------------------------------- +// Route Meta +// --------------------------------------------------------------------------- + +/// Operational metadata on a route entry. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct RouteMeta { + /// Entity tags — drive policy group inheritance. + #[serde(default)] + pub tags: Vec, + + /// Host-defined grouping (virtual server ID, namespace, etc.). + /// Used for scope matching: route scope must match request scope. + #[serde(default)] + pub scope: Option, + + /// Arbitrary key-value metadata. + #[serde(default)] + pub properties: HashMap, +} + +// --------------------------------------------------------------------------- +// String or List (for tool matching) +// --------------------------------------------------------------------------- + +/// A tool matcher — single name, list of names, or glob pattern. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum StringOrList { + /// Single string (exact name or glob pattern). + Single(String), + /// List of exact names. + List(Vec), +} + +impl Default for StringOrList { + fn default() -> Self { + Self::Single(String::new()) + } +} + +impl StringOrList { + /// Check if this matcher matches the given name. + pub fn matches(&self, name: &str) -> bool { + match self { + Self::Single(pattern) => { + if pattern == "*" { + true + } else if pattern.contains('*') { + let prefix = pattern.trim_end_matches('*'); + name.starts_with(prefix) + } else { + name == pattern + } + } + Self::List(names) => names.iter().any(|n| n == name), + } + } +} + +// --------------------------------------------------------------------------- +// Config Loading +// --------------------------------------------------------------------------- + +/// Load and parse a CPEX config from a YAML file. +pub fn load_config(path: &Path) -> Result { + let content = std::fs::read_to_string(path).map_err(|e| PluginError::Config { + message: format!("failed to read config file '{}': {}", path.display(), e), + })?; + parse_config(&content) +} + +/// Parse a CPEX config from a YAML string. +pub fn parse_config(yaml: &str) -> Result { + let config: CpexConfig = + serde_yaml::from_str(yaml).map_err(|e| PluginError::Config { + message: format!("failed to parse config YAML: {}", e), + })?; + validate_config(&config)?; + Ok(config) +} + +// --------------------------------------------------------------------------- +// Validation +// --------------------------------------------------------------------------- + +/// Validate a parsed config for structural correctness. +fn validate_config(config: &CpexConfig) -> Result<(), PluginError> { + let mut seen_names = HashSet::new(); + for plugin in &config.plugins { + if !seen_names.insert(&plugin.name) { + return Err(PluginError::Config { + message: format!("duplicate plugin name: '{}'", plugin.name), + }); + } + } + + if config.routing_enabled() { + let plugin_names: HashSet<&str> = + config.plugins.iter().map(|p| p.name.as_str()).collect(); + + for (i, route) in config.routes.iter().enumerate() { + let count = [ + route.tool.is_some(), + route.resource.is_some(), + route.prompt.is_some(), + route.llm.is_some(), + ] + .iter() + .filter(|&&m| m) + .count(); + + if count == 0 { + return Err(PluginError::Config { + message: format!( + "route {} has no entity matcher (need tool, resource, prompt, or llm)", + i + ), + }); + } + if count > 1 { + return Err(PluginError::Config { + message: format!("route {} has multiple entity matchers (need exactly one)", i), + }); + } + + for plugin_ref in &route.plugins { + if !plugin_names.contains(plugin_ref.name()) { + return Err(PluginError::Config { + message: format!("route {} references unknown plugin '{}'", i, plugin_ref.name()), + }); + } + } + } + + for (group_name, group) in &config.global.policies { + for plugin_ref in &group.plugins { + if !plugin_names.contains(plugin_ref.name()) { + return Err(PluginError::Config { + message: format!( + "policy group '{}' references unknown plugin '{}'", + group_name, + plugin_ref.name() + ), + }); + } + } + } + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Route Resolution +// --------------------------------------------------------------------------- + +/// Specificity scores for route matching. +const SPECIFICITY_EXACT_NAME: usize = 1000; +const SPECIFICITY_NAME_LIST: usize = 500; +const SPECIFICITY_GLOB: usize = 300; +const SPECIFICITY_WHEN_ONLY: usize = 10; +const SPECIFICITY_WILDCARD: usize = 0; + +/// Resolve which plugins should fire for a given entity. +/// +/// When routing is disabled, returns all plugin names. When enabled, +/// matches the entity against routes and collects plugins from the +/// `all` group, defaults, matching policy groups (via merged tags), +/// and the route itself. +/// +/// `request_scope` and `request_tags` come from the host's +/// `MetaExtension` on the request. +pub fn resolve_plugins_for_entity( + config: &CpexConfig, + entity_type: &str, + entity_name: &str, + request_scope: Option<&str>, + request_tags: &HashSet, +) -> Vec { + if !config.routing_enabled() { + return config + .plugins + .iter() + .map(|p| ResolvedPlugin { + name: p.name.clone(), + config_overrides: None, + when: None, + }) + .collect(); + } + + let mut resolved = Vec::new(); + + // 1. Always include plugins from the "all" policy group + if let Some(all_group) = config.global.policies.get("all") { + collect_plugin_refs(&all_group.plugins, &mut resolved, None); + } + + // 2. Include plugins from matching defaults + if let Some(default_group) = config.global.defaults.get(entity_type) { + collect_plugin_refs(&default_group.plugins, &mut resolved, None); + } + + // 3. Find matching route (with scope check) + if let Some(route) = find_matching_route(config, entity_type, entity_name, request_scope) { + // Merge tags: route's static tags + host's runtime tags + let mut merged_tags: HashSet = request_tags.clone(); + if let Some(meta) = &route.meta { + for tag in &meta.tags { + merged_tags.insert(tag.clone()); + } + } + + // Include plugins from all matching policy groups (merged tags) + for tag in &merged_tags { + if tag == "all" { + continue; // already handled above + } + if let Some(group) = config.global.policies.get(tag.as_str()) { + collect_plugin_refs(&group.plugins, &mut resolved, None); + } + } + + // Include route-level plugins, carrying the route's when clause + collect_plugin_refs(&route.plugins, &mut resolved, route.when.as_deref()); + } + + // Deduplicate by name, preserving order. Later overrides win. + let mut seen = HashSet::new(); + let mut deduped = Vec::new(); + for rp in resolved.into_iter().rev() { + if seen.insert(rp.name.clone()) { + deduped.push(rp); + } + } + deduped.reverse(); + deduped +} + +/// A resolved plugin with optional config overrides and when clause. +#[derive(Debug, Clone)] +pub struct ResolvedPlugin { + /// Plugin name. + pub name: String, + + /// Config overrides from the route. + pub config_overrides: Option, + + /// When clause from the route — carried but not evaluated here. + pub when: Option, +} + +/// Collect plugin refs into the resolved list. +fn collect_plugin_refs( + refs: &[PluginRef], + resolved: &mut Vec, + route_when: Option<&str>, +) { + for plugin_ref in refs { + resolved.push(ResolvedPlugin { + name: plugin_ref.name().to_string(), + config_overrides: plugin_ref.overrides().cloned(), + when: route_when.map(String::from), + }); + } +} + +/// Find the best matching route for an entity by specificity. +/// +/// Scope matching: if a route declares a scope, the request must +/// have the same scope. No scope on the route matches any request. +fn find_matching_route<'a>( + config: &'a CpexConfig, + entity_type: &str, + entity_name: &str, + request_scope: Option<&str>, +) -> Option<&'a RouteEntry> { + let mut best: Option<(usize, &RouteEntry)> = None; + + for route in &config.routes { + // Check scope compatibility + let route_scope = route.meta.as_ref().and_then(|m| m.scope.as_deref()); + let scope_bonus = match (route_scope, request_scope) { + (None, _) => 0, // route is global + (Some(rs), Some(rq)) if rs == rq => 100, // scopes match + (Some(_), _) => continue, // scope mismatch — skip + }; + + let base_specificity = match entity_type { + "tool" => { + if let Some(matcher) = &route.tool { + if !matcher.matches(entity_name) { + continue; + } + match matcher { + StringOrList::Single(s) if s == "*" => SPECIFICITY_WILDCARD, + StringOrList::Single(s) if s.contains('*') => SPECIFICITY_GLOB, + StringOrList::List(_) => SPECIFICITY_NAME_LIST, + StringOrList::Single(_) => SPECIFICITY_EXACT_NAME, + } + } else { + continue; + } + } + "resource" => { + if let Some(matcher) = &route.resource { + if !matcher.matches(entity_name) { + continue; + } + match matcher { + StringOrList::Single(s) if s == "*" => SPECIFICITY_WILDCARD, + StringOrList::Single(s) if s.contains('*') => SPECIFICITY_GLOB, + StringOrList::List(_) => SPECIFICITY_NAME_LIST, + StringOrList::Single(_) => SPECIFICITY_EXACT_NAME, + } + } else { + continue; + } + } + "prompt" => { + if let Some(matcher) = &route.prompt { + if !matcher.matches(entity_name) { + continue; + } + match matcher { + StringOrList::Single(s) if s == "*" => SPECIFICITY_WILDCARD, + StringOrList::Single(s) if s.contains('*') => SPECIFICITY_GLOB, + StringOrList::List(_) => SPECIFICITY_NAME_LIST, + StringOrList::Single(_) => SPECIFICITY_EXACT_NAME, + } + } else { + continue; + } + } + "llm" => { + if let Some(matcher) = &route.llm { + if !matcher.matches(entity_name) { + continue; + } + match matcher { + StringOrList::Single(s) if s == "*" => SPECIFICITY_WILDCARD, + StringOrList::Single(s) if s.contains('*') => SPECIFICITY_GLOB, + StringOrList::List(_) => SPECIFICITY_NAME_LIST, + StringOrList::Single(_) => SPECIFICITY_EXACT_NAME, + } + } else { + continue; + } + } + _ => continue, + }; + + let when_bonus = if route.when.is_some() { SPECIFICITY_WHEN_ONLY } else { 0 }; + let total = base_specificity + scope_bonus + when_bonus; + + if best.map_or(true, |(s, _)| total > s) { + best = Some((total, route)); + } + } + + best.map(|(_, route)| route) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Helper: empty tags for tests that don't need them + fn no_tags() -> HashSet { + HashSet::new() + } + + #[test] + fn test_parse_minimal_config() { + let yaml = r#" +plugins: + - name: rate_limiter + kind: builtin + hooks: [tool_pre_invoke] + mode: sequential + priority: 5 + config: + max_requests: 100 +"#; + let config = parse_config(yaml).unwrap(); + assert!(!config.routing_enabled()); + assert_eq!(config.plugins.len(), 1); + assert_eq!(config.plugins[0].name, "rate_limiter"); + } + + #[test] + fn test_no_plugin_settings_defaults_routing_disabled() { + let yaml = r#" +plugins: + - name: test + kind: builtin + hooks: [tool_pre_invoke] +"#; + let config = parse_config(yaml).unwrap(); + assert!(!config.routing_enabled()); + assert_eq!(config.plugin_settings.plugin_timeout, 30); + } + + #[test] + fn test_routing_enabled() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [identity] +plugins: + - name: identity + kind: builtin + hooks: [identity_resolve] +routes: + - tool: get_compensation + meta: + tags: [pii] +"#; + let config = parse_config(yaml).unwrap(); + assert!(config.routing_enabled()); + } + + #[test] + fn test_duplicate_plugin_names_rejected() { + let yaml = r#" +plugins: + - name: dup + kind: builtin + hooks: [tool_pre_invoke] + - name: dup + kind: builtin + hooks: [tool_post_invoke] +"#; + assert!(parse_config(yaml) + .unwrap_err() + .to_string() + .contains("duplicate plugin name")); + } + + #[test] + fn test_route_requires_one_entity_matcher() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: [] +routes: + - meta: + tags: [pii] +"#; + assert!(parse_config(yaml) + .unwrap_err() + .to_string() + .contains("no entity matcher")); + } + + #[test] + fn test_route_rejects_multiple_entity_matchers() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: [] +routes: + - tool: get_compensation + resource: "hr://employees/*" +"#; + assert!(parse_config(yaml) + .unwrap_err() + .to_string() + .contains("multiple entity matchers")); + } + + #[test] + fn test_route_unknown_plugin_rejected() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: known + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + plugins: + - unknown +"#; + assert!(parse_config(yaml) + .unwrap_err() + .to_string() + .contains("unknown plugin 'unknown'")); + } + + #[test] + fn test_policy_group_unknown_plugin_rejected() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [nonexistent] +plugins: [] +routes: [] +"#; + assert!(parse_config(yaml) + .unwrap_err() + .to_string() + .contains("unknown plugin 'nonexistent'")); + } + + #[test] + fn test_resolve_conditions_mode_returns_all() { + let yaml = r#" +plugins: + - name: a + kind: builtin + hooks: [tool_pre_invoke] + - name: b + kind: builtin + hooks: [tool_post_invoke] +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "anything", None, &no_tags()); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert_eq!(names, vec!["a", "b"]); + } + + #[test] + fn test_resolve_routes_inherits_policy_groups() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: + - identity + pii: + plugins: + - apl_policy +plugins: + - name: identity + kind: builtin + hooks: [identity_resolve] + - name: apl_policy + kind: builtin + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_compensation + meta: + tags: [pii] +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "get_compensation", None, &no_tags()); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert!(names.contains(&"identity")); + assert!(names.contains(&"apl_policy")); + } + + #[test] + fn test_resolve_no_matching_route_gets_all_only() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: + - identity +plugins: + - name: identity + kind: builtin + hooks: [identity_resolve] +routes: + - tool: get_compensation +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "unknown_tool", None, &no_tags()); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert_eq!(names, vec!["identity"]); + } + + #[test] + fn test_exact_match_beats_glob() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: specific + kind: builtin + hooks: [tool_pre_invoke] + - name: general + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: "hr-*" + plugins: + - general + - tool: hr-compensation + plugins: + - specific +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "hr-compensation", None, &no_tags()); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert!(names.contains(&"specific")); + assert!(!names.contains(&"general")); + } + + #[test] + fn test_plugin_ref_bare_name() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: rate_limiter + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + plugins: + - rate_limiter +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "get_compensation", None, &no_tags()); + assert_eq!(resolved[0].name, "rate_limiter"); + assert!(resolved[0].config_overrides.is_none()); + } + + #[test] + fn test_plugin_ref_with_overrides() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: rate_limiter + kind: builtin + hooks: [tool_pre_invoke] + config: + max_requests: 100 +routes: + - tool: get_compensation + plugins: + - rate_limiter: + config: + max_requests: 10 +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "get_compensation", None, &no_tags()); + assert_eq!(resolved[0].name, "rate_limiter"); + assert!(resolved[0].config_overrides.is_some()); + let overrides = resolved[0].config_overrides.as_ref().unwrap(); + assert_eq!(overrides["config"]["max_requests"], 10); + } + + #[test] + fn test_plugin_ref_mixed_bare_and_overrides() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: rate_limiter + kind: builtin + hooks: [tool_pre_invoke] + - name: pii_scanner + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + plugins: + - rate_limiter + - pii_scanner: + config: + sensitivity: high +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "get_compensation", None, &no_tags()); + assert_eq!(resolved.len(), 2); + assert_eq!(resolved[0].name, "rate_limiter"); + assert!(resolved[0].config_overrides.is_none()); + assert_eq!(resolved[1].name, "pii_scanner"); + assert!(resolved[1].config_overrides.is_some()); + } + + #[test] + fn test_deduplication_preserves_order() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [a, b] + pii: + plugins: [b, c] +plugins: + - name: a + kind: builtin + hooks: [tool_pre_invoke] + - name: b + kind: builtin + hooks: [tool_pre_invoke] + - name: c + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + meta: + tags: [pii] +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity(&config, "tool", "get_compensation", None, &no_tags()); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert_eq!(names, vec!["a", "b", "c"]); + } + + #[test] + fn test_glob_matches() { + let matcher = StringOrList::Single("hr-*".to_string()); + assert!(matcher.matches("hr-compensation")); + assert!(matcher.matches("hr-benefits")); + assert!(!matcher.matches("finance-report")); + } + + #[test] + fn test_wildcard_matches_everything() { + let matcher = StringOrList::Single("*".to_string()); + assert!(matcher.matches("anything")); + } + + #[test] + fn test_list_matches_any_member() { + let matcher = StringOrList::List(vec![ + "get_compensation".to_string(), + "get_benefits".to_string(), + ]); + assert!(matcher.matches("get_compensation")); + assert!(matcher.matches("get_benefits")); + assert!(!matcher.matches("send_email")); + } + + #[test] + fn test_validation_skipped_when_routing_disabled() { + let yaml = r#" +plugins: + - name: test + kind: builtin + hooks: [tool_pre_invoke] +routes: + - meta: + tags: [pii] +"#; + let config = parse_config(yaml); + assert!(config.is_ok()); + } + + // -- Scope matching tests -- + + #[test] + fn test_scope_match_selects_scoped_route() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: scoped_plugin + kind: builtin + hooks: [tool_pre_invoke] + - name: global_plugin + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + meta: + scope: hr-services + plugins: + - scoped_plugin + - tool: get_compensation + plugins: + - global_plugin +"#; + let config = parse_config(yaml).unwrap(); + + // With matching scope — scoped route wins (more specific) + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", Some("hr-services"), &no_tags(), + ); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert!(names.contains(&"scoped_plugin")); + assert!(!names.contains(&"global_plugin")); + + // Without scope — global route matches + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", None, &no_tags(), + ); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert!(names.contains(&"global_plugin")); + assert!(!names.contains(&"scoped_plugin")); + + // With different scope — global route matches (scoped doesn't) + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", Some("billing"), &no_tags(), + ); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + assert!(names.contains(&"global_plugin")); + assert!(!names.contains(&"scoped_plugin")); + } + + // -- Tag merging tests -- + + #[test] + fn test_host_tags_merged_with_route_tags() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + pii: + plugins: [pii_plugin] + runtime_tag: + plugins: [runtime_plugin] +plugins: + - name: pii_plugin + kind: builtin + hooks: [tool_pre_invoke] + - name: runtime_plugin + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + meta: + tags: [pii] +"#; + let config = parse_config(yaml).unwrap(); + + // Host provides a runtime tag that matches a policy group + let mut host_tags = HashSet::new(); + host_tags.insert("runtime_tag".to_string()); + + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", None, &host_tags, + ); + let names: Vec<&str> = resolved.iter().map(|r| r.name.as_str()).collect(); + + // Both route's static tag (pii) and host's runtime tag activate their groups + assert!(names.contains(&"pii_plugin")); + assert!(names.contains(&"runtime_plugin")); + } + + // -- When clause carried tests -- + + #[test] + fn test_when_clause_carried_on_resolved_plugins() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: conditional_plugin + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + when: "args.include_ssn == true" + plugins: + - conditional_plugin +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", None, &no_tags(), + ); + assert_eq!(resolved[0].name, "conditional_plugin"); + assert_eq!(resolved[0].when.as_deref(), Some("args.include_ssn == true")); + } + + #[test] + fn test_when_clause_not_on_policy_group_plugins() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [global_plugin] +plugins: + - name: global_plugin + kind: builtin + hooks: [tool_pre_invoke] + - name: route_plugin + kind: builtin + hooks: [tool_pre_invoke] +routes: + - tool: get_compensation + when: "args.sensitive == true" + plugins: + - route_plugin +"#; + let config = parse_config(yaml).unwrap(); + let resolved = resolve_plugins_for_entity( + &config, "tool", "get_compensation", None, &no_tags(), + ); + + // global_plugin has no when clause (from all group) + let global = resolved.iter().find(|r| r.name == "global_plugin").unwrap(); + assert!(global.when.is_none()); -// TODO: Implement CpexConfig, GlobalConfig, RouteEntry serde models + // route_plugin carries the route's when clause + let route = resolved.iter().find(|r| r.name == "route_plugin").unwrap(); + assert_eq!(route.when.as_deref(), Some("args.sensitive == true")); + } +} diff --git a/crates/cpex-core/src/error.rs b/crates/cpex-core/src/error.rs index 4b684d54..fd253429 100644 --- a/crates/cpex-core/src/error.rs +++ b/crates/cpex-core/src/error.rs @@ -24,6 +24,12 @@ use thiserror::Error; /// Covers plugin execution failures, policy violations, timeouts, /// and configuration issues. Each variant carries enough context /// for the caller to log, report, or recover. +/// +/// Mirrors the Python framework's `PluginErrorModel` with: +/// - `code` — business-logic error code (e.g., `"rate_limit_exceeded"`) +/// - `details` — structured diagnostic data for logging +/// - `proto_error_code` — protocol-level error code for the host to +/// map back to the wire format (MCP JSON-RPC, HTTP status, etc.) #[derive(Debug, Error)] pub enum PluginError { /// A plugin raised an execution error. @@ -31,8 +37,17 @@ pub enum PluginError { Execution { plugin_name: String, message: String, + /// Business-logic error code (e.g., `"invalid_token"`). #[source] source: Option>, + /// Business-logic error code set by the plugin. + code: Option, + /// Structured diagnostic data for logging or debugging. + details: HashMap, + /// Protocol-level error code for the host to map to the wire + /// format. MCP: JSON-RPC codes (e.g., -32603). HTTP: status + /// codes. The host interprets this; CPEX just carries it. + proto_error_code: Option, }, /// A plugin exceeded its execution timeout. @@ -40,6 +55,8 @@ pub enum PluginError { Timeout { plugin_name: String, timeout_ms: u64, + /// Protocol-level error code for the host. + proto_error_code: Option, }, /// A plugin returned a policy violation (deny). @@ -93,6 +110,12 @@ pub struct PluginViolation { /// Name of the plugin that produced the violation. /// Set by the framework after the plugin returns, not by the plugin itself. pub plugin_name: Option, + + /// Protocol-level error code for the host to map to the wire format. + /// MCP: JSON-RPC codes (e.g., -32603). HTTP: status codes (e.g., 403). + /// Set by the plugin; the host interprets it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub proto_error_code: Option, } impl PluginViolation { @@ -104,6 +127,7 @@ impl PluginViolation { description: None, details: HashMap::new(), plugin_name: None, + proto_error_code: None, } } @@ -118,6 +142,12 @@ impl PluginViolation { self.details = details; self } + + /// Attach a protocol-level error code. + pub fn with_proto_error_code(mut self, code: i64) -> Self { + self.proto_error_code = Some(code); + self + } } impl std::fmt::Display for PluginViolation { diff --git a/crates/cpex-core/src/executor.rs b/crates/cpex-core/src/executor.rs index 4b1188ef..9a6cdcd2 100644 --- a/crates/cpex-core/src/executor.rs +++ b/crates/cpex-core/src/executor.rs @@ -26,6 +26,7 @@ use std::any::Any; use std::collections::HashMap; +use std::fmt; use std::sync::Arc; use std::time::Duration; @@ -67,24 +68,35 @@ impl Default for ExecutorConfig { /// Aggregate result from a full hook invocation across all phases. /// /// Wraps the final payload, extensions, any violation, and the -/// context table. The caller should pass `context_table` into the -/// next hook invocation to preserve per-plugin local state across -/// hooks in the same request lifecycle. +/// context table. Immutable by design — policy decisions cannot be +/// tampered with after the executor returns them. +/// +/// The caller should pass `context_table` into the next hook +/// invocation to preserve per-plugin local state across hooks in +/// the same request lifecycle. +/// +/// Background tasks are returned separately as [`BackgroundTasks`] +/// to keep the policy result immutable. #[derive(Debug)] pub struct PipelineResult { - /// Whether the pipeline completed without a deny. - pub allowed: bool, + /// Whether the pipeline should continue processing. + /// `false` means a plugin denied — the pipeline was halted. + pub continue_processing: bool, /// The final payload after all modifications (type-erased). /// `None` if the pipeline was denied before any modifications. - pub payload: Option>, + pub modified_payload: Option>, /// The final extensions after all modifications. - pub extensions: Extensions, + /// `None` if no plugin modified extensions. + pub modified_extensions: Option, /// The violation that caused a deny, if any. pub violation: Option, + /// Optional metadata aggregated from plugins (telemetry, diagnostics). + pub metadata: Option, + /// Plugin contexts indexed by plugin ID. Thread this into the /// next hook invocation to preserve per-plugin `local_state`. pub context_table: PluginContextTable, @@ -98,10 +110,11 @@ impl PipelineResult { context_table: PluginContextTable, ) -> Self { Self { - allowed: true, - payload: Some(payload), - extensions, + continue_processing: true, + modified_payload: Some(payload), + modified_extensions: Some(extensions), violation: None, + metadata: None, context_table, } } @@ -113,13 +126,91 @@ impl PipelineResult { context_table: PluginContextTable, ) -> Self { Self { - allowed: false, - payload: None, - extensions, + continue_processing: false, + modified_payload: None, + modified_extensions: Some(extensions), violation: Some(violation), + metadata: None, context_table, } } + + /// Whether this result represents a denial. + pub fn is_denied(&self) -> bool { + !self.continue_processing + } +} + +// --------------------------------------------------------------------------- +// Background Tasks +// --------------------------------------------------------------------------- + +/// Handles to fire-and-forget background tasks spawned by the executor. +/// +/// Returned separately from [`PipelineResult`] so that the policy +/// result stays immutable. If not awaited, tasks complete on their +/// own in the background. Call `wait_for_background_tasks()` when you +/// need to ensure tasks have finished (tests, graceful shutdown, +/// audit flush). +pub struct BackgroundTasks { + tasks: Vec<(String, tokio::task::JoinHandle<()>)>, +} + +impl BackgroundTasks { + /// Create an empty set of background tasks. + pub fn empty() -> Self { + Self { tasks: Vec::new() } + } + + /// Create from a list of (plugin_name, handle) pairs. + fn from_handles(tasks: Vec<(String, tokio::task::JoinHandle<()>)>) -> Self { + Self { tasks } + } + + /// Whether there are any background tasks. + pub fn is_empty(&self) -> bool { + self.tasks.is_empty() + } + + /// Number of background tasks. + pub fn len(&self) -> usize { + self.tasks.len() + } + + /// Wait for all fire-and-forget background tasks to complete. + /// + /// Returns a list of errors from any tasks that panicked. + /// An empty list means all tasks completed successfully. + /// + /// Consumes `self` — each task handle can only be awaited once. + /// + /// If not called, background tasks still complete on their own. + /// Use this for tests, graceful shutdown, or when you need to + /// ensure audit/logging tasks have flushed before proceeding. + pub async fn wait_for_background_tasks(self) -> Vec { + let mut errors = Vec::new(); + for (plugin_name, handle) in self.tasks { + if let Err(e) = handle.await { + errors.push(crate::error::PluginError::Execution { + plugin_name, + message: format!("background task panicked: {}", e), + source: None, + code: None, + details: std::collections::HashMap::new(), + proto_error_code: None, + }); + } + } + errors + } +} + +impl fmt::Debug for BackgroundTasks { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BackgroundTasks") + .field("count", &self.tasks.len()) + .finish() + } } // --------------------------------------------------------------------------- @@ -158,19 +249,26 @@ impl Executor { /// /// # Returns /// - /// A `PipelineResult` with the final payload, extensions, violation, - /// and the updated context table for threading into the next hook. + /// A tuple of: + /// - `PipelineResult` — immutable policy result with payload, + /// extensions, violation, and context table. + /// - `BackgroundTasks` — handles to fire-and-forget tasks. Call + /// `wait_for_background_tasks()` to await them, or drop to let + /// them complete in the background. pub async fn execute( &self, entries: &[HookEntry], payload: Box, extensions: Extensions, context_table: Option, - ) -> PipelineResult { + ) -> (PipelineResult, BackgroundTasks) { let mut ctx_table = context_table.unwrap_or_default(); if entries.is_empty() { - return PipelineResult::allowed_with(payload, extensions, ctx_table); + return ( + PipelineResult::allowed_with(payload, extensions, ctx_table), + BackgroundTasks::empty(), + ); } // Group entries by mode (from trusted_config) @@ -193,7 +291,10 @@ impl Executor { ) .await { - return PipelineResult::denied(v, current_extensions, ctx_table); + return ( + PipelineResult::denied(v, current_extensions, ctx_table), + BackgroundTasks::empty(), + ); } // Phase 2: TRANSFORM — serial, chained, can modify, cannot block @@ -218,17 +319,23 @@ impl Executor { .run_concurrent_phase(&concurrent, &*current_payload, ¤t_extensions, &ctx_table) .await { - return PipelineResult::denied(violation, current_extensions, ctx_table); + return ( + PipelineResult::denied(violation, current_extensions, ctx_table), + BackgroundTasks::empty(), + ); } // Phase 5: FIRE_AND_FORGET — background, read-only, ignore results - self.spawn_fire_and_forget( + let bg_handles = self.spawn_fire_and_forget( &fire_and_forget, &*current_payload, &ctx_table, ); - PipelineResult::allowed_with(current_payload, current_extensions, ctx_table) + ( + PipelineResult::allowed_with(current_payload, current_extensions, ctx_table), + BackgroundTasks::from_handles(bg_handles), + ) } // ----------------------------------------------------------------------- @@ -547,14 +654,18 @@ impl Executor { /// Each handler runs in its own `tokio::spawn` — the pipeline does /// not wait for them. Errors and timeouts are logged but have no /// effect on the pipeline result. + /// + /// Returns the plugin name and join handle for each spawned task + /// so they can be stored on `PipelineResult` for optional awaiting + /// via `wait_for_background_tasks()`. fn spawn_fire_and_forget( &self, entries: &[HookEntry], payload: &dyn PluginPayload, ctx_table: &PluginContextTable, - ) { + ) -> Vec<(String, tokio::task::JoinHandle<()>)> { if entries.is_empty() { - return; + return Vec::new(); } let timeout_dur = Duration::from_secs(self.config.timeout_seconds); @@ -564,14 +675,17 @@ impl Executor { .map(|c| c.global_state.clone()) .unwrap_or_default(); + let mut handles = Vec::with_capacity(entries.len()); + for entry in entries { let plugin_name = entry.plugin_ref.name().to_string(); let handler = Arc::clone(&entry.handler); let owned_payload = payload.clone_boxed(); let mut ctx = PluginContext::with_global_state(global_state.clone()); let dur = timeout_dur; + let name_for_log = plugin_name.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let filtered = FilteredExtensions::default(); let result = timeout( dur, @@ -582,14 +696,18 @@ impl Executor { match result { Ok(Ok(_)) => {} // discard Ok(Err(e)) => { - warn!("FIRE_AND_FORGET plugin '{}' error (ignored): {}", plugin_name, e); + warn!("FIRE_AND_FORGET plugin '{}' error (ignored): {}", name_for_log, e); } Err(_) => { - warn!("FIRE_AND_FORGET plugin '{}' timed out (ignored)", plugin_name); + warn!("FIRE_AND_FORGET plugin '{}' timed out (ignored)", name_for_log); } } }); + + handles.push((plugin_name, handle)); } + + handles } } @@ -719,8 +837,8 @@ mod tests { Extensions::default(), PluginContextTable::new(), ); - assert!(result.allowed); - assert!(result.payload.is_some()); + assert!(result.continue_processing); + assert!(result.modified_payload.is_some()); assert!(result.violation.is_none()); } @@ -732,8 +850,8 @@ mod tests { Extensions::default(), PluginContextTable::new(), ); - assert!(!result.allowed); - assert!(result.payload.is_none()); + assert!(!result.continue_processing); + assert!(result.modified_payload.is_none()); assert!(result.violation.is_some()); } @@ -743,10 +861,10 @@ mod tests { let payload: Box = Box::new(TestPayload { value: "test".into(), }); - let result = executor + let (result, _) = executor .execute(&[], payload, Extensions::default(), None) .await; - assert!(result.allowed); - assert!(result.payload.is_some()); + assert!(result.continue_processing); + assert!(result.modified_payload.is_some()); } } diff --git a/crates/cpex-core/src/factory.rs b/crates/cpex-core/src/factory.rs new file mode 100644 index 00000000..e77f80ca --- /dev/null +++ b/crates/cpex-core/src/factory.rs @@ -0,0 +1,141 @@ +// Location: ./crates/cpex-core/src/factory.rs +// Copyright 2025 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Teryl Taylor +// +// Plugin factory registry. +// +// Provides a factory pattern for creating plugin instances from +// config. The host registers factories by `kind` name before +// loading config. When the manager processes a config file, it +// looks up the factory for each plugin's `kind` and calls create(). +// +// This decouples plugin instantiation from the manager — the +// manager doesn't know how to create a "builtin" vs "wasm" vs +// "python" plugin. The factory does. +// +// Mirrors the Python framework's PluginLoader in +// cpex/framework/loader/plugin.py. + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::error::PluginError; +use crate::plugin::{Plugin, PluginConfig}; +use crate::registry::AnyHookHandler; + +// --------------------------------------------------------------------------- +// Plugin Factory Trait +// --------------------------------------------------------------------------- + +/// Factory for creating plugin instances from config. +/// +/// The host registers factories by `kind` name before loading +/// config. When the manager processes a config file, it looks up +/// the factory for each plugin's `kind` and calls `create()`. +/// +/// The factory returns both the plugin and its handler because it +/// knows the concrete types — which handler traits the plugin +/// implements and which hooks it handles. +/// +/// # Examples +/// +/// ```rust,ignore +/// struct RateLimiterFactory; +/// +/// impl PluginFactory for RateLimiterFactory { +/// fn create(&self, config: &PluginConfig) +/// -> Result +/// { +/// let plugin = Arc::new(RateLimiter::from_config(config)?); +/// let handler = Arc::new(TypedHandlerAdapter::::new( +/// Arc::clone(&plugin), +/// )); +/// Ok(PluginInstance { plugin, handler }) +/// } +/// } +/// +/// let mut factories = PluginFactoryRegistry::new(); +/// factories.register("security/rate_limit", Box::new(RateLimiterFactory)); +/// ``` +pub trait PluginFactory: Send + Sync { + /// Create a plugin instance and its handler from config. + /// + /// The `config` is the plugin's entry from the YAML file. + fn create(&self, config: &PluginConfig) -> Result; +} + +/// A created plugin instance — the plugin and its type-erased handlers. +/// +/// Each handler is paired with the hook name it handles. A plugin +/// that implements multiple hook types (e.g., `ToolPreInvoke` and +/// `ToolPostInvoke`) returns one entry per hook. +pub struct PluginInstance { + /// The plugin implementation. + pub plugin: Arc, + + /// Type-erased handlers paired with their hook names. + /// Each entry maps a hook name to the adapter for that hook type. + pub handlers: Vec<(&'static str, Arc)>, +} + +// --------------------------------------------------------------------------- +// Plugin Factory Registry +// --------------------------------------------------------------------------- + +/// Registry of plugin factories keyed by `kind` name. +/// +/// The host populates this before calling `PluginManager::from_config()`. +/// Each factory knows how to create plugins of a specific kind. +/// +/// # Examples +/// +/// ```rust,ignore +/// let mut factories = PluginFactoryRegistry::new(); +/// factories.register("builtin/rate_limit", Box::new(RateLimiterFactory)); +/// factories.register("builtin/identity", Box::new(IdentityFactory)); +/// +/// let manager = PluginManager::from_config(path, &factories)?; +/// ``` +pub struct PluginFactoryRegistry { + factories: HashMap>, +} + +impl PluginFactoryRegistry { + /// Create an empty factory registry. + pub fn new() -> Self { + Self { + factories: HashMap::new(), + } + } + + /// Register a factory for a given `kind` name. + pub fn register( + &mut self, + kind: impl Into, + factory: Box, + ) { + self.factories.insert(kind.into(), factory); + } + + /// Look up a factory by `kind` name. + pub fn get(&self, kind: &str) -> Option<&dyn PluginFactory> { + self.factories.get(kind).map(|f| f.as_ref()) + } + + /// Whether a factory exists for the given `kind`. + pub fn has(&self, kind: &str) -> bool { + self.factories.contains_key(kind) + } + + /// All registered kind names. + pub fn kinds(&self) -> Vec<&str> { + self.factories.keys().map(|s| s.as_str()).collect() + } +} + +impl Default for PluginFactoryRegistry { + fn default() -> Self { + Self::new() + } +} diff --git a/crates/cpex-core/src/hooks/payload.rs b/crates/cpex-core/src/hooks/payload.rs index c25f0247..f46d89c6 100644 --- a/crates/cpex-core/src/hooks/payload.rs +++ b/crates/cpex-core/src/hooks/payload.rs @@ -39,11 +39,16 @@ use serde::{Deserialize, Serialize}; /// This is a Phase 1 stub with minimal fields. Phase 3 adds the /// full CMF extension types (SecurityExtension with MonotonicSet, /// DelegationExtension with scope-narrowing chain, HttpExtension -/// with Guarded, MetaExtension, etc.). +/// with Guarded, etc.). /// /// Mirrors Python's `cpex.framework.extensions.Extensions`. #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct Extensions { + /// Host-provided operational metadata — entity identification, + /// tags, scope, and arbitrary properties. Immutable. + #[serde(default)] + pub meta: Option, + /// Security labels (monotonic — add-only in the full implementation). #[serde(default)] pub labels: std::collections::HashSet, @@ -53,6 +58,42 @@ pub struct Extensions { pub custom: HashMap, } +/// Host-provided operational metadata about the entity being processed. +/// +/// Carries entity identification (type + name) for route resolution, +/// operational tags for policy group inheritance, scope for host-defined +/// grouping, and arbitrary properties for policy conditions. +/// +/// Immutable — set by the host before invoking the hook. Plugins +/// can read but not modify. +/// +/// Mirrors Python's `cpex.framework.extensions.meta.MetaExtension`. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct MetaExtension { + /// Entity type: "tool", "resource", "prompt", "llm". + /// Used by the manager for route resolution. + #[serde(default)] + pub entity_type: Option, + + /// Entity name: "get_compensation", "hr://employees/*", etc. + /// Used by the manager for route resolution. + #[serde(default)] + pub entity_name: Option, + + /// Operational tags — drive policy group inheritance. + /// Merged with static tags from the matching route's `meta.tags`. + #[serde(default)] + pub tags: std::collections::HashSet, + + /// Host-defined grouping (virtual server ID, namespace, etc.). + #[serde(default)] + pub scope: Option, + + /// Arbitrary key-value metadata. + #[serde(default)] + pub properties: HashMap, +} + /// Capability-filtered view of Extensions for a specific plugin. /// /// Built by the framework before dispatching to each plugin. Fields @@ -63,6 +104,9 @@ pub struct Extensions { /// the Python `filter_extensions()` implementation. #[derive(Debug, Clone, Default)] pub struct FilteredExtensions { + /// Meta extension (always visible — immutable, no capability needed). + pub meta: Option, + /// Security labels (visible with `read_labels` capability). pub labels: Option>, diff --git a/crates/cpex-core/src/lib.rs b/crates/cpex-core/src/lib.rs index 2743b238..fbede921 100644 --- a/crates/cpex-core/src/lib.rs +++ b/crates/cpex-core/src/lib.rs @@ -17,6 +17,7 @@ // - [`manager`] — PluginManager lifecycle and hook dispatch // - [`registry`] — PluginInstanceRegistry and HookRegistry // - [`config`] — Unified YAML configuration parsing +// - [`factory`] — Plugin factory registry for config-driven instantiation // - [`context`] — PluginContext (local_state + global_state) // - [`error`] — Error types, violations, and result types @@ -24,6 +25,7 @@ pub mod config; pub mod context; pub mod error; pub mod executor; +pub mod factory; pub mod hooks; pub mod manager; pub mod plugin; diff --git a/crates/cpex-core/src/manager.rs b/crates/cpex-core/src/manager.rs index a2a6e8a3..0810bbba 100644 --- a/crates/cpex-core/src/manager.rs +++ b/crates/cpex-core/src/manager.rs @@ -24,13 +24,18 @@ // Mirrors the Python framework's PluginManager in // cpex/framework/manager.py. -use std::sync::Arc; +use std::hash::{Hash, Hasher}; +use std::path::Path; +use std::sync::{Arc, RwLock}; -use tracing::{error, info}; +use hashbrown::HashMap; +use tracing::{error, info, warn}; +use crate::config::{self, CpexConfig}; use crate::context::PluginContextTable; use crate::error::PluginError; -use crate::executor::{Executor, ExecutorConfig, PipelineResult}; +use crate::executor::{BackgroundTasks, Executor, ExecutorConfig, PipelineResult}; +use crate::factory::PluginFactoryRegistry; use crate::hooks::adapter::TypedHandlerAdapter; use crate::hooks::payload::{Extensions, PluginPayload}; use crate::hooks::trait_def::{HookHandler, HookTypeDef, PluginResult}; @@ -90,6 +95,44 @@ impl Default for ManagerConfig { /// The manager wraps each plugin in a `PluginRef` with an authoritative /// config from the config loader. The executor reads all scheduling /// decisions from `PluginRef.trusted_config` — never from the plugin. +/// Cache key for resolved routing entries. +/// +/// Includes entity type, name, hook name, and scope so that +/// the same tool on different scopes or at different hook points +/// caches separately. +/// +/// Custom Hash/Eq implementations hash on `&str` slices so that +/// `raw_entry` lookups with borrowed strings produce the same hash +/// as the owned key — enabling zero-allocation cache hits. +#[derive(Debug, Clone)] +struct RouteCacheKey { + entity_type: String, + entity_name: String, + hook_name: String, + scope: Option, +} + +impl Hash for RouteCacheKey { + fn hash(&self, state: &mut H) { + self.entity_type.as_str().hash(state); + self.entity_name.as_str().hash(state); + self.hook_name.as_str().hash(state); + self.scope.as_deref().hash(state); + } +} + +impl PartialEq for RouteCacheKey { + fn eq(&self, other: &Self) -> bool { + self.entity_type == other.entity_type + && self.entity_name == other.entity_name + && self.hook_name == other.hook_name + && self.scope == other.scope + } +} + +impl Eq for RouteCacheKey {} + + pub struct PluginManager { /// Plugin registry — stores PluginRefs and hook-to-handler mappings. registry: PluginRegistry, @@ -97,6 +140,23 @@ pub struct PluginManager { /// Executor — stateless 5-phase pipeline engine. executor: Executor, + /// Parsed CPEX config (when loaded from file). Used for route resolution. + cpex_config: Option, + + /// Factory registry — owned by the manager. Used for initial + /// instantiation and for creating override instances when routes + /// override a plugin's base config. + factories: PluginFactoryRegistry, + + /// Cache of resolved hook entries per (entity, hook, scope). + /// Populated on first access, invalidated on config reload. + /// Uses Arc so cache reads are refcount bumps (~1ns), not data copies. + route_cache: RwLock>>>, + + /// Hasher builder for zero-allocation cache lookups via raw_entry. + cache_hasher: hashbrown::DefaultHashBuilder, + + /// Whether initialize() has been called. initialized: bool, } @@ -104,13 +164,160 @@ pub struct PluginManager { impl PluginManager { /// Create a new PluginManager with the given configuration. pub fn new(config: ManagerConfig) -> Self { + let cache_hasher = hashbrown::DefaultHashBuilder::default(); Self { registry: PluginRegistry::new(), executor: Executor::new(config.executor), + cpex_config: None, + factories: PluginFactoryRegistry::new(), + route_cache: RwLock::new(HashMap::with_hasher(cache_hasher.clone())), + cache_hasher, initialized: false, } } + // ----------------------------------------------------------------------- + // Factory Registration + // ----------------------------------------------------------------------- + + /// Register a plugin factory for a given `kind` name. + /// + /// The host calls this to tell the manager how to create plugins + /// of a specific kind. Must be called before `load_config()`. + /// + /// # Examples + /// + /// ```rust,ignore + /// let mut manager = PluginManager::default(); + /// manager.register_factory("builtin", Box::new(BuiltinFactory)); + /// manager.register_factory("security/rate_limit", Box::new(RateLimiterFactory)); + /// manager.load_config(Path::new("plugins.yaml"))?; + /// ``` + pub fn register_factory( + &mut self, + kind: impl Into, + factory: Box, + ) { + self.factories.register(kind, factory); + } + + // ----------------------------------------------------------------------- + // Config Loading + // ----------------------------------------------------------------------- + + /// Load plugins from a YAML config file. + /// + /// Parses the config, looks up each plugin's `kind` in the + /// factory registry, instantiates the plugins, and registers + /// them. Factories must be registered via `register_factory()` + /// before calling this method. + /// + /// # Examples + /// + /// ```rust,ignore + /// let mut manager = PluginManager::default(); + /// manager.register_factory("builtin", Box::new(BuiltinFactory)); + /// manager.load_config_file(Path::new("plugins/config.yaml"))?; + /// manager.initialize().await?; + /// ``` + pub fn load_config_file(&mut self, path: &Path) -> Result<(), PluginError> { + let cpex_config = config::load_config(path)?; + self.load_config(cpex_config) + } + + /// Load plugins from a parsed config. + /// + /// Looks up each plugin's `kind` in the factory registry, + /// instantiates the plugins, and registers them with their + /// hook names from the config. + pub fn load_config(&mut self, cpex_config: CpexConfig) -> Result<(), PluginError> { + // Update executor settings from config + self.executor = Executor::new(ExecutorConfig { + timeout_seconds: cpex_config.plugin_settings.plugin_timeout, + short_circuit_on_deny: cpex_config.plugin_settings.short_circuit_on_deny, + }); + + // Instantiate and register each plugin from config + for plugin_config in &cpex_config.plugins { + let factory = self.factories.get(&plugin_config.kind).ok_or_else(|| { + PluginError::Config { + message: format!( + "no factory registered for plugin kind '{}' (plugin '{}')", + plugin_config.kind, plugin_config.name + ), + } + })?; + + let instance = factory.create(plugin_config)?; + + self.registry + .register_multi_handler( + instance.plugin, + plugin_config.clone(), + instance.handlers, + ) + .map_err(|msg| PluginError::Config { message: msg })?; + + info!( + "Registered plugin '{}' (kind: '{}') for hooks: {:?}", + plugin_config.name, plugin_config.kind, plugin_config.hooks + ); + } + + // Clear routing cache — config changed + self.clear_routing_cache(); + + // Store config for route resolution + self.cpex_config = Some(cpex_config); + + Ok(()) + } + + /// Create a PluginManager from a parsed config (convenience). + /// + /// Uses the passed factory registry for initial instantiation. + /// Note: for route-level config overrides to create new instances + /// at runtime, use `register_factory()` + `load_config()` instead + /// so the manager owns the factories. + pub fn from_config( + cpex_config: CpexConfig, + factories: &PluginFactoryRegistry, + ) -> Result { + let mut manager = Self::new(ManagerConfig::default()); + + // Instantiate and register each plugin + for plugin_config in &cpex_config.plugins { + let factory = factories.get(&plugin_config.kind).ok_or_else(|| { + PluginError::Config { + message: format!( + "no factory registered for plugin kind '{}' (plugin '{}')", + plugin_config.kind, plugin_config.name + ), + } + })?; + + let instance = factory.create(plugin_config)?; + + manager + .registry + .register_multi_handler( + instance.plugin, + plugin_config.clone(), + instance.handlers, + ) + .map_err(|msg| PluginError::Config { message: msg })?; + } + + // Update executor from config settings + manager.executor = Executor::new(ExecutorConfig { + timeout_seconds: cpex_config.plugin_settings.plugin_timeout, + short_circuit_on_deny: cpex_config.plugin_settings.short_circuit_on_deny, + }); + + manager.cpex_config = Some(cpex_config); + Ok(manager) + } + // ----------------------------------------------------------------------- // Registration // ----------------------------------------------------------------------- @@ -244,6 +451,9 @@ impl PluginManager { plugin_name, message: format!("initialization failed: {}", e), source: Some(Box::new(e)), + code: None, + details: std::collections::HashMap::new(), + proto_error_code: None, }); } @@ -304,33 +514,51 @@ impl PluginManager { /// /// # Returns /// - /// A `PipelineResult` with the final payload, extensions, violation, - /// and the updated context table. + /// A tuple of `(PipelineResult, BackgroundTasks)`. The result + /// contains the final payload, extensions, violation, and context + /// table. Background tasks can be awaited or dropped. pub async fn invoke_by_name( &self, hook_name: &str, payload: Box, extensions: Extensions, context_table: Option, - ) -> PipelineResult { + ) -> (PipelineResult, BackgroundTasks) { let hook_type = HookType::new(hook_name); - let entries = self.registry.entries_for_hook(&hook_type); + let all_entries = self.registry.entries_for_hook(&hook_type); + + if all_entries.is_empty() { + return ( + PipelineResult::allowed_with( + payload, + extensions, + context_table.unwrap_or_default(), + ), + BackgroundTasks::empty(), + ); + } + + let entries = self.filter_entries_by_route(all_entries, &extensions, hook_name); if entries.is_empty() { - return PipelineResult::allowed_with( - payload, - extensions, - context_table.unwrap_or_default(), + return ( + PipelineResult::allowed_with( + payload, + extensions, + context_table.unwrap_or_default(), + ), + BackgroundTasks::empty(), ); } self.executor - .execute(entries, payload, extensions, context_table) + .execute(&entries, payload, extensions, context_table) .await } // ----------------------------------------------------------------------- // Hook Invocation — Typed (invoke::) + // ----------------------------------------------------------------------- /// Invoke a typed hook. @@ -340,6 +568,11 @@ impl PluginManager { /// Dispatch goes through the same registry and 5-phase executor /// as `invoke_by_name()`. /// + /// When routing is enabled, the entity is identified from + /// `extensions.meta` (entity_type + entity_name). Only plugins + /// matching the resolved route fire. When routing is disabled + /// or meta is absent, all registered plugins fire. + /// /// # Type Parameters /// /// - `H` — the hook type (implements `HookTypeDef`). @@ -347,38 +580,239 @@ impl PluginManager { /// # Arguments /// /// * `payload` — the typed payload. - /// * `extensions` — the full extensions. + /// * `extensions` — the full extensions (includes meta for routing). /// * `context_table` — optional context table from a previous hook. /// /// # Returns /// - /// A `PipelineResult` with the final payload (type-erased — - /// caller downcasts via `as_any()`), extensions, violation, and - /// the updated context table. + /// A tuple of `(PipelineResult, BackgroundTasks)`. pub async fn invoke( &self, payload: H::Payload, extensions: Extensions, context_table: Option, - ) -> PipelineResult { + ) -> (PipelineResult, BackgroundTasks) { let hook_type = HookType::new(H::NAME); - let entries = self.registry.entries_for_hook(&hook_type); + let all_entries = self.registry.entries_for_hook(&hook_type); + + if all_entries.is_empty() { + let boxed: Box = Box::new(payload); + return ( + PipelineResult::allowed_with( + boxed, + extensions, + context_table.unwrap_or_default(), + ), + BackgroundTasks::empty(), + ); + } + + let entries = self.filter_entries_by_route(all_entries, &extensions, H::NAME); if entries.is_empty() { let boxed: Box = Box::new(payload); - return PipelineResult::allowed_with( - boxed, - extensions, - context_table.unwrap_or_default(), + return ( + PipelineResult::allowed_with( + boxed, + extensions, + context_table.unwrap_or_default(), + ), + BackgroundTasks::empty(), ); } let boxed: Box = Box::new(payload); self.executor - .execute(entries, boxed, extensions, context_table) + .execute(&entries, boxed, extensions, context_table) .await } + // ----------------------------------------------------------------------- + // Route Filtering + // ----------------------------------------------------------------------- + + /// Filter hook entries based on route resolution, with caching. + /// + /// When routing is enabled and extensions.meta provides entity + /// identification, resolves the route and returns only the entries + /// for plugins that match. Results are cached by + /// `(entity_type, entity_name, hook_name, scope)` — subsequent + /// calls for the same key return an `Arc` to the cached entries + /// (refcount bump, no data copy). + /// + /// When routing is disabled or meta is absent, returns all entries. + fn filter_entries_by_route( + &self, + entries: &[crate::registry::HookEntry], + extensions: &Extensions, + hook_name: &str, + ) -> Arc> { + // If no config or routing disabled, return all + let cpex_config = match &self.cpex_config { + Some(c) if c.routing_enabled() => c, + _ => return Arc::new(entries.to_vec()), + }; + + // Extract entity info from meta extension + let meta = match &extensions.meta { + Some(m) => m, + None => return Arc::new(entries.to_vec()), + }; + + let (entity_type, entity_name) = match (&meta.entity_type, &meta.entity_name) { + (Some(t), Some(n)) => (t.as_str(), n.as_str()), + _ => return Arc::new(entries.to_vec()), + }; + + let request_scope = meta.scope.as_deref(); + + // Fast path: zero-allocation cache lookup with raw_entry + let hash = { + use std::hash::BuildHasher; + let mut hasher = self.cache_hasher.build_hasher(); + entity_type.hash(&mut hasher); + entity_name.hash(&mut hasher); + hook_name.hash(&mut hasher); + request_scope.hash(&mut hasher); + hasher.finish() + }; + { + let cache = self.route_cache.read().unwrap(); + if let Some((_, cached)) = cache.raw_entry().from_hash(hash, |key| { + key.entity_type == entity_type + && key.entity_name == entity_name + && key.hook_name == hook_name + && key.scope.as_deref() == request_scope + }) { + return Arc::clone(cached); + } + } + + // Slow path: resolve, filter, and cache (allocations only here) + let resolved = config::resolve_plugins_for_entity( + cpex_config, + entity_type, + entity_name, + request_scope, + &meta.tags, + ); + + // Filter entries to resolved plugins, preserving resolution order. + // If a plugin has config overrides and we have a factory for its kind, + // create a new instance with the merged config. + let mut filtered = Vec::new(); + for resolved_plugin in &resolved { + if let Some(entry) = entries.iter().find(|e| e.plugin_ref.name() == resolved_plugin.name) { + if let Some(overrides) = &resolved_plugin.config_overrides { + // Try to create an override instance + if let Some(override_entry) = self.create_override_instance(entry, overrides) { + filtered.push(override_entry); + continue; + } + } + filtered.push(entry.clone()); + } + } + + let cached = Arc::new(filtered); + + // Store in cache — owned key allocated only on cache miss + let cache_key = RouteCacheKey { + entity_type: entity_type.to_string(), + entity_name: entity_name.to_string(), + hook_name: hook_name.to_string(), + scope: meta.scope.clone(), + }; + { + let mut cache = self.route_cache.write().unwrap(); + cache.insert(cache_key, Arc::clone(&cached)); + } + + cached + } + + /// Create an override plugin instance with merged config. + /// + /// When a route overrides a plugin's config, we create a new + /// instance via the factory with the merged config. Returns + /// None if no factory is available for the plugin's kind. + fn create_override_instance( + &self, + base_entry: &crate::registry::HookEntry, + overrides: &serde_json::Value, + ) -> Option { + let base_config = base_entry.plugin_ref.trusted_config(); + let kind = &base_config.kind; + + let factory = self.factories.get(kind)?; + + // Merge: start with base config, overlay with overrides + let mut merged_config = base_config.clone(); + if let Some(override_config) = overrides.get("config") { + // Merge the plugin-specific config section + if let Some(base_plugin_config) = &merged_config.config { + let mut merged = base_plugin_config.clone(); + if let (Some(base_obj), Some(override_obj)) = + (merged.as_object_mut(), override_config.as_object()) + { + for (key, value) in override_obj { + base_obj.insert(key.clone(), value.clone()); + } + } + merged_config.config = Some(merged); + } else { + merged_config.config = Some(override_config.clone()); + } + } + + // Create new instance with merged config + let target_hook = base_entry.handler.hook_type_name(); + match factory.create(&merged_config) { + Ok(instance) => { + // Find the handler matching the current hook + let handler = instance + .handlers + .into_iter() + .find(|(name, _)| *name == target_hook) + .map(|(_, h)| h); + + if let Some(handler) = handler { + let plugin_ref = + crate::registry::PluginRef::new(instance.plugin, merged_config); + Some(crate::registry::HookEntry { + plugin_ref, + handler, + }) + } else { + warn!( + "Override instance for '{}' has no handler for hook '{}'", + base_config.name, target_hook + ); + None + } + } + Err(e) => { + error!( + "Failed to create override instance for '{}': {}", + base_config.name, e + ); + None // fall back to base instance + } + } + } + + /// Clear the routing cache. Call when config is reloaded or + /// plugins are registered/unregistered. + pub fn clear_routing_cache(&self) { + let mut cache = self.route_cache.write().unwrap(); + cache.clear(); + } + + /// Number of entries in the routing cache. + pub fn routing_cache_size(&self) -> usize { + self.route_cache.read().unwrap().len() + } + // ----------------------------------------------------------------------- // Query Methods // ----------------------------------------------------------------------- @@ -511,6 +945,9 @@ mod tests { plugin_name: "error-plugin".into(), message: "simulated failure".into(), source: None, + code: None, + details: std::collections::HashMap::new(), + proto_error_code: None, }) } @@ -574,12 +1011,12 @@ mod tests { }); - let result = mgr + let (result, _) = mgr .invoke_by_name("test_hook", payload, Extensions::default(), None) .await; - assert!(result.allowed); - assert!(result.payload.is_some()); + assert!(result.continue_processing); + assert!(result.modified_payload.is_some()); } #[tokio::test] @@ -597,11 +1034,11 @@ mod tests { }); - let result = mgr + let (result, _) = mgr .invoke_by_name("test_hook", payload, Extensions::default(), None) .await; - assert!(result.allowed); + assert!(result.continue_processing); } #[tokio::test] @@ -618,11 +1055,11 @@ mod tests { }); - let result = mgr + let (result, _) = mgr .invoke_by_name("test_hook", payload, Extensions::default(), None) .await; - assert!(!result.allowed); + assert!(!result.continue_processing); assert_eq!(result.violation.as_ref().unwrap().code, "denied"); } @@ -640,11 +1077,11 @@ mod tests { }; - let result = mgr + let (result, _) = mgr .invoke::(payload, Extensions::default(), None) .await; - assert!(result.allowed); + assert!(result.continue_processing); } #[tokio::test] @@ -687,12 +1124,12 @@ mod tests { }); - let result = mgr + let (result, _) = mgr .invoke_by_name("test_hook", payload, Extensions::default(), None) .await; // Audit mode — deny is suppressed, pipeline continues - assert!(result.allowed); + assert!(result.continue_processing); } #[tokio::test] @@ -718,8 +1155,8 @@ mod tests { // First invocation — flaky plugin errors, gets disabled, pipeline continues // because on_error is Disable (not Fail). allow-plugin still runs. let payload: Box = Box::new(TestPayload { value: "first".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; - assert!(result.allowed); + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + assert!(result.continue_processing); // Verify the plugin is now disabled let plugin_ref = mgr.get_plugin("flaky-plugin").unwrap(); @@ -729,8 +1166,8 @@ mod tests { // Second invocation — flaky plugin should be skipped entirely // (group_by_mode filters it out). Only allow-plugin runs. let payload2: Box = Box::new(TestPayload { value: "second".into() }); - let result2 = mgr.invoke_by_name("test_hook", payload2, Extensions::default(), None).await; - assert!(result2.allowed); + let (result2, _) = mgr.invoke_by_name("test_hook", payload2, Extensions::default(), None).await; + assert!(result2.continue_processing); } #[tokio::test] @@ -750,8 +1187,8 @@ mod tests { // First invocation — plugin errors, ignored, pipeline continues let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; - assert!(result.allowed); + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + assert!(result.continue_processing); // Plugin should NOT be disabled — still in its original mode let plugin_ref = mgr.get_plugin("flaky-plugin").unwrap(); @@ -776,8 +1213,8 @@ mod tests { // Invocation — plugin errors, pipeline halts with a violation let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; - assert!(!result.allowed); + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + assert!(!result.continue_processing); assert_eq!(result.violation.as_ref().unwrap().code, "plugin_error"); assert_eq!( result.violation.as_ref().unwrap().plugin_name.as_deref(), @@ -848,10 +1285,10 @@ mod tests { let payload = TestPayload { value: "original".into() }; - let result = mgr.invoke::(payload, Extensions::default(), None).await; + let (result, _) = mgr.invoke::(payload, Extensions::default(), None).await; - assert!(result.allowed); - let final_payload = result.payload.unwrap(); + assert!(result.continue_processing); + let final_payload = result.modified_payload.unwrap(); let typed = final_payload.as_any().downcast_ref::().unwrap(); assert_eq!(typed.value, "original_transformed"); } @@ -902,10 +1339,10 @@ mod tests { let start = std::time::Instant::now(); let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; let elapsed = start.elapsed(); - assert!(result.allowed); + assert!(result.continue_processing); assert_eq!(CALL_COUNT.load(Ordering::SeqCst), 2); // If they ran in parallel, total time should be ~50ms, not ~100ms assert!(elapsed.as_millis() < 90, "concurrent plugins ran serially: {}ms", elapsed.as_millis()); @@ -932,11 +1369,11 @@ mod tests { let start = std::time::Instant::now(); let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; let elapsed = start.elapsed(); // Should have timed out and denied (on_error: Fail) - assert!(!result.allowed); + assert!(!result.continue_processing); assert_eq!(result.violation.as_ref().unwrap().code, "plugin_timeout"); // Should have returned in ~1s, not 5s assert!(elapsed.as_secs() < 3, "timeout didn't fire: {}s", elapsed.as_secs()); @@ -980,14 +1417,15 @@ mod tests { mgr.initialize().await.unwrap(); let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + let (result, bg) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; // Pipeline should return immediately — before the background task finishes - assert!(result.allowed); + assert!(result.continue_processing); assert!(!TASK_COMPLETED.load(Ordering::SeqCst), "fire-and-forget task completed before pipeline returned"); - // Wait for the background task to finish - tokio::time::sleep(std::time::Duration::from_millis(300)).await; + // Wait for background tasks using wait_for_background_tasks() + let errors = bg.wait_for_background_tasks().await; + assert!(errors.is_empty(), "background task had errors: {:?}", errors); assert!(TASK_COMPLETED.load(Ordering::SeqCst), "fire-and-forget task never completed"); } @@ -1052,9 +1490,9 @@ mod tests { mgr.initialize().await.unwrap(); let payload: Box = Box::new(TestPayload { value: "test".into() }); - let result = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + let (result, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; - assert!(result.allowed); + assert!(result.continue_processing); assert!( saw_writer.load(std::sync::atomic::Ordering::SeqCst), "reader plugin did not see writer's global_state change" @@ -1098,8 +1536,8 @@ mod tests { // First invocation — no context table, starts fresh let payload: Box = Box::new(TestPayload { value: "first".into() }); - let result1 = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; - assert!(result1.allowed); + let (result1, _) = mgr.invoke_by_name("test_hook", payload, Extensions::default(), None).await; + assert!(result1.continue_processing); // Check call_count = 1 in the returned context table let table = &result1.context_table; @@ -1108,14 +1546,792 @@ mod tests { // Second invocation — pass the context table from the first call let payload2: Box = Box::new(TestPayload { value: "second".into() }); - let result2 = mgr.invoke_by_name( + let (result2, _) = mgr.invoke_by_name( "test_hook", payload2, Extensions::default(), Some(result1.context_table), ).await; - assert!(result2.allowed); + assert!(result2.continue_processing); // call_count should now be 2 — local_state persisted across invocations let table2 = &result2.context_table; let ctx2 = table2.values().next().expect("context table should have one entry"); assert_eq!(ctx2.get_local("call_count").unwrap().as_u64().unwrap(), 2); } + + // -- Factory-based tests -- + + /// A test factory that creates AllowPlugin instances. + struct AllowPluginFactory; + + impl crate::factory::PluginFactory for AllowPluginFactory { + fn create( + &self, + config: &PluginConfig, + ) -> Result { + let plugin = Arc::new(AllowPlugin { cfg: config.clone() }); + let handler: Arc = Arc::new( + TypedHandlerAdapter::::new(Arc::clone(&plugin)), + ); + Ok(crate::factory::PluginInstance { + plugin, + handlers: vec![("test_hook", handler)], + }) + } + } + + /// A test factory that creates DenyPlugin instances. + struct DenyPluginFactory; + + impl crate::factory::PluginFactory for DenyPluginFactory { + fn create( + &self, + config: &PluginConfig, + ) -> Result { + let plugin = Arc::new(DenyPlugin { cfg: config.clone() }); + let handler: Arc = Arc::new( + TypedHandlerAdapter::::new(Arc::clone(&plugin)), + ); + Ok(crate::factory::PluginInstance { + plugin, + handlers: vec![("test_hook", handler)], + }) + } + } + + #[tokio::test] + async fn test_from_config_creates_manager() { + let yaml = r#" +plugins: + - name: allow_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 + +plugin_settings: + plugin_timeout: 60 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + assert_eq!(mgr.plugin_count(), 1); + assert!(mgr.has_hooks_for("test_hook")); + } + + #[tokio::test] + async fn test_from_config_invokes_correctly() { + let yaml = r#" +plugins: + - name: denier + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/deny", Box::new(DenyPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { + value: "test".into(), + }); + // context_table = None (first invocation) + + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + assert!(!result.continue_processing); + assert_eq!(result.violation.as_ref().unwrap().code, "denied"); + } + + #[tokio::test] + async fn test_from_config_unknown_kind_rejected() { + let yaml = r#" +plugins: + - name: mystery + kind: unknown/type + hooks: [test_hook] +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let factories = PluginFactoryRegistry::new(); // empty — no factories + + let result = PluginManager::from_config(cpex_config, &factories); + match result { + Err(e) => assert!(e.to_string().contains("no factory registered"), "got: {}", e), + Ok(_) => panic!("expected error for unknown kind"), + } + } + + #[tokio::test] + async fn test_from_config_multiple_plugins() { + let yaml = r#" +plugins: + - name: gate + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 5 + - name: fallback + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + factories.register("test/deny", Box::new(DenyPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + assert_eq!(mgr.plugin_count(), 2); + + // Deny plugin has higher priority (5 < 10), so it fires first + let payload: Box = Box::new(TestPayload { + value: "test".into(), + }); + // context_table = None (first invocation) + + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + assert!(!result.continue_processing); // gate denied before fallback could allow + } + + // -- Routing cache tests -- + + #[tokio::test] + async fn test_routing_cache_populated_on_first_invoke() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allow_plugin] +plugins: + - name: allow_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 +routes: + - tool: get_compensation +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + assert_eq!(mgr.routing_cache_size(), 0); + + // First invoke — populates cache + let payload: Box = Box::new(TestPayload { value: "test".into() }); + let ext = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + ..Default::default() + }), + ..Default::default() + }; + // context_table = None (first invocation) + mgr.invoke_by_name("test_hook", payload, ext, None).await; + + assert_eq!(mgr.routing_cache_size(), 1); + + // Second invoke — cache hit, still size 1 + let payload2: Box = Box::new(TestPayload { value: "test2".into() }); + let ext2 = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", payload2, ext2, None).await; + + assert_eq!(mgr.routing_cache_size(), 1); // cache hit — no new entry + } + + #[tokio::test] + async fn test_routing_cache_different_entities_separate() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allow_plugin] +plugins: + - name: allow_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential +routes: + - tool: get_compensation + - tool: send_email +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // Invoke for get_compensation + let p1: Box = Box::new(TestPayload { value: "t".into() }); + let e1 = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", p1, e1, None).await; + + // Invoke for send_email + let p2: Box = Box::new(TestPayload { value: "t".into() }); + let e2 = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("send_email".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", p2, e2, None).await; + + assert_eq!(mgr.routing_cache_size(), 2); + } + + #[tokio::test] + async fn test_routing_cache_cleared() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allow_plugin] +plugins: + - name: allow_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential +routes: + - tool: get_compensation +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + let payload: Box = Box::new(TestPayload { value: "t".into() }); + let ext = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", payload, ext, None).await; + assert_eq!(mgr.routing_cache_size(), 1); + + mgr.clear_routing_cache(); + assert_eq!(mgr.routing_cache_size(), 0); + } + + #[tokio::test] + async fn test_routing_cache_scope_creates_separate_entries() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allow_plugin] +plugins: + - name: allow_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential +routes: + - tool: get_compensation +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut factories = PluginFactoryRegistry::new(); + factories.register("test/allow", Box::new(AllowPluginFactory)); + + let mut mgr = PluginManager::from_config(cpex_config, &factories).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // Same entity, different scopes → separate cache entries + let p1: Box = Box::new(TestPayload { value: "t".into() }); + let e1 = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + scope: Some("hr-server".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", p1, e1, None).await; + + let p2: Box = Box::new(TestPayload { value: "t".into() }); + let e2 = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + scope: Some("billing-server".into()), + ..Default::default() + }), + ..Default::default() + }; + mgr.invoke_by_name("test_hook", p2, e2, None).await; + + assert_eq!(mgr.routing_cache_size(), 2); // different scopes → different cache entries + } + + // -- Override instance tests -- + + #[tokio::test] + async fn test_route_override_creates_new_instance() { + let yaml = r#" +plugin_settings: + routing_enabled: true +plugins: + - name: rate_limiter + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 + config: + max_requests: 100 +routes: + - tool: get_compensation + plugins: + - rate_limiter: + config: + max_requests: 10 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + // Use register_factory + load_config so manager owns factories + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // Invoke with routing — should create override instance + let payload: Box = Box::new(TestPayload { value: "t".into() }); + let ext = Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some("get_compensation".into()), + ..Default::default() + }), + ..Default::default() + }; + // context_table = None (first invocation) + + let (result, _) = mgr + .invoke_by_name("test_hook", payload, ext, None) + .await; + + // Plugin executed (allow plugin returns allowed) + assert!(result.continue_processing); + // Cache populated + assert_eq!(mgr.routing_cache_size(), 1); + } + + #[tokio::test] + async fn test_register_factory_then_load_config() { + let yaml = r#" +plugins: + - name: my_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 + +plugin_settings: + plugin_timeout: 45 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + assert_eq!(mgr.plugin_count(), 1); + assert!(mgr.has_hooks_for("test_hook")); + + let payload: Box = Box::new(TestPayload { value: "t".into() }); + // context_table = None (first invocation) + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + assert!(result.continue_processing); + } + + // -- End-to-end routing tests -- + + /// Helper to build meta extensions for routing tests. + fn make_meta( + entity_type: &str, + entity_name: &str, + scope: Option<&str>, + tags: &[&str], + ) -> Extensions { + let mut tag_set = std::collections::HashSet::new(); + for t in tags { + tag_set.insert(t.to_string()); + } + Extensions { + meta: Some(crate::hooks::payload::MetaExtension { + entity_type: Some(entity_type.into()), + entity_name: Some(entity_name.into()), + scope: scope.map(String::from), + tags: tag_set, + ..Default::default() + }), + ..Default::default() + } + } + + #[tokio::test] + async fn test_routing_full_flow_different_tools_different_plugins() { + // Setup: identity fires for all, apl_policy fires for pii tools, + // rate_limiter fires only for get_compensation route + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [identity] + pii: + plugins: [apl_policy] +plugins: + - name: identity + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 1 + - name: apl_policy + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 + - name: rate_limiter + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 5 +routes: + - tool: get_compensation + meta: + tags: [pii] + plugins: + - rate_limiter + - tool: send_email + plugins: + - rate_limiter +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // get_compensation: identity (all) + apl_policy (pii tag) + rate_limiter (route) + // apl_policy denies → overall denied + let p1: Box = Box::new(TestPayload { value: "t".into() }); + let (r1, _) = mgr + .invoke_by_name("test_hook", p1, make_meta("tool", "get_compensation", None, &[]), None) + .await; + assert!(!r1.continue_processing); // apl_policy (deny) fires due to pii tag + + // send_email: identity (all) + rate_limiter (route) — no pii tag + // both allow → overall allowed + let p2: Box = Box::new(TestPayload { value: "t".into() }); + let (r2, _) = mgr + .invoke_by_name("test_hook", p2, make_meta("tool", "send_email", None, &[]), None) + .await; + assert!(r2.continue_processing); // no deny plugin fires + } + + #[tokio::test] + async fn test_routing_disabled_fires_all_plugins() { + // Same plugins but routing disabled — all fire regardless of entity + let yaml = r#" +plugins: + - name: denier + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 + - name: allower + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 20 +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // Even with meta, routing disabled → all plugins fire → denier wins + let p: Box = Box::new(TestPayload { value: "t".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", p, make_meta("tool", "anything", None, &[]), None) + .await; + assert!(!result.continue_processing); // denier fires (all plugins active) + } + + #[tokio::test] + async fn test_routing_no_meta_fires_all_plugins() { + // Routing enabled but no meta on extensions → fallback to all + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allower] +plugins: + - name: allower + kind: test/allow + hooks: [test_hook] + mode: sequential + - name: denier + kind: test/deny + hooks: [test_hook] + mode: sequential +routes: + - tool: get_compensation + plugins: + - denier +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // No meta → all plugins fire (both allower and denier) + let p: Box = Box::new(TestPayload { value: "t".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", p, Extensions::default(), None) + .await; + // denier has default priority 100, allower has default 100 — order depends on registration + // but at least both fire (not filtered by routing) + // We can't assert allow/deny specifically since both run — just check it executed + assert!(result.continue_processing || !result.continue_processing); // both plugins fired + } + + #[tokio::test] + async fn test_routing_wildcard_catches_unmatched() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [identity] +plugins: + - name: identity + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 1 + - name: specific_plugin + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 + - name: fallback_plugin + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 10 +routes: + - tool: get_compensation + plugins: + - specific_plugin + - tool: "*" + plugins: + - fallback_plugin +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // get_compensation matches exact route → specific_plugin (deny) + let p1: Box = Box::new(TestPayload { value: "t".into() }); + let (r1, _) = mgr + .invoke_by_name("test_hook", p1, make_meta("tool", "get_compensation", None, &[]), None) + .await; + assert!(!r1.continue_processing); // specific_plugin denies + + // unknown_tool matches wildcard → fallback_plugin (allow) + let p2: Box = Box::new(TestPayload { value: "t".into() }); + let (r2, _) = mgr + .invoke_by_name("test_hook", p2, make_meta("tool", "unknown_tool", None, &[]), None) + .await; + assert!(r2.continue_processing); // fallback_plugin allows + } + + #[tokio::test] + async fn test_routing_host_tags_activate_policy_groups() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [identity] + urgent: + plugins: [denier] +plugins: + - name: identity + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 1 + - name: denier + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 +routes: + - tool: get_compensation +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // Without urgent tag → only identity fires → allowed + let p1: Box = Box::new(TestPayload { value: "t".into() }); + let (r1, _) = mgr + .invoke_by_name("test_hook", p1, make_meta("tool", "get_compensation", None, &[]), None) + .await; + assert!(r1.continue_processing); + + // Clear cache so new tags take effect + mgr.clear_routing_cache(); + + // With urgent tag from host → denier also fires → denied + let p2: Box = Box::new(TestPayload { value: "t".into() }); + let (r2, _) = mgr + .invoke_by_name("test_hook", p2, make_meta("tool", "get_compensation", None, &["urgent"]), None) + .await; + assert!(!r2.continue_processing); + } + + #[tokio::test] + async fn test_routing_works_with_typed_invoke() { + let yaml = r#" +plugin_settings: + routing_enabled: true +global: + policies: + all: + plugins: [allower] + pii: + plugins: [denier] +plugins: + - name: allower + kind: test/allow + hooks: [test_hook] + mode: sequential + priority: 1 + - name: denier + kind: test/deny + hooks: [test_hook] + mode: sequential + priority: 10 +routes: + - tool: get_compensation + meta: + tags: [pii] + - tool: send_email +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + let mut mgr = PluginManager::default(); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.register_factory("test/deny", Box::new(DenyPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + // context_table = None (first invocation) + + // Typed invoke for get_compensation — pii tag activates denier → denied + let (r1, _) = mgr + .invoke::( + TestPayload { value: "t".into() }, + make_meta("tool", "get_compensation", None, &[]), + None, + ) + .await; + assert!(!r1.continue_processing); + + // Typed invoke for send_email — no pii tag → only allower → allowed + let (r2, _) = mgr + .invoke::( + TestPayload { value: "t".into() }, + make_meta("tool", "send_email", None, &[]), + None, + ) + .await; + assert!(r2.continue_processing); + } } diff --git a/crates/cpex-core/src/plugin.rs b/crates/cpex-core/src/plugin.rs index 9d4a00a6..b9c13f11 100644 --- a/crates/cpex-core/src/plugin.rs +++ b/crates/cpex-core/src/plugin.rs @@ -89,13 +89,19 @@ pub trait Plugin: Send + Sync { /// /// Called before any hook invocations. Use this to establish /// connections, load resources, or validate configuration. - async fn initialize(&self) -> Result<(), PluginError>; + /// Default implementation does nothing. + async fn initialize(&self) -> Result<(), PluginError> { + Ok(()) + } /// Graceful shutdown. /// /// Called once during teardown. Use this to flush buffers, close /// connections, or release resources. - async fn shutdown(&self) -> Result<(), PluginError>; + /// Default implementation does nothing. + async fn shutdown(&self) -> Result<(), PluginError> { + Ok(()) + } } // --------------------------------------------------------------------------- diff --git a/crates/cpex-core/src/registry.rs b/crates/cpex-core/src/registry.rs index cbc88a9c..fce0466c 100644 --- a/crates/cpex-core/src/registry.rs +++ b/crates/cpex-core/src/registry.rs @@ -278,6 +278,66 @@ impl PluginRegistry { self.register_for_names_inner(plugin, config, handler, names) } + /// Register a plugin with a handler for multiple hook names. + /// + /// Like `register_for_names` but without requiring a `HookTypeDef` + /// type parameter. Used by the config-driven factory path where + /// the hook type is not known at compile time — the factory + /// provides the handler directly. + pub fn register_for_names_with_handler( + &mut self, + plugin: Arc, + config: PluginConfig, + handler: Arc, + names: &[&str], + ) -> Result<(), String> { + self.register_for_names_inner(plugin, config, handler, names) + } + + + /// Register a plugin with multiple handlers, each for a specific hook. + /// + /// Used when a plugin implements multiple hook types with different + /// payloads (e.g., `ToolPreInvoke` and `ToolPostInvoke`). Each + /// handler is registered under its paired hook name. + /// + /// The plugin is registered once in the name index. Each handler + /// gets its own `HookEntry` in the hook index under the specified name. + pub fn register_multi_handler( + &mut self, + plugin: Arc, + config: PluginConfig, + handlers: Vec<(&str, Arc)>, + ) -> Result<(), String> { + let name = config.name.clone(); + + if self.plugins.contains_key(&name) { + return Err(format!("plugin '{}' is already registered", name)); + } + + let plugin_ref = PluginRef::new(plugin, config); + + for (hook_name, handler) in &handlers { + let hook_type = HookType::new(*hook_name); + let entry = HookEntry { + plugin_ref: plugin_ref.clone(), + handler: Arc::clone(handler), + }; + self.hook_index.entry(hook_type).or_default().push(entry); + } + + // Sort each affected hook's entry list by trusted priority + for (hook_name, _) in &handlers { + let hook_type = HookType::new(*hook_name); + if let Some(entries) = self.hook_index.get_mut(&hook_type) { + entries.sort_by_key(|e| e.plugin_ref.priority()); + } + } + + self.plugins.insert(name, plugin_ref); + Ok(()) + } + /// Internal: register handler under one or more hook names. fn register_for_names_inner( &mut self,