Struct BackgroundTasks 

pub struct BackgroundTasks { /* private fields */ }

Handles to fire-and-forget background tasks spawned by the executor.

Returned separately from PipelineResult so that the policy result stays immutable. If not awaited, tasks complete on their own in the background. Call wait_for_background_tasks() when you need to ensure tasks have finished (tests, graceful shutdown, audit flush).

Implementations

impl BackgroundTasks

pub fn empty() -> Self

Create an empty set of background tasks.

pub fn is_empty(&self) -> bool

Whether there are any background tasks.

pub fn len(&self) -> usize

Number of background tasks.

pub async fn wait_for_background_tasks(self) -> Vec<PluginError>

Wait for all fire-and-forget background tasks to complete.

Returns a list of errors from any tasks that panicked. An empty list means all tasks completed successfully.

Consumes self — each task handle can only be awaited once.

If not called, background tasks still complete on their own. Use this for tests, graceful shutdown, or when you need to ensure audit/logging tasks have flushed before proceeding.

Examples found in repository

examples/plugin_demo.rs (line 449)

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 430)

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 ===");
}

Trait Implementations

impl Debug for BackgroundTasks

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.