diff --git a/changelog.d/protobuf_nesting_depth_limit.fix.md b/changelog.d/protobuf_nesting_depth_limit.fix.md
new file mode 100644
index 0000000000000..9151acea087f4
--- /dev/null
+++ b/changelog.d/protobuf_nesting_depth_limit.fix.md
@@ -0,0 +1,3 @@
+Fixed an issue where unusually deeply nested event data or metadata could make disk buffers unreadable or cause vector-to-vector pipelines to retry indefinitely. Vector now detects affected events before buffering or sending while leaving safely nested events unchanged. When when_full = "overflow" is configured, the original event is routed intact to the overflow stage regardless of buffer occupancy; otherwise, only the affected event is dropped.
+
+authors: connoryy ganelo EricaJ6 jonodera97
diff --git a/lib/vector-buffers/benches/common.rs b/lib/vector-buffers/benches/common.rs
index cb2f1ec402116..280be0b519337 100644
--- a/lib/vector-buffers/benches/common.rs
+++ b/lib/vector-buffers/benches/common.rs
@@ -12,7 +12,7 @@ use vector_buffers::{
         builder::TopologyBuilder,
         channel::{BufferReceiver, BufferSender},
     },
-    BufferType, EventCount,
+    BufferType, Bufferable, EventCount,
 };
 use vector_common::byte_size_of::ByteSizeOf;
 use vector_common::finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable};
@@ -56,6 +56,8 @@ impl<const N: usize> Finalizable for Message<N> {
     }
 }
 
+impl<const N: usize> Bufferable for Message<N> {}
+
 #[derive(Debug)]
 pub struct EncodeError;
 
diff --git a/lib/vector-buffers/examples/buffer_perf.rs b/lib/vector-buffers/examples/buffer_perf.rs
index 67d781bd7cf0a..36219db6cdf99 100644
--- a/lib/vector-buffers/examples/buffer_perf.rs
+++ b/lib/vector-buffers/examples/buffer_perf.rs
@@ -67,6 +67,8 @@ impl EventCount for VariableMessage {
     }
 }
 
+impl Bufferable for VariableMessage {}
+
 impl Finalizable for VariableMessage {
     fn take_finalizers(&mut self) -> EventFinalizers {
         std::mem::take(&mut self.finalizers)
diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs
index aaf582c683cfd..3ce7f47d7f4bf 100644
--- a/lib/vector-buffers/src/lib.rs
+++ b/lib/vector-buffers/src/lib.rs
@@ -105,10 +105,62 @@ impl<T> InMemoryBufferable for T where
 /// An item that can be buffered.
 ///
 /// This supertrait serves as the base trait for any item that can be pushed into a buffer.
-pub trait Bufferable: InMemoryBufferable + Encodable {}
+pub trait Bufferable: InMemoryBufferable + Encodable {
+    /// Drops any sub-items that cannot be persisted by the calling backend (e.g. due to
+    /// format-imposed nesting depth limits), reporting them as dropped via the appropriate
+    /// telemetry. Returns `None` if nothing remains worth writing.
+    ///
+    /// # Who calls this
+    ///
+    /// Only persistent backends with wire-format constraints invoke this — today that's
+    /// the disk-v2 sender (`SenderAdapter::send`/`try_send`). In-memory channels skip it
+    /// entirely because they hold the in-memory representation and have no nesting-limit
+    /// risk. A new backend with similar constraints should call this in the same place
+    /// and surface the resulting drops to `BufferSender` so that buffer-usage
+    /// instrumentation stays consistent with what actually lands in the buffer.
+    ///
+    /// # Default behaviour
+    ///
+    /// The default returns `Some(self)` if the item carries any events, and `None` if
+    /// it is already empty. This means an item that arrives empty (`event_count() == 0`)
+    /// is silently *not* persisted — preserving the pre-existing
+    /// "don't write empty records to disk" behaviour the call site used to enforce.
+    /// Types whose owners want empty items to be persisted must override this.
+    ///
+    /// # Skipping this call
+    ///
+    /// If a persistent backend writes an item without first calling `filter_unencodable`,
+    /// any sub-item that exceeds the format's limits will surface as a hard
+    /// [`Encodable::encode`] error and the *entire* item is rejected — including any
+    /// sibling sub-items that would otherwise have encoded fine. The filter is the only
+    /// path that produces graceful per-item drop with telemetry and a `Rejected` event
+    /// status; the encode-level check exists purely as defense-in-depth to ensure a
+    /// corrupt record cannot reach disk if a future caller forgets to filter.
+    fn filter_unencodable(self) -> Option<Self> {
+        if self.event_count() > 0 {
+            Some(self)
+        } else {
+            None
+        }
+    }
 
-// Blanket implementation for anything that is already bufferable.
-impl<T> Bufferable for T where T: InMemoryBufferable + Encodable {}
+    /// Returns whether every sub-item can be persisted by a backend with wire-format
+    /// constraints, without consuming or modifying the item.
+    ///
+    /// This is the non-destructive counterpart to [`Bufferable::filter_unencodable`], and
+    /// exists so routing policy can be decided *before* any filtering happens. In
+    /// particular `WhenFull::Overflow` needs to know that an item can never reach disk, so
+    /// it can hand the item to the overflow stage intact rather than pruning sub-items for
+    /// a write that would not have succeeded at any buffer occupancy.
+    ///
+    /// The default returns `true`, which is correct for any type without format limits.
+    /// Implementors overriding [`Bufferable::filter_unencodable`] must override this too,
+    /// and the two must agree: this returns `false` exactly when `filter_unencodable` would
+    /// drop at least one sub-item.
+    fn is_fully_encodable(&self) -> bool {
+        true
+    }
+}
 
 pub trait EventCount {
     fn event_count(&self) -> usize;
diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs
index 6d73d020937eb..9dbeee9b8535f 100644
--- a/lib/vector-buffers/src/test/messages.rs
+++ b/lib/vector-buffers/src/test/messages.rs
@@ -7,7 +7,12 @@ use vector_common::finalization::{
     AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable,
 };
 
-use crate::{encoding::FixedEncodable, EventCount};
+use crate::{encoding::FixedEncodable, Bufferable, EventCount};
+
+impl Bufferable for SizedRecord {}
+impl Bufferable for UndecodableRecord {}
+impl Bufferable for MultiEventRecord {}
+impl Bufferable for PoisonPillMultiEventRecord {}
 
 macro_rules! message_wrapper {
     ($id:ident: $ty:ty, $event_count:expr) => {
diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs
index 15191ec02eead..92cbb839ae9c7 100644
--- a/lib/vector-buffers/src/topology/channel/sender.rs
+++ b/lib/vector-buffers/src/topology/channel/sender.rs
@@ -40,12 +40,45 @@ impl<T> SenderAdapter<T>
 where
     T: Bufferable,
 {
+    /// Whether this backend can only persist items satisfying [`Bufferable::is_fully_encodable`].
+    ///
+    /// In-memory stages hold the in-memory representation and have no wire format, so they can
+    /// accept any item regardless of its nesting depth. Disk stages encode to protobuf on write
+    /// and cannot.
+    ///
+    /// Callers use this to avoid assuming a stage is constrained: an item that one stage cannot
+    /// encode may be perfectly storable by another, so the check must be asked of the specific
+    /// stage rather than applied to every topology.
+    pub(crate) fn requires_encodable_items(&self) -> bool {
+        match self {
+            Self::InMemory(_) => false,
+            Self::DiskV2(_) => true,
+        }
+    }
+
     pub(crate) async fn send(&mut self, item: T) -> crate::Result<()> {
         match self {
             Self::InMemory(tx) => tx.send(item).await.map_err(Into::into),
             Self::DiskV2(writer) => {
+                let pre_count = item.event_count() as u64;
+                let pre_size = item.size_of() as u64;
                 let mut writer = writer.lock().await;
 
+                let Some(item) = item.filter_unencodable() else {
+                    // The whole item was filtered out (e.g. every sub-item over the
+                    // protobuf nesting budget). Report the drop directly via the
+                    // ledger's usage handle so it shows up in the disk-v2 stage's
+                    // `received` / `dropped` metrics — `BufferSender` does not carry
+                    // its own handle for backends that `provides_instrumentation()`.
+                    writer.track_dropped(pre_count, pre_size);
+                    return Ok(());
+                };
+                if item.event_count() as u64 != pre_count {
+                    let dropped_events = pre_count - item.event_count() as u64;
+                    let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64);
+                    writer.track_dropped(dropped_events, dropped_bytes);
+                }
+
                 writer.write_record(item).await.map(|_| ()).map_err(|e| {
                     // TODO: Could some errors be handled and not be unrecoverable? Right now,
                     // encoding should theoretically be recoverable -- encoded value was too big, or
@@ -69,6 +102,26 @@ where
             Self::DiskV2(writer) => {
                 let mut writer = writer.lock().await;
 
+                // Filtering here is unconditional and independent of current occupancy.
+                // Whether an unencodable item should be dropped or handed to an overflow
+                // stage is a `WhenFull` policy decision, so it is made in `BufferSender`
+                // before the item ever reaches this backend: `WhenFull::Overflow` diverts
+                // items failing `is_fully_encodable` straight to the overflow stage, and
+                // anything arriving here is therefore expected to be persistable. Keeping
+                // the filter unconditional means a given item is treated the same at 99%
+                // full as at 100% full.
+                let pre_count = item.event_count() as u64;
+                let pre_size = item.size_of() as u64;
+                let Some(item) = item.filter_unencodable() else {
+                    writer.track_dropped(pre_count, pre_size);
+                    return Ok(None);
+                };
+                if item.event_count() as u64 != pre_count {
+                    let dropped_events = pre_count - item.event_count() as u64;
+                    let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64);
+                    writer.track_dropped(dropped_events, dropped_bytes);
+                }
+
                 writer.try_write_record(item).await.map_err(|e| {
                     // TODO: Could some errors be handled and not be unrecoverable? Right now,
                     // encoding should theoretically be recoverable -- encoded value was too big, or
@@ -212,7 +265,25 @@ impl<T: Bufferable> BufferSender<T> {
                 }
             }
             WhenFull::Overflow => {
-                if let Some(item) = self.base.try_send(item).await? {
+                // An item the base stage can never encode is routed to the overflow stage
+                // intact, whatever the current occupancy. Deciding this here, rather than
+                // letting the backend filter it, is what makes the behaviour
+                // state-independent: previously an over-nested item was pruned while the
+                // disk had room and forwarded whole once the disk reported full, so the
+                // same item took different paths at 99% and 100%.
+                //
+                // The check is gated on the base stage actually having a wire-format
+                // constraint. A memory stage overflowing to disk can store an over-nested
+                // item perfectly well, so diverting it past memory would send an item the
+                // base could have kept to a stage that must drop it.
+                if self.base.requires_encodable_items() && !item.is_fully_encodable() {
+                    sent_to_base = false;
+                    self.overflow
+                        .as_mut()
+                        .unwrap_or_else(|| unreachable!("overflow must exist"))
+                        .send(item, send_reference)
+                        .await?;
+                } else if let Some(item) = self.base.try_send(item).await? {
                     sent_to_base = false;
                     self.overflow
                         .as_mut()
diff --git a/lib/vector-buffers/src/topology/test_util.rs b/lib/vector-buffers/src/topology/test_util.rs
index ec0acac7416d2..26c8688548906 100644
--- a/lib/vector-buffers/src/topology/test_util.rs
+++ b/lib/vector-buffers/src/topology/test_util.rs
@@ -71,6 +71,8 @@ impl EventCount for Sample {
     }
 }
 
+impl Bufferable for Sample {}
+
 #[derive(Debug)]
 #[allow(dead_code)] // The inner _is_ read by the `Debug` impl, but that's ignored
 pub struct BasicError(pub(crate) String);
diff --git a/lib/vector-buffers/src/variants/disk_v2/ledger.rs b/lib/vector-buffers/src/variants/disk_v2/ledger.rs
index d22accf5db81a..ec4a5bf7a8d9f 100644
--- a/lib/vector-buffers/src/variants/disk_v2/ledger.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/ledger.rs
@@ -387,6 +387,20 @@ where
             .increment_received_event_count_and_byte_size(event_count, record_size);
     }
 
+    /// Tracks events that arrived at the buffer but were rejected before being
+    /// persisted (e.g. `Bufferable::filter_unencodable` dropping over-budget
+    /// sub-items). Bumps both `received` and the unintentional-`dropped` counter
+    /// on the usage handle so `buffer_size = received - sent - dropped` stays
+    /// consistent and operators can see the rejection in buffer-usage metrics.
+    /// `total_buffer_size` is intentionally left alone — these events never
+    /// reached disk.
+    pub fn track_dropped(&self, event_count: u64, byte_size: u64) {
+        self.usage_handle
+            .increment_received_event_count_and_byte_size(event_count, byte_size);
+        self.usage_handle
+            .increment_dropped_event_count_and_byte_size(event_count, byte_size, false);
+    }
+
     /// Tracks the statistics of multiple successful reads.
     pub fn track_reads(&self, event_count: u64, total_record_size: u64) {
         self.decrement_total_buffer_size(total_record_size);
diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs
new file mode 100644
index 0000000000000..ce71f9f7bcd7d
--- /dev/null
+++ b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs
@@ -0,0 +1,337 @@
+//! Buffer-usage accounting around `Bufferable::filter_unencodable`.
+//!
+//! When the disk-v2 sender drops sub-items because they exceed protobuf nesting
+//! limits, those drops must show up as unintentional buffer drops so that
+//! `buffer_size_*` (received minus left) stays consistent with what is actually
+//! queued on disk. Without that, a single rejected event makes the buffer report
+//! one queued event forever.
+
+use std::{error, fmt, time::Duration};
+
+use bytes::{Buf, BufMut};
+use tokio::time::timeout;
+use vector_common::{
+    byte_size_of::ByteSizeOf,
+    finalization::{AddBatchNotifier, BatchNotifier},
+};
+
+use super::create_default_buffer_v2_with_usage;
+use crate::{
+    encoding::FixedEncodable,
+    test::{install_tracing_helpers, with_temp_dir},
+    topology::channel::{limited, BufferSender, SenderAdapter},
+    Bufferable, EventCount, WhenFull,
+};
+
+/// A bufferable carrying a self-declared `event_count` of `events`, whose
+/// `filter_unencodable` shrinks it to `post_filter` events (or drops it entirely
+/// when `post_filter == 0`). Lets the test pin "before vs after filter" sizing
+/// without needing the full `EventArray` machinery.
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct FilterableBatch {
+    events: u32,
+    post_filter: u32,
+}
+
+impl AddBatchNotifier for FilterableBatch {
+    fn add_batch_notifier(&mut self, batch: BatchNotifier) {
+        drop(batch);
+    }
+}
+
+impl ByteSizeOf for FilterableBatch {
+    fn allocated_bytes(&self) -> usize {
+        0
+    }
+}
+impl EventCount for FilterableBatch {
+    fn event_count(&self) -> usize {
+        self.events as usize
+    }
+}
+
+#[derive(Debug)]
+struct CodecError;
+impl fmt::Display for CodecError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{self:?}")
+    }
+}
+impl error::Error for CodecError {}
+
+impl FixedEncodable for FilterableBatch {
+    type EncodeError = CodecError;
+    type DecodeError = CodecError;
+    fn encode<B: BufMut>(self, buf: &mut B) -> Result<(), Self::EncodeError> {
+        if buf.remaining_mut() < 8 {
+            return Err(CodecError);
+        }
+        buf.put_u32(self.events);
+        buf.put_u32(self.post_filter);
+        Ok(())
+    }
+    fn decode<B: Buf>(mut buf: B) -> Result<Self, Self::DecodeError> {
+        Ok(FilterableBatch {
+            events: buf.get_u32(),
+            post_filter: buf.get_u32(),
+        })
+    }
+    fn encoded_size(&self) -> Option<usize> {
+        Some(8)
+    }
+}
+
+impl Bufferable for FilterableBatch {
+    fn is_fully_encodable(&self) -> bool {
+        self.post_filter == self.events
+    }
+
+    fn filter_unencodable(self) -> Option<Self> {
+        if self.post_filter == 0 {
+            None
+        } else {
+            Some(FilterableBatch {
+                events: self.post_filter,
+                post_filter: self.post_filter,
+            })
+        }
+    }
+}
+
+/// A partial-filter drop on a disk-v2 send must show up as an unintentional buffer
+/// drop, so `buffer_size_*` stays consistent with what actually landed on disk.
+///
+/// Note: we deliberately do NOT attach `with_usage_instrumentation` to the
+/// `BufferSender`. In production, `TopologyBuilder::build` skips that call for
+/// disk-v2 because the stage `provides_instrumentation()` itself via the ledger.
+/// This test reflects that production wiring: filter-drop accounting goes
+/// through `Ledger::track_dropped`, and `usage` (returned by the helper) IS the
+/// ledger's handle.
+#[tokio::test]
+async fn filter_drops_are_reported_as_unintentional_buffer_drops() {
+    let _a = install_tracing_helpers();
+
+    with_temp_dir(|dir| {
+        let data_dir = dir.to_path_buf();
+
+        async move {
+            let (writer, _reader, _ledger, usage) =
+                create_default_buffer_v2_with_usage::<_, FilterableBatch>(data_dir).await;
+            let mut sender = BufferSender::new(SenderAdapter::from(writer), WhenFull::Block);
+
+            // 10 events arrive, filter keeps 3. Flush after each send so the
+            // ledger's `track_write` actually reaches the usage handle (it only
+            // fires when buffered writes are flushed to disk).
+            sender
+                .send(
+                    FilterableBatch {
+                        events: 10,
+                        post_filter: 3,
+                    },
+                    None,
+                )
+                .await
+                .expect("send should succeed");
+            sender.flush().await.expect("flush should succeed");
+
+            let snapshot = usage.snapshot();
+            assert_eq!(
+                snapshot.received_event_count, 10,
+                "received counts both the 7 filter-dropped events (via track_dropped) \
+                 and the 3 events flushed to disk (via track_write)",
+            );
+            assert_eq!(
+                snapshot.dropped_event_count, 7,
+                "filter drops show up under the disk-v2 stage's unintentional dropped count \
+                 so buffer_size stays consistent (received - sent - dropped = 3 queued)",
+            );
+            assert_eq!(
+                snapshot.dropped_event_count_intentional, 0,
+                "no buffer-fullness drops here",
+            );
+
+            // 5 events arrive, filter drops them all (nothing reaches disk).
+            sender
+                .send(
+                    FilterableBatch {
+                        events: 5,
+                        post_filter: 0,
+                    },
+                    None,
+                )
+                .await
+                .expect("send should succeed");
+
+            let snapshot = usage.snapshot();
+            assert_eq!(
+                snapshot.received_event_count, 15,
+                "fully-filtered item still bumps received via track_dropped",
+            );
+            assert_eq!(
+                snapshot.dropped_event_count, 12,
+                "all 5 events from the fully-filtered item are reported as unintentional drops",
+            );
+        }
+    })
+    .await;
+}
+
+/// Under `WhenFull::Overflow`, an item the base stage cannot encode must reach the
+/// overflow stage *intact* while the base stage still has room.
+///
+/// This is the near-full half of the state-independence guarantee: the routing decision
+/// is made from the item alone, so it does not matter how full the base stage is.
+#[tokio::test]
+async fn unencodable_item_overflows_intact_when_base_has_room() {
+    let _a = install_tracing_helpers();
+
+    with_temp_dir(|dir| {
+        let data_dir = dir.to_path_buf();
+
+        async move {
+            let (writer, _reader, _ledger, _usage) =
+                create_default_buffer_v2_with_usage::<_, FilterableBatch>(data_dir).await;
+
+            let (overflow_tx, mut overflow_rx) = limited(100);
+            let mut sender = BufferSender::with_overflow(
+                SenderAdapter::from(writer),
+                BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+            );
+
+            // The disk stage is empty, so it has ample room. The item is wholly
+            // unencodable, so it must still be handed to the overflow stage rather than
+            // filtered away.
+            sender
+                .send(
+                    FilterableBatch {
+                        events: 5,
+                        post_filter: 0,
+                    },
+                    None,
+                )
+                .await
+                .expect("send should succeed");
+
+            let received = overflow_rx.next().await.expect("item must reach overflow");
+            assert_eq!(
+                received,
+                FilterableBatch {
+                    events: 5,
+                    post_filter: 0,
+                },
+                "overflow must receive the item intact, with no sub-items pruned",
+            );
+        }
+    })
+    .await;
+}
+
+/// The already-full half of the same guarantee: an unencodable item reaches the overflow
+/// stage intact when the base stage is at capacity, and by the same route.
+///
+/// The base here is an in-memory stage rather than disk, because it can be driven to a
+/// known-full state deterministically. Reliably forcing disk-v2's `is_buffer_full()` to
+/// `true` under the minimum-size config requires careful record/buffer size tuning, since
+/// `can_write_record` generally short-circuits writes before `total_buffer_size` reaches
+/// `max_buffer_size`. That substitution is sound for this property: the unencodable-item
+/// decision is taken in `BufferSender` from `Bufferable::is_fully_encodable` before any
+/// backend is consulted, so the base stage's type and occupancy are both immaterial. That
+/// is precisely the invariant being asserted.
+#[tokio::test]
+async fn unencodable_item_overflows_intact_when_base_is_full() {
+    let _a = install_tracing_helpers();
+
+    let (base_tx, _base_rx) = limited::<FilterableBatch>(1);
+    let (overflow_tx, mut overflow_rx) = limited(100);
+
+    let mut sender = BufferSender::with_overflow(
+        SenderAdapter::from(base_tx),
+        BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+    );
+
+    // Fill the base stage so any further send would be rejected for fullness.
+    sender
+        .send(
+            FilterableBatch {
+                events: 1,
+                post_filter: 1,
+            },
+            None,
+        )
+        .await
+        .expect("first send should occupy the base stage");
+
+    sender
+        .send(
+            FilterableBatch {
+                events: 5,
+                post_filter: 0,
+            },
+            None,
+        )
+        .await
+        .expect("send should succeed");
+
+    let received = overflow_rx.next().await.expect("item must reach overflow");
+    assert_eq!(
+        received,
+        FilterableBatch {
+            events: 5,
+            post_filter: 0,
+        },
+        "a full base stage must not change how an unencodable item is routed",
+    );
+}
+
+/// A base stage without a wire-format constraint must keep an unencodable item rather than
+/// pass it to the overflow stage.
+///
+/// The encodability check is a property of the *base* stage, not of the item alone. In a
+/// `memory -> disk` overflow topology the memory stage can hold an arbitrarily nested item
+/// safely, so diverting it past memory would hand an item the base could have kept to a
+/// stage that has no choice but to drop it. This is the mirror image of the
+/// `disk -> memory` cases above and guards against reintroducing that assumption.
+#[tokio::test]
+async fn unencodable_item_stays_in_base_when_base_has_no_encoding_constraint() {
+    let _a = install_tracing_helpers();
+
+    let (base_tx, mut base_rx) = limited::<FilterableBatch>(100);
+    let (overflow_tx, mut overflow_rx) = limited(100);
+
+    let mut sender = BufferSender::with_overflow(
+        SenderAdapter::from(base_tx),
+        BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+    );
+
+    // The base is in-memory and empty, so it can hold this item despite the item being
+    // unencodable for a protobuf-backed stage.
+    sender
+        .send(
+            FilterableBatch {
+                events: 5,
+                post_filter: 0,
+            },
+            None,
+        )
+        .await
+        .expect("send should succeed");
+
+    let received = timeout(Duration::from_secs(5), base_rx.next())
+        .await
+        .expect("item must stay in the base stage rather than be diverted to overflow")
+        .expect("base stage should yield the item");
+    assert_eq!(
+        received,
+        FilterableBatch {
+            events: 5,
+            post_filter: 0,
+        },
+        "an unconstrained base stage must keep the item intact",
+    );
+    assert!(
+        timeout(Duration::from_millis(50), overflow_rx.next())
+            .await
+            .is_err(),
+        "the overflow stage must not be involved when the base can hold the item",
+    );
+}
diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs
index 4302d6d461c17..e2e121c26ffbd 100644
--- a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs
@@ -749,6 +749,8 @@ async fn reader_throws_error_when_record_is_undecodable_via_metadata() {
         }
     }
 
+    impl crate::Bufferable for ControllableRecord {}
+
     with_temp_dir(|dir| {
         let data_dir = dir.to_path_buf();
 
diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs b/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs
index 2319fbf97dce8..a3e57a46e60a2 100644
--- a/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/tests/mod.rs
@@ -24,6 +24,7 @@ type FilesystemUnderTest = ProductionFilesystem;
 
 mod acknowledgements;
 mod basic;
+mod filter_metrics;
 mod initialization;
 mod invariants;
 mod known_errors;
diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs
index 3b13e35ca47e0..17f6d1a3cdb67 100644
--- a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs
@@ -9,7 +9,7 @@ use vector_common::finalization::{
 use crate::{
     encoding::FixedEncodable,
     variants::disk_v2::{record::RECORD_HEADER_LEN, tests::align16},
-    EventCount,
+    Bufferable, EventCount,
 };
 
 #[derive(Debug)]
@@ -108,6 +108,8 @@ impl ByteSizeOf for Record {
     }
 }
 
+impl Bufferable for Record {}
+
 impl FixedEncodable for Record {
     type EncodeError = EncodeError;
     type DecodeError = DecodeError;
diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs
index fc8f60672e1af..685bdcb494d46 100644
--- a/lib/vector-buffers/src/variants/disk_v2/writer.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs
@@ -993,6 +993,16 @@ where
         total_buffer_size >= max_buffer_size
     }
 
+    /// Records sub-items that arrived at the buffer but were dropped before
+    /// reaching disk (e.g. `Bufferable::filter_unencodable` rejecting events
+    /// the protobuf decoder cannot handle). Delegates to the ledger's usage
+    /// handle so the rejection shows up under the disk-v2 stage's
+    /// `received` / `dropped` metrics in production, where the
+    /// `BufferSender` does not carry its own usage instrumentation.
+    pub(crate) fn track_dropped(&self, event_count: u64, byte_size: u64) {
+        self.ledger.track_dropped(event_count, byte_size);
+    }
+
     /// Ensures this writer is ready to attempt writer the next record.
     #[instrument(skip(self), level = "debug")]
     async fn ensure_ready_for_write(&mut self) -> io::Result<()> {
diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs
index 27c779212e04b..85a509ffc20bf 100644
--- a/lib/vector-core/src/event/mod.rs
+++ b/lib/vector-core/src/event/mod.rs
@@ -10,6 +10,7 @@ pub use log_event::LogEvent;
 pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, WithMetadata};
 pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
 pub use r#ref::{EventMutRef, EventRef};
+pub use ser::{event_exceeds_max_nesting_cost, MAX_VALUE_NESTING_FRAMES};
 use serde::{Deserialize, Serialize};
 pub use trace::TraceEvent;
 use vector_buffers::EventCount;
diff --git a/lib/vector-core/src/event/ser.rs b/lib/vector-core/src/event/ser.rs
index be41910466976..4cf0e5d31d490 100644
--- a/lib/vector-core/src/event/ser.rs
+++ b/lib/vector-core/src/event/ser.rs
@@ -3,13 +3,154 @@ use enumflags2::{bitflags, BitFlags, FromBitsError};
 use prost::Message;
 use snafu::Snafu;
 use vector_buffers::encoding::{AsMetadata, Encodable};
+use vector_buffers::{Bufferable, EventCount};
+use vector_common::internal_event::{self, ComponentEventsDropped, UNINTENTIONAL};
+use vrl::value::Value;
 
-use super::{proto, Event, EventArray};
+use super::{proto, Event, EventArray, EventStatus};
+
+/// Per-level prost recursion frame cost of an [`Value::Object`].
+///
+/// Decoding an object level walks `Value → ValueMap → map_entry (synthetic) → Value`,
+/// adding three message-decode frames before reaching the child Value.
+pub(crate) const OBJECT_FRAME_COST: usize = 3;
+
+/// Per-level prost recursion frame cost of an [`Value::Array`].
+///
+/// Decoding an array level walks `Value → ValueArray → Value`, adding two message-decode
+/// frames before reaching the child Value.
+pub(crate) const ARRAY_FRAME_COST: usize = 2;
+
+/// Per-leaf prost recursion frame cost of a [`Value::Timestamp`].
+///
+/// Unlike other scalar variants, `Value::Timestamp` is encoded as a nested
+/// `google.protobuf.Timestamp` message, so decoding it consumes one additional frame
+/// beyond the enclosing `Value`. Without this cost, a timestamp leaf under 32 object
+/// levels would sneak past the gate at cost 96 and trip prost's recursion limit on
+/// decode at cost 97.
+pub(crate) const TIMESTAMP_FRAME_COST: usize = 1;
+
+/// Maximum prost recursion frame cost accepted for any arbitrary [`Value`].
+///
+/// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting level
+/// consumes 3 frames for [`Value::Object`], 2 for [`Value::Array`], or 1 for a
+/// [`Value::Timestamp`] leaf, plus a fixed overhead for the proto wrappers outside the
+/// Value tree.
+///
+/// Some protobuf paths (`Log.fields` and `Trace.fields`) can carry 99 frames, but the
+/// `Log.value` and metadata paths are only safe through 96. We use that highest common
+/// safe limit for every value so validation does not depend on its event type, root type,
+/// or destination protobuf field.
+pub const MAX_VALUE_NESTING_FRAMES: usize = 96;
+
+/// Walks a [`Value`] tree accumulating prost recursion frame cost, returning
+/// `Err(over_budget_cost)` as soon as any branch exceeds `budget`.
+///
+/// Object levels weigh [`OBJECT_FRAME_COST`] frames each, array levels weigh
+/// [`ARRAY_FRAME_COST`], and timestamp leaves weigh [`TIMESTAMP_FRAME_COST`] (because
+/// they decode into a nested `google.protobuf.Timestamp` message); other scalar leaves
+/// are free. Performs an early-exit traversal so well-formed events incur a single
+/// descent of the deepest branch only.
+///
+/// # Errors
+///
+/// Returns `Err(actual_cost)` if any branch's cumulative frame cost exceeds `budget`.
+pub(crate) fn check_value_nesting_cost(
+    value: &Value,
+    accumulated: usize,
+    budget: usize,
+) -> Result<(), usize> {
+    let level_cost = match value {
+        Value::Object(_) => OBJECT_FRAME_COST,
+        Value::Array(_) => ARRAY_FRAME_COST,
+        Value::Timestamp(_) => TIMESTAMP_FRAME_COST,
+        _ => 0,
+    };
+    let next = accumulated + level_cost;
+    if next > budget {
+        return Err(next);
+    }
+    match value {
+        Value::Object(map) => {
+            for v in map.values() {
+                check_value_nesting_cost(v, next, budget)?;
+            }
+        }
+        Value::Array(arr) => {
+            for v in arr {
+                check_value_nesting_cost(v, next, budget)?;
+            }
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+/// Checks whether an event's nesting frame cost exceeds the safe limits for protobuf encoding.
+///
+/// Returns `Some((cost, budget))` identifying the path that violated its budget, or `None`
+/// if the event is within bounds.
+///
+/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`].
+///
+/// For metrics, only metadata is checked since metric values have a fixed structure.
+pub fn event_exceeds_max_nesting_cost(event: &Event) -> Option<(usize, usize)> {
+    let check = |value: &Value| {
+        check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES)
+            .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES))
+    };
+    match event {
+        Event::Log(log) => check(log.value())
+            .and_then(|()| check(log.metadata().value()))
+            .err(),
+        Event::Trace(trace) => check(trace.value())
+            .and_then(|()| check(trace.metadata().value()))
+            .err(),
+        Event::Metric(metric) => check(metric.metadata().value()).err(),
+    }
+}
+
+/// Checks all events in an `EventArray` for nesting cost violations.
+///
+/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`]. For metrics,
+/// only metadata is checked since metric values have a fixed structure.
+fn check_event_array_nesting_cost(events: &EventArray) -> Result<(), EncodeError> {
+    let check = |value: &Value| {
+        check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).map_err(|cost| {
+            EncodeError::NestingTooDeep {
+                cost,
+                budget: MAX_VALUE_NESTING_FRAMES,
+            }
+        })
+    };
+    match events {
+        EventArray::Logs(logs) => {
+            for log in logs {
+                check(log.value())?;
+                check(log.metadata().value())?;
+            }
+        }
+        EventArray::Traces(traces) => {
+            for trace in traces {
+                check(trace.value())?;
+                check(trace.metadata().value())?;
+            }
+        }
+        EventArray::Metrics(metrics) => {
+            for metric in metrics {
+                check(metric.metadata().value())?;
+            }
+        }
+    }
+    Ok(())
+}
 
 #[derive(Debug, Snafu)]
 pub enum EncodeError {
     #[snafu(display("the provided buffer was too small to fully encode this item"))]
     BufferTooSmall,
+    #[snafu(display("event nesting cost {cost} exceeds protobuf budget of {budget}"))]
+    NestingTooDeep { cost: usize, budget: usize },
 }
 
 #[derive(Debug, Snafu)]
@@ -91,10 +232,28 @@ impl Encodable for EventArray {
         metadata.contains(EventEncodableMetadataFlags::DiskBufferV1CompatibilityMode)
     }
 
+    /// # Errors
+    ///
+    /// Returns `EncodeError::NestingTooDeep` if any contained event's value or metadata
+    /// exceeds [`MAX_VALUE_NESTING_FRAMES`]. This is **all-or-nothing**: a single
+    /// over-budget event fails the entire batch, because a partially-encoded
+    /// `EventArray` reaching disk would trip prost's recursion limit on decode and
+    /// corrupt the buffer.
+    ///
+    /// Callers that want graceful per-item drop with telemetry and
+    /// `EventStatus::Rejected` must run [`Bufferable::filter_unencodable`] first.
+    /// `SenderAdapter::send`/`try_send` already does this on the disk-v2 path, so the
+    /// `NestingTooDeep` arm is unreachable from any current production call site — it
+    /// is defense-in-depth for a future caller that bypasses `SenderAdapter`.
+    ///
+    /// Returns `EncodeError::BufferTooSmall` if the buffer cannot hold the encoded
+    /// output.
     fn encode<B>(self, buffer: &mut B) -> Result<(), Self::EncodeError>
     where
         B: BufMut,
     {
+        check_event_array_nesting_cost(&self)?;
+
         proto::EventArray::from(self)
             .encode(buffer)
             .map_err(|_| EncodeError::BufferTooSmall)
@@ -117,3 +276,59 @@ impl Encodable for EventArray {
         }
     }
 }
+
+impl Bufferable for EventArray {
+    /// Reuses the same budget walk as the encode-time gate, so the routing decision and
+    /// the eventual encode can never disagree about what is persistable.
+    fn is_fully_encodable(&self) -> bool {
+        check_event_array_nesting_cost(self).is_ok()
+    }
+
+    fn filter_unencodable(self) -> Option<Self> {
+        let exceeds =
+            |value: &Value| check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).is_err();
+        let mut dropped = 0;
+        let filtered = match self {
+            EventArray::Logs(mut logs) => {
+                logs.retain(|log| {
+                    let too_deep = exceeds(log.value()) || exceeds(log.metadata().value());
+                    if too_deep {
+                        log.metadata().update_status(EventStatus::Rejected);
+                        dropped += 1;
+                    }
+                    !too_deep
+                });
+                EventArray::Logs(logs)
+            }
+            EventArray::Traces(mut traces) => {
+                traces.retain(|trace| {
+                    let too_deep = exceeds(trace.value()) || exceeds(trace.metadata().value());
+                    if too_deep {
+                        trace.metadata().update_status(EventStatus::Rejected);
+                        dropped += 1;
+                    }
+                    !too_deep
+                });
+                EventArray::Traces(traces)
+            }
+            EventArray::Metrics(mut metrics) => {
+                metrics.retain(|metric| {
+                    let too_deep = exceeds(metric.metadata().value());
+                    if too_deep {
+                        metric.metadata().update_status(EventStatus::Rejected);
+                        dropped += 1;
+                    }
+                    !too_deep
+                });
+                EventArray::Metrics(metrics)
+            }
+        };
+        if dropped > 0 {
+            internal_event::emit(ComponentEventsDropped::<UNINTENTIONAL> {
+                count: dropped,
+                reason: "Event nesting cost exceeds maximum for protobuf encoding.",
+            });
+        }
+        (filtered.event_count() > 0).then_some(filtered)
+    }
+}
diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs
index aaab559da3184..3db9f2d69ee49 100644
--- a/lib/vector-core/src/event/test/serialization.rs
+++ b/lib/vector-core/src/event/test/serialization.rs
@@ -1,11 +1,19 @@
 use bytes::{Buf, BufMut, BytesMut};
+use chrono::TimeZone;
+use prost::Message;
 use quickcheck::{QuickCheck, TestResult};
 use regex::Regex;
 use similar_asserts::assert_eq;
 use vector_buffers::encoding::Encodable;
+use vector_buffers::Bufferable;
 
 use super::*;
 use crate::config::log_schema;
+use crate::event::event_exceeds_max_nesting_cost;
+use crate::event::ser::{
+    check_value_nesting_cost, ARRAY_FRAME_COST, MAX_VALUE_NESTING_FRAMES, OBJECT_FRAME_COST,
+    TIMESTAMP_FRAME_COST,
+};
 
 fn encode_value<T: Encodable, B: BufMut>(value: T, buffer: &mut B) {
     value.encode(buffer).expect("encoding should not fail");
@@ -96,3 +104,671 @@ fn type_serialization() {
     assert_eq!(map["bool"], json!(true));
     assert_eq!(map["string"], json!("thisisastring"));
 }
+
+// ---------------------------------------------------------------------------
+// Nesting validation tests
+// ---------------------------------------------------------------------------
+//
+// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting
+// level consumes a path-dependent number of prost recursion frames:
+//
+//   - `Value::Object` level: Value + ValueMap + map_entry = 3 frames
+//   - `Value::Array` level:  Value + ValueArray          = 2 frames
+//
+// Encoding paths have different fixed proto-wrapper overhead before the Value tree:
+//
+//   - `Log.fields` and `Trace.fields` can carry 99 Value frames.
+//   - `Log.value` and metadata can carry 96 Value frames.
+//
+// The gate uses the highest common safe limit, MAX_VALUE_NESTING_FRAMES (96), for every
+// arbitrary Value. The boundary tests verify both that common limit and the extra
+// headroom on the wider wire paths.
+//
+// The saturated-event tests create events with ALL Value-carrying fields at the common
+// max frame cost simultaneously. The proto conversion code populates every
+// field (including deprecated ones like Log.metadata), so a single roundtrip per event
+// type covers every proto path automatically.
+
+/// Maximum number of object-only nesting levels that fit the common Value budget.
+const MAX_OBJECT_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / OBJECT_FRAME_COST;
+
+/// Maximum number of array-only nesting levels that fit the common Value budget.
+const MAX_ARRAY_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / ARRAY_FRAME_COST;
+
+/// Creates a Value with the specified number of nested Object wrapping levels.
+///
+/// Returns a Value that is `wrapping_levels` nested Objects deep, with a string leaf.
+fn create_nested_value(wrapping_levels: usize) -> Value {
+    let mut value = Value::from("innermost");
+    for _ in 0..wrapping_levels {
+        let mut map = ObjectMap::new();
+        map.insert("nested".into(), value);
+        value = Value::Object(map);
+    }
+    value
+}
+
+/// Creates a Value with the specified number of nested Array wrapping levels.
+fn create_nested_array(wrapping_levels: usize) -> Value {
+    let mut value = Value::from("innermost");
+    for _ in 0..wrapping_levels {
+        value = Value::Array(vec![value]);
+    }
+    value
+}
+
+/// Creates a Value with the specified number of nested Object wrapping levels around
+/// the supplied leaf. Used to probe leaf-specific frame costs (e.g. `Value::Timestamp`).
+fn create_nested_value_with_leaf(wrapping_levels: usize, leaf: Value) -> Value {
+    let mut value = leaf;
+    for _ in 0..wrapping_levels {
+        let mut map = ObjectMap::new();
+        map.insert("nested".into(), value);
+        value = Value::Object(map);
+    }
+    value
+}
+
+/// A fixed [`Value::Timestamp`] for use as a leaf in nesting tests.
+fn ts_leaf() -> Value {
+    Value::Timestamp(
+        chrono::Utc
+            .timestamp_opt(1_700_000_000, 0)
+            .single()
+            .unwrap(),
+    )
+}
+
+/// Create a [`LogEvent`] with every arbitrary Value at `value_depth`.
+fn create_saturated_log(value_depth: usize) -> LogEvent {
+    let mut event = LogEvent::default();
+    event.insert("data", create_nested_value(value_depth - 1));
+    *event.metadata_mut().value_mut() = create_nested_value(value_depth);
+    event
+}
+
+/// Create a [`TraceEvent`] with every arbitrary Value at `value_depth`.
+fn create_saturated_trace(value_depth: usize) -> TraceEvent {
+    let mut trace = TraceEvent::default();
+    trace.insert("data", create_nested_value(value_depth - 1));
+    *trace.metadata_mut().value_mut() = create_nested_value(value_depth);
+    trace
+}
+
+/// Create a Metric with metadata at `value_depth`.
+/// (Metric values have fixed structure — only metadata carries arbitrary Values.)
+fn create_saturated_metric(value_depth: usize) -> Metric {
+    let mut metric = Metric::new(
+        "test",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    );
+    *metric.metadata_mut().value_mut() = create_nested_value(value_depth);
+    metric
+}
+
+/// Build all three `EventArray` variants with every arbitrary Value at the same depth.
+fn saturated_event_arrays(value_depth: usize) -> Vec<(&'static str, EventArray)> {
+    vec![
+        (
+            "Log",
+            EventArray::Logs(LogArray::from(vec![create_saturated_log(value_depth)])),
+        ),
+        (
+            "Trace",
+            EventArray::Traces(TraceArray::from(vec![create_saturated_trace(value_depth)])),
+        ),
+        (
+            "Metric",
+            EventArray::Metrics(MetricArray::from(vec![create_saturated_metric(
+                value_depth,
+            )])),
+        ),
+    ]
+}
+
+/// Build all three Event variants for `EventWrapper` encoding.
+fn saturated_events(value_depth: usize) -> Vec<(&'static str, Event)> {
+    vec![
+        ("Log", Event::Log(create_saturated_log(value_depth))),
+        ("Trace", Event::Trace(create_saturated_trace(value_depth))),
+        (
+            "Metric",
+            Event::Metric(create_saturated_metric(value_depth)),
+        ),
+    ]
+}
+
+/// Verify that the common Value budget roundtrips through every protobuf path and that
+/// increasing every Value by one object level exceeds at least one wire-path limit.
+#[test]
+fn max_nesting_budget_is_safe_for_all_paths() {
+    for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE) {
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        assert!(
+            proto::EventArray::decode(buf.freeze()).is_ok(),
+            "EventArray decode FAILED for {name} at the common Value budget.",
+        );
+    }
+
+    for (name, event) in saturated_events(MAX_OBJECT_DEPTH_VALUE) {
+        let wrapper = proto::EventWrapper::from(event);
+        let mut buf = BytesMut::with_capacity(65536);
+        wrapper.encode(&mut buf).unwrap();
+        assert!(
+            proto::EventWrapper::decode(buf.freeze()).is_ok(),
+            "EventWrapper decode FAILED for {name} at the common Value budget.",
+        );
+    }
+
+    let any_fails = saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1)
+        .into_iter()
+        .any(|(_, array)| {
+            let proto_array = proto::EventArray::from(array);
+            let mut buf = BytesMut::with_capacity(65536);
+            proto_array.encode(&mut buf).unwrap();
+            proto::EventArray::decode(buf.freeze()).is_err()
+        });
+    assert!(
+        any_fails,
+        "No path failed one object level above MAX_VALUE_NESTING_FRAMES.",
+    );
+}
+
+/// Verify the nesting gate accepts all event types at the max object-only depth.
+#[test]
+fn nesting_gate_accepts_all_types_at_max_depth() {
+    for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE) {
+        let mut buf = BytesMut::with_capacity(65536);
+        assert!(
+            array.encode(&mut buf).is_ok(),
+            "nesting gate rejected {name} at max object depths",
+        );
+    }
+}
+
+/// Verify the nesting gate rejects every event type above the common Value budget.
+#[test]
+fn nesting_gate_rejects_above_max_depth() {
+    for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1) {
+        let mut buf = BytesMut::with_capacity(65536);
+        assert!(
+            matches!(
+                array.encode(&mut buf),
+                Err(super::super::ser::EncodeError::NestingTooDeep { .. })
+            ),
+            "nesting gate should reject {name} above the common Value budget",
+        );
+    }
+}
+
+/// Verify that the wider `Log.fields` path has one level of headroom over the common
+/// budget while the metadata path is tight, for both object-only and array-only values.
+///
+/// Object-only `Log.fields`:     depth 33 succeeds, 34 fails.
+/// Object-only `metadata_full`:  depth 32 succeeds, 33 fails.
+/// Array-only  `Log.fields`:     depth 49 succeeds, 50 fails.
+/// Array-only  `metadata_full`:  depth 48 succeeds, 49 fails.
+#[test]
+fn per_path_boundaries() {
+    let roundtrip_value = |value: Value| -> bool {
+        let mut event = LogEvent::default();
+        event.insert("data", value);
+        let array = EventArray::Logs(LogArray::from(vec![event]));
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        proto::EventArray::decode(buf.freeze()).is_ok()
+    };
+
+    let roundtrip_metadata = |value: Value| -> bool {
+        let mut event = LogEvent::from("flat");
+        *event.metadata_mut().value_mut() = value;
+        let array = EventArray::Logs(LogArray::from(vec![event]));
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        proto::EventArray::decode(buf.freeze()).is_ok()
+    };
+
+    // `Log.fields` accepts 33 object levels (cost 99), one more than the common limit.
+    // The "data" key contributes the outer object level.
+    assert!(
+        roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE)),
+        "Log.fields should succeed one object level above the common budget"
+    );
+    assert!(
+        !roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE + 1)),
+        "Log.fields should fail at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE + 2
+    );
+
+    // `metadata_full` is tight at the common limit of 32 object levels (cost 96).
+    assert!(
+        roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_VALUE)),
+        "metadata_full should succeed at the common object-depth limit"
+    );
+    assert!(
+        !roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_VALUE + 1)),
+        "metadata_full should fail at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE + 1
+    );
+
+    // The outer object plus 48 nested arrays costs 99 frames on `Log.fields`.
+    assert!(
+        roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE)),
+        "Log.fields should succeed with one array level of headroom"
+    );
+    assert!(
+        !roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE + 1)),
+        "Log.fields should fail at array depth {}",
+        MAX_ARRAY_DEPTH_VALUE + 2
+    );
+
+    // `metadata_full` is tight at 48 array levels (cost 96).
+    assert!(
+        roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_VALUE)),
+        "metadata_full should succeed at the common array-depth limit"
+    );
+    assert!(
+        !roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_VALUE + 1)),
+        "metadata_full should fail at array depth {}",
+        MAX_ARRAY_DEPTH_VALUE + 1
+    );
+}
+
+/// Non-object log roots are encoded through `Log.value`, not the legacy `Log.fields`
+/// map. Its lower wire limit establishes the common budget used for every Value.
+#[test]
+fn value_budget_matches_tightest_wire_path() {
+    let make_log = |array_depth| LogEvent::from(create_nested_array(array_depth));
+    let raw_roundtrip = |log: LogEvent| {
+        let array = EventArray::Logs(LogArray::from(vec![log]));
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        proto::EventArray::decode(buf.freeze()).is_ok()
+    };
+
+    assert!(
+        raw_roundtrip(make_log(MAX_ARRAY_DEPTH_VALUE)),
+        "Log.value should roundtrip at its array-depth limit",
+    );
+    assert!(
+        !raw_roundtrip(make_log(MAX_ARRAY_DEPTH_VALUE + 1)),
+        "Log.value should fail prost decoding past its array-depth limit",
+    );
+
+    let accepted = Event::Log(make_log(MAX_ARRAY_DEPTH_VALUE));
+    assert!(event_exceeds_max_nesting_cost(&accepted).is_none());
+    let accepted = EventArray::Logs(LogArray::from(vec![accepted.into_log()]));
+    let mut buf = BytesMut::with_capacity(65536);
+    accepted
+        .encode(&mut buf)
+        .expect("the last decodable Log.value depth should pass the gate");
+
+    let rejected = Event::Log(make_log(MAX_ARRAY_DEPTH_VALUE + 1));
+    assert_eq!(
+        event_exceeds_max_nesting_cost(&rejected),
+        Some((98, MAX_VALUE_NESTING_FRAMES)),
+    );
+    let rejected = EventArray::Logs(LogArray::from(vec![rejected.into_log()]));
+    assert!(
+        rejected.clone().filter_unencodable().is_none(),
+        "the buffer filter should drop an undecodable Log.value root",
+    );
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        matches!(
+            rejected.encode(&mut buf),
+            Err(super::super::ser::EncodeError::NestingTooDeep {
+                cost: 98,
+                budget: MAX_VALUE_NESTING_FRAMES,
+            })
+        ),
+        "the encode-time gate should reject an undecodable Log.value root",
+    );
+}
+
+/// Verify that array-only nesting deeper than the object-only cap (32) is accepted by
+/// the gate — this is the regression that the frame-cost check addresses. Previously a
+/// uniform depth-33 cap dropped array-only events that prost would happily roundtrip.
+#[test]
+fn nesting_gate_accepts_deep_array_nesting() {
+    // Forty arrays below the outer log object cost 83 frames, comfortably under the
+    // 96-frame Value budget but over the old uniform depth limit.
+    let mut event = LogEvent::default();
+    event.insert("data", create_nested_array(40));
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        array.encode(&mut buf).is_ok(),
+        "nesting gate should accept array-only nesting at depth 40",
+    );
+}
+
+/// Verify the gate correctly accounts for mixed array/object nesting via the per-variant
+/// frame weights. Uses the metadata path because it has no outer wrapping object, making
+/// the arithmetic match the inserted Value's cost directly.
+#[test]
+fn nesting_gate_handles_mixed_array_object_nesting() {
+    // Alternating levels (innermost-Array, then Object, then Array, ...). For N levels,
+    // cost = ceil(N/2)*ARRAY_FRAME_COST + floor(N/2)*OBJECT_FRAME_COST.
+    let build_alternating = |total_levels: usize| -> Value {
+        let mut value = Value::from("leaf");
+        for i in 0..total_levels {
+            if i % 2 == 0 {
+                value = Value::Array(vec![value]);
+            } else {
+                let mut map = ObjectMap::new();
+                map.insert("k".into(), value);
+                value = Value::Object(map);
+            }
+        }
+        value
+    };
+
+    // 38 alternating levels: 19 array (cost 38) + 19 object (cost 57) = 95 frames.
+    // Under the common Value budget of 96. Fits.
+    let mut event = LogEvent::from("flat");
+    *event.metadata_mut().value_mut() = build_alternating(38);
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        array.encode(&mut buf).is_ok(),
+        "nesting gate should accept 38 alternating metadata levels (cost 95)",
+    );
+
+    // 39 alternating levels: 20 array (cost 40) + 19 object (cost 57) = 97 frames.
+    // Over the common Value budget of 96. Fails.
+    let mut event = LogEvent::from("flat");
+    *event.metadata_mut().value_mut() = build_alternating(39);
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        matches!(
+            array.encode(&mut buf),
+            Err(super::super::ser::EncodeError::NestingTooDeep { .. })
+        ),
+        "nesting gate should reject 39 alternating metadata levels (cost 97)",
+    );
+}
+
+/// Verify the gate rejects a `Value::Timestamp` leaf sitting at the deepest object
+/// position the budget would otherwise allow, and that the underlying proto roundtrip
+/// would in fact fail there — confirming the timestamp leaf is not free.
+#[test]
+fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() {
+    let roundtrip_log = |value: Value| -> bool {
+        let mut event = LogEvent::default();
+        event.insert("data", value);
+        let array = EventArray::Logs(LogArray::from(vec![event]));
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        proto::EventArray::decode(buf.freeze()).is_ok()
+    };
+    let roundtrip_metadata = |value: Value| -> bool {
+        let mut event = LogEvent::from("flat");
+        *event.metadata_mut().value_mut() = value;
+        let array = EventArray::Logs(LogArray::from(vec![event]));
+        let proto_array = proto::EventArray::from(array);
+        let mut buf = BytesMut::with_capacity(65536);
+        proto_array.encode(&mut buf).unwrap();
+        proto::EventArray::decode(buf.freeze()).is_ok()
+    };
+
+    // `Log.fields` can carry 33 object levels (cost 99), but a Timestamp leaf raises
+    // that cost to 100 and fails decode. The gate rejects it under the common limit too.
+    let event_data_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf());
+    assert!(
+        !roundtrip_log(event_data_ts.clone()),
+        "depth {} with Timestamp leaf is expected to fail prost decode",
+        MAX_OBJECT_DEPTH_VALUE + 1,
+    );
+
+    let mut event = LogEvent::default();
+    event.insert("data", event_data_ts);
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        matches!(
+            array.encode(&mut buf),
+            Err(super::super::ser::EncodeError::NestingTooDeep { .. })
+        ),
+        "gate should reject event-data Timestamp leaf above the common budget",
+    );
+
+    // Metadata reaches its wire boundary at the common 32-object limit.
+    let metadata_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf());
+    assert!(
+        !roundtrip_metadata(metadata_ts.clone()),
+        "metadata depth {MAX_OBJECT_DEPTH_VALUE} with Timestamp leaf is expected to fail prost decode"
+    );
+
+    let mut event = LogEvent::from("flat");
+    *event.metadata_mut().value_mut() = metadata_ts;
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        matches!(
+            array.encode(&mut buf),
+            Err(super::super::ser::EncodeError::NestingTooDeep { .. })
+        ),
+        "gate should reject metadata Timestamp leaf at object depth {MAX_OBJECT_DEPTH_VALUE}",
+    );
+}
+
+/// Verify the gate still admits Timestamp leaves one level shallower than the boundary
+/// — they cost exactly one frame, no more — and that those payloads roundtrip cleanly
+/// through prost.
+#[test]
+fn nesting_gate_accepts_timestamp_leaf_below_max_object_depth() {
+    // One object level below the common limit plus a Timestamp leaf.
+    let mut event = LogEvent::default();
+    event.insert(
+        "data",
+        create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 2, ts_leaf()),
+    );
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        array.encode(&mut buf).is_ok(),
+        "gate should accept event-data Timestamp leaf at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE - 1,
+    );
+    assert!(
+        proto::EventArray::decode(buf.freeze()).is_ok(),
+        "prost should decode event-data Timestamp leaf at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE - 1,
+    );
+
+    // Metadata: one shallower.
+    let mut event = LogEvent::from("flat");
+    *event.metadata_mut().value_mut() =
+        create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf());
+    let array = EventArray::Logs(LogArray::from(vec![event]));
+    let mut buf = BytesMut::with_capacity(65536);
+    assert!(
+        array.encode(&mut buf).is_ok(),
+        "gate should accept metadata Timestamp leaf at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE - 1,
+    );
+    assert!(
+        proto::EventArray::decode(buf.freeze()).is_ok(),
+        "prost should decode metadata Timestamp leaf at object depth {}",
+        MAX_OBJECT_DEPTH_VALUE - 1,
+    );
+}
+
+/// Verify `filter_unencodable` keeps the valid events and drops only the over-budget
+/// ones, returning a smaller `EventArray` rather than failing the whole batch.
+#[test]
+fn filter_unencodable_drops_only_over_budget_events() {
+    let good = || {
+        let mut event = LogEvent::default();
+        event.insert("data", "ok");
+        event
+    };
+    let bad = || {
+        let mut event = LogEvent::default();
+        event.insert("data", create_nested_value(MAX_OBJECT_DEPTH_VALUE));
+        event
+    };
+
+    let logs = vec![good(), bad(), good(), bad(), good()];
+    let array = EventArray::Logs(LogArray::from(logs));
+
+    let filtered = array
+        .filter_unencodable()
+        .expect("3 good events should survive filtering");
+    assert_eq!(filtered.event_count(), 3, "only good events should remain");
+
+    let EventArray::Logs(surviving) = filtered else {
+        panic!("variant should be preserved");
+    };
+    for log in &surviving {
+        assert_eq!(
+            log.value().get("data").and_then(|v| v.as_bytes()),
+            Some(&bytes::Bytes::from_static(b"ok")),
+            "only good events should remain",
+        );
+    }
+}
+
+/// Verify that the public per-event entry point used by both the native codec and the
+/// vector sink charges `Value::Timestamp` for one frame, just like the buffer gate.
+/// Without this, a deep object chain ending in a timestamp could pass the codec check
+/// and fail prost decode on the receiving end.
+#[test]
+fn event_exceeds_max_nesting_cost_charges_timestamp_leaf() {
+    let log_at_max_with_ts = {
+        let mut event = LogEvent::default();
+        event.insert(
+            "data",
+            create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()),
+        );
+        Event::Log(event)
+    };
+    assert!(
+        event_exceeds_max_nesting_cost(&log_at_max_with_ts).is_some(),
+        "depth {MAX_OBJECT_DEPTH_VALUE} log with Timestamp leaf must be rejected",
+    );
+
+    let trace_at_max_with_ts = {
+        let mut trace = TraceEvent::default();
+        trace.insert(
+            "data",
+            create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()),
+        );
+        Event::Trace(trace)
+    };
+    assert!(
+        event_exceeds_max_nesting_cost(&trace_at_max_with_ts).is_some(),
+        "depth {MAX_OBJECT_DEPTH_VALUE} trace with Timestamp leaf must be rejected",
+    );
+
+    let metric_at_max_with_ts = {
+        let mut metric = Metric::new(
+            "test",
+            MetricKind::Incremental,
+            MetricValue::Counter { value: 1.0 },
+        );
+        *metric.metadata_mut().value_mut() =
+            create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf());
+        Event::Metric(metric)
+    };
+    assert!(
+        event_exceeds_max_nesting_cost(&metric_at_max_with_ts).is_some(),
+        "metric with metadata-Timestamp leaf at depth {MAX_OBJECT_DEPTH_VALUE} must be rejected",
+    );
+
+    // And one shallower stays under the budget.
+    let log_below_max_with_ts = {
+        let mut event = LogEvent::default();
+        event.insert(
+            "data",
+            create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 2, ts_leaf()),
+        );
+        Event::Log(event)
+    };
+    assert!(
+        event_exceeds_max_nesting_cost(&log_below_max_with_ts).is_none(),
+        "depth {} log with Timestamp leaf must be accepted",
+        MAX_OBJECT_DEPTH_VALUE - 1,
+    );
+}
+
+/// Unit-level check that `check_value_nesting_cost` charges `TIMESTAMP_FRAME_COST`
+/// for a `Value::Timestamp` leaf, independent of nesting.
+#[test]
+fn check_value_nesting_cost_charges_timestamp_leaf() {
+    let ts = ts_leaf();
+    assert!(check_value_nesting_cost(&ts, 0, TIMESTAMP_FRAME_COST).is_ok());
+    assert!(check_value_nesting_cost(&ts, 0, TIMESTAMP_FRAME_COST - 1).is_err());
+
+    // A single object level containing a timestamp leaf: OBJECT_FRAME_COST + TIMESTAMP_FRAME_COST.
+    let mut map = ObjectMap::new();
+    map.insert("ts".into(), ts);
+    let nested = Value::Object(map);
+    let expected = OBJECT_FRAME_COST + TIMESTAMP_FRAME_COST;
+    assert!(check_value_nesting_cost(&nested, 0, expected).is_ok());
+    assert!(check_value_nesting_cost(&nested, 0, expected - 1).is_err());
+}
+
+/// Verify flat events pass without issues.
+#[test]
+fn nesting_gate_accepts_flat_events() {
+    let mut log = LogEvent::from("hello world");
+    log.insert("foo", "bar");
+    let events = EventArray::Logs(LogArray::from(vec![log]));
+    let mut buf = BytesMut::with_capacity(1024);
+    assert!(events.encode(&mut buf).is_ok());
+
+    let mut trace = TraceEvent::default();
+    trace.insert("foo", "bar");
+    let events = EventArray::Traces(TraceArray::from(vec![trace]));
+    let mut buf = BytesMut::with_capacity(1024);
+    assert!(events.encode(&mut buf).is_ok());
+
+    let metric = Metric::new(
+        "test_counter",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    );
+    let events = EventArray::Metrics(MetricArray::from(vec![metric]));
+    let mut buf = BytesMut::with_capacity(1024);
+    assert!(events.encode(&mut buf).is_ok());
+}
+
+#[test]
+fn check_value_nesting_cost_with_configurable_budget() {
+    // Five nested objects: 5 levels × 3 frames per object = 15 frame cost.
+    let mut value = Value::from("leaf");
+    for _ in 0..5 {
+        let mut map = ObjectMap::new();
+        map.insert("n".into(), value);
+        value = Value::Object(map);
+    }
+
+    assert!(check_value_nesting_cost(&value, 0, 15).is_ok());
+    assert!(check_value_nesting_cost(&value, 0, 14).is_err());
+    assert!(check_value_nesting_cost(&value, 0, 30).is_ok());
+
+    let flat = Value::from("hello");
+    assert!(check_value_nesting_cost(&flat, 0, 0).is_ok());
+}
+
+#[test]
+fn check_value_nesting_cost_with_mixed_variants() {
+    // Outer array (2) → inner object (3) → inner array (2) → leaf = 7 frame cost.
+    let inner = Value::Array(vec![Value::from("leaf")]);
+    let mut map = ObjectMap::new();
+    map.insert("arr".into(), inner);
+    let value = Value::Array(vec![Value::Object(map)]);
+
+    assert!(check_value_nesting_cost(&value, 0, 7).is_ok());
+    assert!(check_value_nesting_cost(&value, 0, 6).is_err());
+}
diff --git a/src/sinks/vector/sink.rs b/src/sinks/vector/sink.rs
index b5d97541c74e7..7ca963b1b0470 100644
--- a/src/sinks/vector/sink.rs
+++ b/src/sinks/vector/sink.rs
@@ -6,7 +6,12 @@ use prost::Message;
 use tower::Service;
 use vector_lib::request_metadata::GroupedCountByteSize;
 use vector_lib::stream::{batcher::data::BatchReduce, BatcherSettings, DriverResponse};
-use vector_lib::{config::telemetry, ByteSizeOf, EstimatedJsonEncodedSizeOf};
+use vector_lib::{
+    config::telemetry,
+    event::event_exceeds_max_nesting_cost,
+    internal_event::{ComponentEventsDropped, UNINTENTIONAL},
+    ByteSizeOf, EstimatedJsonEncodedSizeOf,
+};
 
 use super::service::VectorRequest;
 use crate::{
@@ -57,6 +62,34 @@ where
 {
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         input
+            .filter_map(|event| {
+                std::future::ready(
+                    if let Some((cost, budget)) = event_exceeds_max_nesting_cost(&event) {
+                        let reason = format!(
+                            "Event nesting cost {cost} exceeds protobuf budget of {budget}."
+                        );
+                        emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                            count: 1,
+                            reason: &reason,
+                        });
+                        match event {
+                            Event::Log(log) => log
+                                .metadata()
+                                .update_status(vector_lib::event::EventStatus::Rejected),
+                            Event::Metric(metric) => metric
+                                .metadata()
+                                .update_status(vector_lib::event::EventStatus::Rejected),
+                            Event::Trace(trace) => trace
+                                .metadata()
+                                .update_status(vector_lib::event::EventStatus::Rejected),
+                        }
+
+                        None
+                    } else {
+                        Some(event)
+                    },
+                )
+            })
             .map(|mut event| {
                 let mut byte_size = telemetry().create_request_count_byte_size();
                 byte_size.add_event(&event, event.estimated_json_encoded_size_of());
@@ -116,3 +149,103 @@ where
         self.run_inner(input).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use bytes::BytesMut;
+    use prost::Message;
+    use vector_lib::event::{
+        event_exceeds_max_nesting_cost, Event, LogEvent, ObjectMap, Value, MAX_VALUE_NESTING_FRAMES,
+    };
+
+    use super::EventWrapper;
+    use crate::proto::vector as proto_vector;
+
+    fn build_nested_value(wrapping_levels: usize) -> Value {
+        let mut v = Value::from("leaf");
+        for _ in 0..wrapping_levels {
+            let mut m = ObjectMap::new();
+            m.insert("nested".into(), v);
+            v = Value::Object(m);
+        }
+        v
+    }
+
+    /// Empirical check: an event sitting *exactly* at the value budget accepted by
+    /// `event_exceeds_max_nesting_cost` must roundtrip through the vector sink's
+    /// actual wire shape — `PushEventsRequest { events: [EventWrapper] }` — and
+    /// not fail decode at the receiver. If this test fails, the value budget is
+    /// too high for the gRPC path and needs to be reduced for the outer request
+    /// wrapper.
+    #[test]
+    fn push_events_request_decode_at_value_budget() {
+        // 31 nested objects under "data" key → 32 effective object levels in
+        // `log.value()` (one outer Object from the inserted key), cost = 96.
+        let mut log = LogEvent::default();
+        log.insert("data", build_nested_value(31));
+        let event = Event::Log(log);
+        assert!(
+            event_exceeds_max_nesting_cost(&event).is_none(),
+            "test setup invariant: event must sit exactly at the value budget \
+             (cost {MAX_VALUE_NESTING_FRAMES})",
+        );
+
+        let request = proto_vector::PushEventsRequest {
+            events: vec![EventWrapper::from(event)],
+        };
+
+        let mut buf = BytesMut::with_capacity(65536);
+        request.encode(&mut buf).expect("encode should succeed");
+
+        proto_vector::PushEventsRequest::decode(buf.freeze())
+            .expect("PushEventsRequest decode should succeed at the accepted value budget");
+    }
+
+    /// An object-root log one step past the common budget still fits the wider
+    /// `Log.fields` wire path, but the sink gate applies the same limit to every Value.
+    #[test]
+    fn push_events_request_rejects_one_past_common_value_budget() {
+        // 32 nested objects under "data" → 33 effective object levels, cost 99.
+        let mut log = LogEvent::default();
+        log.insert("data", build_nested_value(32));
+        let event = Event::Log(log);
+        assert_eq!(
+            event_exceeds_max_nesting_cost(&event),
+            Some((MAX_VALUE_NESTING_FRAMES + 3, MAX_VALUE_NESTING_FRAMES)),
+        );
+
+        let request = proto_vector::PushEventsRequest {
+            events: vec![EventWrapper::from(event)],
+        };
+
+        let mut buf = BytesMut::with_capacity(65536);
+        request.encode(&mut buf).expect("encode should succeed");
+
+        proto_vector::PushEventsRequest::decode(buf.freeze()).expect(
+            "the object-root wire path can decode 99 frames even though the common \
+             Value gate rejects it",
+        );
+    }
+
+    #[test]
+    fn push_events_request_decode_with_metadata_at_value_budget() {
+        let mut log = LogEvent::from("flat");
+        *log.metadata_mut().value_mut() = build_nested_value(32);
+        let event = Event::Log(log);
+        assert!(
+            event_exceeds_max_nesting_cost(&event).is_none(),
+            "test setup invariant: metadata must sit exactly at the Value budget \
+             (cost {MAX_VALUE_NESTING_FRAMES})",
+        );
+
+        let request = proto_vector::PushEventsRequest {
+            events: vec![EventWrapper::from(event)],
+        };
+
+        let mut buf = BytesMut::with_capacity(65536);
+        request.encode(&mut buf).expect("encode should succeed");
+
+        proto_vector::PushEventsRequest::decode(buf.freeze())
+            .expect("PushEventsRequest decode should succeed at the accepted metadata budget");
+    }
+}