diff --git a/litellm/router_rust/core.py b/litellm/router_rust/core.py
index 3e222313f32..eef454862f9 100644
--- a/litellm/router_rust/core.py
+++ b/litellm/router_rust/core.py
@@ -4378,6 +4378,9 @@ class _Driver:
         self._hop_first: Final[dict[int, BaseException]] = {}  # mutable-ok: filled as the call runs
         # Each level's latest pick error: the one object it logged, recorded and raises.
         self._level_picks: Final[dict[int, BaseException]] = {}  # mutable-ok: filled as the call runs
+        # Failures a retry or a later hop raised past the request's first logged failure, which the
+        # attempt itself recorded: the failed hop's fallback cooldown leaves them alone.
+        self._recorded_blocked: Final[set[int]] = set()  # mutable-ok: filled as the call runs
         self._served: tuple[str | None, Mapping[str, object]] = (None, MappingProxyType({}))
         # The latest attempt's deployment and call, which a blocked response is raised from.
         self._latest_call: tuple[Mapping[str, object], AttemptCall] | None = None
@@ -4804,7 +4807,10 @@ class _Driver:
                     self._level_errors.get(at + 1),
                     self._walk.hop_kwargs(event),
                 )
-                if event.get("succeeded") is not True:
+                if (
+                    event.get("succeeded") is not True
+                    and id(self._level_errors.get(at + 1)) not in self._recorded_blocked
+                ):
                     self._core.trigger_fallback_cooldown(
                         self._request,
                         self._level_errors.get(at + 1),
@@ -4967,6 +4973,28 @@ class _Driver:
             or _raised_before_provider(error),
         }
 
+    def _records_blocked(
+        self, failed: Mapping[str, object], call: "AttemptCall | None", error: BaseException, in_call: bool
+    ) -> bool:
+        """Record a failure the shared logging object skipped (a retry or later hop), like the first one."""
+        from litellm.router_utils.fallback_event_handlers import (
+            _failure_callbacks_blocked_logging_object,  # pyright: ignore[reportPrivateUsage]  # the fallback path's own rule
+        )
+
+        if not in_call or call is None or failed.get("outside_deployment") is True:
+            return False
+        if self._plan.sync:
+            if not failure_already_logged(call.kwargs, "sync_failure"):
+                return False
+        elif _failure_callbacks_blocked_logging_object(call.kwargs) is None:
+            return False
+        if not self._core.outcome_callbacks(call.kwargs, sync=self._plan.sync).get("failure", True):
+            return False
+        if self._core.blocked_failure_exempt(error, call.kwargs, self._plan.kind in _GENERIC_API_KINDS):
+            return False
+        self._recorded_blocked.add(id(error))
+        return True
+
     def _caller_entra_failure(self, error: BaseException, sent: Mapping[str, object], deployment_id: object) -> bool:
         """``is_caller_entra_credential_failure``: Entra ID rejected a credential the caller sent, not
         the deployment's own, so the failure is the caller's."""
@@ -5128,6 +5156,7 @@ class _Driver:
         # Whether a failure raised now runs the failure callbacks: none raised before the litellm
         # call or after it answered does, nor any but the first on a shared logging object.
         logs_failure: bool = False  # rebind-ok: set around the litellm call
+        in_call: bool = False  # rebind-ok: set around the litellm call, whichever attempt logs its failure
         reading_stream: StreamAttempt | None = None  # rebind-ok: set while the first chunk is read
         refused_call: Final = self._default_call_error(step)
         if refused_call is not None:
@@ -5160,6 +5189,7 @@ class _Driver:
                         return await self._refused(step, deployment, call, refused)
                 first_to_log: Final = not failure_already_logged(call.kwargs, "async_failure")
                 logs_failure = first_to_log  # rebind-ok: whether a failure here is this attempt's to log
+                in_call = True  # rebind-ok: see above
                 pending: Final[object] = (
                     self._core._endpoints.create_file_everywhere(  # pyright: ignore[reportPrivateUsage]  # see above
                         _with_target_order(call.kwargs, step.get("target_order")),
@@ -5173,12 +5203,15 @@ class _Driver:
                 awaited: Final = self._plan.kind == "generic" or inspect.isawaitable(pending)
                 answered: Final[object] = await pending if awaited else pending  # pyright: ignore[reportGeneralTypeIssues, reportUnknownVariableType]  # see above
                 logs_failure = False  # rebind-ok: see above
+                in_call = False  # rebind-ok: see above
                 self._core._endpoints.check_response(self._plan, answered, call)  # pyright: ignore[reportPrivateUsage, reportUnknownArgumentType]  # see above
                 stream: Final = self._stream_attempt(step, call.kwargs, slot)
                 logs_failure = first_to_log  # rebind-ok: the stream's own failure logging, for its first chunk
+                in_call = True  # rebind-ok: see above
                 reading_stream = stream  # rebind-ok: None until the attempt opens a stream
                 response: Final = await open_attempt_stream(stream, answered)  # pyright: ignore[reportUnknownArgumentType]  # see above
                 logs_failure = False  # rebind-ok: see above
+                in_call = False  # rebind-ok: see above
                 # `self.success_calls[model_name] += 1`, inside the try: a counter replaced with a
                 # plain dict fails the attempt.
                 self._core._endpoints.counters.succeeded(self._plan, call)  # pyright: ignore[reportPrivateUsage]  # see above
@@ -5194,7 +5227,10 @@ class _Driver:
                     )
             failed: Final = self._failed(step, deployment, call, error)
             if reading_stream is None or not logs_failure:
-                return {**failed, "recorded": not logs_failure}
+                return {
+                    **failed,
+                    "recorded": not logs_failure and not self._records_blocked(failed, call, error, in_call),
+                }
             # `_dispatch_stream_failure`: the stream logs its failure in a task of its own.
             reading_stream.report_first_chunk_failure(error)
             return {**failed, "recorded": True}
@@ -5212,6 +5248,7 @@ class _Driver:
         deployment: Mapping[str, object] = as_str_mapping(step.get("deployment")) or {}  # rebind-ok: see attempt
         call: AttemptCall | None = None  # rebind-ok: see attempt
         logs_failure: bool = False  # rebind-ok: see attempt
+        in_call: bool = False  # rebind-ok: see attempt
         self._reading_first_chunk = False
         refused_call: Final = self._default_call_error(step)
         if refused_call is not None:
@@ -5228,22 +5265,27 @@ class _Driver:
                 return {**self._failed(step, deployment, call, refused), "recorded": True}
             first_to_log: Final = not failure_already_logged(call.kwargs, "sync_failure")
             logs_failure = first_to_log  # rebind-ok: see attempt
+            in_call = True  # rebind-ok: see attempt
             answered: Final = self._core._host.invoke(call.host_op)(**call.kwargs)  # pyright: ignore[reportPrivateUsage]  # a sync endpoint
             logs_failure = False  # rebind-ok: see attempt
+            in_call = False  # rebind-ok: see attempt
             if inspect.isawaitable(answered):
                 raise _NotSynchronous(f"the host's {call.host_op} endpoint is not synchronous")
             self._core._endpoints.check_response(self._plan, answered, call)  # pyright: ignore[reportPrivateUsage]  # see above
             stream: Final = self._stream_attempt(step, call.kwargs)
             logs_failure = first_to_log  # rebind-ok: see attempt
+            in_call = True  # rebind-ok: see attempt
             self._reading_first_chunk = True
             response: Final = open_attempt_stream_sync(stream, answered)
             self._reading_first_chunk = False
             logs_failure = False  # rebind-ok: see attempt
+            in_call = False  # rebind-ok: see attempt
             self._core._endpoints.counters.succeeded(self._plan, call)  # pyright: ignore[reportPrivateUsage]  # see attempt
         except _NotSynchronous:
             raise
         except Exception as error:  # noqa: BLE001  # see attempt
-            return {**self._failed(step, deployment, call, error), "recorded": not logs_failure}
+            failed: Final = self._failed(step, deployment, call, error)
+            return {**failed, "recorded": not logs_failure and not self._records_blocked(failed, call, error, in_call)}
         return self._succeeded(call, response, "", stream)
 
     async def select(self, step: Mapping[str, object]) -> object:
diff --git a/litellm/router_rust/surface/health.py b/litellm/router_rust/surface/health.py
index 0653518d0d7..80f7e148185 100644
--- a/litellm/router_rust/surface/health.py
+++ b/litellm/router_rust/surface/health.py
@@ -298,6 +298,33 @@ class RustHealthCallbacks:
             "caller_scoped": _is_per_user_provider_request(logged) or self.per_user_deployment(deployment_id),
         }
 
+    def blocked_failure_exempt(self, exception: BaseException, kwargs: Mapping[str, object], generic: bool) -> bool:
+        """The exits of ``_trigger_cooldown_for_failed_deployment`` for a failure the request's shared
+        logging object does not log: the selection tag is not one of them; a stream stall and a
+        generic call's 404 are."""
+        from litellm.litellm_core_utils.litellm_logging import Logging
+        from litellm.router_utils.cooldown_handlers import (
+            cast_exception_status_to_int,
+            is_advisor_orchestration_failure,
+            is_caller_timeout_408,
+            is_stream_stall,
+        )
+        from litellm.router_utils.fallback_event_handlers import (
+            _REQUEST_SCOPED_STATUS_CODES,  # pyright: ignore[reportPrivateUsage]  # the fallback path's own rule
+        )
+
+        exception_status: Final[object] = getattr(exception, "status_code", "")
+        logging_obj: Final = kwargs.get("litellm_logging_obj")
+        details: Final[Mapping[str, object]] = (
+            logging_obj.model_call_details if isinstance(logging_obj, Logging) else {}
+        )
+        return (
+            (isinstance(exception, Exception) and is_advisor_orchestration_failure(exception))
+            or is_stream_stall(exception)
+            or (generic and cast_exception_status_to_int(exception_status) in _REQUEST_SCOPED_STATUS_CODES)  # pyright: ignore[reportArgumentType]  # the status as raised
+            or is_caller_timeout_408(details, exception_status, ended=datetime.now())  # pyright: ignore[reportArgumentType]  # noqa: DTZ005  # see logged_call
+        )
+
     def trigger_fallback_cooldown(
         self, kwargs: Mapping[str, object], exception: object, generic: bool, *, own_loop: bool = False
     ) -> None:
diff --git a/tests/unit/test_router/core/test_usage_and_failure_counting_parity.py b/tests/unit/test_router/core/test_usage_and_failure_counting_parity.py
index 8e79303eec7..3ef5b0c10c7 100644
--- a/tests/unit/test_router/core/test_usage_and_failure_counting_parity.py
+++ b/tests/unit/test_router/core/test_usage_and_failure_counting_parity.py
@@ -81,6 +81,16 @@ class SharedLoggingHost(EndpointHost):
         details["async_failure_logged_attempt"] = details.get("current_attempt_api_call_start_time")
         await super().log_async(call, error, response)
 
+    def log_sync(self, call: Call, error: BaseException | None, response: object) -> None:
+        logging_obj: Final = call.kwargs.get("litellm_logging_obj")
+        if error is None or not isinstance(logging_obj, LiteLLMLogging):
+            super().log_sync(call, error, response)
+            return
+        if logging_obj.model_call_details.get("has_logged_sync_failure") is True:
+            return
+        logging_obj.model_call_details["has_logged_sync_failure"] = True
+        super().log_sync(call, error, response)
+
 
 @pytest.fixture
 def host() -> SharedLoggingHost:
@@ -165,7 +175,7 @@ async def test_a_realtime_session_counts_no_request_against_a_limited_deployment
 # --- P3: what a shared logging object lets through -------------------------------------------------
 
 
-async def test_only_the_first_failure_on_a_shared_logging_object_counts_a_request(
+async def test_every_failure_on_a_shared_logging_object_counts_a_request(
     make_core: MakeCore, host: SharedLoggingHost
 ) -> None:
     core: Final = chain(make_core)
@@ -174,8 +184,8 @@ async def test_only_the_first_failure_on_a_shared_logging_object_counts_a_reques
     _ = await attempt(core, "a", litellm_logging_obj=request_logging())
 
     assert await core.health().usage("a") == (None, 1)
-    assert await core.health().usage("b") == (None, None)
-    assert await core.health().usage("c") == (None, None)
+    assert await core.health().usage("b") == (None, 1)
+    assert await core.health().usage("c") == (None, 1)
 
 
 async def test_every_failure_counts_a_request_when_each_call_logs_on_its_own(
@@ -191,13 +201,55 @@ async def test_every_failure_counts_a_request_when_each_call_logs_on_its_own(
     assert await core.health().usage("c") == (None, 1)
 
 
-async def test_a_blocked_retry_in_the_same_group_records_nothing(make_core: MakeCore, host: SharedLoggingHost) -> None:
+async def test_a_retry_past_the_first_logged_failure_is_recorded(make_core: MakeCore, host: SharedLoggingHost) -> None:
     core: Final = make_core(model_list=[deployment("g", "a"), deployment("g", "b")], num_retries=1, allowed_fails=5)
     host.answer("*", fresh(server_error))
 
     _ = await attempt(core, "g", litellm_logging_obj=request_logging())
 
-    assert core.health().failures_this_minute("a") + core.health().failures_this_minute("b") == 1
+    assert core.health().failures_this_minute("a") + core.health().failures_this_minute("b") == 2
+
+
+def test_a_sync_retry_past_the_first_logged_failure_is_recorded(make_core: MakeCore, host: SharedLoggingHost) -> None:
+    core: Final = make_core(model_list=[deployment("g", "a"), deployment("g", "b")], num_retries=1, allowed_fails=5)
+    host.answer("*", fresh(server_error))
+
+    _ = attempt_sync(core, "g", litellm_logging_obj=request_logging())
+
+    assert core.health().failures_this_minute("a") + core.health().failures_this_minute("b") == 2
+
+
+async def test_a_retried_deployment_is_cooled_down_past_the_first_logged_failure(
+    make_core: MakeCore, host: SharedLoggingHost
+) -> None:
+    core: Final = make_core(
+        model_list=[deployment("g", "a"), deployment("g", "b")], num_retries=1, allowed_fails=0, cooldown_time=60
+    )
+    host.answer("*", fresh(rate_limited))
+
+    _ = await attempt(core, "g", litellm_logging_obj=request_logging())
+
+    assert sorted(cooled_down(core, ["a", "b"])) == ["a", "b"]
+
+
+async def test_a_hop_whose_group_is_cooled_records_its_first_failure(
+    make_core: MakeCore, host: SharedLoggingHost
+) -> None:
+    core: Final = make_core(
+        model_list=[deployment("a", "a"), deployment("b", "b"), deployment("c", "c")],
+        fallbacks=[{"a": ["b"]}, {"b": ["c"]}],
+        num_retries=0,
+        allowed_fails=0,
+        cooldown_time=60,
+    )
+    host.answer("openai/a", fresh(server_error))
+    host.answer("openai/b", fresh(rate_limited))
+    _ = await attempt(core, "a", litellm_logging_obj=request_logging())
+    assert sorted(cooled_down(core, ["a", "b"])) == ["a", "b"]
+
+    _ = await attempt(core, "a", litellm_logging_obj=request_logging())
+
+    assert core.health().failures_this_minute("b") == 1
 
 
 def test_a_sync_failure_counts_no_request(make_core: MakeCore, host: SharedLoggingHost) -> None:
@@ -253,12 +305,22 @@ def selection_failure() -> Exception:
 
 
 @pytest.mark.parametrize(
-    ("builder", "stalled", "counted"),
-    [(server_error, False, 1), (selection_failure, False, 1), (timeout, True, 0), (timeout, False, 1)],
+    ("builder", "stalled", "counted", "requests"),
+    [
+        (server_error, False, 1, 1),
+        (selection_failure, False, 1, None),
+        (timeout, True, 0, None),
+        (timeout, False, 1, 1),
+    ],
     ids=["server-error", "selection-tag-not-exempt", "stream-stall-exempt", "plain-timeout"],
 )
 async def test_a_failed_hop_past_the_first_logged_failure_takes_the_fallback_paths_exits(
-    make_core: MakeCore, host: SharedLoggingHost, builder: ErrorBuilder, stalled: bool, counted: int
+    make_core: MakeCore,
+    host: SharedLoggingHost,
+    builder: ErrorBuilder,
+    stalled: bool,
+    counted: int,
+    requests: int | None,
 ) -> None:
     core: Final = make_core(
         model_list=[deployment("a", "a"), deployment("b", "b")], fallbacks=[{"a": ["b"]}], num_retries=0
@@ -271,7 +333,7 @@ async def test_a_failed_hop_past_the_first_logged_failure_takes_the_fallback_pat
     _ = await attempt(core, "a", litellm_logging_obj=request_logging())
 
     assert core.health().failures_this_minute("b") == counted
-    assert await core.health().usage("b") == (None, None)
+    assert await core.health().usage("b") == (None, requests)
 
 
 @pytest.mark.parametrize(("logged", "counted"), [(True, 0), (False, 1)], ids=["generic", "chat"])