diff --git a/litellm/integrations/newrelic/newrelic_metrics.py b/litellm/integrations/newrelic/newrelic_metrics.py
index 7db26eff0fc..25dbfc2bdb2 100644
--- a/litellm/integrations/newrelic/newrelic_metrics.py
+++ b/litellm/integrations/newrelic/newrelic_metrics.py
@@ -207,17 +207,16 @@ class NewRelicMetricsLogger(CustomBatchLogger):
         of racing it. Each pass attempts the whole current queue in
         ``batch_size`` chunks, unlike the periodic path it does not stop at the
         first failing chunk, so a persistently failing head never starves the
-        tail, and a record appended mid-drain is snapshotted and delivered by a
-        later pass. Only after ``_MAX_DRAIN_PASSES`` against a permanently
-        failing destination is the remainder dropped, and then only the records
-        this drain actually tried: a record a callback appended after our last
-        snapshot is left for its own serialized drain, so it is never wiped
-        un-tried and never stranded.
+        tail. Only after ``_MAX_DRAIN_PASSES`` against a permanently failing
+        destination is the remainder dropped, and then only the records that were
+        queued when this drain began, so every dropped record got the full retry
+        budget: a record a callback appended mid-drain is not in that snapshot,
+        so it is left for its own serialized drain rather than dropped after
+        fewer attempts, and is never stranded.
         """
         async with self._drain_lock:
-            attempted: tuple[NewRelicMetricRecord, ...] = ()  # rebind-ok: each pass's pre-attempt snapshot
+            attempted: Final = tuple(self.log_queue)
             for _pass in range(NEWRELIC_METRICS_MAX_DRAIN_PASSES):
-                attempted = tuple(self.log_queue)
                 await self._drain_flush_once()
                 if not self.log_queue:
                     return
diff --git a/tests/test_litellm/integrations/newrelic/test_newrelic_metrics.py b/tests/test_litellm/integrations/newrelic/test_newrelic_metrics.py
index f28e030f77f..9c75e0b0a47 100644
--- a/tests/test_litellm/integrations/newrelic/test_newrelic_metrics.py
+++ b/tests/test_litellm/integrations/newrelic/test_newrelic_metrics.py
@@ -716,6 +716,34 @@ async def test_terminal_drop_leaves_untried_late_arrival_for_next_drain():
     assert logger.log_queue == [late_record], "the un-tried late arrival is left for its own drain, not dropped"
 
 
+@pytest.mark.asyncio
+async def test_record_appended_on_an_early_pass_is_not_dropped_short_of_the_retry_budget():
+    """A record a callback appends during an early drain pass entered the queue
+    after this drain's snapshot, so it has not seen the full retry budget. The
+    terminal drop must clear only records queued when the drain began, leaving
+    the early-pass arrival for its own serialized drain instead of dropping it
+    after fewer than the configured attempts."""
+    logger = _make_logger()
+    logger.stop()
+    early_record = _record(model="early-pass-arrival")
+    posts = {"n": 0}
+
+    async def _fail_and_append_on_first_pass(url, data=None, headers=None, **kw):
+        posts["n"] += 1
+        # One record queued at start means the first pass's post is the 1st;
+        # append during it, before this drain's later passes.
+        if posts["n"] == 1:
+            logger.log_queue.append(early_record)
+        resp = _response(503)
+        raise HTTPStatusError("err", request=resp.request, response=resp)
+
+    with patch("asyncio.sleep", new=AsyncMock()):
+        logger.async_client.post = _fail_and_append_on_first_pass
+        logger.log_queue.append(_record(model="doomed"))
+        await logger._drain_with_retry()
+    assert logger.log_queue == [early_record], "the early-pass arrival is left for its own drain, not dropped short"
+
+
 @pytest.mark.asyncio
 async def test_post_stop_drains_are_serialized():
     """A callback that appends to a stopped logger and starts its own drain must