feat(events): add chain_reorg and safe_target chain events

crates/blockchain/src/events.rs

@@ -22,16 +22,20 @@ use tracing::warn;
 ///
 /// These are the names consumers address events by (the SSE `event:` line and,
 /// later, `?topics=` filtering), kept separate from [`ChainEvent`] so the
-/// payload stays flat with the topic travelling out-of-band. The names match
-/// the Ethereum beacon-API eventstream topics (`head`, `block`,
-/// `finalized_checkpoint`); `justified_checkpoint` is an ethlambda extension
-/// with no beacon analog.
+/// payload stays flat with the topic travelling out-of-band. Names match the
+/// Ethereum beacon-API eventstream topics where an analog exists (`head`,
+/// `block`, `finalized_checkpoint`, `chain_reorg`); `justified_checkpoint` and
+/// `safe_target` are ethlambda extensions with no direct beacon topic.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum Topic {
     Head,
     Block,
     JustifiedCheckpoint,
     FinalizedCheckpoint,
+    /// Fork choice switched to a head that is not a descendant of the old one.
+    ChainReorg,
+    /// The interval-3 fork-choice safe attestation target advanced.
+    SafeTarget,
 }
 
 impl Topic {
@@ -41,6 +45,8 @@ impl Topic {
             Topic::Block => "block",
             Topic::JustifiedCheckpoint => "justified_checkpoint",
             Topic::FinalizedCheckpoint => "finalized_checkpoint",
+            Topic::ChainReorg => "chain_reorg",
+            Topic::SafeTarget => "safe_target",
         }
     }
 }
@@ -60,6 +66,8 @@ impl FromStr for Topic {
             "block" => Ok(Topic::Block),
             "justified_checkpoint" => Ok(Topic::JustifiedCheckpoint),
             "finalized_checkpoint" => Ok(Topic::FinalizedCheckpoint),
+            "chain_reorg" => Ok(Topic::ChainReorg),
+            "safe_target" => Ok(Topic::SafeTarget),
             other => Err(UnknownTopic(other.to_string())),
         }
     }
@@ -90,6 +98,19 @@ pub enum ChainEvent {
     JustifiedCheckpoint { slot: u64, block: H256, state: H256 },
     /// The finalized checkpoint advanced.
     FinalizedCheckpoint { slot: u64, block: H256, state: H256 },
+    /// Fork choice switched to a head off the old head's chain. `slot` is the
+    /// new head's slot; `depth` the number of blocks rolled back. Mirrors the
+    /// beacon `chain_reorg` payload minus `epoch`/`execution_optimistic`.
+    ChainReorg {
+        slot: u64,
+        depth: u64,
+        old_head_block: H256,
+        old_head_state: H256,
+        new_head_block: H256,
+        new_head_state: H256,
+    },
+    /// The interval-3 safe attestation target advanced (ethlambda-specific).
+    SafeTarget { slot: u64, block: H256 },
 }
 
 impl ChainEvent {
@@ -99,6 +120,8 @@ impl ChainEvent {
             ChainEvent::Block { .. } => Topic::Block,
             ChainEvent::JustifiedCheckpoint { .. } => Topic::JustifiedCheckpoint,
             ChainEvent::FinalizedCheckpoint { .. } => Topic::FinalizedCheckpoint,
+            ChainEvent::ChainReorg { .. } => Topic::ChainReorg,
+            ChainEvent::SafeTarget { .. } => Topic::SafeTarget,
         }
     }
 }
@@ -182,6 +205,7 @@ pub(crate) struct ChainEventSnapshot {
     head: H256,
     justified: Checkpoint,
     finalized: Checkpoint,
+    safe_target: H256,
 }
 
 impl ChainEventSnapshot {
@@ -194,16 +218,20 @@ impl ChainEventSnapshot {
             finalized: store
                 .latest_finalized()
                 .expect("latest finalized checkpoint exists"),
+            safe_target: store.safe_target().expect("safe target exists"),
         }
     }
 
     /// Emit one event per value that changed since the snapshot, in a fixed
-    /// order: `head` → `justified_checkpoint` → `finalized_checkpoint`.
-    /// (`block` is emitted separately by the import path, ahead of this diff.)
+    /// order: `chain_reorg` → `head` → `justified_checkpoint` →
+    /// `finalized_checkpoint` → `safe_target`. (`block` is emitted separately by
+    /// the import path, ahead of this diff.)
     ///
-    /// `wall_clock_slot` is the caller's current slot, used only to gate the
-    /// `head` event against [`HEAD_EVENT_RECENCY_SLOTS`]; the other events are
-    /// ungated.
+    /// `wall_clock_slot` is the caller's current slot, used to gate the `head`
+    /// and `chain_reorg` events against [`HEAD_EVENT_RECENCY_SLOTS`] (catch-up
+    /// head walks are noise); `justified_checkpoint`, `finalized_checkpoint`,
+    /// and `safe_target` are ungated (they coalesce to the latest value, which
+    /// is the only one that matters).
     pub(crate) fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) {
         let head = store.head().expect("head block exists");
         if head != self.head {
@@ -215,6 +243,29 @@ impl ChainEventSnapshot {
             {
                 // Skip stale heads (catch-up/backfill): see HEAD_EVENT_RECENCY_SLOTS.
                 if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot {
+                    // A head change that leaves the old head's chain is a reorg;
+                    // surface it just before the `head` event (beacon ordering).
+                    // `reorg_depth` returns None for a plain extension. Reading
+                    // the old-head header can fail only on genuine store
+                    // inconsistency; that just drops the reorg detail, never the
+                    // head event. Note: during a multi-move catch-up tick the
+                    // diff sees only net (pre, post) heads, so a transient reorg
+                    // that reverts within one call is invisible here (store
+                    // metrics still count it).
+                    if let Some(depth) = crate::store::reorg_depth(self.head, head, store)
+                        && let Some(old_header) = store
+                            .get_block_header(&self.head)
+                            .expect("block header read should succeed")
+                    {
+                        events.emit(ChainEvent::ChainReorg {
+                            slot: header.slot,
+                            depth,
+                            old_head_block: self.head,
+                            old_head_state: old_header.state_root,
+                            new_head_block: head,
+                            new_head_state: header.state_root,
+                        });
+                    }
                     events.emit(ChainEvent::Head {
                         slot: header.slot,
                         block: head,
@@ -264,6 +315,27 @@ impl ChainEventSnapshot {
                 );
             }
         }
+
+        // Safe target is a fork-choice block root (not a checkpoint), advanced
+        // at interval 3. Report its slot from the header; a missing header only
+        // drops this event.
+        let safe_target = store.safe_target().expect("safe target exists");
+        if safe_target != self.safe_target {
+            if let Some(header) = store
+                .get_block_header(&safe_target)
+                .expect("block header read should succeed")
+            {
+                events.emit(ChainEvent::SafeTarget {
+                    slot: header.slot,
+                    block: safe_target,
+                });
+            } else {
+                warn!(
+                    safe_target_root = %ShortRoot(&safe_target.0),
+                    "Safe target header missing while emitting event; skipping"
+                );
+            }
+        }
     }
 }
 
@@ -297,11 +369,13 @@ mod tests {
         }
     }
 
-    const ALL_TOPICS: [Topic; 4] = [
+    const ALL_TOPICS: [Topic; 6] = [
         Topic::Head,
         Topic::Block,
         Topic::JustifiedCheckpoint,
         Topic::FinalizedCheckpoint,
+        Topic::ChainReorg,
+        Topic::SafeTarget,
     ];
 
     #[tokio::test]
@@ -356,6 +430,18 @@ mod tests {
                 },
                 Topic::FinalizedCheckpoint,
             ),
+            (
+                ChainEvent::ChainReorg {
+                    slot: 1,
+                    depth: 1,
+                    old_head_block: block,
+                    old_head_state: state,
+                    new_head_block: block,
+                    new_head_state: state,
+                },
+                Topic::ChainReorg,
+            ),
+            (ChainEvent::SafeTarget { slot: 1, block }, Topic::SafeTarget),
         ];
         for (event, topic) in cases {
             assert_eq!(event.topic(), topic);
@@ -516,4 +602,79 @@ mod tests {
         }
         assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
     }
+
+    /// Swinging the head onto a sibling chain fires `chain_reorg` ahead of the
+    /// `head` event, with the rolled-back depth. (A plain extension is covered
+    /// by `chain_event_diff_emits_recent_head`, which sees no reorg.)
+    #[test]
+    fn chain_event_diff_emits_chain_reorg_before_head() {
+        let mut store = test_store();
+        let genesis = store.head().expect("store head exists");
+        let bus = EventBus::new(16);
+        let mut rx = bus.subscribe();
+
+        // Two forks off genesis: A(1) ← A2(2), and B(1) on its own.
+        let a = H256([1u8; 32]);
+        let a2 = H256([2u8; 32]);
+        let b = H256([3u8; 32]);
+        insert_test_block(&mut store, a, 1, genesis, H256([11u8; 32]));
+        insert_test_block(&mut store, a2, 2, a, H256([22u8; 32]));
+        insert_test_block(&mut store, b, 1, genesis, H256([33u8; 32]));
+
+        // Head starts on A2; snapshot; then fork choice picks B, off A2's chain.
+        store
+            .update_checkpoints(ForkCheckpoints::head_only(a2))
+            .expect("head A2");
+        let snapshot = ChainEventSnapshot::capture(&store);
+        store
+            .update_checkpoints(ForkCheckpoints::head_only(b))
+            .expect("head B");
+
+        // Wall clock at B's slot so the recency gate lets both events through.
+        snapshot.diff_and_emit(&store, &bus, 1);
+
+        match rx.try_recv().unwrap() {
+            ChainEvent::ChainReorg {
+                slot,
+                depth,
+                old_head_block,
+                new_head_block,
+                ..
+            } => {
+                assert_eq!(slot, 1, "reorg slot is the new head's slot");
+                assert_eq!(depth, 1, "A2 rolls back one block to the A/B fork point");
+                assert_eq!(old_head_block, a2);
+                assert_eq!(new_head_block, b);
+            }
+            other => panic!("expected chain_reorg first, got: {other:?}"),
+        }
+        match rx.try_recv().unwrap() {
+            ChainEvent::Head { block, .. } => assert_eq!(block, b),
+            other => panic!("expected head after reorg, got: {other:?}"),
+        }
+        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
+    }
+
+    /// A safe-target move (interval-3 output) emits `safe_target` and nothing
+    /// else when head/justified/finalized are unchanged.
+    #[test]
+    fn chain_event_diff_emits_safe_target() {
+        let mut store = test_store();
+        let genesis = store.head().expect("store head exists");
+        let bus = EventBus::new(16);
+        let mut rx = bus.subscribe();
+
+        let target = H256([6u8; 32]);
+        insert_test_block(&mut store, target, 2, genesis, H256([66u8; 32]));
+        let snapshot = ChainEventSnapshot::capture(&store);
+        store.set_safe_target(target).expect("set safe target");
+
+        snapshot.diff_and_emit(&store, &bus, 2);
+
+        match rx.try_recv().unwrap() {
+            ChainEvent::SafeTarget { slot, block } => assert_eq!((slot, block), (2, target)),
+            other => panic!("expected safe_target, got: {other:?}"),
+        }
+        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
+    }
 }

crates/blockchain/src/store.rs

@@ -1177,7 +1177,10 @@ pub fn verify_block_signatures(
 /// slot head to the lower slot head's slot, we don't arrive at the lower slot head.
 /// Returns `Some(depth)` where depth is the number of blocks walked back, or `None`
 /// if no reorg occurred.
-fn reorg_depth(old_head: H256, new_head: H256, store: &Store) -> Option<u64> {
+///
+/// `pub(crate)` so the actor-layer chain-event diff can reuse the same
+/// detection for the `chain_reorg` event, keeping one definition of "reorg".
+pub(crate) fn reorg_depth(old_head: H256, new_head: H256, store: &Store) -> Option<u64> {
     if new_head == old_head {
         return None;
     }

docs/rpc.md

@@ -86,16 +86,18 @@ SSZ-encoded `SignedBlock` at the latest finalized checkpoint. The genesis/anchor
 
 ### `GET /lean/v0/events`
 
-Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Four event types:
+Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Six event types:
 
-Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`.
+Payload fields mirror the Ethereum beacon-API eventstream where an analog exists: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. `justified_checkpoint` and `safe_target` are ethlambda extensions with no beacon topic.
 
 | Event | Payload | Emitted when |
 |-------|---------|--------------|
 | `head` | `{ "slot": 128, "block": "0x…", "state": "0x…" }` | Fork choice selects a new head within `HEAD_EVENT_RECENCY_SLOTS` (32 slots) of the wall clock; no head events fire during catch-up |
 | `block` | `{ "slot": 128, "block": "0x…" }` | A block is imported into the store |
 | `justified_checkpoint` | `{ "slot": 120, "block": "0x…", "state": "0x…" }` | The justified checkpoint advances |
 | `finalized_checkpoint` | `{ "slot": 96, "block": "0x…", "state": "0x…" }` | The finalized checkpoint advances |
+| `chain_reorg` | `{ "slot": 128, "depth": 2, "old_head_block": "0x…", "old_head_state": "0x…", "new_head_block": "0x…", "new_head_state": "0x…" }` | Fork choice switches to a head off the old head's chain (same recency gate as `head`); beacon shape minus `epoch`/`execution_optimistic` |
+| `safe_target` | `{ "slot": 127, "block": "0x…" }` | The interval-3 safe attestation target advances |
 
 The topic name travels only on the SSE `event:` line; the `data:` line carries the flat JSON payload. Example frame:
 
@@ -112,7 +114,7 @@ A **required** comma-separated list of event names selects which events to strea
 curl -N 'http://127.0.0.1:5052/lean/v0/events?topics=head,finalized_checkpoint'
 ```
 
-Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want.
+Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`, `chain_reorg`, `safe_target`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want.
 
 | Status | Condition |
 |--------|-----------|