fixes: added epoch and updated documentation around sequencing.

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

Author: terylt

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

@@ -284,6 +284,7 @@ impl AuditLogger {
             // `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("epoch".into(), json!(decisions.epoch()));
                 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()));
@@ -309,6 +310,7 @@ impl AuditLogger {
                     "state": format!("{:?}", effect.state),
                     "caused_by": effect.plugin_name,
                     "details": effect.details,
+                    "epoch": effect.epoch,
                     "stream_id": effect.stream_id,
                     "stream_seq": effect.stream_seq,
                     "emission_seq": effect.emission_seq,
@@ -495,11 +497,12 @@ mod tests {
     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.set_stream(1_700_000_000, "decision".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["epoch"], 1_700_000_000u64);
+        assert_eq!(record["stream_id"], "decision");
         assert_eq!(record["stream_seq"], 7);
         assert_eq!(record["emission_seq"], 42);
     }

crates/cpex-core/src/decision.rs

@@ -122,6 +122,7 @@ pub struct DecisionLog {
     span: Option<Span>,
     input_labels: Vec<String>,
     input_hash: Option<String>,
+    epoch: Option<u64>,
     stream_id: Option<String>,
     stream_seq: Option<u64>,
     emission_seq: Option<u64>,
@@ -190,30 +191,60 @@ 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) {
+    /// Stamp the audit-stream identity + sequence numbers, assigned by the
+    /// executor at emission. The two counters are **distinct claims** — don't
+    /// use one for the other's job:
+    ///
+    /// - `epoch` — the executor's boot time (Unix nanoseconds), captured once
+    ///   at startup. It scopes the counters so a verifier tells a *counter
+    ///   reset* (new, larger epoch) from *records lost* (a gap within an
+    ///   epoch); being ordered, `(epoch, emission_seq)` is a total order across
+    ///   restarts, computable from the record alone. Cross-epoch tail-loss is
+    ///   not provable from the counters alone — that is what a durable sink
+    ///   (the ledger) is for.
+    /// - `stream_id` — the per-type stream (`"decision"`), the entry-type a
+    ///   merged consumer keys on.
+    /// - `stream_seq` — a **completeness** claim. Dense (gap-free) within
+    ///   `(epoch, stream_id)`; a gap means a record was dropped.
+    /// - `emission_seq` — an **ordering** claim *only*. Monotonic across all
+    ///   streams within the epoch (decisions and effects share it) for
+    ///   reconstructing interleaved order. A single-stream consumer sees it
+    ///   *sparse* by design — the gaps are the other stream's records, never a
+    ///   loss signal.
+    pub fn set_stream(
+        &mut self,
+        epoch: u64,
+        stream_id: String,
+        stream_seq: u64,
+        emission_seq: u64,
+    ) {
+        self.epoch = Some(epoch);
         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`).
+    /// The executor boot epoch (Unix nanoseconds) this record was emitted in.
+    /// Orderable, so a new/larger value marks a restart — a reset is
+    /// distinguishable from a loss, and it extends `emission_seq` to a total
+    /// order across restarts.
+    pub fn epoch(&self) -> Option<u64> {
+        self.epoch
+    }
+
+    /// The per-type 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.
+    /// **Completeness** counter — dense within `(epoch, stream_id)`; a gap is a
+    /// dropped record.
     pub fn stream_seq(&self) -> Option<u64> {
         self.stream_seq
     }
 
-    /// Global monotonic sequence across decisions and effects — interleaved
-    /// order.
+    /// **Ordering** counter — monotonic across decisions and effects within the
+    /// epoch. Sparse for a single-stream consumer by design; not a loss signal.
     pub fn emission_seq(&self) -> Option<u64> {
         self.emission_seq
     }
@@ -339,11 +370,13 @@ mod tests {
     #[test]
     fn stream_and_sequences_stamp_and_read_back() {
         let mut log = DecisionLog::new();
+        assert!(log.epoch().is_none());
         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"));
+        log.set_stream(1_700_000_000, "decision".into(), 7, 42);
+        assert_eq!(log.epoch(), Some(1_700_000_000));
+        assert_eq!(log.stream_id(), Some("decision"));
         assert_eq!(log.stream_seq(), Some(7));
         assert_eq!(log.emission_seq(), Some(42));
     }

crates/cpex-core/src/effect.rs

@@ -63,17 +63,23 @@ 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`).
+    /// The executor's boot time (Unix nanoseconds), scoping the sequences so a
+    /// verifier tells a counter reset (new, larger epoch) from records lost.
+    /// Ordered, so `(epoch, emission_seq)` totally orders records across
+    /// restarts. Stamped at emission.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub epoch: Option<u64>,
+    /// The per-type stream this record belongs to (`"effect"`). Stamped at
+    /// emission.
     #[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.
+    /// **Completeness** counter — dense within `(epoch, stream_id)`; a gap means
+    /// an effect record 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.
+    /// **Ordering** counter — monotonic across decisions and effects within the
+    /// epoch, for interleaved order. Sparse for an effects-only consumer by
+    /// design; not a loss signal. Stamped at emission.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub emission_seq: Option<u64>,
 }
@@ -93,6 +99,7 @@ impl EffectRecord {
             state: EffectState::Prepared,
             details: HashMap::new(),
             plugin_name: None,
+            epoch: None,
             stream_id: None,
             stream_seq: None,
             emission_seq: None,

crates/cpex-core/src/executor.rs

@@ -311,14 +311,15 @@ pub struct Executor {
     /// `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>,
+    /// Audit stream identity + counters. `epoch` is the executor's boot time
+    /// (Unix nanos), captured once — it scopes the counters so a restart is
+    /// distinguishable from a loss and orders records across restarts. Each
+    /// record carries its per-type counter (`decision_seq` / `effect_seq`,
+    /// gap-free → completeness) and the shared `emission_seq` (global across
+    /// both → interleaved order). The counters are `Arc` so copy-on-write
+    /// snapshot mutations stay on the same stream.
+    epoch: u64,
     decision_seq: Arc<AtomicU64>,
-    effect_stream_id: Arc<str>,
     effect_seq: Arc<AtomicU64>,
     emission_seq: Arc<AtomicU64>,
 }
@@ -330,9 +331,14 @@ impl Executor {
             config,
             audit_handlers: Vec::new(),
             effect_log: None,
-            decision_stream_id: Arc::from(format!("dec-{}", uuid::Uuid::new_v4().simple())),
+            // Boot time in Unix nanoseconds — an orderable epoch that needs no
+            // persistence. A new executor (restart or config reload) gets a
+            // larger value, so a verifier tells a reset from a loss.
+            epoch: std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_nanos() as u64)
+                .unwrap_or(0),
             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)),
         }
@@ -383,7 +389,8 @@ impl Executor {
     /// `PipelineResult.decision_log`.
     fn stamp_decision_stream(&self, decisions: &mut DecisionLog) {
         decisions.set_stream(
-            self.decision_stream_id.to_string(),
+            self.epoch,
+            "decision".to_string(),
             self.decision_seq.fetch_add(1, Ordering::Relaxed),
             self.emission_seq.fetch_add(1, Ordering::Relaxed),
         );
@@ -692,7 +699,7 @@ 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(),
+                    epoch: self.epoch,
                     stream_seq: self.effect_seq.clone(),
                     emission_seq: self.emission_seq.clone(),
                 }));
@@ -1344,10 +1351,10 @@ 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>,
+    /// Boot epoch + counters (shared with the executor). Each emitted record is
+    /// stamped with `epoch`, `stream_seq` (gap-free within the effect stream),
+    /// and the global `emission_seq` (interleaved order vs decisions).
+    epoch: u64,
     stream_seq: Arc<AtomicU64>,
     emission_seq: Arc<AtomicU64>,
 }
@@ -1373,7 +1380,8 @@ impl EffectEmitter for AuditEffectEmitter {
         // 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.epoch = Some(self.epoch);
+        stamped.stream_id = Some("effect".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));
 
@@ -1597,7 +1605,7 @@ mod tests {
             plugin_name: "delegator".into(),
             timeout: Duration::from_secs(5),
             durable: Some(Arc::new(FailingLog)),
-            stream_id: Arc::from("eff-test"),
+            epoch: 0,
             stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
             emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         };
@@ -1616,7 +1624,7 @@ mod tests {
             plugin_name: "delegator".into(),
             timeout: Duration::from_secs(5),
             durable: Some(Arc::new(OkLog)),
-            stream_id: Arc::from("eff-test"),
+            epoch: 0,
             stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
             emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         };

crates/cpex-core/src/manager.rs

@@ -2585,16 +2585,8 @@ plugins:
         assert_eq!(l[0].0, "effect:Prepared");
         assert_eq!(l[2].0, "decision");
         // Distinct per-type streams.
-        assert!(
-            l[0].1.starts_with("eff-"),
-            "effect stream id; got {}",
-            l[0].1
-        );
-        assert!(
-            l[2].1.starts_with("dec-"),
-            "decision stream id; got {}",
-            l[2].1
-        );
+        assert_eq!(l[0].1, "effect", "effect stream id");
+        assert_eq!(l[2].1, "decision", "decision stream id");
     }
 
     /// `on_effect` fires on the `begin_effect` (prepared) leg — a sink observes

docs/content/docs/auditing.md

@@ -135,7 +135,8 @@ line per effect:
   ],
   "span":    { "trace_id": "…", "span_id": "…", "parent_span_id": "…" },
   "taint":   { "input": ["PII"], "final": ["PII", "secret"] },
-  "content": { "input_hash": "sha256:…", "output_hash": "sha256:…" }
+  "content": { "input_hash": "sha256:…", "output_hash": "sha256:…" },
+  "epoch": 1723680000000000000, "stream_id": "decision", "stream_seq": 413, "emission_seq": 913
 }
 
 // effect
@@ -144,7 +145,8 @@ line per effect:
   "effect": {
     "kind": "token_mint", "state": "confirmed", "key": "…",
     "caused_by": "oauth-delegator",
-    "details": { "audience": "workday-api", "scope": "read_compensation" }
+    "details": { "audience": "workday-api", "scope": "read_compensation" },
+    "epoch": 1723680000000000000, "stream_id": "effect", "stream_seq": 7, "emission_seq": 912
   }
 }
 ```
@@ -153,6 +155,35 @@ Fields appear only when present: `span` always; `taint` when labels exist;
 `content` only when content provenance is enabled; `subject` when a subject is
 resolved.
 
+### Sequence numbers — completeness vs. order
+
+Four fields — `epoch`, `stream_id`, `stream_seq`, `emission_seq` — let a
+downstream store prove properties about the stream it received. Two are
+**claims** a verifier checks; two **scope** those claims. Don't use one claim
+for the other's job:
+
+- `stream_seq` is a **completeness** claim. It is dense (gap-free) within its
+  `(epoch, stream_id)`. **A gap means a record was dropped** — a consumer of one
+  stream can prove nothing was silently lost.
+- `emission_seq` is an **ordering** claim only. It is monotonic across *both*
+  streams within an epoch, so a consumer that merges decisions and effects can
+  reconstruct their interleave (an effect emits during a request, so it carries
+  a lower `emission_seq` than the decision that closed the request). **A
+  single-stream consumer sees it sparse by design — the gaps are the other
+  stream's records, not a loss.** Do not detect loss from `emission_seq`.
+- `stream_id` scopes `stream_seq` — it names the per-type stream, `"decision"`
+  or `"effect"` (the entry-type a merged consumer keys on). Decisions and
+  effects each get their own dense counter, so a consumer of just one still has
+  gap-free completeness.
+- `epoch` scopes both counters. It is the executor's boot time (Unix
+  nanoseconds), so a *new, larger* value marks a restart: `stream_seq` proves
+  completeness within an epoch, and across a restart the epoch changes, so a
+  verifier tells a **counter reset from records lost** — and `(epoch,
+  emission_seq)` is a total order across restarts. Detecting loss of the *tail*
+  of a previous epoch (a crash between emit and persist) is not possible from
+  the counters alone — that is what a durable sink (an append-only ledger) is
+  for.
+
 ### Destinations
 
 | `destination` | Behaviour |