diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go
index 5474fd95d..e4797db4a 100644
--- a/weed/server/volume_server_block.go
+++ b/weed/server/volume_server_block.go
@@ -1345,6 +1345,28 @@ func (bs *BlockService) observePrimaryShipperConnectivity(path string) {
 		// entries are available.
 		_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
 			connected = vol.TryReconnectShippers()
+			if !connected {
+				// Bridge 1: check if any shipper reached NeedsRebuild during
+				// the proactive reconnect. Emit per-replica NeedsRebuildObserved
+				// to the core, then initiate primary-direct rebuild for that
+				// specific replica. The master is NOT involved in the rebuild
+				// decision — the primary owns data-control recovery. The master
+				// sees the result via subsequent heartbeat projection.
+				for _, st := range vol.ReplicaShipperStates() {
+					if st.State == "needs_rebuild" && st.DataAddr != "" {
+						replicaID := bs.resolveReplicaIDForShipper(path, st.DataAddr)
+						glog.V(0).Infof("block service: shipper %s (replica=%s) needs rebuild — primary-direct rebuild",
+							st.DataAddr, replicaID)
+						if replicaID != "" {
+							bs.applyCoreEvent(engine.NeedsRebuildObserved{
+								ID:        path,
+								ReplicaID: replicaID,
+								Reason:    "gap_exceeds_retained_wal",
+							})
+						}
+					}
+				}
+			}
 			return nil
 		})
 	}
@@ -1353,6 +1375,31 @@ func (bs *BlockService) observePrimaryShipperConnectivity(path string) {
 	bs.observePrimaryShipperConnectivityStatus(path, connected)
 }
 
+// resolveReplicaIDForShipper maps a shipper's data address back to the
+// replica ID. Uses the shipper group's per-shipper ReplicaID which was
+// set via SetReplicaAddrs from the ServerID. Works for RF=2 and RF=3+.
+func (bs *BlockService) resolveReplicaIDForShipper(path, dataAddr string) string {
+	if bs == nil || bs.blockStore == nil || dataAddr == "" {
+		return ""
+	}
+	var replicaID string
+	_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
+		sg := vol.GetShipperGroup()
+		if sg == nil {
+			return nil
+		}
+		for i := 0; i < sg.Len(); i++ {
+			s := sg.Shipper(i)
+			if s != nil && s.DataAddr() == dataAddr {
+				replicaID = s.ReplicaID()
+				return nil
+			}
+		}
+		return nil
+	})
+	return replicaID
+}
+
 func (bs *BlockService) observePrimaryShipperConnectivityStatus(path string, connected bool) {
 	if bs == nil || bs.v2Core == nil || !connected {
 		return
diff --git a/weed/storage/blockvol/wal_shipper.go b/weed/storage/blockvol/wal_shipper.go
index 2878c9508..e958e24e3 100644
--- a/weed/storage/blockvol/wal_shipper.go
+++ b/weed/storage/blockvol/wal_shipper.go
@@ -111,6 +111,11 @@ func (s *WALShipper) ReplicaID() string {
 	return s.replicaID
 }
 
+// DataAddr returns the shipper's configured data channel address.
+func (s *WALShipper) DataAddr() string {
+	return s.dataAddr
+}
+
 // SetLiveShippingPolicy installs a host-provided gate for current live-tail
 // shipping. The callback is consulted before any network dial or send occurs.
 func (s *WALShipper) SetLiveShippingPolicy(fn func(replicaID string, entryLSN uint64) (allow bool, reason string)) {