fix: harden audit seam per review — startup recovery, panic containment, config-gated effects, one audit record per invocation

Signed-off-by: Teryl Taylor <terylt@ibm.com>

Author: terylt

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,6 +25,31 @@ 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() {
+            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

builtins/plugins/delegator-oauth/src/delegator.rs

@@ -333,8 +333,17 @@ impl OAuthDelegator {
             },
         };
         // Best-effort completion: the act already happened, so a completion
-        // write failure is not fatal (recovery reconciles by the intent's key).
-        let _ = ext.complete_effect(&intent, state).await;
+        // write failure is not fatal (recovery reconciles by the intent's
+        // key). It must not be silent, though — mirror the core primitive and
+        // log, so a persistently failing WAL is visible rather than hidden.
+        if let Err(e) = ext.complete_effect(&intent, state).await {
+            tracing::warn!(
+                effect_key = %intent.key,
+                error = %e,
+                "failed to write token-mint effect completion; \
+                 recovery will reconcile by key"
+            );
+        }
 
         outcome
     }

crates/cpex-core/src/config.rs

@@ -738,6 +738,22 @@ pub(crate) fn validate_config(config: &CpexConfig) -> Result<(), Box<PluginError
                 message: format!("duplicate plugin name: '{}'", plugin.name),
             }));
         }
+
+        // `emit_effect` is only honored in modes whose phase wires a live
+        // effect emitter (Sequential / Transform). In any other mode the
+        // capability would silently no-op at runtime — the mint runs with no
+        // write-ahead record and no fail-closed guarantee — so reject the
+        // combination here rather than let it fail open.
+        if plugin.capabilities.contains("emit_effect") && !plugin.mode.grants_effect_emitter() {
+            return Err(Box::new(PluginError::Config {
+                message: format!(
+                    "plugin '{}' declares the 'emit_effect' capability with mode '{}', \
+                     which cannot emit effects (only sequential/transform can); \
+                     effect calls would silently no-op",
+                    plugin.name, plugin.mode
+                ),
+            }));
+        }
     }
 
     if config.routing_enabled() {
@@ -1189,6 +1205,39 @@ plugins:
             .contains("duplicate plugin name"));
     }
 
+    #[test]
+    fn emit_effect_rejected_in_non_emitting_modes() {
+        // Concurrent / audit / fire_and_forget don't wire an effect emitter,
+        // so `emit_effect` there would silently no-op — reject at config.
+        for mode in ["concurrent", "audit", "fire_and_forget"] {
+            let yaml = format!(
+                "plugins:\n  - name: minter\n    kind: builtin\n    mode: {mode}\n    \
+                 hooks: [tool_pre_invoke]\n    capabilities: [emit_effect]\n"
+            );
+            let err = parse_config(&yaml).unwrap_err().to_string().to_lowercase();
+            assert!(
+                err.contains("emit_effect") && err.contains(mode),
+                "mode {mode} should be rejected; got: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn emit_effect_allowed_in_emitting_modes() {
+        // Sequential / transform run through the serial phase, which grants
+        // the emitter — so `emit_effect` is honored and must pass validation.
+        for mode in ["sequential", "transform"] {
+            let yaml = format!(
+                "plugins:\n  - name: minter\n    kind: builtin\n    mode: {mode}\n    \
+                 hooks: [tool_pre_invoke]\n    capabilities: [emit_effect]\n"
+            );
+            assert!(
+                parse_config(&yaml).is_ok(),
+                "mode {mode} should be allowed to emit effects"
+            );
+        }
+    }
+
     #[test]
     fn parses_effect_log_settings() {
         let yaml = r#"

crates/cpex-core/src/decision.rs

@@ -34,6 +34,16 @@ pub enum PluginAction {
     ModifiedPayload,
     /// Wrote to an extension slot it was capable of writing.
     ModifiedExtensions,
+    /// Signalled a block from a non-blocking phase (Transform), so the deny
+    /// was suppressed. Recorded as its own action — never as `Allowed` — so
+    /// the record reflects the plugin's actual decision, not the discarded
+    /// intent. A downstream mapping (e.g. an OCSF `ai_operation` disposition)
+    /// must not read this as an allow.
+    DenyIgnored,
+    /// Cancelled mid-flight because another concurrent branch short-circuited
+    /// the phase. An intentional abort, not a failure — distinct from
+    /// [`PluginAction::Error`] so it doesn't read as a crash.
+    Aborted,
     /// Failed. The string is the error rendered by the executor; whether
     /// this halts the pipeline is decided by the plugin's `on_error`.
     Error(String),

crates/cpex-core/src/effect.rs

@@ -56,6 +56,10 @@ pub struct EffectRecord {
     pub description: String,
     /// Idempotency / reconciliation key threaded into the external call, so
     /// an `unknown` outcome can be resolved against the participant later.
+    /// Must be **unique per attempt** — recovery resolves keys across the
+    /// whole WAL, so a key reused across retries would let an earlier
+    /// attempt's terminal record mask a later attempt's orphan (see
+    /// [`FileEffectLog::recover`]).
     pub key: String,
     /// Where in its lifecycle this record is.
     pub state: EffectState,
@@ -410,6 +414,15 @@ impl FileEffectLog {
 
             // A key is resolved iff some record for it reached a terminal
             // state. Everything else (prepared-only, unknown) is unresolved.
+            //
+            // INVARIANT: a `key` is a unique per-attempt id, not a reused
+            // idempotency key. Resolution matches across the whole file, so a
+            // plugin that reused one stable key across retries would let a
+            // terminal record from an earlier attempt mask a later attempt's
+            // orphaned `prepared` as resolved. The OAuth delegator satisfies
+            // this with a fresh UUID per mint; a future plugin that wants
+            // stable idempotency keys must add per-attempt scoping here (e.g.
+            // an attempt counter alongside the key) before relying on recovery.
             let resolved: std::collections::HashSet<&str> = records
                 .iter()
                 .filter(|r| matches!(r.state, EffectState::Confirmed | EffectState::Rejected))

crates/cpex-core/src/executor.rs

@@ -396,6 +396,45 @@ impl Executor {
         );
     }
 
+    /// Emit a single allow decision record for an invocation that resolved to
+    /// zero plugins, keeping the audit stream dense at one record per
+    /// invocation. **Cheap no-op when no audit sink is attached** — an
+    /// unaudited host pays only a length check, building no record and
+    /// consuming no sequence number. The manager calls this at its zero-plugin
+    /// short-circuits (which return before reaching `execute`), and `execute`
+    /// calls it for a direct empty invocation. Captures the same span /
+    /// input-label / input-hash provenance a normal run records at entry, then
+    /// stamps and emits.
+    pub(crate) async fn emit_empty_allow(
+        &self,
+        payload: &dyn PluginPayload,
+        extensions: &Extensions,
+    ) {
+        if self.audit_handlers.is_empty() {
+            return;
+        }
+        let mut decisions = DecisionLog::new();
+        let request = extensions.request.as_ref();
+        decisions.set_span(crate::decision::Span::for_request(
+            request.and_then(|r| r.trace_id.as_deref()),
+            request.and_then(|r| r.span_id.as_deref()),
+        ));
+        if let Some(sec) = extensions.security.as_ref() {
+            let mut labels: Vec<String> = sec.labels.iter().cloned().collect();
+            labels.sort_unstable();
+            decisions.set_input_labels(labels);
+        }
+        if self.config.capture_content_provenance {
+            let hash = payload
+                .audit_bytes()
+                .map(|b| crate::hooks::payload::content_hash(&b));
+            decisions.set_input_hash(hash);
+        }
+        decisions.finalize(Verdict::Allow);
+        self.stamp_decision_stream(&mut decisions);
+        self.emit_audit(payload, extensions, &decisions).await;
+    }
+
     async fn emit_audit(
         &self,
         payload: &dyn PluginPayload,
@@ -465,7 +504,15 @@ impl Executor {
     ) -> (PipelineResult, BackgroundTasks) {
         let mut ctx_table = context_table.unwrap_or_default();
 
+        // A hook that resolves to zero plugins is a normal case (nothing is
+        // configured for this entity). It still emits exactly one allow record
+        // so the audit stream stays dense at one record per invocation — but
+        // `emit_empty_allow` is a no-op when no sink is attached, so an
+        // unaudited host pays nothing. (The manager short-circuits most
+        // zero-plugin invocations before reaching here and calls
+        // `emit_empty_allow` itself; this covers a direct `execute(&[], …)`.)
         if entries.is_empty() {
+            self.emit_empty_allow(&*payload, &extensions).await;
             return (
                 PipelineResult::allowed_with(payload, extensions, ctx_table),
                 BackgroundTasks::empty(),
@@ -705,13 +752,40 @@ impl Executor {
                 }));
             }
 
-            // Execute with timeout — handler borrows payload, gets filtered extensions
+            // Execute with timeout — handler borrows payload, gets filtered
+            // extensions. Contain a panic the same way the concurrent phase
+            // does (`catch_unwind`): a panic between `begin_effect` and
+            // `complete_effect` would otherwise unwind the whole request
+            // future. Collapsing it into a `PluginError` lets `on_error`
+            // decide and keeps the pipeline's bookkeeping intact; the orphaned
+            // WAL entry is left for recovery to reconcile as `unknown` rather
+            // than crashing the request.
+            use futures::FutureExt;
             let timeout_dur = Duration::from_secs(self.config.timeout_seconds);
             let result = timeout(
                 timeout_dur,
-                entry.handler.invoke(&**payload, &filtered, &mut ctx),
+                std::panic::AssertUnwindSafe(entry.handler.invoke(&**payload, &filtered, &mut ctx))
+                    .catch_unwind(),
             )
-            .await;
+            .await
+            .map(|caught| {
+                caught.unwrap_or_else(|panic| {
+                    let msg = panic
+                        .downcast_ref::<&'static str>()
+                        .map(|s| s.to_string())
+                        .or_else(|| panic.downcast_ref::<String>().cloned())
+                        .unwrap_or_else(|| "unknown panic".to_string());
+                    error!("{} plugin '{}' panicked: {}", phase_label, plugin_name, msg);
+                    Err(Box::new(crate::error::PluginError::Execution {
+                        plugin_name: plugin_name.to_string(),
+                        message: format!("task panicked: {msg}"),
+                        source: None,
+                        code: Some("panic".into()),
+                        details: std::collections::HashMap::new(),
+                        proto_error_code: None,
+                    }))
+                })
+            });
 
             match result {
                 Ok(Ok(result_box)) => {
@@ -724,8 +798,21 @@ impl Executor {
                             }
                         }
 
+                        // A block signalled from a non-blocking phase
+                        // (Transform): suppressed by the phase contract
+                        // (can_modify, not can_block), but recorded as the
+                        // plugin's actual intent — never a plain allow.
+                        // Enforcement is unchanged (the pipeline proceeds);
+                        // this plugin's modifications are skipped, since it
+                        // asked to stop rather than shape.
+                        let deny_ignored =
+                            !erased.continue_processing && !can_block && erased.violation.is_some();
+                        if deny_ignored {
+                            action = PluginAction::DenyIgnored;
+                        }
+
                         // Accept modifications
-                        if can_modify {
+                        if can_modify && !deny_ignored {
                             if let Some(mp) = erased.modified_payload {
                                 *payload = mp;
                                 action = PluginAction::ModifiedPayload;
@@ -832,12 +919,26 @@ impl Executor {
                     // If extract failed or no modifications — payload unchanged
                 },
                 Ok(Err(e)) => {
+                    // A contained panic (from the `catch_unwind` above) carries
+                    // code "panic". Surface it with the same "plugin_panic"
+                    // violation code the concurrent phase uses, so a host or
+                    // sink can distinguish a panic from an ordinary plugin error
+                    // by code, regardless of which phase it happened in.
+                    let is_panic = matches!(
+                        e.as_ref(),
+                        crate::error::PluginError::Execution { code: Some(c), .. }
+                            if c.as_str() == "panic"
+                    );
                     error!("{} plugin '{}' failed: {}", phase_label, plugin_name, e);
                     action = PluginAction::Error(e.to_string());
                     match on_error {
                         OnError::Fail if can_block => {
                             let mut v = crate::error::PluginViolation::new(
-                                "plugin_error",
+                                if is_panic {
+                                    "plugin_panic"
+                                } else {
+                                    "plugin_error"
+                                },
                                 format!("Plugin '{}' failed: {}", plugin_name, e),
                             );
                             v.plugin_name = Some(plugin_name.to_string());
@@ -1141,8 +1242,9 @@ impl Executor {
                 },
                 BranchOutcome::TimedOut => PluginAction::Error("timed out".to_string()),
                 BranchOutcome::Panicked(s) => PluginAction::Error(format!("panicked: {s}")),
-                // Cancelled because another branch short-circuited the phase.
-                BranchOutcome::Aborted => PluginAction::Error("aborted".to_string()),
+                // Cancelled because another branch short-circuited the phase —
+                // an intentional abort, recorded as such rather than an error.
+                BranchOutcome::Aborted => PluginAction::Aborted,
             };
             decisions.record(plugin_name, entry.plugin_ref.trusted_config().mode, action);