cpex-core 0.2.2

// Location: ./crates/cpex-core/src/extensions/container.rs
// Copyright 2025
// SPDX-License-Identifier: Apache-2.0
// Authors: Teryl Taylor
//
// Extensions and OwnedExtensions — typed containers for all
// extension data passed separately from the payload to handlers.
//
// Extensions is fully immutable (all Arc<T>) — zero-copy shareable.
// OwnedExtensions is the plugin's writeable workspace, created by
// cow_copy(), returned in PluginResult::modify_extensions().

use std::collections::HashMap;
use std::sync::Arc;

use serde::{Deserialize, Serialize};

use super::agent::AgentExtension;
use super::completion::CompletionExtension;
use super::delegation::DelegationExtension;
use super::framework::FrameworkExtension;
use super::guarded::{Guarded, WriteToken};
use super::http::HttpExtension;
use super::llm::LLMExtension;
use super::mcp::MCPExtension;
use super::meta::MetaExtension;
use super::provenance::ProvenanceExtension;
use super::raw_credentials::RawCredentialsExtension;
use super::request::RequestExtension;
use super::security::SecurityExtension;

// ---------------------------------------------------------------------------
// Extensions — all Arc, fully immutable, zero-copy shareable
// ---------------------------------------------------------------------------

/// Typed container for all message extensions.
///
/// All slots are `Arc<T>` — fully immutable, zero-copy shareable.
/// Cloning is all refcount bumps. `filter_extensions()` creates a
/// filtered view by setting unwanted slots to `None` (still all Arc,
/// no deep copies). Plugins receive `&Extensions` (zero cost).
///
/// To modify, plugins call `cow_copy()` which returns an
/// `OwnedExtensions` with mutable/monotonic/guarded slots cloned
/// out of Arc and write tokens propagated.
///
/// Mirrors Python's `cpex.framework.extensions.Extensions`.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct Extensions {
    /// Execution environment and request tracing (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub request: Option<Arc<RequestExtension>>,

    /// Agent execution context — session, conversation, lineage (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub agent: Option<Arc<AgentExtension>>,

    /// HTTP headers (frozen as Arc — unfrozen in OwnedExtensions).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub http: Option<Arc<HttpExtension>>,

    /// Security — labels, classification, subject (frozen as Arc).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub security: Option<Arc<SecurityExtension>>,

    /// Delegation chain (frozen as Arc).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub delegation: Option<Arc<DelegationExtension>>,

    /// Raw credential material — Layer 3 of the credential storage
    /// model (see `RawCredentialsExtension` docs). Capability-gated;
    /// `filter_extensions` strips this slot for plugins without
    /// `read_inbound_credentials` / `read_delegated_tokens`. Token
    /// fields inside this extension are `#[serde(skip)]`, so any
    /// serialization (logs, audit dumps, hot-reload snapshots) drops
    /// secret material even when the slot itself survives. The
    /// out-of-process consequence — remote / WASM plugins can't see
    /// raw tokens at all — is intentional and documented on
    /// `RawCredentialsExtension`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub raw_credentials: Option<Arc<RawCredentialsExtension>>,

    /// MCP entity metadata (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub mcp: Option<Arc<MCPExtension>>,

    /// LLM completion information (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub completion: Option<Arc<CompletionExtension>>,

    /// Origin and message threading (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub provenance: Option<Arc<ProvenanceExtension>>,

    /// Model identity and capabilities (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub llm: Option<Arc<LLMExtension>>,

    /// Agentic framework context (immutable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub framework: Option<Arc<FrameworkExtension>>,

    /// Host-provided operational metadata (immutable).
    #[serde(default)]
    pub meta: Option<Arc<MetaExtension>>,

    /// Custom extensions (frozen as Arc — unfrozen in OwnedExtensions).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub custom: Option<Arc<HashMap<String, serde_json::Value>>>,

    /// Write tokens — set by the executor per plugin, NOT serialized.
    /// Used by `cow_copy()` to propagate write access to OwnedExtensions.
    #[serde(skip)]
    pub http_write_token: Option<WriteToken>,
    #[serde(skip)]
    pub labels_write_token: Option<WriteToken>,
    #[serde(skip)]
    pub delegation_write_token: Option<WriteToken>,
}

impl Clone for Extensions {
    /// All Arc bumps — zero data copies. Write tokens are NOT cloned.
    fn clone(&self) -> Self {
        Self {
            request: self.request.clone(),
            agent: self.agent.clone(),
            http: self.http.clone(),
            security: self.security.clone(),
            delegation: self.delegation.clone(),
            raw_credentials: self.raw_credentials.clone(),
            mcp: self.mcp.clone(),
            completion: self.completion.clone(),
            provenance: self.provenance.clone(),
            llm: self.llm.clone(),
            framework: self.framework.clone(),
            meta: self.meta.clone(),
            custom: self.custom.clone(),
            http_write_token: None,
            labels_write_token: None,
            delegation_write_token: None,
        }
    }
}

impl Extensions {
    /// Create a copy-on-write owned copy for modification.
    ///
    /// Immutable slots share the same `Arc` (refcount bump, ~1ns).
    /// Mutable/monotonic/guarded slots are cloned out of Arc into
    /// owned values — the plugin can modify them directly.
    /// Write tokens are propagated from the original.
    ///
    /// # Usage
    ///
    /// ```ignore
    /// fn handle(&self, payload: &P, ext: &Extensions, ctx: &mut PluginContext) -> PluginResult<P> {
    ///     let mut owned = ext.cow_copy();
    ///     owned.security.as_mut().unwrap().add_label("CHECKED");
    ///     if let Some(ref token) = owned.http_write_token {
    ///         owned.http.as_mut().unwrap().write(token).set_header("X-Foo", "bar");
    ///     }
    ///     PluginResult::modify_extensions(owned)
    /// }
    /// ```
    pub fn cow_copy(&self) -> OwnedExtensions {
        OwnedExtensions {
            // Immutable — same Arc pointers
            request: self.request.clone(),
            agent: self.agent.clone(),
            mcp: self.mcp.clone(),
            completion: self.completion.clone(),
            provenance: self.provenance.clone(),
            llm: self.llm.clone(),
            framework: self.framework.clone(),
            meta: self.meta.clone(),
            raw_credentials: self.raw_credentials.clone(),

            // Mutable/monotonic/guarded — cloned out of Arc into owned
            http: self.http.as_ref().map(|arc| Guarded::new((**arc).clone())),
            security: self.security.as_ref().map(|arc| (**arc).clone()),
            delegation: self.delegation.as_ref().map(|arc| (**arc).clone()),
            custom: self.custom.as_ref().map(|arc| (**arc).clone()),

            // Write tokens — propagated from the original
            http_write_token: if self.http_write_token.is_some() {
                Some(WriteToken::new())
            } else {
                None
            },
            labels_write_token: if self.labels_write_token.is_some() {
                Some(WriteToken::new())
            } else {
                None
            },
            delegation_write_token: if self.delegation_write_token.is_some() {
                Some(WriteToken::new())
            } else {
                None
            },
        }
    }

    /// Validate that immutable slots were not tampered with.
    ///
    /// A slot that is `None` in modified (because capability filtering
    /// hid it from the plugin) is always valid — the plugin never saw
    /// it. Only flag as tampering when both are `Some` with different
    /// Arc pointers, or when the original is `None` but modified is
    /// `Some` (the plugin fabricated a slot it shouldn't have).
    pub fn validate_immutable(&self, modified: &OwnedExtensions) -> bool {
        fn ptr_eq_opt<T>(a: &Option<Arc<T>>, b: &Option<Arc<T>>) -> bool {
            match (a, b) {
                (Some(a), Some(b)) => Arc::ptr_eq(a, b),
                (None, None) => true,
                (_, None) => true,        // plugin never saw it — not tampering
                (None, Some(_)) => false, // plugin fabricated a slot
            }
        }

        ptr_eq_opt(&self.request, &modified.request)
            && ptr_eq_opt(&self.agent, &modified.agent)
            && ptr_eq_opt(&self.mcp, &modified.mcp)
            && ptr_eq_opt(&self.completion, &modified.completion)
            && ptr_eq_opt(&self.provenance, &modified.provenance)
            && ptr_eq_opt(&self.llm, &modified.llm)
            && ptr_eq_opt(&self.framework, &modified.framework)
            && ptr_eq_opt(&self.meta, &modified.meta)
        // NOTE: `raw_credentials` is INTENTIONALLY excluded from the
        // immutable check. Framework orchestrators (apl-cpex's
        // DelegationPluginInvoker) legitimately write
        // `delegated_tokens.*` via the shared Mutex during route
        // evaluation, producing a new Arc by the time the synthetic
        // handler returns. Per-plugin write authority is enforced at
        // the capability layer (`write_delegated_tokens` /
        // `write_inbound_credentials`), not at this pointer-equality
        // gate. Until cap-tier-aware merge lands, treat raw_credentials
        // as merge-able like `security` and `delegation`.
    }

    /// Merge an OwnedExtensions back into this Extensions.
    pub fn merge_owned(&mut self, owned: OwnedExtensions) {
        self.http = owned.http.map(|g| Arc::new(g.into_inner()));
        self.security = owned.security.map(Arc::new);
        self.delegation = owned.delegation.map(Arc::new);
        self.custom = owned.custom.map(Arc::new);
        // `raw_credentials` is shared by Arc in `OwnedExtensions` —
        // plugins don't mutate it directly. But framework orchestrators
        // (apl-cpex's DelegationPluginInvoker) DO write delegated_tokens
        // / inbound_tokens through the shared `Arc<Mutex<Extensions>>`
        // before the synthetic handler returns. We must propagate
        // those writes back so callers of `invoke_named` see the
        // minted tokens in `PipelineResult.modified_extensions`.
        // Without this, `delegate(...)` steps silently lose their
        // results at the executor merge boundary.
        if owned.raw_credentials.is_some() {
            self.raw_credentials = owned.raw_credentials;
        }
    }
}

// ---------------------------------------------------------------------------
// OwnedExtensions — plugin's writeable workspace
// ---------------------------------------------------------------------------

/// Owned copy of extensions for plugin modification.
///
/// Returned by `Extensions::cow_copy()`. Immutable slots share
/// the same `Arc` pointers as the original (zero copy). Mutable,
/// monotonic, and guarded slots are cloned into owned values that
/// the plugin can modify directly.
///
/// Plugins return this in `PluginResult::modify_extensions()`.
/// The executor validates (immutable unchanged, monotonic superset)
/// and merges back into the pipeline's `Extensions`.
///
/// Hosts never see this type — the executor converts to `Extensions`
/// before building `PipelineResult`.
#[derive(Debug)]
pub struct OwnedExtensions {
    // Immutable — same Arc pointers as original
    pub request: Option<Arc<RequestExtension>>,
    pub agent: Option<Arc<AgentExtension>>,
    pub mcp: Option<Arc<MCPExtension>>,
    pub completion: Option<Arc<CompletionExtension>>,
    pub provenance: Option<Arc<ProvenanceExtension>>,
    pub llm: Option<Arc<LLMExtension>>,
    pub framework: Option<Arc<FrameworkExtension>>,
    pub meta: Option<Arc<MetaExtension>>,
    /// Raw credentials are shared by Arc here too — write tokens for
    /// `inbound_tokens` and `delegated_tokens` mutation paths land in
    /// slice 2 (IdentityResolve) and slice 3 (TokenDelegate). Until
    /// then, no plugin writes through `OwnedExtensions.raw_credentials`.
    pub raw_credentials: Option<Arc<RawCredentialsExtension>>,

    // Mutable/monotonic/guarded — owned, modifiable
    pub http: Option<Guarded<HttpExtension>>,
    pub security: Option<SecurityExtension>,
    pub delegation: Option<DelegationExtension>,
    pub custom: Option<HashMap<String, serde_json::Value>>,

    // Write tokens — propagated from executor
    pub http_write_token: Option<WriteToken>,
    pub labels_write_token: Option<WriteToken>,
    pub delegation_write_token: Option<WriteToken>,
}
#[cfg(test)]
mod tests {
    use super::*;
    use crate::extensions::{
        DelegationExtension, HttpExtension, RequestExtension, SecurityExtension,
    };

    fn make_extensions() -> Extensions {
        let mut security = SecurityExtension::default();
        security.add_label("PII");

        let mut http = HttpExtension::default();
        http.set_header("Authorization", "Bearer token");

        Extensions {
            request: Some(Arc::new(RequestExtension {
                request_id: Some("req-001".into()),
                ..Default::default()
            })),
            security: Some(Arc::new(security)),
            http: Some(Arc::new(http)),
            delegation: Some(Arc::new(DelegationExtension::default())),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                ..Default::default()
            })),
            ..Default::default()
        }
    }

    #[test]
    fn test_cow_copy_shares_immutable_arcs() {
        let ext = make_extensions();
        let cow = ext.cow_copy();

        // Immutable slots share the same Arc — zero copy
        assert!(Arc::ptr_eq(
            ext.request.as_ref().unwrap(),
            cow.request.as_ref().unwrap()
        ));
        assert!(Arc::ptr_eq(
            ext.meta.as_ref().unwrap(),
            cow.meta.as_ref().unwrap()
        ));
    }

    #[test]
    fn test_cow_copy_deep_clones_mutable_slots() {
        let ext = make_extensions();
        let cow = ext.cow_copy();

        // Mutable/monotonic slots are deep cloned — independent copies
        assert!(cow.security.is_some());
        assert!(cow.http.is_some());
        assert!(cow.delegation.is_some());

        // Modifying the COW copy doesn't affect the original
        cow.security.as_ref().unwrap().has_label("PII");
    }

    #[test]
    fn test_cow_copy_propagates_write_tokens() {
        let mut ext = make_extensions();

        // No tokens on the original → no tokens on COW
        let cow_no_tokens = ext.cow_copy();
        assert!(cow_no_tokens.http_write_token.is_none());
        assert!(cow_no_tokens.labels_write_token.is_none());
        assert!(cow_no_tokens.delegation_write_token.is_none());

        // Executor sets tokens based on capabilities
        ext.http_write_token = Some(WriteToken::new());
        ext.labels_write_token = Some(WriteToken::new());

        // COW copy propagates only the tokens that exist
        let cow_with_tokens = ext.cow_copy();
        assert!(cow_with_tokens.http_write_token.is_some());
        assert!(cow_with_tokens.labels_write_token.is_some());
        assert!(cow_with_tokens.delegation_write_token.is_none()); // wasn't set
    }

    #[test]
    fn test_cow_copy_write_token_enables_guarded_write() {
        let mut ext = make_extensions();
        ext.http_write_token = Some(WriteToken::new());

        let mut cow = ext.cow_copy();

        // Can read without token
        assert_eq!(
            cow.http
                .as_ref()
                .unwrap()
                .read()
                .get_header("Authorization"),
            Some("Bearer token")
        );

        // Can write with token from COW
        let token = cow.http_write_token.as_ref().unwrap();
        cow.http
            .as_mut()
            .unwrap()
            .write(token)
            .set_header("X-Custom", "value");

        assert_eq!(
            cow.http.as_ref().unwrap().read().get_header("X-Custom"),
            Some("value")
        );

        // Original unchanged
        assert!(ext.http.as_ref().unwrap().get_header("X-Custom").is_none());
    }

    #[test]
    fn test_cow_copy_monotonic_label_insert() {
        let mut ext = make_extensions();
        ext.labels_write_token = Some(WriteToken::new());

        let mut cow = ext.cow_copy();

        // Can add labels on the COW copy
        cow.security.as_mut().unwrap().add_label("HIPAA");
        assert!(cow.security.as_ref().unwrap().has_label("HIPAA"));

        // Original unchanged
        assert!(!ext.security.as_ref().unwrap().has_label("HIPAA"));
    }

    #[test]
    fn test_validate_immutable_passes_for_cow() {
        let ext = make_extensions();
        let cow = ext.cow_copy();

        // COW copy shares immutable Arcs → validation passes
        assert!(ext.validate_immutable(&cow));
    }

    #[test]
    fn test_validate_immutable_fails_when_tampered() {
        let ext = make_extensions();
        let mut cow = ext.cow_copy();

        // Tamper with an immutable slot
        cow.request = Some(Arc::new(RequestExtension {
            request_id: Some("TAMPERED".into()),
            ..Default::default()
        }));

        // Validation fails — different Arc pointer
        assert!(!ext.validate_immutable(&cow));
    }

    #[test]
    fn test_validate_immutable_both_none_passes() {
        let ext = Extensions::default();
        let cow = ext.cow_copy();
        assert!(ext.validate_immutable(&cow));
    }

    #[test]
    fn test_clone_drops_write_tokens() {
        let mut ext = make_extensions();
        ext.http_write_token = Some(WriteToken::new());
        ext.labels_write_token = Some(WriteToken::new());
        ext.delegation_write_token = Some(WriteToken::new());

        // Regular clone drops all tokens
        let cloned = ext.clone();
        assert!(cloned.http_write_token.is_none());
        assert!(cloned.labels_write_token.is_none());
        assert!(cloned.delegation_write_token.is_none());

        // cow_copy propagates them
        let cow = ext.cow_copy();
        assert!(cow.http_write_token.is_some());
        assert!(cow.labels_write_token.is_some());
        assert!(cow.delegation_write_token.is_some());
    }

    #[test]
    fn test_cow_copy_modify_multiple_fields() {
        use crate::extensions::delegation::DelegationHop;
        use crate::extensions::DelegationExtension;

        // Build extensions with security, http, delegation, custom
        let mut security = SecurityExtension::default();
        security.add_label("PII");

        let mut http = HttpExtension::default();
        http.set_header("Authorization", "Bearer token");

        let mut ext = Extensions {
            security: Some(Arc::new(security)),
            http: Some(Arc::new(http)),
            delegation: Some(Arc::new(DelegationExtension::default())),
            custom: Some(Arc::new(
                [("existing".to_string(), serde_json::json!("value"))].into(),
            )),
            meta: Some(Arc::new(MetaExtension {
                entity_type: Some("tool".into()),
                ..Default::default()
            })),
            ..Default::default()
        };

        // Executor sets all write tokens
        ext.http_write_token = Some(WriteToken::new());
        ext.labels_write_token = Some(WriteToken::new());
        ext.delegation_write_token = Some(WriteToken::new());

        // Plugin does one cow_copy, modifies multiple fields
        let mut cow = ext.cow_copy();

        // 1. Add security labels (monotonic)
        cow.security.as_mut().unwrap().add_label("CHECKED");
        cow.security.as_mut().unwrap().add_label("COMPLIANT");

        // 2. Inject HTTP headers (guarded)
        let token = cow.http_write_token.as_ref().unwrap();
        cow.http
            .as_mut()
            .unwrap()
            .write(token)
            .set_header("X-Checked", "true");
        cow.http
            .as_mut()
            .unwrap()
            .write(token)
            .set_header("X-Policy", "v2");

        // 3. Append delegation hop (monotonic)
        cow.delegation.as_mut().unwrap().append_hop(DelegationHop {
            subject_id: "service-a".into(),
            scopes_granted: vec!["read_hr".into()],
            ..Default::default()
        });

        // 4. Add custom data (mutable, no token needed)
        cow.custom
            .as_mut()
            .unwrap()
            .insert("audit.timestamp".into(), serde_json::json!("2026-04-29"));

        // Verify COW copy has all modifications
        let sec = cow.security.as_ref().unwrap();
        assert!(sec.has_label("PII")); // original
        assert!(sec.has_label("CHECKED")); // added
        assert!(sec.has_label("COMPLIANT")); // added

        let http = cow.http.as_ref().unwrap().read();
        assert_eq!(http.get_header("Authorization"), Some("Bearer token")); // original
        assert_eq!(http.get_header("X-Checked"), Some("true")); // added
        assert_eq!(http.get_header("X-Policy"), Some("v2")); // added

        assert_eq!(cow.delegation.as_ref().unwrap().chain.len(), 1);
        assert_eq!(
            cow.delegation.as_ref().unwrap().chain[0].subject_id,
            "service-a"
        );

        assert_eq!(
            cow.custom.as_ref().unwrap().get("existing").unwrap(),
            "value"
        );
        assert_eq!(
            cow.custom.as_ref().unwrap().get("audit.timestamp").unwrap(),
            "2026-04-29"
        );

        // Verify original is unchanged
        assert!(!ext.security.as_ref().unwrap().has_label("CHECKED"));
        assert!(ext.http.as_ref().unwrap().get_header("X-Checked").is_none());
        assert!(ext.delegation.as_ref().unwrap().chain.is_empty());
        assert!(!ext.custom.as_ref().unwrap().contains_key("audit.timestamp"));

        // Immutable slots still valid
        assert!(ext.validate_immutable(&cow));
    }

    #[test]
    fn test_validate_immutable_passes_when_slot_filtered_out() {
        // Bug fix regression: when capability filtering hides a slot
        // from the plugin (e.g., agent=None in owned because plugin
        // lacks read_agent), validate_immutable must NOT treat that
        // as tampering.
        let ext = make_extensions();
        let mut cow = ext.cow_copy();

        // Simulate capability filtering hiding the agent slot
        cow.agent = None;

        // Validation should pass — plugin never saw the slot
        assert!(ext.validate_immutable(&cow));
    }

    #[test]
    fn test_validate_immutable_fails_when_slot_fabricated() {
        // If the original has no agent but the plugin returns one,
        // that's fabrication — should fail.
        let ext = Extensions::default(); // no agent
        let mut cow = ext.cow_copy();

        cow.agent = Some(Arc::new(AgentExtension {
            agent_id: Some("fabricated".into()),
            ..Default::default()
        }));

        assert!(!ext.validate_immutable(&cow));
    }

    #[test]
    fn test_validate_immutable_passes_multiple_slots_filtered() {
        // Multiple immutable slots filtered out — all should pass
        let ext = make_extensions();
        let mut cow = ext.cow_copy();

        cow.agent = None;
        cow.mcp = None;
        cow.completion = None;
        cow.framework = None;

        assert!(ext.validate_immutable(&cow));
    }

    #[test]
    fn test_merge_owned_preserves_http_response_headers() {
        // Bug fix regression: merge_owned must preserve response
        // headers written by a plugin through Guarded write access.
        let mut http = HttpExtension::default();
        http.set_request_header("Authorization", "Bearer tok");

        let mut ext = Extensions {
            http: Some(Arc::new(http)),
            ..Default::default()
        };
        ext.http_write_token = Some(WriteToken::new());

        let mut cow = ext.cow_copy();

        // Plugin writes response headers through the guard
        let token = cow.http_write_token.as_ref().unwrap();
        let h = cow.http.as_mut().unwrap().write(token);
        h.set_response_header("X-Tool-Name", "get_compensation");
        h.set_response_header("X-Status", "success");

        // Merge back
        ext.merge_owned(cow);

        // Response headers must be present after merge
        let merged_http = ext.http.as_ref().unwrap();
        assert_eq!(
            merged_http.get_response_header("X-Tool-Name"),
            Some("get_compensation")
        );
        assert_eq!(merged_http.get_response_header("X-Status"), Some("success"));
        // Original request headers preserved
        assert_eq!(
            merged_http.get_request_header("Authorization"),
            Some("Bearer tok")
        );
    }

    #[test]
    fn test_merge_owned_with_filtered_security() {
        // A plugin without read_labels gets empty labels in its
        // filtered view. After cow_copy + merge_owned, the pipeline's
        // security labels must be preserved (not overwritten with empty).
        let mut security = SecurityExtension::default();
        security.add_label("PII");
        security.add_label("HR");

        let ext = Extensions {
            security: Some(Arc::new(security)),
            ..Default::default()
        };

        // Simulate: plugin has no read_labels, so filtered security
        // has empty labels. cow_copy of filtered would have empty labels.
        let mut cow = ext.cow_copy();

        // Plugin's owned security has the labels (from cow_copy of full ext)
        // But in the real flow, it would be from the filtered ext.
        // Simulate filtered: clear labels
        cow.security.as_mut().unwrap().labels = crate::extensions::MonotonicSet::new();

        // merge_owned replaces pipeline security with owned
        let mut ext_mut = ext.clone();
        ext_mut.merge_owned(cow);

        // After merge, the security comes from the owned (which had empty labels)
        // This is expected — the executor's monotonic check should prevent
        // this case. merge_owned itself is just a field replacement.
        let merged_sec = ext_mut.security.as_ref().unwrap();
        assert!(!merged_sec.has_label("PII")); // replaced by owned
    }

    #[test]
    fn test_merge_owned_none_http_preserves_pipeline() {
        // If owned.http is None (plugin had no read_headers capability),
        // merge_owned replaces with None. The executor should only call
        // merge_owned when the plugin actually modified something.
        let mut http = HttpExtension::default();
        http.set_request_header("X-Original", "value");

        let mut ext = Extensions {
            http: Some(Arc::new(http)),
            ..Default::default()
        };

        let mut cow = ext.cow_copy();
        cow.http = None; // simulate filtered-out HTTP

        ext.merge_owned(cow);

        // HTTP is now None — this is the raw merge behavior.
        // The executor guards against this by only calling merge_owned
        // when the plugin returned modify_extensions.
        assert!(ext.http.is_none());
    }

    #[test]
    fn test_read_only_plugin_zero_cost() {
        // Plugin that only reads — no cow_copy, no clone
        let ext = make_extensions();

        // Read security labels
        let has_pii = ext
            .security
            .as_ref()
            .map(|s| s.has_label("PII"))
            .unwrap_or(false);
        assert!(has_pii);

        // Read HTTP headers
        let auth = ext
            .http
            .as_ref()
            .and_then(|h| h.get_header("Authorization"));
        assert_eq!(auth, Some("Bearer token"));

        // Read meta
        let entity = ext.meta.as_ref().and_then(|m| m.entity_type.as_deref());
        assert_eq!(entity, Some("tool"));

        // No cow_copy called — zero allocations for read-only access
    }
}