cpex_core/

plugin.rs

// Location: ./crates/cpex-core/src/plugin.rs
// Copyright 2025
// SPDX-License-Identifier: Apache-2.0
// Authors: Teryl Taylor
//
// Plugin trait and supporting types.
//
// Defines the core Plugin trait that all plugin implementations satisfy —
// native Rust, WASM hosts, Python bridge hosts, and dlopen'd shared
// libraries. Also defines PluginConfig (YAML-declared plugin settings),
// PluginMode (5-phase execution modes), and OnError (failure behavior).
//
// The Plugin trait handles lifecycle only (initialize, shutdown, config).
// Hook-specific logic is defined by handler traits generated by the
// define_hook! macro (see hooks/macros.rs). A plugin implements Plugin
// for lifecycle + one or more handler traits for the hooks it handles.
//
// The manager wraps each plugin in a PluginRef with an authoritative
// config from the config loader — the plugin's own config() is for
// the plugin's reading only, never used by the executor for scheduling.

use std::collections::HashSet;
use std::fmt;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::error::PluginError;

// ---------------------------------------------------------------------------
// Plugin Trait
// ---------------------------------------------------------------------------

/// Core plugin interface — lifecycle management only.
///
/// Every plugin in the CPEX framework — regardless of language or
/// deployment model — implements this trait. It covers lifecycle
/// (initialize, shutdown) and identity (config). Hook-specific logic
/// is defined separately by handler traits generated by `define_hook!`.
///
/// # Lifecycle
///
/// 1. `initialize()` — called once after loading, before any hooks fire.
/// 2. Hook handlers — called on each hook invocation (defined by handler traits).
/// 3. `shutdown()` — called once during graceful teardown.
///
/// # Hook Handlers
///
/// A plugin implements one or more handler traits alongside Plugin:
///
/// ```rust,ignore
/// impl Plugin for MyPlugin {
///     fn config(&self) -> &PluginConfig { &self.config }
///     async fn initialize(&self) -> Result<(), Box<PluginError>> { Ok(()) }
///     async fn shutdown(&self) -> Result<(), Box<PluginError>> { Ok(()) }
/// }
///
/// impl CmfHookHandler for MyPlugin {
///     fn cmf_hook(&self, payload: MessagePayload, ext: &Extensions, ctx: &PluginContext) -> PluginResult<MessagePayload> {
///         PluginResult::allow()
///     }
/// }
/// ```
///
/// # Trust Model
///
/// The manager wraps each plugin in a `PluginRef` with an authoritative
/// config from the config loader. The executor reads scheduling decisions
/// (mode, priority, hooks, capabilities) from the `PluginRef` — never
/// from `plugin.config()`. The plugin's own `config()` is available for
/// the plugin's reading during hook execution.
///
/// # Implementors
///
/// - Native Rust plugins (implement directly)
/// - `cpex-hosts::wasm` (bridges to WASM guest via wasmtime)
/// - `cpex-hosts::python` (bridges to Python plugin classes via PyO3)
/// - `cpex-hosts::native` (bridges to dlopen'd shared libraries)
#[async_trait]
pub trait Plugin: Send + Sync {
    /// Returns the plugin's configuration.
    ///
    /// Available for the plugin's own reading during hook execution.
    /// The manager/executor never reads this — they use the authoritative
    /// config from `PluginRef.trusted_config()`.
    fn config(&self) -> &PluginConfig;

    /// One-time initialization after loading.
    ///
    /// Called before any hook invocations. Use this to establish
    /// connections, load resources, or validate configuration.
    /// Default implementation does nothing.
    async fn initialize(&self) -> Result<(), Box<PluginError>> {
        Ok(())
    }

    /// Graceful shutdown.
    ///
    /// Called once during teardown. Use this to flush buffers, close
    /// connections, or release resources.
    /// Default implementation does nothing.
    async fn shutdown(&self) -> Result<(), Box<PluginError>> {
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Plugin Configuration
// ---------------------------------------------------------------------------

/// Declared plugin configuration from the unified YAML config.
///
/// Controls how the framework loads, schedules, and gates the plugin.
/// Corresponds to a single entry in the `plugins:` list in config YAML.
///
/// The manager holds the authoritative copy in `PluginRef.trusted_config`.
/// The plugin receives its own copy for reading via `Plugin::config()`.
///
/// # Examples
///
/// ```yaml
/// plugins:
///   - name: apl-policy
///     kind: builtin
///     hooks: [tool_pre_invoke, tool_post_invoke]
///     mode: sequential
///     priority: 10
///     on_error: fail
///     capabilities: [read_security, append_labels]
///     config:
///       policy_file: apl/demo/hr_policy.yaml
/// ```
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PluginConfig {
    /// Unique plugin name.
    pub name: String,

    /// Plugin kind — determines how the framework loads it.
    ///
    /// - `"builtin"` — compiled into the runtime
    /// - `"native://path/to/lib.so"` — dlopen'd shared library
    /// - `"wasm://path/to/plugin.wasm"` — wasmtime sandbox
    /// - `"python://module.path.ClassName"` — PyO3 bridge
    /// - `"external"` — MCP/gRPC/Unix socket transport
    pub kind: String,

    /// Human-readable description.
    #[serde(default)]
    pub description: Option<String>,

    /// Plugin author or team.
    #[serde(default)]
    pub author: Option<String>,

    /// Semantic version string.
    #[serde(default)]
    pub version: Option<String>,

    /// Hook names this plugin handles.
    #[serde(default)]
    pub hooks: Vec<String>,

    /// Execution mode — determines scheduling behavior and authority.
    #[serde(default)]
    pub mode: PluginMode,

    /// Execution priority — lower numbers execute first within each mode.
    #[serde(default = "default_priority")]
    pub priority: i32,

    /// Error handling behavior when the plugin fails.
    #[serde(default)]
    pub on_error: OnError,

    /// Declared capabilities for extension visibility gating.
    ///
    /// Controls which extensions the plugin can see and modify.
    /// Extensions not covered by declared capabilities appear as
    /// `None` in the filtered view.
    #[serde(default)]
    pub capabilities: HashSet<String>,

    /// Tags for categorization and searchability.
    #[serde(default)]
    pub tags: Vec<String>,

    /// Legacy conditions for when the plugin should execute.
    ///
    /// Each condition narrows the plugin's scope by server, tenant,
    /// tool name, prompt name, etc. If any condition in the list
    /// matches, the plugin runs. If the list is empty (default),
    /// the plugin runs unconditionally.
    ///
    /// **Backward compatibility:** Conditions are the legacy mechanism
    /// for scoping plugins. When the host uses the unified routing
    /// system (`routes:` in config YAML), routing rules handle scope
    /// matching and conditions on the plugin are ignored. The two
    /// mechanisms should not be used together on the same plugin.
    #[serde(default)]
    pub conditions: Vec<PluginCondition>,

    /// Plugin-specific configuration (opaque to the framework).
    #[serde(default)]
    pub config: Option<serde_json::Value>,
}

impl PluginConfig {
    /// Whether this plugin's `conditions` allow it to fire for the given
    /// request `Extensions`. Used in legacy mode (`routing_enabled: false`)
    /// to filter which plugins run per request — mirrors the Python
    /// implementation's per-plugin condition filtering.
    ///
    /// Semantics:
    /// - Empty `conditions` Vec → fire always (no restriction).
    /// - Non-empty → fire if ANY condition matches (OR across the list,
    ///   AND within each individual condition).
    ///
    /// Field-source mapping (see project memory `project_conditions_field_mapping`):
    /// - `server_ids` ← `extensions.mcp.{tool|resource|prompt}.server_id`
    /// - `tenant_ids` ← `extensions.security.subject.claims["tenant"]`
    /// - `tools|prompts|resources` ← `extensions.meta.entity_name` (when matching `entity_type`)
    /// - `agents` ← `extensions.agent.agent_id`
    /// - `user_patterns` ← `extensions.security.subject.id` (glob match)
    /// - `content_types` ← `extensions.mcp.resource.mime_type`
    pub fn passes_conditions(&self, extensions: &crate::hooks::payload::Extensions) -> bool {
        if self.conditions.is_empty() {
            return true;
        }

        // Source values once from the extensions tree.
        let server_id = extensions.mcp.as_ref().and_then(|m| {
            m.tool
                .as_ref()
                .and_then(|t| t.server_id.as_deref())
                .or_else(|| m.resource.as_ref().and_then(|r| r.server_id.as_deref()))
                .or_else(|| m.prompt.as_ref().and_then(|p| p.server_id.as_deref()))
        });
        let tenant_id = extensions
            .security
            .as_ref()
            .and_then(|s| s.subject.as_ref())
            .and_then(|sub| sub.claims.get("tenant"))
            .map(|s| s.as_str());
        let entity_name = extensions
            .meta
            .as_ref()
            .and_then(|m| m.entity_name.as_deref());
        let entity_type = extensions
            .meta
            .as_ref()
            .and_then(|m| m.entity_type.as_deref());
        let (tool, prompt, resource) = match entity_type {
            Some("tool") => (entity_name, None, None),
            Some("prompt") => (None, entity_name, None),
            Some("resource") => (None, None, entity_name),
            _ => (None, None, None),
        };
        let agent = extensions
            .agent
            .as_ref()
            .and_then(|a| a.agent_id.as_deref());
        let user = extensions
            .security
            .as_ref()
            .and_then(|s| s.subject.as_ref())
            .and_then(|sub| sub.id.as_deref());
        let content_type = extensions
            .mcp
            .as_ref()
            .and_then(|m| m.resource.as_ref())
            .and_then(|r| r.mime_type.as_deref());

        let ctx = MatchContext {
            server_id,
            tenant_id,
            tool,
            prompt,
            resource,
            agent,
            user,
            content_type,
        };
        self.conditions.iter().any(|c| c.matches(&ctx))
    }
}

fn default_priority() -> i32 {
    100
}

// ---------------------------------------------------------------------------
// Plugin Condition (legacy scoping)
// ---------------------------------------------------------------------------

/// Condition for when a plugin should execute.
///
/// Narrows plugin scope to specific servers, tenants, tools, prompts,
/// resources, or agents. All fields are optional — only specified
/// fields participate in matching. Within a field, any match suffices
/// (OR semantics). Across fields, all must match (AND semantics).
///
/// This is the legacy scoping mechanism. The unified routing system
/// (`routes:` in config) supersedes this — when routes are used,
/// conditions are ignored.
///
/// Mirrors Python's `PluginCondition` in `cpex/framework/models.py`.
///
/// # Examples
///
/// ```
/// use cpex_core::plugin::PluginCondition;
///
/// // Only run for specific tools on specific servers
/// let cond = PluginCondition {
///     server_ids: Some(vec!["server-1".into(), "server-2".into()].into_iter().collect()),
///     tools: Some(vec!["get_compensation".into()].into_iter().collect()),
///     ..Default::default()
/// };
/// assert!(cond.server_ids.as_ref().unwrap().contains("server-1"));
/// assert!(cond.tools.as_ref().unwrap().contains("get_compensation"));
/// ```
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PluginCondition {
    /// Set of server IDs — plugin runs only on these servers.
    #[serde(default)]
    pub server_ids: Option<HashSet<String>>,

    /// Set of tenant IDs — plugin runs only for these tenants.
    #[serde(default)]
    pub tenant_ids: Option<HashSet<String>>,

    /// Set of tool names — plugin runs only for these tools.
    #[serde(default)]
    pub tools: Option<HashSet<String>>,

    /// Set of prompt names — plugin runs only for these prompts.
    #[serde(default)]
    pub prompts: Option<HashSet<String>>,

    /// Set of resource identifiers — plugin runs only for these resources.
    #[serde(default)]
    pub resources: Option<HashSet<String>>,

    /// Set of agent identifiers — plugin runs only for these agents.
    #[serde(default)]
    pub agents: Option<HashSet<String>>,

    /// User patterns (glob or regex) — plugin runs only for matching users.
    #[serde(default)]
    pub user_patterns: Option<Vec<String>>,

    /// Content types — plugin runs only for these content types.
    #[serde(default)]
    pub content_types: Option<Vec<String>>,
}

/// Bundle of optional context values used to evaluate a `PluginCondition`.
///
/// Each field corresponds to one of the condition's gates. `None` means
/// "no value sourced from the extensions tree"; the condition then
/// rejects when the corresponding `Some(set)` is set on the condition
/// (i.e., the gate was specified but couldn't be evaluated).
///
/// Replaces an 8-arg `matches(...)` call where every arg was
/// `Option<&str>` and could be misordered silently.
#[derive(Debug, Default, Clone, Copy)]
pub struct MatchContext<'a> {
    pub server_id: Option<&'a str>,
    pub tenant_id: Option<&'a str>,
    pub tool: Option<&'a str>,
    pub prompt: Option<&'a str>,
    pub resource: Option<&'a str>,
    pub agent: Option<&'a str>,
    pub user: Option<&'a str>,
    pub content_type: Option<&'a str>,
}

impl PluginCondition {
    /// Whether this condition matches the given context.
    ///
    /// A field that is `None` is treated as "any" (no restriction).
    /// A `Some(set)` field matches if the given value is in the set
    /// (exact match for ID-shaped fields; glob match via `wildmatch`
    /// for `user_patterns`).
    /// All specified fields must match — AND semantics within one condition.
    pub fn matches(&self, ctx: &MatchContext<'_>) -> bool {
        let MatchContext {
            server_id,
            tenant_id,
            tool,
            prompt,
            resource,
            agent,
            user,
            content_type,
        } = *ctx;
        let check_set = |field: &Option<HashSet<String>>, value: Option<&str>| -> bool {
            match field {
                None => true, // not specified — matches anything
                Some(set) => match value {
                    Some(v) => set.contains(v),
                    None => false, // field required but no value provided
                },
            }
        };

        // user_patterns: list of globs. Match if any pattern matches the user.
        let check_patterns = |field: &Option<Vec<String>>, value: Option<&str>| -> bool {
            match field {
                None => true,
                Some(patterns) => match value {
                    Some(v) => patterns
                        .iter()
                        .any(|p| wildmatch::WildMatch::new(p).matches(v)),
                    None => false,
                },
            }
        };

        // content_types: list of exact strings.
        let check_list = |field: &Option<Vec<String>>, value: Option<&str>| -> bool {
            match field {
                None => true,
                Some(list) => match value {
                    Some(v) => list.iter().any(|s| s == v),
                    None => false,
                },
            }
        };

        check_set(&self.server_ids, server_id)
            && check_set(&self.tenant_ids, tenant_id)
            && check_set(&self.tools, tool)
            && check_set(&self.prompts, prompt)
            && check_set(&self.resources, resource)
            && check_set(&self.agents, agent)
            && check_patterns(&self.user_patterns, user)
            && check_list(&self.content_types, content_type)
    }
}

// ---------------------------------------------------------------------------
// Plugin Mode
// ---------------------------------------------------------------------------

/// Execution mode — determines a plugin's scheduling behavior and authority.
///
/// The 5-phase model defines both what a plugin *can do* (block, modify)
/// and *how it runs* (serial, parallel, background). Scheduling is derived
/// from mode; plugin authors don't control it directly.
///
/// # Execution Order
///
/// ```text
/// SEQUENTIAL → TRANSFORM → AUDIT → CONCURRENT → FIRE_AND_FORGET
/// ```
///
/// # Mode Capabilities
///
/// | Mode           | Can Block? | Can Modify? | Execution       |
/// |----------------|------------|-------------|-----------------|
/// | Sequential     | Yes        | Yes         | Serial, chained |
/// | Transform      | No         | Yes         | Serial, chained |
/// | Audit          | No         | No          | Serial          |
/// | Concurrent     | Yes        | No          | Parallel        |
/// | FireAndForget  | No         | No          | Background      |
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum PluginMode {
    /// Policy enforcement + transformation. Serial, chained. Can block and modify.
    #[default]
    Sequential,

    /// Data shaping (PII redaction, normalization). Serial, chained. Can modify, cannot block.
    Transform,

    /// Observation and logging. Serial, read-only. Cannot block or modify.
    Audit,

    /// Independent policy gates. Parallel, fail-fast. Can block, cannot modify.
    Concurrent,

    /// Telemetry and async side effects. Background tasks. Cannot block or modify.
    FireAndForget,

    /// Plugin is disabled — skipped during execution.
    Disabled,
}

impl PluginMode {
    /// Whether this mode allows the plugin to block the pipeline.
    pub fn can_block(&self) -> bool {
        matches!(self, Self::Sequential | Self::Concurrent)
    }

    /// Whether this mode allows the plugin to modify the payload.
    pub fn can_modify(&self) -> bool {
        matches!(self, Self::Sequential | Self::Transform)
    }

    /// Whether the framework waits for this plugin to complete.
    pub fn is_awaited(&self) -> bool {
        !matches!(self, Self::FireAndForget | Self::Disabled)
    }
}

impl fmt::Display for PluginMode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Sequential => write!(f, "sequential"),
            Self::Transform => write!(f, "transform"),
            Self::Audit => write!(f, "audit"),
            Self::Concurrent => write!(f, "concurrent"),
            Self::FireAndForget => write!(f, "fire_and_forget"),
            Self::Disabled => write!(f, "disabled"),
        }
    }
}

// ---------------------------------------------------------------------------
// Error Handling Mode
// ---------------------------------------------------------------------------

/// Error handling behavior when a plugin fails.
///
/// Independent of [`PluginMode`] — any mode can use any error behavior.
/// Controls whether plugin failures halt the pipeline, are logged and
/// skipped, or cause the plugin to be auto-disabled.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum OnError {
    /// Pipeline halts and error propagates. Fail-safe enforcement.
    #[default]
    Fail,

    /// Error logged, pipeline continues. For non-critical plugins.
    Ignore,

    /// Plugin auto-disabled after error. Prevents repeated failures.
    Disable,
}

impl fmt::Display for OnError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Fail => write!(f, "fail"),
            Self::Ignore => write!(f, "ignore"),
            Self::Disable => write!(f, "disable"),
        }
    }
}