feat(replay): wire latest_processed_slot for RPC getSlot

- Add SlotTree.tip() to get highest slot among fork leaves (bypass mode)
- Update latest_processed_slot inside vote handling to match Agave behavior
- Update latest_processed_slot in bypassConsensus() from SlotTree.tip()
- Change ForkChoiceProcessedSlot to use store() since slot can decrease
Note: Bypass mode uses highest fork tip, while consensus mode matches
Agave's semantics where processed slot only updates when voting.

Author: prestonsn

src/replay/consensus/core.zig

@@ -675,6 +675,12 @@ pub const TowerConsensus = struct {
                 slot_leaders,
                 vote_sockets,
             );
+
+            // Update the latest processed slot to the bank being voted on.
+            // This matches Agave's behavior: the processed slot is updated inside
+            // handle_votable_bank(), which is only called when vote_bank.is_some().
+            // See: https://github.com/anza-xyz/agave/blob/5e900421520a10933642d5e9a21e191a70f9b125/core/src/replay_stage.rs#L2683
+            slot_tracker.latest_processed_slot.set(voted.slot);
         }
 
         // Reset onto a fork

src/replay/service.zig

@@ -609,7 +609,6 @@ fn freezeCompletedSlots(state: *ReplayState, results: []const ReplayResult) !boo
                     slot,
                     last_entry_hash,
                 ));
-                state.slot_tracker.latest_confirmed_slot.update(slot);
                 processed_a_slot = true;
             } else {
                 state.logger.info().logf("partially replayed slot: {}", .{slot});
@@ -622,6 +621,20 @@ fn freezeCompletedSlots(state: *ReplayState, results: []const ReplayResult) !boo
 
 /// bypass the tower bft consensus protocol, simply rooting slots with SlotTree.reRoot
 fn bypassConsensus(state: *ReplayState) !void {
+    // NOTE: Processed slot semantics differ from Agave when Sig is in bypass-consensus mode.
+    // In bypass mode, `latest_processed_slot` is set to the highest slot among all fork
+    // leaves (SlotTree.tip()).
+    //
+    // This differs from Agave's behavior: the processed slot is only updated
+    // when `vote_bank.is_some()` (i.e., when the validator has selected a bank
+    // to vote on after passing all consensus checks like lockout, threshold, and
+    // switch proof). If the validator is locked out or fails
+    // threshold checks, the processed slot is NOT updated and can go stale.
+    // See: https://github.com/anza-xyz/agave/blob/5e900421520a10933642d5e9a21e191a70f9b125/core/src/replay_stage.rs#L2683
+    //
+    // TowerConsensus implements Agave's processed slot semantics when consensus is enabled.
+    state.slot_tracker.latest_processed_slot.set(state.slot_tree.tip());
+
     if (state.slot_tree.reRoot(state.allocator)) |new_root| {
         const slot_tracker = &state.slot_tracker;
 

src/replay/trackers.zig

@@ -14,8 +14,11 @@ const SlotState = sig.core.SlotState;
 pub const ForkChoiceProcessedSlot = struct {
     slot: std.atomic.Value(Slot) = .init(0),
 
-    pub fn update(self: *@This(), new_slot: Slot) void {
-        _ = self.slot.fetchMax(new_slot, .monotonic);
+    /// Set the current processed slot (heaviest fork tip).
+    /// Uses store() because this can decrease when the fork choice
+    /// switches to a different fork with a lower slot.
+    pub fn set(self: *@This(), new_slot: Slot) void {
+        self.slot.store(new_slot, .monotonic);
     }
 
     pub fn get(self: *const @This()) Slot {
@@ -246,6 +249,16 @@ pub const SlotTree = struct {
     const List = std.ArrayListUnmanaged;
     const min_age = 32;
 
+    /// Returns the highest slot among all fork tips (leaves).
+    /// In bypass mode (without ForkChoice), this represents the "processed" slot.
+    pub fn tip(self: *const SlotTree) Slot {
+        var max_slot: Slot = self.root.slot;
+        for (self.leaves.items) |leaf| {
+            max_slot = @max(max_slot, leaf.slot);
+        }
+        return max_slot;
+    }
+
     pub fn deinit(const_self: SlotTree, allocator: Allocator) void {
         var self = const_self;
         self.root.destroyRecursivelyDownstream(allocator);