Struct PluginManager 

pub struct PluginManager { /* private fields */ }

Implementations

impl PluginManager

pub fn new(config: ManagerConfig) -> Self

Create a new PluginManager with the given configuration.

pub fn config_generation(&self) -> u64

Monotonic counter that increments on every runtime snapshot swap (registry mutation, config (re)load). External orchestrators (e.g. apl-cpex’s dispatch-plan cache) pair their cached values with the generation seen at build time; a mismatch on lookup signals “evict + rebuild.” Acquire pairs with the Release fetch_add in mutate_runtime / try_mutate_runtime so observing a higher generation guarantees visibility of the new snapshot.

pub fn register_factory( &self, kind: impl Into<String>, factory: Box<dyn PluginFactory>, )

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

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"))?;

Examples found in repository

examples/plugin_demo.rs (line 422)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

examples/cmf_capabilities_demo.rs (line 337)

async fn main() {
    println!("=== CMF Capabilities Demo ===\n");

    // Load config from YAML file — capabilities declared per plugin
    let config_path = "crates/cpex-core/examples/cmf_capabilities_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 mgr = PluginManager::default();
    mgr.register_factory("builtin/identity-checker", Box::new(IdentityCheckerFactory));
    mgr.register_factory("builtin/header-injector", Box::new(HeaderInjectorFactory));
    mgr.register_factory("builtin/audit-logger", Box::new(AuditLoggerFactory));
    mgr.load_config(cpex_config).unwrap();
    mgr.initialize().await.unwrap();

    // --- Build CMF Message ---
    let payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Assistant,
            content: vec![
                ContentPart::Text {
                    text: "Looking up compensation.".into(),
                },
                ContentPart::ToolCall {
                    content: ToolCall {
                        tool_call_id: "tc_001".into(),
                        name: "get_compensation".into(),
                        arguments: [("employee_id".to_string(), serde_json::json!(42))].into(),
                        namespace: None,
                    },
                },
            ],
            channel: None,
        },
    };

    // --- Build Extensions with security, HTTP, meta ---
    let mut security = SecurityExtension::default();
    security.add_label("PII");
    security.add_label("HR_DATA");
    security.classification = Some("confidential".into());
    security.subject = Some(cpex_core::extensions::security::SubjectExtension {
        id: Some("alice".into()),
        subject_type: Some(cpex_core::extensions::security::SubjectType::User),
        roles: ["hr_admin".to_string()].into(),
        permissions: ["read_compensation".to_string()].into(),
        ..Default::default()
    });

    let mut http = HttpExtension::default();
    http.set_header("Authorization", "Bearer eyJ...");
    http.set_header("X-Request-ID", "req-abc-123");

    let ext = Extensions {
        request: Some(Arc::new(RequestExtension {
            environment: Some("production".into()),
            request_id: Some("req-abc-123".into()),
            ..Default::default()
        })),
        security: Some(Arc::new(security)),
        http: Some(Arc::new(http)),
        meta: Some(Arc::new(MetaExtension {
            entity_type: Some("tool".into()),
            entity_name: Some("get_compensation".into()),
            tags: ["pii".to_string(), "hr".to_string()].into(),
            ..Default::default()
        })),
        ..Default::default()
    };

    // --- Pre-invoke: type-safe dispatch via invoke_named ---
    println!("=== Phase 1: cmf.tool_pre_invoke ===\n");

    // invoke_named<CmfHook> gives compile-time payload type checking
    // while routing to the specific "cmf.tool_pre_invoke" hook name
    let (pre_result, bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_pre_invoke",
            payload,
            ext,
            None, // first hook — no context table
        )
        .await;

    println!();
    if pre_result.continue_processing {
        println!("Pre-invoke result: ALLOWED");
        if let Some(ref modified_ext) = pre_result.modified_extensions {
            if let Some(ref sec) = modified_ext.security {
                let labels: Vec<&String> = sec.labels.iter().collect();
                println!("  Labels after pre-invoke: {:?}", labels);
            }
            if let Some(ref http) = modified_ext.http {
                println!("  Headers after pre-invoke: {:?}", http.request_headers);
            }
        }
    } else {
        println!(
            "Pre-invoke result: DENIED — {}",
            pre_result.violation.as_ref().unwrap().reason
        );
        bg.wait_for_background_tasks().await;
        println!("\n=== Demo complete ===");
        return;
    }
    bg.wait_for_background_tasks().await;

    // --- Simulate tool execution ---
    println!("\n--- Tool 'get_compensation' executes... ---");
    println!("  Result: {{\"salary\": 150000, \"currency\": \"USD\"}}\n");

    // --- Post-invoke: different CMF message with tool result ---
    println!("=== Phase 2: cmf.tool_post_invoke ===\n");

    let post_payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Tool,
            content: vec![ContentPart::ToolResult {
                content: cpex_core::cmf::ToolResult {
                    tool_call_id: "tc_001".into(),
                    tool_name: "get_compensation".into(),
                    content: serde_json::json!({"salary": 150000, "currency": "USD"}),
                    is_error: false,
                },
            }],
            channel: None,
        },
    };

    // Build post-invoke extensions — carry forward any modifications
    // from pre-invoke via the context table
    let post_ext = pre_result.modified_extensions.unwrap_or_else(|| {
        // Rebuild if no modifications
        let mut security = SecurityExtension::default();
        security.add_label("PII");
        Extensions {
            security: Some(Arc::new(security)),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                entity_name: Some("get_compensation".into()),
                ..Default::default()
            })),
            ..Default::default()
        }
    });

    // Thread the context table from pre-invoke to preserve plugin state
    let (post_result, post_bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_post_invoke",
            post_payload,
            post_ext,
            Some(pre_result.context_table),
        )
        .await;

    println!();
    if post_result.continue_processing {
        println!("Post-invoke result: ALLOWED");
    } else {
        println!(
            "Post-invoke result: DENIED — {}",
            post_result.violation.as_ref().unwrap().reason
        );
    }

    post_bg.wait_for_background_tasks().await;
    println!("\n=== Demo complete ===");
}

pub fn load_config_file(&self, path: &Path) -> Result<(), Box<PluginError>>

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

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( &self, cpex_config: CpexConfig, ) -> Result<(), Box<PluginError>>

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.

Examples found in repository

examples/plugin_demo.rs (line 426)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

examples/cmf_capabilities_demo.rs (line 340)

async fn main() {
    println!("=== CMF Capabilities Demo ===\n");

    // Load config from YAML file — capabilities declared per plugin
    let config_path = "crates/cpex-core/examples/cmf_capabilities_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 mgr = PluginManager::default();
    mgr.register_factory("builtin/identity-checker", Box::new(IdentityCheckerFactory));
    mgr.register_factory("builtin/header-injector", Box::new(HeaderInjectorFactory));
    mgr.register_factory("builtin/audit-logger", Box::new(AuditLoggerFactory));
    mgr.load_config(cpex_config).unwrap();
    mgr.initialize().await.unwrap();

    // --- Build CMF Message ---
    let payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Assistant,
            content: vec![
                ContentPart::Text {
                    text: "Looking up compensation.".into(),
                },
                ContentPart::ToolCall {
                    content: ToolCall {
                        tool_call_id: "tc_001".into(),
                        name: "get_compensation".into(),
                        arguments: [("employee_id".to_string(), serde_json::json!(42))].into(),
                        namespace: None,
                    },
                },
            ],
            channel: None,
        },
    };

    // --- Build Extensions with security, HTTP, meta ---
    let mut security = SecurityExtension::default();
    security.add_label("PII");
    security.add_label("HR_DATA");
    security.classification = Some("confidential".into());
    security.subject = Some(cpex_core::extensions::security::SubjectExtension {
        id: Some("alice".into()),
        subject_type: Some(cpex_core::extensions::security::SubjectType::User),
        roles: ["hr_admin".to_string()].into(),
        permissions: ["read_compensation".to_string()].into(),
        ..Default::default()
    });

    let mut http = HttpExtension::default();
    http.set_header("Authorization", "Bearer eyJ...");
    http.set_header("X-Request-ID", "req-abc-123");

    let ext = Extensions {
        request: Some(Arc::new(RequestExtension {
            environment: Some("production".into()),
            request_id: Some("req-abc-123".into()),
            ..Default::default()
        })),
        security: Some(Arc::new(security)),
        http: Some(Arc::new(http)),
        meta: Some(Arc::new(MetaExtension {
            entity_type: Some("tool".into()),
            entity_name: Some("get_compensation".into()),
            tags: ["pii".to_string(), "hr".to_string()].into(),
            ..Default::default()
        })),
        ..Default::default()
    };

    // --- Pre-invoke: type-safe dispatch via invoke_named ---
    println!("=== Phase 1: cmf.tool_pre_invoke ===\n");

    // invoke_named<CmfHook> gives compile-time payload type checking
    // while routing to the specific "cmf.tool_pre_invoke" hook name
    let (pre_result, bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_pre_invoke",
            payload,
            ext,
            None, // first hook — no context table
        )
        .await;

    println!();
    if pre_result.continue_processing {
        println!("Pre-invoke result: ALLOWED");
        if let Some(ref modified_ext) = pre_result.modified_extensions {
            if let Some(ref sec) = modified_ext.security {
                let labels: Vec<&String> = sec.labels.iter().collect();
                println!("  Labels after pre-invoke: {:?}", labels);
            }
            if let Some(ref http) = modified_ext.http {
                println!("  Headers after pre-invoke: {:?}", http.request_headers);
            }
        }
    } else {
        println!(
            "Pre-invoke result: DENIED — {}",
            pre_result.violation.as_ref().unwrap().reason
        );
        bg.wait_for_background_tasks().await;
        println!("\n=== Demo complete ===");
        return;
    }
    bg.wait_for_background_tasks().await;

    // --- Simulate tool execution ---
    println!("\n--- Tool 'get_compensation' executes... ---");
    println!("  Result: {{\"salary\": 150000, \"currency\": \"USD\"}}\n");

    // --- Post-invoke: different CMF message with tool result ---
    println!("=== Phase 2: cmf.tool_post_invoke ===\n");

    let post_payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Tool,
            content: vec![ContentPart::ToolResult {
                content: cpex_core::cmf::ToolResult {
                    tool_call_id: "tc_001".into(),
                    tool_name: "get_compensation".into(),
                    content: serde_json::json!({"salary": 150000, "currency": "USD"}),
                    is_error: false,
                },
            }],
            channel: None,
        },
    };

    // Build post-invoke extensions — carry forward any modifications
    // from pre-invoke via the context table
    let post_ext = pre_result.modified_extensions.unwrap_or_else(|| {
        // Rebuild if no modifications
        let mut security = SecurityExtension::default();
        security.add_label("PII");
        Extensions {
            security: Some(Arc::new(security)),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                entity_name: Some("get_compensation".into()),
                ..Default::default()
            })),
            ..Default::default()
        }
    });

    // Thread the context table from pre-invoke to preserve plugin state
    let (post_result, post_bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_post_invoke",
            post_payload,
            post_ext,
            Some(pre_result.context_table),
        )
        .await;

    println!();
    if post_result.continue_processing {
        println!("Post-invoke result: ALLOWED");
    } else {
        println!(
            "Post-invoke result: DENIED — {}",
            post_result.violation.as_ref().unwrap().reason
        );
    }

    post_bg.wait_for_background_tasks().await;
    println!("\n=== Demo complete ===");
}

pub fn register_visitor(&self, visitor: Arc<dyn ConfigVisitor>)

Register an external config visitor. Visitors run during load_config_yaml (after plugin instantiation) and can install per-route handler overrides via annotate_route. Visitor order matches registration order. Multiple visitors are allowed — they typically don’t share state, so order rarely matters.

pub fn load_config_yaml( self: &Arc<Self>, yaml: &str, ) -> Result<(), Box<PluginError>>

Load a unified-config YAML string. Parses the YAML twice — once into a typed CpexConfig for plugin instantiation, once into a raw serde_yaml::Value so visitors can inspect orchestrator- specific blocks (e.g. apl:) that cpex-core itself doesn’t model. Calls existing load_config(cpex_config) first, then walks each registered visitor over the raw YAML’s sections in the documented hierarchy order:

1. visit_global(global_yaml)
2. visit_default(entity_type, default_yaml) per global.defaults entry
3. visit_policy_bundle(tag, bundle_yaml) per global.policies entry
4. visit_route(route_yaml, parsed_route) per routes[] entry

All sections for one visitor run before the next visitor starts, giving each visitor a consistent view of its own accumulated state. A visitor returning Err aborts the load — the plugin snapshot stays at the post-load_config state (partial load is not rolled back; operators should treat any error from this method as a hard stop).

pub fn from_config( cpex_config: CpexConfig, factories: &PluginFactoryRegistry, ) -> Result<Self, Box<PluginError>>

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 register_handler<H, P>( &self, plugin: Arc<P>, config: PluginConfig, ) -> Result<(), Box<PluginError>>

where H: HookTypeDef, H::Result: Into<PluginResult<H::Payload>>, P: Plugin + HookHandler<H> + 'static,

Register a plugin handler for its primary hook name.

This is the preferred registration method. The framework creates the type-erased adapter internally — no AnyHookHandler needed.

Type Parameters

• H — the hook type (implements HookTypeDef).
• P — the plugin type (implements Plugin + HookHandler<H>).

Arguments

• plugin — the plugin implementation.
• config — authoritative config from the config loader.

Examples

manager.register_handler::<CmfHook, _>(plugin, config)?;

pub fn register_handler_for_names<H, P>( &self, plugin: Arc<P>, config: PluginConfig, names: &[&str], ) -> Result<(), Box<PluginError>>

where H: HookTypeDef, H::Result: Into<PluginResult<H::Payload>>, P: Plugin + HookHandler<H> + 'static,

Register a plugin handler for multiple hook names.

This is the CMF pattern — one handler covers multiple hook names (cmf.tool_pre_invoke, cmf.llm_input, etc.).

Examples

manager.register_handler_for_names::<CmfHook, _>(
    plugin, config,
    &["cmf.tool_pre_invoke", "cmf.llm_input", "cmf.llm_output"],
)?;

pub fn register_raw<H: HookTypeDef>( &self, plugin: Arc<dyn Plugin>, config: PluginConfig, handler: Arc<dyn AnyHookHandler>, ) -> Result<(), Box<PluginError>>

Register with an explicit AnyHookHandler (advanced use).

For cases where the automatic adapter doesn’t fit — e.g., Python/WASM bridge hosts that implement AnyHookHandler directly. Most callers should use register_handler instead.

pub async fn initialize(&self) -> Result<(), Box<PluginError>>

Initialize all registered plugins.

Calls plugin.initialize() on each registered plugin. Must be called before invoking any hooks. Idempotent — calling twice has no effect.

Examples found in repository

examples/plugin_demo.rs (line 429)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

examples/cmf_capabilities_demo.rs (line 341)

async fn main() {
    println!("=== CMF Capabilities Demo ===\n");

    // Load config from YAML file — capabilities declared per plugin
    let config_path = "crates/cpex-core/examples/cmf_capabilities_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 mgr = PluginManager::default();
    mgr.register_factory("builtin/identity-checker", Box::new(IdentityCheckerFactory));
    mgr.register_factory("builtin/header-injector", Box::new(HeaderInjectorFactory));
    mgr.register_factory("builtin/audit-logger", Box::new(AuditLoggerFactory));
    mgr.load_config(cpex_config).unwrap();
    mgr.initialize().await.unwrap();

    // --- Build CMF Message ---
    let payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Assistant,
            content: vec![
                ContentPart::Text {
                    text: "Looking up compensation.".into(),
                },
                ContentPart::ToolCall {
                    content: ToolCall {
                        tool_call_id: "tc_001".into(),
                        name: "get_compensation".into(),
                        arguments: [("employee_id".to_string(), serde_json::json!(42))].into(),
                        namespace: None,
                    },
                },
            ],
            channel: None,
        },
    };

    // --- Build Extensions with security, HTTP, meta ---
    let mut security = SecurityExtension::default();
    security.add_label("PII");
    security.add_label("HR_DATA");
    security.classification = Some("confidential".into());
    security.subject = Some(cpex_core::extensions::security::SubjectExtension {
        id: Some("alice".into()),
        subject_type: Some(cpex_core::extensions::security::SubjectType::User),
        roles: ["hr_admin".to_string()].into(),
        permissions: ["read_compensation".to_string()].into(),
        ..Default::default()
    });

    let mut http = HttpExtension::default();
    http.set_header("Authorization", "Bearer eyJ...");
    http.set_header("X-Request-ID", "req-abc-123");

    let ext = Extensions {
        request: Some(Arc::new(RequestExtension {
            environment: Some("production".into()),
            request_id: Some("req-abc-123".into()),
            ..Default::default()
        })),
        security: Some(Arc::new(security)),
        http: Some(Arc::new(http)),
        meta: Some(Arc::new(MetaExtension {
            entity_type: Some("tool".into()),
            entity_name: Some("get_compensation".into()),
            tags: ["pii".to_string(), "hr".to_string()].into(),
            ..Default::default()
        })),
        ..Default::default()
    };

    // --- Pre-invoke: type-safe dispatch via invoke_named ---
    println!("=== Phase 1: cmf.tool_pre_invoke ===\n");

    // invoke_named<CmfHook> gives compile-time payload type checking
    // while routing to the specific "cmf.tool_pre_invoke" hook name
    let (pre_result, bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_pre_invoke",
            payload,
            ext,
            None, // first hook — no context table
        )
        .await;

    println!();
    if pre_result.continue_processing {
        println!("Pre-invoke result: ALLOWED");
        if let Some(ref modified_ext) = pre_result.modified_extensions {
            if let Some(ref sec) = modified_ext.security {
                let labels: Vec<&String> = sec.labels.iter().collect();
                println!("  Labels after pre-invoke: {:?}", labels);
            }
            if let Some(ref http) = modified_ext.http {
                println!("  Headers after pre-invoke: {:?}", http.request_headers);
            }
        }
    } else {
        println!(
            "Pre-invoke result: DENIED — {}",
            pre_result.violation.as_ref().unwrap().reason
        );
        bg.wait_for_background_tasks().await;
        println!("\n=== Demo complete ===");
        return;
    }
    bg.wait_for_background_tasks().await;

    // --- Simulate tool execution ---
    println!("\n--- Tool 'get_compensation' executes... ---");
    println!("  Result: {{\"salary\": 150000, \"currency\": \"USD\"}}\n");

    // --- Post-invoke: different CMF message with tool result ---
    println!("=== Phase 2: cmf.tool_post_invoke ===\n");

    let post_payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Tool,
            content: vec![ContentPart::ToolResult {
                content: cpex_core::cmf::ToolResult {
                    tool_call_id: "tc_001".into(),
                    tool_name: "get_compensation".into(),
                    content: serde_json::json!({"salary": 150000, "currency": "USD"}),
                    is_error: false,
                },
            }],
            channel: None,
        },
    };

    // Build post-invoke extensions — carry forward any modifications
    // from pre-invoke via the context table
    let post_ext = pre_result.modified_extensions.unwrap_or_else(|| {
        // Rebuild if no modifications
        let mut security = SecurityExtension::default();
        security.add_label("PII");
        Extensions {
            security: Some(Arc::new(security)),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                entity_name: Some("get_compensation".into()),
                ..Default::default()
            })),
            ..Default::default()
        }
    });

    // Thread the context table from pre-invoke to preserve plugin state
    let (post_result, post_bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_post_invoke",
            post_payload,
            post_ext,
            Some(pre_result.context_table),
        )
        .await;

    println!();
    if post_result.continue_processing {
        println!("Post-invoke result: ALLOWED");
    } else {
        println!(
            "Post-invoke result: DENIED — {}",
            post_result.violation.as_ref().unwrap().reason
        );
    }

    post_bg.wait_for_background_tasks().await;
    println!("\n=== Demo complete ===");
}

pub async fn shutdown(&self)

Shutdown all registered plugins.

Calls plugin.shutdown() on each registered plugin in reverse registration order. Errors are logged but do not halt the shutdown process — all plugins get a chance to clean up. Shut the manager down. Terminal: after shutdown() returns, no further register_* / invoke_* should be called. New fire-and-forget tasks spawned after close() will not be tracked (the TaskTracker is single-shot by design).

Examples found in repository

examples/plugin_demo.rs (line 553)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

pub async fn invoke_by_name( &self, hook_name: &str, payload: Box<dyn PluginPayload>, extensions: Extensions, context_table: Option<PluginContextTable>, ) -> (PipelineResult, BackgroundTasks)

Invoke a hook by name with a type-erased payload.

This is the dynamic dispatch path used by Python/Go/WASM callers via FFI or PyO3 bindings. The hook name is resolved from the registry and dispatched through the 5-phase executor.

Arguments

• hook_name — the hook name string (e.g., "cmf.tool_pre_invoke").
• payload — the payload as Box<dyn PluginPayload>.
• extensions — the full extensions (filtered per plugin by the executor).
• context_table — optional context table from a previous hook invocation. Pass None on the first hook call; thread the returned table into subsequent calls to preserve per-plugin state.

Returns

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<H: HookTypeDef>( &self, payload: H::Payload, extensions: Extensions, context_table: Option<PluginContextTable>, ) -> (PipelineResult, BackgroundTasks)

Invoke a typed hook.

This is the compile-time dispatch path used by Rust callers. The hook type H determines the payload and result types. 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).

Arguments

• payload — the typed payload.
• extensions — the full extensions (includes meta for routing).
• context_table — optional context table from a previous hook.

Returns

A tuple of (PipelineResult, BackgroundTasks).

Examples found in repository

examples/plugin_demo.rs (line 446)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

pub async fn invoke_named<H: HookTypeDef>( &self, hook_name: &str, payload: H::Payload, extensions: Extensions, context_table: Option<PluginContextTable>, ) -> (PipelineResult, BackgroundTasks)

Invoke a typed hook by explicit name.

Combines compile-time payload type checking (from H) with runtime hook name routing (from hook_name). Use this when a single hook type (e.g., CmfHook) covers multiple hook names (e.g., cmf.tool_pre_invoke, cmf.tool_post_invoke).

Type Parameters

• H — the hook type (provides payload type checking).

Arguments

• hook_name — the hook name for dispatch routing.
• payload — the typed payload (compile-time checked against H::Payload).
• extensions — the full extensions.
• context_table — optional context table from a previous hook.

Examples

// Compile-time: payload must be MessagePayload (from CmfHook)
// Runtime: dispatches to plugins registered under "cmf.tool_pre_invoke"
let (result, bg) = mgr.invoke_named::<CmfHook>(
    "cmf.tool_pre_invoke", payload, ext, None,
).await;

Examples found in repository

examples/cmf_capabilities_demo.rs (lines 405-410)

async fn main() {
    println!("=== CMF Capabilities Demo ===\n");

    // Load config from YAML file — capabilities declared per plugin
    let config_path = "crates/cpex-core/examples/cmf_capabilities_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 mgr = PluginManager::default();
    mgr.register_factory("builtin/identity-checker", Box::new(IdentityCheckerFactory));
    mgr.register_factory("builtin/header-injector", Box::new(HeaderInjectorFactory));
    mgr.register_factory("builtin/audit-logger", Box::new(AuditLoggerFactory));
    mgr.load_config(cpex_config).unwrap();
    mgr.initialize().await.unwrap();

    // --- Build CMF Message ---
    let payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Assistant,
            content: vec![
                ContentPart::Text {
                    text: "Looking up compensation.".into(),
                },
                ContentPart::ToolCall {
                    content: ToolCall {
                        tool_call_id: "tc_001".into(),
                        name: "get_compensation".into(),
                        arguments: [("employee_id".to_string(), serde_json::json!(42))].into(),
                        namespace: None,
                    },
                },
            ],
            channel: None,
        },
    };

    // --- Build Extensions with security, HTTP, meta ---
    let mut security = SecurityExtension::default();
    security.add_label("PII");
    security.add_label("HR_DATA");
    security.classification = Some("confidential".into());
    security.subject = Some(cpex_core::extensions::security::SubjectExtension {
        id: Some("alice".into()),
        subject_type: Some(cpex_core::extensions::security::SubjectType::User),
        roles: ["hr_admin".to_string()].into(),
        permissions: ["read_compensation".to_string()].into(),
        ..Default::default()
    });

    let mut http = HttpExtension::default();
    http.set_header("Authorization", "Bearer eyJ...");
    http.set_header("X-Request-ID", "req-abc-123");

    let ext = Extensions {
        request: Some(Arc::new(RequestExtension {
            environment: Some("production".into()),
            request_id: Some("req-abc-123".into()),
            ..Default::default()
        })),
        security: Some(Arc::new(security)),
        http: Some(Arc::new(http)),
        meta: Some(Arc::new(MetaExtension {
            entity_type: Some("tool".into()),
            entity_name: Some("get_compensation".into()),
            tags: ["pii".to_string(), "hr".to_string()].into(),
            ..Default::default()
        })),
        ..Default::default()
    };

    // --- Pre-invoke: type-safe dispatch via invoke_named ---
    println!("=== Phase 1: cmf.tool_pre_invoke ===\n");

    // invoke_named<CmfHook> gives compile-time payload type checking
    // while routing to the specific "cmf.tool_pre_invoke" hook name
    let (pre_result, bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_pre_invoke",
            payload,
            ext,
            None, // first hook — no context table
        )
        .await;

    println!();
    if pre_result.continue_processing {
        println!("Pre-invoke result: ALLOWED");
        if let Some(ref modified_ext) = pre_result.modified_extensions {
            if let Some(ref sec) = modified_ext.security {
                let labels: Vec<&String> = sec.labels.iter().collect();
                println!("  Labels after pre-invoke: {:?}", labels);
            }
            if let Some(ref http) = modified_ext.http {
                println!("  Headers after pre-invoke: {:?}", http.request_headers);
            }
        }
    } else {
        println!(
            "Pre-invoke result: DENIED — {}",
            pre_result.violation.as_ref().unwrap().reason
        );
        bg.wait_for_background_tasks().await;
        println!("\n=== Demo complete ===");
        return;
    }
    bg.wait_for_background_tasks().await;

    // --- Simulate tool execution ---
    println!("\n--- Tool 'get_compensation' executes... ---");
    println!("  Result: {{\"salary\": 150000, \"currency\": \"USD\"}}\n");

    // --- Post-invoke: different CMF message with tool result ---
    println!("=== Phase 2: cmf.tool_post_invoke ===\n");

    let post_payload = MessagePayload {
        message: Message {
            schema_version: cpex_core::cmf::constants::SCHEMA_VERSION.into(),
            role: Role::Tool,
            content: vec![ContentPart::ToolResult {
                content: cpex_core::cmf::ToolResult {
                    tool_call_id: "tc_001".into(),
                    tool_name: "get_compensation".into(),
                    content: serde_json::json!({"salary": 150000, "currency": "USD"}),
                    is_error: false,
                },
            }],
            channel: None,
        },
    };

    // Build post-invoke extensions — carry forward any modifications
    // from pre-invoke via the context table
    let post_ext = pre_result.modified_extensions.unwrap_or_else(|| {
        // Rebuild if no modifications
        let mut security = SecurityExtension::default();
        security.add_label("PII");
        Extensions {
            security: Some(Arc::new(security)),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                entity_name: Some("get_compensation".into()),
                ..Default::default()
            })),
            ..Default::default()
        }
    });

    // Thread the context table from pre-invoke to preserve plugin state
    let (post_result, post_bg) = mgr
        .invoke_named::<CmfHook>(
            "cmf.tool_post_invoke",
            post_payload,
            post_ext,
            Some(pre_result.context_table),
        )
        .await;

    println!();
    if post_result.continue_processing {
        println!("Post-invoke result: ALLOWED");
    } else {
        println!(
            "Post-invoke result: DENIED — {}",
            post_result.violation.as_ref().unwrap().reason
        );
    }

    post_bg.wait_for_background_tasks().await;
    println!("\n=== Demo complete ===");
}

pub fn find_plugin_entries(&self, plugin_name: &str) -> Vec<(String, HookEntry)>

Find every (hook_name, HookEntry) pair belonging to the named plugin. Returns an empty Vec if the plugin isn’t registered.

Used by external orchestrators (notably apl-cpex) that decide the per-route plugin lineup themselves and need handler refs + trusted_config to build pre-resolved dispatch plans. Cheaper than going through invoke_named per request because the caller can cache the resulting entries — pair the result with config_generation to invalidate the cache on snapshot swaps.

Bypasses route/entity filtering — caller has already decided this plugin should run. APL’s routes: is itself the authoritative lineup; cpex-core’s condition-based routing is a parallel model for non-APL hosts.

pub async fn invoke_entries<H: HookTypeDef>( &self, entries: &[HookEntry], payload: H::Payload, extensions: Extensions, context_table: Option<PluginContextTable>, ) -> (PipelineResult, BackgroundTasks)

Dispatch a caller-supplied slice of HookEntries through the executor’s full 5-phase pipeline (sequential, transform, audit, concurrent, fire-and-forget). All on_error / timeout / mode / write-token machinery applies.

Bypasses hook-name lookup and route/entity filtering — caller has already resolved the lineup (typically via find_plugin_entries + a per-route dispatch plan). The H: HookTypeDef parameter enforces payload type at compile time; mismatched payloads fail to compile, same as invoke_named.

Returns (PipelineResult, BackgroundTasks) identical in shape to invoke_named so callers can swap between the two paths without rewriting downstream result handling.

pub fn annotate_route<H>( &self, entity_type: impl Into<String>, entity_name: impl Into<String>, scope: Option<String>, hook_name: impl Into<String>, handler: Arc<H>, config: PluginConfig, )

where H: Plugin + AnyHookHandler + 'static,

Override the resolved plugin list for one (entity_type, entity_name) pair on the listed hooks with a single synthetic handler. The handler takes responsibility for any further plugin dispatch within itself (typically by calling invoke_entries against the same registry’s other entries — i.e. APL’s plugin(name) → CmfPluginInvoker → invoke_entries flow).

This is the integration point external orchestrators (APL, future Rego/Cedar-direct/Custom) use to drive plugins via their own semantics instead of cpex-core’s imperative routes.*.plugins: chain. Bumps the config generation so cached dispatch plans in downstream caches invalidate.

config provides the trusted_config for the synthetic plugin — the executor reads mode, on_error, capabilities, etc. from it the same way it does for any other registered plugin. Capabilities should be a superset of what the orchestrator needs to read from Extensions (cpex-core’s per-plugin filter still applies to the synthetic handler).

The underlying plugins: chain for this route is not removed — those plugins stay discoverable via find_plugin_entries so the orchestrator can dispatch into them by name.

pub fn remove_route_annotation( &self, entity_type: &str, entity_name: &str, scope: Option<&str>, hook_name: &str, )

Remove a route annotation for a specific hook. No-op when no annotation exists for the key. Bumps the generation so downstream caches invalidate.

pub async fn build_override_entries( &self, plugin_name: &str, config_override: Option<&Value>, capabilities_override: Option<&HashSet<String>>, on_error_override: Option<OnError>, ) -> Vec<(String, HookEntry)>

Build per-hook HookEntrys for a plugin with optional route- level overrides. Used by external orchestrators (notably apl-cpex’s dispatch plan) that need to splice per-route plugin variants — different config, narrower capabilities, different on_error — into the dispatch lineup while keeping cpex-core the source of truth for instantiation and isolation.

Behavior:

• All three overrides None: returns the base entries unchanged. Caller can use them as-is.
• Only capabilities_override / on_error_override set (config_override is None): builds new PluginRefs sharing the base plugin Arc with a merged TrustedConfig (override caps / on_error replace base values) and an independent circuit breaker. Cheap — no factory call.
• config_override set: invokes the registered factory for the plugin’s kind with a merged PluginConfig (override config replaces base config wholesale per unified-config spec — not deep merge), calls initialize() on the new instance, and wraps every returned handler in a new PluginRef with a fresh circuit breaker.

Returns an empty Vec when:

• the plugin name isn’t registered in the manager,
• the factory for the plugin’s kind is missing,
• the factory’s create errors,
• or initialize() fails on the new instance.

Each of those is a configuration / wiring fault the caller should treat as NotFound at dispatch time. The method logs the underlying error before returning empty so debugging surfaces in operator logs rather than as a silent miss.

pub fn clear_routing_cache(&self)

Clear the routing cache. Call when config is reloaded or plugins are registered/unregistered. Also resets the “cache full” warn-once latch so the next fill cycle can warn again.

pub fn routing_cache_size(&self) -> usize

Number of entries in the routing cache.

pub fn has_hooks_for(&self, hook_name: &str) -> bool

Whether anything would run for the given hook name — either a registered plugin handler OR a route annotation targeting that hook.

Route annotations (installed by APL from a route’s policy: / args: / result: blocks) must be counted here: a route whose only handler for a phase is an annotation (e.g. a response-side result: { ssn: redact(...) } on cmf.tool_post_invoke, with no globally-registered post-invoke plugin) would otherwise report “no hooks” and be skipped by out-of-process hosts that use this as a fast-skip gate — silently dropping the route’s policy for that phase.

Examples found in repository

examples/plugin_demo.rs (line 434)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

pub fn get_plugin(&self, name: &str) -> Option<Arc<PluginRef>>

Look up a plugin by name. Returns an Arc<PluginRef> clone — works with the snapshot-based dispatch model where the registry sits behind a transient Arc<RuntimeSnapshot> guard. Arc<PluginRef> derefs to PluginRef, so callers can chain methods directly: mgr.get_plugin("name").unwrap().is_disabled() still compiles.

pub fn plugin_count(&self) -> usize

Total number of registered plugins.

Examples found in repository

examples/plugin_demo.rs (line 431)

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 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.register_factory("builtin/remote_authz", Box::new(RemoteAuthzFactory));
    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::<ToolPreInvoke>(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 ctx_table = cpex_core::context::PluginContextTable::new();
    ctx_table
        .global_state
        .insert("pii_clearance".into(), serde_json::Value::Bool(true));
    let (result, bg) = mgr
        .invoke::<ToolPreInvoke>(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::<ToolPostInvoke>(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::<ToolPreInvoke>(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::<ToolPreInvoke>(payload, ext, None).await;
    print_result("some_other_tool (wildcard)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 5: Awaiting plugin — cache hit ---
    // RemoteAuthz's `handle` is `async fn` and reads from a tokio
    // RwLock. Its initialize() pre-loaded an ACL containing "alice"
    // and "bob"; this call exercises the cache-hit fast path.
    println!("=== Scenario 5: query_external_data (async plugin, cache hit) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "alice".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (alice — in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 6: Awaiting plugin — cache miss path with .await ---
    // "charlie" is not in the cached ACL, so RemoteAuthz takes the
    // cache-miss branch and `.await`s a simulated remote call before
    // denying.
    println!("=== Scenario 6: query_external_data (async plugin, cache miss) ===\n");
    let payload = ToolInvokePayload {
        tool_name: "query_external_data".into(),
        user: "charlie".into(),
        arguments: "dataset=sales".into(),
    };
    let ext = make_tool_extensions("query_external_data", &[]);
    let (result, bg) = mgr.invoke::<ToolPreInvoke>(payload, ext, None).await;
    print_result("query_external_data (charlie — not in ACL)", &result);
    bg.wait_for_background_tasks().await;

    // --- Scenario 7: No user identity ---
    println!("=== Scenario 7: 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::<ToolPreInvoke>(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 ===");
}

pub fn plugin_names(&self) -> Vec<String>

All registered plugin names (owned, not borrowed from the registry).

pub fn is_initialized(&self) -> bool

Whether the manager has been initialized.

pub fn unregister(&self, name: &str) -> Option<Arc<PluginRef>>

Unregister a plugin by name.

Trait Implementations

impl Default for PluginManager

fn default() -> Self

Returns the “default value” for a type.