feat: added sequence numbering to events, documented auditing await behavior and audit bytes canonicalization.

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

Author: terylt

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

@@ -278,6 +278,16 @@ impl AuditLogger {
                     json!({ "input_hash": input_hash, "output_hash": output_hash }),
                 );
             }
+
+            // Stream identity + sequences. `stream_seq` is gap-free within the
+            // decision stream (a consumer proves none was dropped); the global
+            // `emission_seq` orders this record against the effect records a
+            // consumer merges into the same chain.
+            if let Some(stream_seq) = decisions.stream_seq() {
+                map.insert("stream_id".into(), json!(decisions.stream_id()));
+                map.insert("stream_seq".into(), json!(stream_seq));
+                map.insert("emission_seq".into(), json!(decisions.emission_seq()));
+            }
         }
         record
     }
@@ -299,6 +309,9 @@ impl AuditLogger {
                     "state": format!("{:?}", effect.state),
                     "caused_by": effect.plugin_name,
                     "details": effect.details,
+                    "stream_id": effect.stream_id,
+                    "stream_seq": effect.stream_seq,
+                    "emission_seq": effect.emission_seq,
                 }),
             );
         }
@@ -478,6 +491,19 @@ mod tests {
         assert!(record["content"]["output_hash"].is_null());
     }
 
+    #[test]
+    fn decision_record_includes_stream_and_sequences() {
+        let plugin = AuditLogger::new(cfg()).unwrap();
+        let mut log = DecisionLog::new();
+        log.set_stream("dec-abc".into(), 7, 42);
+        log.finalize(Verdict::Allow);
+
+        let record = plugin.build_decision_record(None, &Extensions::default(), &log);
+        assert_eq!(record["stream_id"], "dec-abc");
+        assert_eq!(record["stream_seq"], 7);
+        assert_eq!(record["emission_seq"], 42);
+    }
+
     #[test]
     fn no_content_field_without_input_hash() {
         // Provenance off (input_hash None) → no content field at all.

crates/cpex-core/src/audit.rs

@@ -31,6 +31,21 @@ pub trait AuditHandler: Send + Sync {
     /// Observe one finished pipeline invocation. Must not block or mutate
     /// anything the pipeline depends on — its return is `()` by design.
     ///
+    /// **Awaited at the verdict return point — a stable contract, not
+    /// fire-and-forget.** The executor `await`s this call *before* it returns
+    /// the pipeline result. That is deliberate: a crash cannot lose a verdict
+    /// that was emitted, so downstream evidence chains need no drop-detection
+    /// for the steady state. Consumers rely on this — a future change to
+    /// fire-and-forget would be a silent semantics break, so it must not be
+    /// made lightly.
+    ///
+    /// The cost of that guarantee is that **sink latency sits on the request
+    /// path** (bounded per sink by the plugin timeout with panic containment,
+    /// and sinks run sequentially). Keep `handle` cheap — serialize / hash /
+    /// append. A slower sink (a network destination, say) should hand off to
+    /// an internal queue on its own side of this boundary rather than block
+    /// here.
+    ///
     /// * `payload` — the message as it stood at the verdict.
     /// * `extensions` — the final extensions (identity, delegation, labels…).
     /// * `decisions` — what each plugin did and how the pipeline ruled.

crates/cpex-core/src/decision.rs

@@ -122,6 +122,9 @@ pub struct DecisionLog {
     span: Option<Span>,
     input_labels: Vec<String>,
     input_hash: Option<String>,
+    stream_id: Option<String>,
+    stream_seq: Option<u64>,
+    emission_seq: Option<u64>,
 }
 
 impl DecisionLog {
@@ -187,6 +190,34 @@ impl DecisionLog {
         self.input_hash.as_deref()
     }
 
+    /// Stamp the stream identity and sequence numbers, assigned by the executor
+    /// at emission. `stream_id` scopes `stream_seq` — a gap-free counter within
+    /// the *decision* stream, so a consumer of decisions alone can prove none
+    /// was dropped. `emission_seq` is the *global* counter across decisions and
+    /// effects alike, so a consumer that merges both streams can reconstruct
+    /// their interleaved order.
+    pub fn set_stream(&mut self, stream_id: String, stream_seq: u64, emission_seq: u64) {
+        self.stream_id = Some(stream_id);
+        self.stream_seq = Some(stream_seq);
+        self.emission_seq = Some(emission_seq);
+    }
+
+    /// The decision stream this record belongs to (scopes `stream_seq`).
+    pub fn stream_id(&self) -> Option<&str> {
+        self.stream_id.as_deref()
+    }
+
+    /// Monotonic, gap-free sequence within the decision stream — completeness.
+    pub fn stream_seq(&self) -> Option<u64> {
+        self.stream_seq
+    }
+
+    /// Global monotonic sequence across decisions and effects — interleaved
+    /// order.
+    pub fn emission_seq(&self) -> Option<u64> {
+        self.emission_seq
+    }
+
     /// The ordered steps taken this invocation.
     pub fn steps(&self) -> &[DecisionStep] {
         &self.steps
@@ -304,4 +335,16 @@ mod tests {
         log.set_input_hash(Some("sha256:abc".into()));
         assert_eq!(log.input_hash(), Some("sha256:abc"));
     }
+
+    #[test]
+    fn stream_and_sequences_stamp_and_read_back() {
+        let mut log = DecisionLog::new();
+        assert!(log.stream_id().is_none());
+        assert!(log.stream_seq().is_none());
+        assert!(log.emission_seq().is_none());
+        log.set_stream("dec-abc".into(), 7, 42);
+        assert_eq!(log.stream_id(), Some("dec-abc"));
+        assert_eq!(log.stream_seq(), Some(7));
+        assert_eq!(log.emission_seq(), Some(42));
+    }
 }

crates/cpex-core/src/effect.rs

@@ -63,6 +63,19 @@ pub struct EffectRecord {
     pub details: HashMap<String, serde_json::Value>,
     /// Which plugin caused the effect. Set by the framework, not self-reported.
     pub plugin_name: Option<String>,
+    /// The effect stream this record belongs to, stamped by the framework at
+    /// emission (scopes `stream_seq`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stream_id: Option<String>,
+    /// Monotonic, gap-free sequence within the effect stream — a consumer of
+    /// effects alone can prove none was dropped. Stamped at emission.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stream_seq: Option<u64>,
+    /// Global monotonic sequence across decisions and effects — lets a consumer
+    /// that merges both streams reconstruct their interleaved order. Stamped at
+    /// emission.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub emission_seq: Option<u64>,
 }
 
 impl EffectRecord {
@@ -80,6 +93,9 @@ impl EffectRecord {
             state: EffectState::Prepared,
             details: HashMap::new(),
             plugin_name: None,
+            stream_id: None,
+            stream_seq: None,
+            emission_seq: None,
         }
     }
 

crates/cpex-core/src/executor.rs

@@ -26,6 +26,7 @@
 
 use std::any::Any;
 use std::fmt;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -309,6 +310,17 @@ pub struct Executor {
     /// `begin_effect` is not crash-safe or fail-closed. Installed from
     /// `plugin_settings.effect_log_path` or programmatically. Opt-in.
     effect_log: Option<Arc<dyn DurableEffectLog>>,
+
+    /// Audit stream identity + counters, fresh per executor lifetime (a new
+    /// identity on config reload). Each emitted record carries its per-stream
+    /// counter (`decision_seq` / `effect_seq`, gap-free → completeness) and the
+    /// shared `emission_seq` (global across both → interleaved order). `Arc`
+    /// so copy-on-write snapshot mutations stay on the same stream.
+    decision_stream_id: Arc<str>,
+    decision_seq: Arc<AtomicU64>,
+    effect_stream_id: Arc<str>,
+    effect_seq: Arc<AtomicU64>,
+    emission_seq: Arc<AtomicU64>,
 }
 
 impl Executor {
@@ -318,6 +330,11 @@ impl Executor {
             config,
             audit_handlers: Vec::new(),
             effect_log: None,
+            decision_stream_id: Arc::from(format!("dec-{}", uuid::Uuid::new_v4().simple())),
+            decision_seq: Arc::new(AtomicU64::new(0)),
+            effect_stream_id: Arc::from(format!("eff-{}", uuid::Uuid::new_v4().simple())),
+            effect_seq: Arc::new(AtomicU64::new(0)),
+            emission_seq: Arc::new(AtomicU64::new(0)),
         }
     }
 
@@ -356,6 +373,22 @@ impl Executor {
 
     /// Invoke every audit sink with the finalized decision, once per pipeline
     /// run. Observation-only — the executor ignores whatever they return.
+    /// Assign this decision's stream identity + sequence numbers. The executor
+    /// writes its **own** record here — a step distinct from the read-only
+    /// handoff in [`Self::emit_audit`] (which takes `&DecisionLog`), so a sink
+    /// never receives anything mutable. `decision_seq` is gap-free within the
+    /// decision stream (completeness); `emission_seq` is the shared global
+    /// counter across decisions and effects (interleaved order). Stamped even
+    /// with no sinks — it's a property of the stream and rides on
+    /// `PipelineResult.decision_log`.
+    fn stamp_decision_stream(&self, decisions: &mut DecisionLog) {
+        decisions.set_stream(
+            self.decision_stream_id.to_string(),
+            self.decision_seq.fetch_add(1, Ordering::Relaxed),
+            self.emission_seq.fetch_add(1, Ordering::Relaxed),
+        );
+    }
+
     async fn emit_audit(
         &self,
         payload: &dyn PluginPayload,
@@ -494,6 +527,7 @@ impl Executor {
             .await
         {
             decisions.finalize(Verdict::Deny(v.clone()));
+            self.stamp_decision_stream(&mut decisions);
             self.emit_audit(&*current_payload, &current_extensions, &decisions)
                 .await;
             return (
@@ -542,6 +576,7 @@ impl Executor {
             .await
         {
             decisions.finalize(Verdict::Deny(violation.clone()));
+            self.stamp_decision_stream(&mut decisions);
             self.emit_audit(&*current_payload, &current_extensions, &decisions)
                 .await;
             return (
@@ -564,6 +599,7 @@ impl Executor {
         );
 
         decisions.finalize(Verdict::Allow);
+        self.stamp_decision_stream(&mut decisions);
         self.emit_audit(&*current_payload, &current_extensions, &decisions)
             .await;
         (
@@ -656,6 +692,9 @@ impl Executor {
                     // The configured WAL (opt-in). `None` → ordering-only, not
                     // fail-closed; `Some` → durable-before-fanout, fail-closed.
                     durable: self.effect_log.clone(),
+                    stream_id: self.effect_stream_id.clone(),
+                    stream_seq: self.effect_seq.clone(),
+                    emission_seq: self.emission_seq.clone(),
                 }));
             }
 
@@ -1305,6 +1344,12 @@ struct AuditEffectEmitter {
     /// before fanning out and fails closed if that write fails. `None` until
     /// slice 3b wires a real WAL — then emit is ordering-only.
     durable: Option<Arc<dyn DurableEffectLog>>,
+    /// Effect stream identity + counters (shared with the executor). Each
+    /// emitted record is stamped with `stream_seq` (gap-free within the effect
+    /// stream) and the global `emission_seq` (interleaved order vs decisions).
+    stream_id: Arc<str>,
+    stream_seq: Arc<AtomicU64>,
+    emission_seq: Arc<AtomicU64>,
 }
 
 impl std::fmt::Debug for AuditEffectEmitter {
@@ -1322,9 +1367,15 @@ impl EffectEmitter for AuditEffectEmitter {
         use futures::FutureExt;
         use std::panic::AssertUnwindSafe;
 
-        // Stamp the causing plugin — set by the framework, not self-reported.
+        // Stamp the causing plugin + stream identity/sequences — all set by the
+        // framework, not self-reported. `stream_seq` is gap-free within the
+        // effect stream (completeness); `emission_seq` is the shared global
+        // counter across decisions and effects (interleaved order).
         let mut stamped = effect.clone();
         stamped.plugin_name = Some(self.plugin_name.clone());
+        stamped.stream_id = Some(self.stream_id.to_string());
+        stamped.stream_seq = Some(self.stream_seq.fetch_add(1, Ordering::Relaxed));
+        stamped.emission_seq = Some(self.emission_seq.fetch_add(1, Ordering::Relaxed));
 
         // Write-ahead: durably record BEFORE any observer sees it. Fail
         // closed — if the durable write fails, return Err and do NOT fan out;
@@ -1546,6 +1597,9 @@ mod tests {
             plugin_name: "delegator".into(),
             timeout: Duration::from_secs(5),
             durable: Some(Arc::new(FailingLog)),
+            stream_id: Arc::from("eff-test"),
+            stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
+            emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         };
         let res = emitter.emit(&effect, &Extensions::default()).await;
         assert!(res.is_err(), "durable write failed → emit fails closed");
@@ -1562,6 +1616,9 @@ mod tests {
             plugin_name: "delegator".into(),
             timeout: Duration::from_secs(5),
             durable: Some(Arc::new(OkLog)),
+            stream_id: Arc::from("eff-test"),
+            stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
+            emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         };
         let res2 = emitter2.emit(&effect, &Extensions::default()).await;
         assert!(res2.is_ok());

crates/cpex-core/src/hooks/payload.rs

@@ -91,11 +91,22 @@ pub trait PluginPayload: Send + Sync + 'static {
     /// or `None` for payloads that can't or shouldn't be serialized (the
     /// default). The bytes feed a content hash — **only the digest is
     /// retained, never the bytes** — so a node's provenance is recorded
-    /// without re-spilling its (possibly sensitive) content. Must be
-    /// *canonical* (stable across processes) for the hashes to compare;
-    /// `impl_plugin_payload!(_, audit_serialize)` derives that via sorted-key
-    /// JSON. Computed only when content provenance is enabled, so the default
-    /// keeps the hot path free.
+    /// without re-spilling its (possibly sensitive) content. Computed only
+    /// when content provenance is enabled, so the default keeps the hot path
+    /// free.
+    ///
+    /// **Byte-stability (what a consumer may assume).**
+    /// `impl_plugin_payload!(_, audit_serialize)` derives this by round-tripping
+    /// through `serde_json::Value` — whose `Map` is a `BTreeMap`, so object keys
+    /// are sorted. Identical content therefore serializes to identical bytes
+    /// across runs and processes, and **two equal digests mean "same content"
+    /// within a deployment**. It is *sorted-key JSON, not full RFC 8785 (JCS)*:
+    /// number formatting follows `serde_json` and is stable within a
+    /// `serde_json` version but is not guaranteed by a canonicalization spec
+    /// across toolchains. So treat digest equality as same-content within a
+    /// build; do not assume cross-toolchain canonicalization. A hand-written
+    /// `audit_bytes` must preserve this property (a canonical, deterministic
+    /// encoding) or its hashes will not be comparable.
     fn audit_bytes(&self) -> Option<Vec<u8>> {
         None
     }