feat: change Plugin handler to async for performance

crates/cpex-core/examples/cmf_capabilities_demo.rs

@@ -42,7 +42,7 @@ impl Plugin for IdentityChecker {
 }
 
 impl HookHandler<CmfHook> for IdentityChecker {
-    fn handle(
+    async fn handle(
         &self,
         payload: &MessagePayload,
         extensions: &Extensions,
@@ -136,7 +136,7 @@ impl Plugin for HeaderInjector {
 }
 
 impl HookHandler<CmfHook> for HeaderInjector {
-    fn handle(
+    async fn handle(
         &self,
         _payload: &MessagePayload,
         extensions: &Extensions,
@@ -201,7 +201,7 @@ impl Plugin for AuditLogger {
 }
 
 impl HookHandler<CmfHook> for AuditLogger {
-    fn handle(
+    async fn handle(
         &self,
         payload: &MessagePayload,
         extensions: &Extensions,

crates/cpex-core/examples/plugin_demo.rs

@@ -76,7 +76,7 @@ impl Plugin for IdentityResolver {
 }
 
 impl HookHandler<ToolPreInvoke> for IdentityResolver {
-    fn handle(
+    async fn handle(
         &self,
         payload: &ToolInvokePayload,
         _extensions: &Extensions,
@@ -98,7 +98,7 @@ impl HookHandler<ToolPreInvoke> for IdentityResolver {
 }
 
 impl HookHandler<ToolPostInvoke> for IdentityResolver {
-    fn handle(
+    async fn handle(
         &self,
         payload: &ToolInvokePayload,
         _extensions: &Extensions,
@@ -126,7 +126,7 @@ impl Plugin for PiiGuard {
 }
 
 impl HookHandler<ToolPreInvoke> for PiiGuard {
-    fn handle(
+    async fn handle(
         &self,
         payload: &ToolInvokePayload,
         _extensions: &Extensions,
@@ -171,7 +171,7 @@ impl Plugin for AuditLogger {
 }
 
 impl HookHandler<ToolPreInvoke> for AuditLogger {
-    fn handle(
+    async fn handle(
         &self,
         payload: &ToolInvokePayload,
         _extensions: &Extensions,
@@ -186,7 +186,7 @@ impl HookHandler<ToolPreInvoke> for AuditLogger {
 }
 
 impl HookHandler<ToolPostInvoke> for AuditLogger {
-    fn handle(
+    async fn handle(
         &self,
         payload: &ToolInvokePayload,
         _extensions: &Extensions,
@@ -200,6 +200,89 @@ impl HookHandler<ToolPostInvoke> for AuditLogger {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Awaiting plugin example — RemoteAuthz
+// ---------------------------------------------------------------------------
+//
+// `HookHandler<H>` is async by design — `handle` is `async fn`.
+// Plugins that don't need to `.await` anything still write
+// `async fn handle` and return synchronously; this plugin shows the
+// other direction, where the body genuinely awaits per-invocation
+// work. The realistic version would call a remote authz service
+// (gRPC, HTTP, OPA, Cedarling, etc.); here we simulate the network
+// round-trip with a small `tokio::time::sleep` so the demo runs
+// offline.
+//
+// Key things this shows:
+//   1. Per-request latency state is *cached at init* — the handler
+//      consults the in-memory ACL and only "calls out" on a miss.
+//      Hot-path I/O is the most common source of latency regressions
+//      in plugins, so prefer initialize-time loading wherever you can.
+//   2. Registration uses the exact same factory pattern as any other
+//      plugin — `TypedHandlerAdapter::<H, _>` and the same
+//      `register_factory` call. There is no separate async path.
+struct RemoteAuthz {
+    cfg: PluginConfig,
+    /// ACL "fetched" at init. Populated in Plugin::initialize.
+    allowed_users: tokio::sync::RwLock<std::collections::HashSet<String>>,
+}
+
+#[async_trait]
+impl Plugin for RemoteAuthz {
+    fn config(&self) -> &PluginConfig {
+        &self.cfg
+    }
+    /// Pretend we're loading the ACL from a remote service. In a real
+    /// plugin this would be `client.fetch_acl().await`; we simulate
+    /// the round-trip with a small sleep so the demo runs offline.
+    async fn initialize(&self) -> Result<(), Box<PluginError>> {
+        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+        let mut acl = self.allowed_users.write().await;
+        acl.extend(["alice", "bob"].iter().map(|s| s.to_string()));
+        println!(
+            "  [remote-authz] initialized — ACL cached ({} users)",
+            acl.len()
+        );
+        Ok(())
+    }
+    async fn shutdown(&self) -> Result<(), Box<PluginError>> {
+        println!("  [remote-authz] shutdown");
+        Ok(())
+    }
+}
+
+impl HookHandler<ToolPreInvoke> for RemoteAuthz {
+    async fn handle(
+        &self,
+        payload: &ToolInvokePayload,
+        _extensions: &Extensions,
+        _ctx: &mut PluginContext,
+    ) -> PluginResult<ToolInvokePayload> {
+        // Cache hit path — fast.
+        let acl = self.allowed_users.read().await;
+        if acl.contains(&payload.user) {
+            println!(
+                "  [remote-authz] OK (cache hit): user '{}' allowed",
+                payload.user
+            );
+            return PluginResult::allow();
+        }
+        drop(acl); // release read lock before the fake remote call
+                   // Cache miss path — simulate a remote authz check. In a real
+                   // plugin this is where you'd `.await` a gRPC or HTTP call.
+                   // The latency cost is real and shows up on the request path.
+        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+        println!(
+            "  [remote-authz] DENIED (cache miss + remote check): user '{}'",
+            payload.user
+        );
+        PluginResult::deny(PluginViolation::new(
+            "remote_authz_denied",
+            format!("User '{}' not in remote ACL", payload.user),
+        ))
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Step 3: Create plugin factories
 // ---------------------------------------------------------------------------
@@ -264,6 +347,27 @@ impl PluginFactory for AuditLoggerFactory {
     }
 }
 
+/// Factory for the async plugin. Note the factory body is identical
+/// in shape to the sync factories above — `TypedHandlerAdapter` and
+/// the `register_factory` path don't care that the underlying handler
+/// is async. The framework hides the choice.
+struct RemoteAuthzFactory;
+impl PluginFactory for RemoteAuthzFactory {
+    fn create(&self, config: &PluginConfig) -> Result<PluginInstance, Box<PluginError>> {
+        let plugin = Arc::new(RemoteAuthz {
+            cfg: config.clone(),
+            allowed_users: tokio::sync::RwLock::new(std::collections::HashSet::new()),
+        });
+        Ok(PluginInstance {
+            plugin: plugin.clone(),
+            handlers: vec![(
+                "tool_pre_invoke",
+                Arc::new(TypedHandlerAdapter::<ToolPreInvoke, _>::new(plugin)),
+            )],
+        })
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Step 4: Build extensions with MetaExtension for routing
 // ---------------------------------------------------------------------------
@@ -318,6 +422,7 @@ async fn main() {
     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");
@@ -401,8 +506,38 @@ async fn main() {
     print_result("some_other_tool (wildcard)", &result);
     bg.wait_for_background_tasks().await;
 
-    // --- Scenario 5: No user identity ---
-    println!("=== Scenario 5: list_departments (no user identity) ===\n");
+    // --- 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(),

crates/cpex-core/examples/plugin_demo.yaml

@@ -15,6 +15,10 @@ global:
     # "pii" group — activated when a route has the "pii" tag
     pii:
       plugins: [pii-guard]
+    # "external_authz" group — activated when a route has the
+    # "needs_remote_authz" tag. Fires the async RemoteAuthz plugin.
+    external_authz:
+      plugins: [remote-authz]
 
 plugins:
   - name: identity-resolver
@@ -33,6 +37,17 @@ plugins:
     config:
       clearance_level: confidential
 
+  # Awaiting plugin — its `handle` body uses `.await` for a
+  # (simulated) remote authz call on cache miss. Wired in exactly
+  # the same way as plugins whose `handle` body has no `.await`;
+  # registration is identical either way.
+  - name: remote-authz
+    kind: builtin/remote_authz
+    hooks: [tool_pre_invoke]
+    mode: sequential
+    priority: 30
+    on_error: fail
+
   - name: audit-logger
     kind: builtin/audit
     hooks: [tool_pre_invoke, tool_post_invoke]
@@ -53,6 +68,16 @@ routes:
     plugins:
       - audit-logger
 
+  # Tool that requires remote authz — triggers the async plugin
+  # on top of the standard "all" policy stack. The "external_authz"
+  # tag matches the policy group of the same name, which fires
+  # remote-authz.
+  - tool: query_external_data
+    meta:
+      tags: [external_authz]
+    plugins:
+      - audit-logger
+
   # Wildcard — catch-all for unmatched tools
   - tool: "*"
     plugins:

crates/cpex-core/src/hooks/adapter.rs

@@ -3,14 +3,20 @@
 // SPDX-License-Identifier: Apache-2.0
 // Authors: Teryl Taylor
 //
-// TypedHandlerAdapter — bridges typed HookHandler<H> to type-erased
-// AnyHookHandler.
+// TypedHandlerAdapter — bridges typed HookHandler<H> to the
+// type-erased AnyHookHandler.
 //
 // This is framework plumbing that plugin authors never see. When a
 // plugin is registered via `manager.register_handler::<H, P>()`, the
 // manager creates a TypedHandlerAdapter internally. The adapter
 // translates between Box<dyn PluginPayload> (what the executor passes)
-// and the concrete payload type (what the handler expects).
+// and the concrete payload type (what the handler expects), and awaits
+// the typed handler's future before re-erasing the result.
+//
+// `HookHandler<H>` is async-by-default (native AFIT). Plugins that
+// don't await anything still write `async fn handle(...)`; the
+// compiler emits a trivially-ready future that LLVM inlines, so the
+// `.await` here is a no-op for sync-style plugins.
 
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -33,6 +39,11 @@ use crate::registry::AnyHookHandler;
 /// Created automatically by `PluginManager::register_handler()`. Plugin
 /// authors never instantiate this directly.
 ///
+/// `HookHandler<H>` is async (native AFIT), so the adapter awaits the
+/// returned future before re-erasing the result. Plugins that don't
+/// `.await` anything compile to a ready future that LLVM inlines, so
+/// they pay no observable cost over a plain function call.
+///
 /// # Type Parameters
 ///
 /// - `H` — the hook type (implements `HookTypeDef`).
@@ -73,12 +84,18 @@ where
     P: Plugin + HookHandler<H> + 'static,
 {
     /// Downcast the type-erased payload to the concrete type and call
-    /// the plugin's typed `handle()` method.
+    /// the plugin's typed `handle()` method, awaiting the returned
+    /// future.
     ///
     /// The framework retains ownership of the payload — the handler
     /// receives a borrow (`&H::Payload`) and clones only if it needs
     /// to modify. The result is erased back to `ErasedResultFields`
     /// for the executor.
+    ///
+    /// For plugins whose body contains no `.await`, the compiler emits
+    /// a trivially-ready future and LLVM inlines this `.await` to a
+    /// direct return — there is no observable runtime cost over a
+    /// plain function call.
     async fn invoke(
         &self,
         payload: &dyn PluginPayload,
@@ -97,7 +114,7 @@ where
                     ),
                 })?;
 
-        let result = self.plugin.handle(typed_ref, extensions, ctx);
+        let result = self.plugin.handle(typed_ref, extensions, ctx).await;
         let plugin_result: PluginResult<H::Payload> = result.into();
 
         Ok(erase_result(plugin_result))