feat(marmot-app): record an epoch_stall_backfill_armed forensic audit row at the epoch-stall arm decision

crates/marmot-app/src/client/audit.rs

@@ -189,6 +189,21 @@ impl AppClient {
         );
     }
 
+    /// Record an `epoch_stall_backfill_armed` forensic audit row at the arm
+    /// decision: the epoch the group was stalled at (`stalled_epoch`) and the
+    /// distinct-undecryptable threshold that armed the backfill (`threshold`).
+    /// Group-scoped, so `group_ref` carries the stalled group id.
+    pub(crate) fn record_epoch_stall_backfill_armed(&self, group_id: &GroupId, stalled_epoch: u64) {
+        self.runtime.session().record_audit_event(
+            Some(group_id),
+            None,
+            AuditEventKind::EpochStallBackfillArmed {
+                stalled_epoch,
+                threshold: self.epoch_stall.threshold() as u64,
+            },
+        );
+    }
+
     /// Apply the audit-logging switch to this live session by swapping the
     /// recorder in place: a file-backed recorder when `enabled`, or a no-op
     /// recorder when off. Dropping the prior recorder flushes and closes any

crates/marmot-app/src/client/epoch_stall.rs

@@ -64,6 +64,14 @@ impl EpochStallDetector {
         }
     }
 
+    /// The distinct-undecryptable count at which this detector arms a backfill.
+    /// Reported on the `epoch_stall_backfill_armed` audit row so the row is
+    /// honest even when the detector was built with a non-default threshold
+    /// (unit tests, or a future configurable value).
+    pub(crate) fn threshold(&self) -> usize {
+        self.threshold
+    }
+
     /// Record that an account-wide full-history replay was just triggered. One
     /// replay re-fetches every group's history, so suppress a further backfill
     /// for every currently-tracked group at its current epoch: N groups stuck at

crates/marmot-app/src/client/sync.rs

@@ -337,10 +337,16 @@ impl AppClient {
         let Ok(record) = self.runtime.group_record(&group_id) else {
             return;
         };
-        if self
-            .epoch_stall
-            .observe_undecryptable(group_id, message_id_hex.to_owned(), record.epoch)
-        {
+        if self.epoch_stall.observe_undecryptable(
+            group_id.clone(),
+            message_id_hex.to_owned(),
+            record.epoch,
+        ) {
+            // Record the arm decision before the replay side effect runs (the
+            // worker seam calls run_pending_epoch_backfill after this returns).
+            // Best-effort, fire-and-forget: recording can never block or fail
+            // the backfill.
+            self.record_epoch_stall_backfill_armed(&group_id, record.epoch.0);
             self.epoch_backfill_pending = true;
         }
     }