feat: first-class decision & effect auditing (the audit seam)

CHANGELOG.md

@@ -18,6 +18,15 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 ### Added
 
 - **Out-of-process host for existing Python CPEX plugins.** A new `cpex-hosts-python` crate registers `kind: isolated_venv`, running an unmodified Python CPEX plugin in its own cached virtualenv as a subprocess instead of in-process through the PyO3 bindings. Each plugin gets a venv keyed by a SHA-256 fingerprint of its requirements + manifest (rebuilt when either changes, `rmtree`d rather than upgraded in place so a removed dependency actually disappears), and the host drives the Python framework's `worker.py` over a newline-delimited JSON stdio protocol. Hook payloads, `context`, and the capability-filtered `Extensions` view cross as JSON; returns come back as a serialized `PluginResult`, with `modified_extensions` merged through the executor's existing copy-on-write tier validation — the host implements no tier logic of its own. Failure modes the executor cannot otherwise distinguish (venv build failure, worker death mid-flight, a task over `max_content_size`, per-invocation timeout) map to distinct `PluginError`s carrying a stable `code` and structured `details`, so the executor's configured `on_error` policy applies unchanged. Pure Rust plus a subprocess — no libpython link, so the crate is in `default-members` and a plain `cargo build` does not require a Python dev install. The wire contract is pinned in `docs/specs/extensions-wire-contract.md`; CMF §3 remains normative for the extension slots themselves. (#149)
+- **First-class decision and effect auditing.** CPEX can now audit its own enforcement — every allow, deny, and modify. A new `AuditHook` family, auto-attached by the `PluginManager`, fires at the executor's verdict return points. Each carries a `DecisionLog` — executor-owned and handed only to audit sinks with the ordered plugin steps, the terminal verdict, the invocation's W3C trace span (`trace_id` / `span_id` / `parent_span_id`, child-span model: a fresh span whose parent is the request's span, for causal-DAG reconstruction), the taint labels the request arrived with, and, opt-in, a content hash of the payload at entry. Irreversible external effects (a token mint, an approval grant) are audited as their own events through a capability-gated, write-ahead protocol: a plugin holding `emit_effect` calls `ext.begin_effect` to durably record intent *before* the act (fail-closed — no durable record, no act) and `ext.complete_effect` to record the outcome (`confirmed` / `rejected` / `unknown`). A durable `FileEffectLog` write-ahead log (append + `fsync`, serialized against concurrent writers, self-compacting at a configurable threshold) makes the intent crash-safe; startup recovery (`PluginManager::recover_effects`) compacts completed effects and reconciles crash-orphaned ones against the issuing participant through an `EffectReconciler` seam (the default logs and leaves them `unknown`). `Extensions::perform_effect` brackets the two-phase protocol so a caller cannot skip, reorder, or forget it. Opt-in throughout: no effect WAL and no content hashing unless `plugin_settings.effect_log_path` / `plugin_settings.capture_content_provenance` are set. (#166)
+- **The OAuth delegator emits write-ahead audit for the tokens it mints.** `cpex-plugin-delegator-oauth` now brackets both mint legs — the workload `client_assertion` base-token mint and the RFC 8693 exchange — with `begin_effect` / `complete_effect`, mapping a successful exchange to `confirmed`, a definitive IdP rejection to `rejected`, and a timeout or unreachable IdP to `unknown`. Effects are emitted only when the operator grants the plugin `emit_effect` and configures an effect WAL; otherwise the mint path is unchanged. (#166)
+- **The reference `audit-logger` renders the new provenance.** Decision records now include the invocation `span`, a `taint` object (the labels the request arrived with vs. the labels after the pipeline — their difference is the taint this node added), and, when content provenance is enabled, a `content` object with the input and output payload hashes (`sha256:…`, digests only, never the content itself). (#166)
+- **Host-settable audit stream identity.** The stream a decision or effect record belongs to can now be named by the embedding host, so a platform's records are attributable to *its* stream rather than CPEX's bare per-type labels. `plugin_settings.audit_stream_namespace` (YAML) prefixes both streams — `gw-1` yields `gw-1:decision` / `gw-1:effect`, keeping the two per-type counters independently gap-free so each stream's completeness claim stays intact. The epoch — the executor-generation id that lets a consumer tell a restart from record loss — can be overridden programmatically (`plugin_settings.audit_epoch` in code, or `ExecutorConfig::audit_epoch` for a host embedding the executor directly) for a host that supplies its own monotonic boot/generation counter; it is deliberately *not* a YAML value, since a static file epoch could not stay monotonic across boots. Both default to CPEX's existing behavior (bare labels, wall-clock epoch). Because a reload builds a fresh executor with the stream counters reset, a reload whose epoch does not advance is warned rather than left to silently collide with the previous generation's records. (#166)
+- **A plugin that performs effects but was not granted `emit_effect` is flagged at startup.** A plugin declares that it mints or performs irreversible effects via `Plugin::emits_effects()`. If the operator omits the `emit_effect` capability grant, that plugin's write-ahead calls no-op and its mint runs unaudited — so the `PluginManager` now logs a warning at startup naming the plugin, turning an invisible audit gap into a visible one. Non-fatal: running an effect-causing plugin unaudited stays a valid operator choice. (#166)
+
+### Changed
+
+- **Serial-phase plugin panics are now contained, not fatal to the call.** A panic in a plugin's hook body is caught by the executor's serial phase (matching the concurrent phase, which already contained panics), converted to a plugin error, and run through the plugin's `on_error` policy — so with the default `on_error: fail` it becomes a fail-closed **deny**, recorded in the decision stream with the violation code `plugin_panic`, rather than aborting the invocation with no record. This extends "completeness by construction" to crashes. **Breaking (FFI / language bindings):** a host that keyed on `RC_PANIC` to detect a *plugin-body* panic now sees a normal deny instead — the panic detail is carried on the deny violation (code `plugin_panic`, distinguishable from an ordinary `plugin_error`). `RC_PANIC` still fires for panics *outside* a plugin body (config load, plugin registration, executor internals). (#166)
 
 ### Fixed
 

bindings/python/src/manager.rs

@@ -358,15 +358,16 @@ mod tests {
         manager
     }
 
-    /// The `tokio::spawn` in `invoke_hook` must catch a panicking plugin and
-    /// surface `JoinError::is_panic()` rather than aborting the process or
-    /// leaking the panic to the pyo3_async_runtimes dispatch task.
-    ///
-    /// This is the Rust-level regression test for the panic-isolation
-    /// guarantee; the Python-level guarantee is that `invoke_hook` raises
-    /// `RuntimeError` rather than `pyo3_async_runtimes.RustPanic`.
+    /// A panicking plugin in the serial phase is contained by the executor
+    /// (`catch_unwind`) and handled by `on_error` — with the default
+    /// `on_error=Fail` it becomes a fail-closed deny, exactly like the
+    /// concurrent phase. So the spawned invoke **completes** (no
+    /// `JoinError::is_panic`), the result is a deny, and the panic is recorded
+    /// in the violation with code `plugin_panic` and its message preserved. The
+    /// `tokio::spawn`/`JoinError` net in `invoke_hook` remains for panics
+    /// outside a contained plugin body.
     #[tokio::test]
-    async fn invoke_on_panicking_plugin_returns_join_error_is_panic() {
+    async fn invoke_on_panicking_plugin_is_contained_as_deny() {
         let manager = build_panicking_manager();
         manager.initialize().await.expect("initialize");
 
@@ -383,27 +384,29 @@ mod tests {
         })
         .await;
 
+        // The panic is contained, so the spawned task completes rather than
+        // unwinding.
         assert!(
-            join_result.is_err(),
-            "spawned task should have failed due to panic"
+            join_result.is_ok(),
+            "the invoke task should complete, not unwind, on a contained panic"
         );
-        let join_err = join_result.unwrap_err();
+        let (result, _bg) = join_result.unwrap();
+
+        // Default on_error=Fail turns the panic into a fail-closed deny ...
         assert!(
-            join_err.is_panic(),
-            "JoinError should report is_panic()=true, not a cancellation"
+            !result.continue_processing,
+            "a contained plugin panic denies the request"
+        );
+        // ... coded plugin_panic, with the panic message preserved.
+        let violation = result.violation.expect("a deny carries a violation");
+        assert_eq!(
+            violation.code, "plugin_panic",
+            "a contained plugin panic is coded plugin_panic"
         );
-
-        // Verify the panic message is extractable — this is the same downcast
-        // logic used in invoke_hook to build the RuntimeError message.
-        let payload = join_err.into_panic();
-        let msg = payload
-            .downcast_ref::<&str>()
-            .copied()
-            .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
-            .unwrap_or("unknown panic");
         assert!(
-            msg.contains("simulated panic"),
-            "panic message should propagate, got: {msg}"
+            violation.reason.contains("simulated panic"),
+            "panic message should propagate, got: {}",
+            violation.reason
         );
     }
 

builtins/plugins/audit-logger/src/factory.rs

@@ -25,16 +25,35 @@ impl PluginFactory for AuditLoggerFactory {
     fn create(&self, config: &PluginConfig) -> Result<PluginInstance, Box<PluginError>> {
         let logger = Arc::new(AuditLogger::new(config.clone())?);
 
+        // Make the inferred mode explicit in the startup log. Audit-only (no
+        // `hooks:`) is the recommended sink mode, so this stays at info rather
+        // than warn — but it names the mode unambiguously and points at the
+        // typo case, so an operator who *meant* to list hooks and lost them to
+        // a YAML slip can catch it in the logs rather than silently getting a
+        // sink. (An explicit config flag would remove the inference entirely —
+        // tracked for the sink-mode discussion.)
         if config.hooks.is_empty() {
-            return Err(Box::new(PluginError::Config {
-                message: format!(
-                    "plugin '{}' (cpex-plugin-audit-logger): `hooks:` must list at \
-                     least one CMF hook to audit (e.g. cmf.tool_pre_invoke)",
-                    config.name
-                ),
-            }));
+            tracing::info!(
+                plugin = %config.name,
+                "audit-logger '{}' running in audit-only sink mode (no `hooks:` listed) — \
+                 auto-attaches to the executor verdict path; if you meant to observe specific \
+                 hooks, list them under `hooks:`",
+                config.name,
+            );
+        } else {
+            tracing::info!(
+                plugin = %config.name,
+                hooks = ?config.hooks,
+                "audit-logger '{}' running as a CMF post-hook observer on {:?}",
+                config.name,
+                config.hooks,
+            );
         }
 
+        // With no `hooks:` listed the logger runs in audit-only mode — it
+        // registers no CMF post-hook handlers and instead auto-attaches as a
+        // decision-audit sink (see `Plugin::as_audit_handler`). Listing hooks
+        // keeps the legacy per-hook observation behavior.
         let handlers: Vec<_> = config
             .hooks
             .iter()