From 153790be7d0464efcd4cf5d0c8272b4632d0caec Mon Sep 17 00:00:00 2001 From: mkoushni Date: Sun, 25 Oct 2026 14:00:42 +0200 Subject: [PATCH 1/2] feat(ws): review fix Signed-off-by: mkoushni --- crates/ppe-core/src/engine.rs | 263 +++++++--- crates/ppe-core/src/hooks/metadata.rs | 18 +- crates/ppe-core/src/hooks/payload.rs | 12 + crates/ppe-core/src/hooks/types.rs | 1 + crates/ppe-core/src/lib.rs | 3 + crates/ppe-core/src/metadata.rs | 662 ++++++++++++++++++++++++++ crates/ppe-core/src/plugin.rs | 3 +- crates/ppe-core/src/prelude.rs | 3 + docs/content/pipeline.md | 7 + 9 files changed, 897 insertions(+), 75 deletions(-) create mode 100644 crates/ppe-core/src/metadata.rs diff --git a/crates/ppe-core/src/engine.rs b/crates/ppe-core/src/engine.rs index 31fc0e6c..13d5716d 100644 --- a/crates/ppe-core/src/engine.rs +++ b/crates/ppe-core/src/engine.rs @@ -22,6 +22,7 @@ // config. Trust flows: // config loader → engine → PluginRef → executor +use std::borrow::Cow; use std::hash::{Hash, Hasher}; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; @@ -65,6 +66,24 @@ pub const VIOLATION_ASSERTION_MISSING: &str = "auth.assertion_missing"; /// deny: the request never reached a rule. pub const VIOLATION_UNIDENTIFIED_REQUEST: &str = "unidentified_request"; +fn entries_for_target<'a>( + entries: &'a [crate::registry::HookEntry], + payload: &dyn PluginPayload, +) -> Cow<'a, [crate::registry::HookEntry]> { + payload.target_plugin().map_or_else( + || Cow::Borrowed(entries), + |target| { + Cow::Owned( + entries + .iter() + .filter(|entry| entry.plugin_ref.name() == target) + .cloned() + .collect(), + ) + }, + ) +} + /// Why a request could not be resolved against the route table. /// /// Resolving no route is not an error: the request falls back to the global @@ -2266,13 +2285,38 @@ impl PolicyEngine { // change our view mid-pipeline. let snapshot = self.load_runtime(); let hook_type = HookType::new(hook_name); - let all_entries = snapshot.registry.entries_for_hook(&hook_type); + let hook_entries = snapshot.registry.entries_for_hook(&hook_type); + let targeted_entries = entries_for_target(hook_entries, payload.as_ref()); + + if targeted_entries.is_empty() + && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + { + let violation = crate::error::PluginViolation::new( + "plugin.not_found", + format!("hook '{hook_name}' has no registered plugin instance named '{target}'"), + ); + return ( + self.finish( + &snapshot, + Some(hook_name), + None, + Some(payload), + PipelineResult::denied( + violation, + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + } // Same caveat as `invoke_named`: route annotations can produce a // dispatch entry without any plugin being registered on the // hook directly, so we can only short-circuit when both the // registry and the annotation map are empty. - if all_entries.is_empty() && snapshot.route_annotations.is_empty() { + if targeted_entries.is_empty() && snapshot.route_annotations.is_empty() { // Route filtering has not run, so the route is resolved here: a // contract written on a route holds whether or not this hook has a // plugin on it. @@ -2294,27 +2338,35 @@ impl PolicyEngine { ); } - let (entries, matched) = match self - .filter_entries_by_route(&snapshot, all_entries, &extensions, hook_name) - .await - { - Ok(resolved) => resolved, - Err(cause) => { - return ( - self.finish( - &snapshot, - Some(hook_name), - None, - Some(payload), - PipelineResult::denied( - cause.violation(), - extensions, - context_table.unwrap_or_default(), - ), - ) - .await, - BackgroundTasks::empty(), - ); + let (entries, matched) = match targeted_entries { + // The caller already selected this semantic operation's one source. + // Policy-mode route resolution deliberately returns no ordinary + // plugin chain, so applying it here would discard the target. + Cow::Owned(owned) => (Arc::new(owned), None), + Cow::Borrowed(all_entries) => { + match self + .filter_entries_by_route(&snapshot, all_entries, &extensions, hook_name) + .await + { + Ok(resolved) => resolved, + Err(cause) => { + return ( + self.finish( + &snapshot, + Some(hook_name), + None, + Some(payload), + PipelineResult::denied( + cause.violation(), + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + }, + } }, }; if entries.is_empty() { @@ -2396,12 +2448,40 @@ impl PolicyEngine { ) -> (PipelineResult, BackgroundTasks) { let snapshot = self.load_runtime(); let hook_type = HookType::new(H::NAME); - let all_entries = snapshot.registry.entries_for_hook(&hook_type); + let hook_entries = snapshot.registry.entries_for_hook(&hook_type); + let targeted_entries = entries_for_target(hook_entries, &payload); + + if targeted_entries.is_empty() + && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + { + let violation = crate::error::PluginViolation::new( + "plugin.not_found", + format!( + "hook '{}' has no registered plugin instance named '{target}'", + H::NAME + ), + ); + return ( + self.finish( + &snapshot, + Some(H::NAME), + None, + Some(Box::new(payload)), + PipelineResult::denied( + violation, + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + } // See `invoke_named` for why we don't short-circuit on - // `all_entries.is_empty()` alone — route annotations can fire + // `targeted_entries.is_empty()` alone — route annotations can fire // without a directly-registered plugin. - if all_entries.is_empty() && snapshot.route_annotations.is_empty() { + if targeted_entries.is_empty() && snapshot.route_annotations.is_empty() { let boxed: Box = Box::new(payload); let matched = resolve_contract_route(&snapshot, &extensions); return ( @@ -2421,27 +2501,35 @@ impl PolicyEngine { ); } - let (entries, matched) = match self - .filter_entries_by_route(&snapshot, all_entries, &extensions, H::NAME) - .await - { - Ok(resolved) => resolved, - Err(cause) => { - return ( - self.finish( - &snapshot, - Some(H::NAME), - None, - Some(Box::new(payload)), - PipelineResult::denied( - cause.violation(), - extensions, - context_table.unwrap_or_default(), - ), - ) - .await, - BackgroundTasks::empty(), - ); + let (entries, matched) = match targeted_entries { + // A targeted semantic operation is already fully resolved by its + // payload and must not be replaced by a route annotation or the + // policy mode's intentionally empty ordinary plugin chain. + Cow::Owned(owned) => (Arc::new(owned), None), + Cow::Borrowed(all_entries) => { + match self + .filter_entries_by_route(&snapshot, all_entries, &extensions, H::NAME) + .await + { + Ok(resolved) => resolved, + Err(cause) => { + return ( + self.finish( + &snapshot, + Some(H::NAME), + None, + Some(Box::new(payload)), + PipelineResult::denied( + cause.violation(), + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + }, + } }, }; if entries.is_empty() { @@ -2522,11 +2610,36 @@ impl PolicyEngine { ) -> (PipelineResult, BackgroundTasks) { let snapshot = self.load_runtime(); let hook_type = HookType::new(hook_name); - let all_entries = snapshot.registry.entries_for_hook(&hook_type); + let hook_entries = snapshot.registry.entries_for_hook(&hook_type); + let targeted_entries = entries_for_target(hook_entries, &payload); + + if targeted_entries.is_empty() + && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + { + let violation = crate::error::PluginViolation::new( + "plugin.not_found", + format!("hook '{hook_name}' has no registered plugin instance named '{target}'"), + ); + return ( + self.finish( + &snapshot, + Some(hook_name), + None, + Some(Box::new(payload)), + PipelineResult::denied( + violation, + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + } // No registered entries AND no route annotations → nothing to // do. Allow-and-pass-through. We can't short-circuit on - // `all_entries.is_empty()` alone, because route annotations + // `targeted_entries.is_empty()` alone, because route annotations // (external-orchestrator handlers from APL / future Rego / // Cedar-direct) can produce a single-entry dispatch even when // no plugin was registered on the hook directly. @@ -2534,7 +2647,7 @@ impl PolicyEngine { // Returning here also means a request whose path cannot be read is // allowed rather than denied on such a hook. Nothing would have // enforced on it either way. - if all_entries.is_empty() && snapshot.route_annotations.is_empty() { + if targeted_entries.is_empty() && snapshot.route_annotations.is_empty() { let boxed: Box = Box::new(payload); let matched = resolve_contract_route(&snapshot, &extensions); return ( @@ -2554,27 +2667,35 @@ impl PolicyEngine { ); } - let (entries, matched) = match self - .filter_entries_by_route(&snapshot, all_entries, &extensions, hook_name) - .await - { - Ok(resolved) => resolved, - Err(cause) => { - return ( - self.finish( - &snapshot, - Some(hook_name), - None, - Some(Box::new(payload)), - PipelineResult::denied( - cause.violation(), - extensions, - context_table.unwrap_or_default(), - ), - ) - .await, - BackgroundTasks::empty(), - ); + let (entries, matched) = match targeted_entries { + // The policy evaluator chose this source by instance name. Route + // filtering governs lifecycle hooks; it must not erase or replace + // an already-resolved semantic operation. + Cow::Owned(owned) => (Arc::new(owned), None), + Cow::Borrowed(all_entries) => { + match self + .filter_entries_by_route(&snapshot, all_entries, &extensions, hook_name) + .await + { + Ok(resolved) => resolved, + Err(cause) => { + return ( + self.finish( + &snapshot, + Some(hook_name), + None, + Some(Box::new(payload)), + PipelineResult::denied( + cause.violation(), + extensions, + context_table.unwrap_or_default(), + ), + ) + .await, + BackgroundTasks::empty(), + ); + }, + } }, }; if entries.is_empty() { diff --git a/crates/ppe-core/src/hooks/metadata.rs b/crates/ppe-core/src/hooks/metadata.rs index d44ecb1f..ecb155d1 100644 --- a/crates/ppe-core/src/hooks/metadata.rs +++ b/crates/ppe-core/src/hooks/metadata.rs @@ -85,6 +85,7 @@ use crate::delegation::hook::DELEGATION_HOOK_METADATA; use crate::elicitation::hook::ELICITATION_HOOK_METADATA; use crate::http_hook::HTTP_HOOK_METADATA; use crate::identity::hook::IDENTITY_HOOK_METADATA; +use crate::metadata::METADATA_HOOK_METADATA; /// Lifecycle position a hook occupies for dispatcher purposes. /// @@ -190,6 +191,7 @@ const HOOK_TABLES: &[&[(&str, HookMetadata)]] = &[ IDENTITY_HOOK_METADATA, DELEGATION_HOOK_METADATA, ELICITATION_HOOK_METADATA, + METADATA_HOOK_METADATA, ]; /// How many rows the per-module tables hold between them, so the flat @@ -311,6 +313,7 @@ mod tests { use crate::hooks::trait_def::HookTypeDef as _; use crate::http_hook::{HOOK_HTTP_REQUEST, HOOK_HTTP_RESPONSE, HttpHook}; use crate::identity::{HOOK_IDENTITY_RESOLVE, IdentityHook}; + use crate::metadata::{HOOK_METADATA_FETCH, MetadataFetchHook}; #[test] fn cmf_tool_pre_invoke_is_pre_phase_for_tool_entity() { @@ -347,10 +350,15 @@ mod tests { #[test] fn an_unphased_hook_is_distinguishable_from_an_absent_one() { - // The three unphased hooks used to be indistinguishable from a + // The unphased hooks used to be indistinguishable from a // name nobody registered, which is how a missing row read as a // deliberate choice. - for name in [HOOK_IDENTITY_RESOLVE, HOOK_TOKEN_DELEGATE, HOOK_ELICIT] { + for name in [ + HOOK_IDENTITY_RESOLVE, + HOOK_TOKEN_DELEGATE, + HOOK_ELICIT, + HOOK_METADATA_FETCH, + ] { let meta = lookup(name).expect("registered"); assert_eq!(meta.phase, HookPhase::Unphased, "{name}"); } @@ -482,7 +490,8 @@ mod tests { assert_eq!(IDENTITY_HOOK_METADATA.len(), 1); assert_eq!(DELEGATION_HOOK_METADATA.len(), 1); assert_eq!(ELICITATION_HOOK_METADATA.len(), 1); - assert_eq!(BUILTIN_HOOK_METADATA.len(), 13); + assert_eq!(METADATA_HOOK_METADATA.len(), 1); + assert_eq!(BUILTIN_HOOK_METADATA.len(), 14); for table in HOOK_TABLES { for (name, _) in *table { assert!( @@ -543,6 +552,7 @@ mod tests { use crate::elicitation::HOOK_ELICIT; use crate::http_hook::{HOOK_HTTP_REQUEST, HOOK_HTTP_RESPONSE}; use crate::identity::HOOK_IDENTITY_RESOLVE; + use crate::metadata::HOOK_METADATA_FETCH; for name in [ HOOK_CMF_TOOL_PRE_INVOKE, @@ -558,6 +568,7 @@ mod tests { HOOK_IDENTITY_RESOLVE, HOOK_TOKEN_DELEGATE, HOOK_ELICIT, + HOOK_METADATA_FETCH, ] { assert!( BUILTIN_HOOK_METADATA.iter().any(|(n, _)| *n == name), @@ -577,6 +588,7 @@ mod tests { (IDENTITY_HOOK_METADATA, IdentityHook::NAME), (DELEGATION_HOOK_METADATA, TokenDelegateHook::NAME), (ELICITATION_HOOK_METADATA, ElicitationHook::NAME), + (METADATA_HOOK_METADATA, MetadataFetchHook::NAME), ] { for (name, meta) in table { assert_eq!(meta.family, Some(family), "{name}"); diff --git a/crates/ppe-core/src/hooks/payload.rs b/crates/ppe-core/src/hooks/payload.rs index 0a451073..fc58115b 100644 --- a/crates/ppe-core/src/hooks/payload.rs +++ b/crates/ppe-core/src/hooks/payload.rs @@ -83,6 +83,18 @@ pub trait PluginPayload: Send + Sync + 'static { /// Downcast to a concrete type via `&mut dyn Any`. fn as_any_mut(&mut self) -> &mut dyn Any; + /// Name of the one plugin instance this payload targets, when dispatch + /// must select an instance rather than run every handler on the hook. + /// + /// Most hook payloads return `None` and use ordinary hook-wide dispatch. + /// Semantic operations such as metadata acquisition override this so + /// [`PolicyEngine::invoke_named`](crate::engine::PolicyEngine::invoke_named) + /// can resolve a configured source by instance name while retaining the + /// normal typed hook, capability filtering, and executor behavior. + fn target_plugin(&self) -> Option<&str> { + None + } + /// Canonical bytes of this payload for content-addressed audit /// provenance, or `None` for a payload that cannot or should not be /// serialized, which is the default. diff --git a/crates/ppe-core/src/hooks/types.rs b/crates/ppe-core/src/hooks/types.rs index 1b90b142..79a854af 100644 --- a/crates/ppe-core/src/hooks/types.rs +++ b/crates/ppe-core/src/hooks/types.rs @@ -143,6 +143,7 @@ mod tests { "http.request", "http.response", "elicit", + "metadata.fetch", // The old list spelled the prompt pair `_fetch`. "cmf.prompt_pre_invoke", "cmf.prompt_post_invoke", diff --git a/crates/ppe-core/src/lib.rs b/crates/ppe-core/src/lib.rs index ab8b690b..b8069c7a 100644 --- a/crates/ppe-core/src/lib.rs +++ b/crates/ppe-core/src/lib.rs @@ -27,6 +27,7 @@ // minting for downstream calls) // - [`elicitation`] — Elicitation hook family (human-in-the-loop: // approval, confirmation, step-up, …) +// - [`metadata`] — named policy-information source dispatch // - [`error`] — Error types, violations, and result types //! The plugin runtime: engine, executor, hook registry, and config parser. @@ -90,6 +91,8 @@ pub mod http_retry; pub mod http_testing; /// The identity resolution hook and its payload. pub mod identity; +/// Typed metadata-source hook and request/result contract. +pub mod metadata; /// The `Plugin` trait and its trusted configuration. pub mod plugin; /// Curated re-exports for plugin authors. diff --git a/crates/ppe-core/src/metadata.rs b/crates/ppe-core/src/metadata.rs new file mode 100644 index 00000000..6973dd3d --- /dev/null +++ b/crates/ppe-core/src/metadata.rs @@ -0,0 +1,662 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! Typed policy dispatch for externally acquired metadata. + +use crate::executor::PipelineResult; +use crate::extensions::AcquiredFacts; +use crate::hooks::payload::PluginPayload; +use crate::hooks::trait_def::PluginResult; + +crate::define_hooks! { + /// The metadata family's rows in the built-in hook metadata table. + METADATA_HOOK_METADATA; + + /// Acquire one record from a named metadata source. This operation is + /// invoked from policy evaluation rather than at a wire lifecycle phase. + HOOK_METADATA_FETCH: "metadata.fetch" => + family: MetadataFetchHook, entity: None, phase: Unphased; +} + +/// Typed request and result accumulator for one metadata source invocation. +/// +/// The instance name selects exactly one plugin registered on +/// [`HOOK_METADATA_FETCH`]. A handler returns its record with +/// [`Self::with_facts`] through [`PluginResult::modify_payload`]. The record is +/// not installed into extensions here; framework policy evaluation validates, +/// stamps, and installs it after dispatch. +#[derive(Clone, Debug, PartialEq)] +pub struct MetadataFetchRequest { + instance: String, + facts: Option, +} + +impl MetadataFetchRequest { + /// Request acquisition from the configured source instance `instance`. + #[must_use] + pub fn new(instance: impl Into) -> Self { + Self { + instance: instance.into(), + facts: None, + } + } + + /// Configured source instance selected for this dispatch. + #[must_use] + pub fn instance(&self) -> &str { + &self.instance + } + + /// Return the source record produced by the handler, when present. + #[must_use] + pub const fn facts(&self) -> Option<&AcquiredFacts> { + self.facts.as_ref() + } + + /// Attach the acquired record returned by a source handler. + #[must_use] + pub fn with_facts(mut self, facts: AcquiredFacts) -> Self { + self.facts = Some(facts); + self + } + + /// Borrow the typed result still held by a metadata dispatch. + /// + /// Returns `None` when the pipeline was denied or the payload is not a + /// [`MetadataFetchRequest`]. The caller that installs the record into + /// extensions clones it there, where ownership is required. + #[must_use] + pub fn from_pipeline_result(result: &PipelineResult) -> Option<&Self> { + result + .modified_payload + .as_ref() + .and_then(|payload| payload.as_any().downcast_ref::()) + } +} + +impl PluginPayload for MetadataFetchRequest { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn target_plugin(&self) -> Option<&str> { + Some(&self.instance) + } +} + +/// Result returned by a typed metadata source handler. +pub type MetadataFetchResult = PluginResult; + +crate::define_hook! { + /// Metadata acquisition hook dispatched to one configured source instance. + MetadataFetchHook, "metadata.fetch" => { + payload: MetadataFetchRequest, + result: MetadataFetchResult, + } +} + +#[cfg(test)] +#[expect(clippy::expect_used, clippy::unwrap_used, reason = "tests")] +mod tests { + use std::collections::HashSet; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + + use chrono::Utc; + use serde_json::json; + + use super::*; + use crate::cmf::{CmfHook, MessagePayload}; + use crate::config::parse_config; + use crate::context::PluginContext; + use crate::engine::PolicyEngine; + use crate::error::PluginError; + use crate::extensions::{Extensions, MetaExtension, MetadataCacheDisposition}; + use crate::factory::{PluginFactory, PluginInstance}; + use crate::hooks::TypedHandlerAdapter; + use crate::hooks::trait_def::HookHandler; + use crate::plugin::{Plugin, PluginConfig, PluginMode}; + use crate::registry::AnyHookHandler; + + struct Source { + config: PluginConfig, + calls: Arc, + saw_metadata: Arc, + forge_metadata: bool, + } + + impl Plugin for Source { + fn config(&self) -> &PluginConfig { + &self.config + } + } + + impl HookHandler for Source { + async fn handle( + &self, + request: &MetadataFetchRequest, + extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> MetadataFetchResult { + self.calls.fetch_add(1, Ordering::Relaxed); + self.saw_metadata + .store(extensions.metadata.is_some(), Ordering::Relaxed); + + let facts = AcquiredFacts::success( + json!({"source": self.config.name}), + Utc::now(), + MetadataCacheDisposition::Disabled, + ); + let response = request.clone().with_facts(facts); + if !self.forge_metadata { + return PluginResult::modify_payload(response); + } + + let mut forged = Extensions::default(); + forged + .install_metadata( + "forged", + "test/source", + AcquiredFacts::success( + json!(true), + Utc::now(), + MetadataCacheDisposition::Disabled, + ), + ) + .unwrap(); + let mut owned = extensions.cow_copy(); + owned.metadata = forged.metadata; + PluginResult::modify(response, owned) + } + } + + fn config(name: &str, read_metadata: bool) -> PluginConfig { + PluginConfig { + name: name.to_owned(), + kind: "test/source".to_owned(), + hooks: vec![HOOK_METADATA_FETCH.to_owned()], + mode: PluginMode::Sequential, + capabilities: if read_metadata { + HashSet::from(["read_metadata".to_owned()]) + } else { + HashSet::new() + }, + ..PluginConfig::default() + } + } + + fn source( + config: PluginConfig, + calls: Arc, + saw_metadata: Arc, + forge_metadata: bool, + ) -> Arc { + Arc::new(Source { + config, + calls, + saw_metadata, + forge_metadata, + }) + } + + #[tokio::test] + async fn direct_registration_dispatches_only_the_named_source() { + let engine = PolicyEngine::default(); + let first_calls = Arc::new(AtomicUsize::new(0)); + let second_calls = Arc::new(AtomicUsize::new(0)); + let second_saw_metadata = Arc::new(AtomicBool::new(false)); + let first = source( + config("first", false), + Arc::clone(&first_calls), + Arc::new(AtomicBool::new(false)), + false, + ); + let second = source( + config("second", false), + Arc::clone(&second_calls), + Arc::clone(&second_saw_metadata), + false, + ); + engine + .register_handler_for_names::( + first, + config("first", false), + &[HOOK_METADATA_FETCH], + ) + .unwrap(); + engine + .register_handler_for_names::( + second, + config("second", false), + &[HOOK_METADATA_FETCH], + ) + .unwrap(); + + let mut extensions = Extensions::default(); + extensions + .install_metadata( + "existing", + "host/source", + AcquiredFacts::success(json!(true), Utc::now(), MetadataCacheDisposition::Disabled), + ) + .unwrap(); + let (result, _) = engine + .invoke_named::( + HOOK_METADATA_FETCH, + MetadataFetchRequest::new("second"), + extensions, + None, + ) + .await; + + assert!( + result.continue_processing, + "dispatch to a registered source should continue", + ); + assert_eq!( + first_calls.load(Ordering::Relaxed), + 0, + "unselected source 'first' must not be dispatched", + ); + assert_eq!( + second_calls.load(Ordering::Relaxed), + 1, + "selected source 'second' must be dispatched once", + ); + assert!( + !second_saw_metadata.load(Ordering::Relaxed), + "source without read_metadata must not see existing metadata", + ); + let response = MetadataFetchRequest::from_pipeline_result(&result) + .expect("dispatch should return the metadata response"); + assert_eq!( + response.instance(), + "second", + "response should name the selected source", + ); + assert_eq!( + response + .facts() + .expect("response should carry acquired facts") + .value(), + Some(&json!({"source": "second"})), + "facts should come from the selected source", + ); + } + + #[tokio::test] + async fn typed_invoke_dispatches_only_the_named_source() { + let engine = PolicyEngine::default(); + let first_calls = Arc::new(AtomicUsize::new(0)); + let second_calls = Arc::new(AtomicUsize::new(0)); + for (name, calls) in [ + ("first", Arc::clone(&first_calls)), + ("second", Arc::clone(&second_calls)), + ] { + engine + .register_handler_for_names::( + source( + config(name, false), + calls, + Arc::new(AtomicBool::new(false)), + false, + ), + config(name, false), + &[HOOK_METADATA_FETCH], + ) + .unwrap(); + } + + let (result, _) = engine + .invoke::( + MetadataFetchRequest::new("second"), + Extensions::default(), + None, + ) + .await; + + assert!( + result.continue_processing, + "typed dispatch to a registered source should continue", + ); + assert_eq!( + first_calls.load(Ordering::Relaxed), + 0, + "unselected source 'first' must not be dispatched", + ); + assert_eq!( + second_calls.load(Ordering::Relaxed), + 1, + "selected source 'second' must be dispatched once", + ); + } + + #[tokio::test] + async fn dynamic_invoke_dispatches_only_the_named_source() { + let engine = PolicyEngine::default(); + let first_calls = Arc::new(AtomicUsize::new(0)); + let second_calls = Arc::new(AtomicUsize::new(0)); + for (name, calls) in [ + ("first", Arc::clone(&first_calls)), + ("second", Arc::clone(&second_calls)), + ] { + engine + .register_handler_for_names::( + source( + config(name, false), + calls, + Arc::new(AtomicBool::new(false)), + false, + ), + config(name, false), + &[HOOK_METADATA_FETCH], + ) + .unwrap(); + } + + let (result, _) = engine + .invoke_by_name( + HOOK_METADATA_FETCH, + Box::new(MetadataFetchRequest::new("second")), + Extensions::default(), + None, + ) + .await; + + assert!( + result.continue_processing, + "dynamic dispatch to a registered source should continue", + ); + assert_eq!( + first_calls.load(Ordering::Relaxed), + 0, + "unselected source 'first' must not be dispatched", + ); + assert_eq!( + second_calls.load(Ordering::Relaxed), + 1, + "selected source 'second' must be dispatched once", + ); + } + + #[tokio::test] + async fn unknown_named_source_denies_with_an_actionable_error() { + let engine = PolicyEngine::default(); + let (result, _) = engine + .invoke_named::( + HOOK_METADATA_FETCH, + MetadataFetchRequest::new("missing"), + Extensions::default(), + None, + ) + .await; + + assert!( + !result.continue_processing, + "unknown source must deny", + ); + let violation = result.violation.expect("unknown source must deny"); + assert_eq!( + violation.code, + "plugin.not_found", + "unknown source denies with plugin.not_found", + ); + assert!( + violation.reason.contains("metadata.fetch"), + "denial should name the metadata.fetch hook", + ); + assert!( + violation.reason.contains("missing"), + "denial should name the missing instance", + ); + } + + #[tokio::test] + async fn capabilities_filter_metadata_and_merge_back_rejects_forged_records() { + let engine = PolicyEngine::default(); + let saw_metadata = Arc::new(AtomicBool::new(false)); + let plugin = source( + config("account", true), + Arc::new(AtomicUsize::new(0)), + Arc::clone(&saw_metadata), + true, + ); + engine + .register_handler_for_names::( + plugin, + config("account", true), + &[HOOK_METADATA_FETCH], + ) + .unwrap(); + let mut extensions = Extensions::default(); + extensions + .install_metadata( + "existing", + "host/source", + AcquiredFacts::success( + json!("trusted"), + Utc::now(), + MetadataCacheDisposition::Disabled, + ), + ) + .unwrap(); + + let (result, _) = engine + .invoke_named::( + HOOK_METADATA_FETCH, + MetadataFetchRequest::new("account"), + extensions, + None, + ) + .await; + + assert!( + saw_metadata.load(Ordering::Relaxed), + "source with read_metadata should see existing metadata", + ); + let returned = result + .modified_extensions + .expect("allowed dispatch returns final extensions"); + let metadata = returned.metadata.expect("parent metadata is retained"); + assert!( + metadata.get("existing").is_some(), + "parent metadata should be retained", + ); + assert!( + metadata.get("forged").is_none(), + "handler must not install a forged metadata record", + ); + } + + struct SourceFactory { + calls: Arc, + } + + impl PluginFactory for SourceFactory { + fn create(&self, config: &PluginConfig) -> Result> { + if config.hooks.as_slice() != [HOOK_METADATA_FETCH] { + return Err(Box::new(PluginError::Config { + message: "test/source requires hooks: [metadata.fetch]".to_owned(), + })); + } + let plugin = source( + config.clone(), + Arc::clone(&self.calls), + Arc::new(AtomicBool::new(false)), + false, + ); + let handler: Arc = + Arc::new(TypedHandlerAdapter::::new( + Arc::clone(&plugin), + )); + Ok(PluginInstance { + plugin, + handlers: vec![(HOOK_METADATA_FETCH, handler)], + }) + } + } + + #[tokio::test] + async fn configuration_factory_registers_a_typed_named_source() { + let calls = Arc::new(AtomicUsize::new(0)); + let engine = PolicyEngine::default(); + engine.register_factory( + "test/source", + Box::new(SourceFactory { + calls: Arc::clone(&calls), + }), + ); + let policy = parse_config( + " +engine_settings: + dispatch: policy +plugins: + - name: account + kind: test/source + hooks: [metadata.fetch] + mode: sequential +routes: + - tool: checkout +", + ) + .unwrap(); + engine.load_config(policy).unwrap(); + + let (result, _) = engine + .invoke_named::( + HOOK_METADATA_FETCH, + MetadataFetchRequest::new("account"), + Extensions { + meta: Some(Arc::new(MetaExtension { + entity_type: Some("tool".to_owned()), + entity_name: Some("checkout".to_owned()), + ..MetaExtension::default() + })), + ..Extensions::default() + }, + None, + ) + .await; + + assert!( + result.continue_processing, + "configured source dispatch should continue", + ); + assert_eq!( + calls.load(Ordering::Relaxed), + 1, + "configured source 'account' must be dispatched once", + ); + assert!( + MetadataFetchRequest::from_pipeline_result(&result) + .is_some_and(|response| response.facts().is_some()), + "configured source should return acquired facts", + ); + } + + struct WrongFamily { + config: PluginConfig, + } + + impl Plugin for WrongFamily { + fn config(&self) -> &PluginConfig { + &self.config + } + } + + impl HookHandler for WrongFamily { + async fn handle( + &self, + _payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + PluginResult::allow() + } + } + + struct WrongFamilyFactory; + + impl PluginFactory for WrongFamilyFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(WrongFamily { + config: config.clone(), + }); + let handler: Arc = + Arc::new(TypedHandlerAdapter::::new( + Arc::clone(&plugin), + )); + Ok(PluginInstance { + plugin, + handlers: vec![(HOOK_METADATA_FETCH, handler)], + }) + } + } + + #[test] + fn metadata_hook_rejects_a_mismatched_handler_family_at_registration() { + let engine = PolicyEngine::default(); + let config = config("wrong", false); + let plugin = Arc::new(WrongFamily { + config: config.clone(), + }); + let error = engine + .register_handler_for_names::(plugin, config, &[HOOK_METADATA_FETCH]) + .expect_err("CMF handler must not register as a metadata source"); + let message = error.to_string(); + assert!( + message.contains("wrong"), + "registration error should name the plugin", + ); + assert!( + message.contains("metadata.fetch"), + "registration error should name the hook", + ); + assert!( + message.contains("cmf"), + "registration error should name the handler family", + ); + } + + #[test] + fn configuration_load_rejects_a_mismatched_handler_family() { + let engine = PolicyEngine::default(); + engine.register_factory("test/wrong", Box::new(WrongFamilyFactory)); + let policy = parse_config( + " +engine_settings: + dispatch: hooks +plugins: + - name: wrong + kind: test/wrong + hooks: [metadata.fetch] + mode: sequential +", + ) + .unwrap(); + let error = engine + .load_config(policy) + .expect_err("config load must reject the mismatched family"); + let message = error.to_string(); + assert!( + message.contains("wrong"), + "config load error should name the plugin", + ); + assert!( + message.contains("metadata.fetch"), + "config load error should name the hook", + ); + assert!( + message.contains("cmf"), + "config load error should name the handler family", + ); + } +} diff --git a/crates/ppe-core/src/plugin.rs b/crates/ppe-core/src/plugin.rs index 3e5f61b6..fb273bb6 100644 --- a/crates/ppe-core/src/plugin.rs +++ b/crates/ppe-core/src/plugin.rs @@ -208,7 +208,8 @@ pub struct PluginConfig { /// [`cmf::constants`][crate::cmf::constants] plus /// [`identity::HOOK_IDENTITY_RESOLVE`][crate::identity::HOOK_IDENTITY_RESOLVE], /// [`delegation::HOOK_TOKEN_DELEGATE`][crate::delegation::HOOK_TOKEN_DELEGATE], - /// and [`elicitation::HOOK_ELICIT`][crate::elicitation::HOOK_ELICIT]. + /// [`elicitation::HOOK_ELICIT`][crate::elicitation::HOOK_ELICIT], and + /// [`metadata::HOOK_METADATA_FETCH`][crate::metadata::HOOK_METADATA_FETCH]. /// /// A host with its own hooks describes them through /// [`register_hook_metadata`][crate::hooks::register_hook_metadata] diff --git a/crates/ppe-core/src/prelude.rs b/crates/ppe-core/src/prelude.rs index bb71dde6..4d12d497 100644 --- a/crates/ppe-core/src/prelude.rs +++ b/crates/ppe-core/src/prelude.rs @@ -75,6 +75,9 @@ pub use crate::plugin::{OnError, Plugin, PluginConfig, PluginMode}; // Hook system pub use crate::hooks::{Extensions, HookHandler, HookTypeDef, PluginPayload, PluginResult}; +pub use crate::metadata::{ + HOOK_METADATA_FETCH, MetadataFetchHook, MetadataFetchRequest, MetadataFetchResult, +}; // Construction. A plugin implements `PluginFactory` and returns a // `PluginInstance`; `TypedHandlerAdapter` is how a typed `HookHandler` is diff --git a/docs/content/pipeline.md b/docs/content/pipeline.md index 52718b84..c9d9ae02 100644 --- a/docs/content/pipeline.md +++ b/docs/content/pipeline.md @@ -25,6 +25,13 @@ The hook names a plugin declares in `hooks:` come from a fixed table: | Identity | `identity.resolve` | | Delegation | `token.delegate` | | Elicitation | `elicit` | +| Metadata | `metadata.fetch` | + +`metadata.fetch` is an unphased PPE policy operation, not an HTTP request or +response lifecycle hook. Its typed request selects one configured source by +plugin instance name. The source receives the ordinary capability-filtered +extension view and returns an acquired-facts record; framework code owns +validation, provenance stamping, and installation of that record. `http.response` is the return half of the generic HTTP path, installed when a global `result:` or `post_invocation:` block exists. Response From 1aa2a9f6e22acac36b0e173d332e46b8eabc4fac Mon Sep 17 00:00:00 2001 From: mkoushni Date: Sun, 25 Oct 2026 14:48:06 +0200 Subject: [PATCH 2/2] fix: address clippy redundant closure warnings and format code Signed-off-by: mkoushni --- crates/ppe-core/src/engine.rs | 6 +++--- crates/ppe-core/src/metadata.rs | 8 ++------ 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/crates/ppe-core/src/engine.rs b/crates/ppe-core/src/engine.rs index 13d5716d..7051dd5e 100644 --- a/crates/ppe-core/src/engine.rs +++ b/crates/ppe-core/src/engine.rs @@ -2289,7 +2289,7 @@ impl PolicyEngine { let targeted_entries = entries_for_target(hook_entries, payload.as_ref()); if targeted_entries.is_empty() - && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + && let Some(target) = payload.target_plugin().map(ToOwned::to_owned) { let violation = crate::error::PluginViolation::new( "plugin.not_found", @@ -2452,7 +2452,7 @@ impl PolicyEngine { let targeted_entries = entries_for_target(hook_entries, &payload); if targeted_entries.is_empty() - && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + && let Some(target) = payload.target_plugin().map(ToOwned::to_owned) { let violation = crate::error::PluginViolation::new( "plugin.not_found", @@ -2614,7 +2614,7 @@ impl PolicyEngine { let targeted_entries = entries_for_target(hook_entries, &payload); if targeted_entries.is_empty() - && let Some(target) = payload.target_plugin().map(|target| target.to_owned()) + && let Some(target) = payload.target_plugin().map(ToOwned::to_owned) { let violation = crate::error::PluginViolation::new( "plugin.not_found", diff --git a/crates/ppe-core/src/metadata.rs b/crates/ppe-core/src/metadata.rs index 6973dd3d..b50769ba 100644 --- a/crates/ppe-core/src/metadata.rs +++ b/crates/ppe-core/src/metadata.rs @@ -399,14 +399,10 @@ mod tests { ) .await; - assert!( - !result.continue_processing, - "unknown source must deny", - ); + assert!(!result.continue_processing, "unknown source must deny",); let violation = result.violation.expect("unknown source must deny"); assert_eq!( - violation.code, - "plugin.not_found", + violation.code, "plugin.not_found", "unknown source denies with plugin.not_found", ); assert!(