diff --git a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py
index a6d59e1..bb9f496 100644
--- a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py
@@ -80,6 +80,7 @@ def artifacts_get(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="artifact commands",
+                logger=logger,
             )
             if require_available := getattr(client, "require_available", None):
                 require_available()
diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py
index 060c960..9b4c118 100644
--- a/packages/tangle-cli/src/tangle_cli/client.py
+++ b/packages/tangle-cli/src/tangle_cli/client.py
@@ -40,6 +40,29 @@
 )
 
 
+class _RetryBudget:
+    """Shared attempt/deadline budget for one logical request.
+
+    The transient-5xx, 429 rate-limit, and 401 auth-refresh retry layers all
+    draw from a single instance so a composed outage cannot multiply their
+    per-layer limits into a large physical request count. ``remaining`` counts
+    the physical requests still permitted; ``deadline`` is a ``time.monotonic``
+    value past which no further retry is attempted.
+    """
+
+    __slots__ = ("remaining", "deadline")
+
+    def __init__(self, max_attempts: int, deadline: float) -> None:
+        self.remaining = max_attempts
+        self.deadline = deadline
+
+    def consume(self) -> None:
+        self.remaining -= 1
+
+    def can_retry(self) -> bool:
+        return self.remaining > 0 and time.monotonic() < self.deadline
+
+
 class TangleApiClient(GeneratedTangleApiOperations):
     """Single public API wrapper for Tangle backends.
 
@@ -50,9 +73,19 @@ class TangleApiClient(GeneratedTangleApiOperations):
 
     _REDIRECT_STATUSES = {301, 302, 303, 307, 308}
     _MAX_REDIRECTS = 5
-    _MAX_RATE_LIMIT_RETRIES = 3
     _RATE_LIMIT_BACKOFF_SECONDS = 1.0
     _MAX_RETRY_AFTER_SECONDS = 60.0
+    _RETRYABLE_GET_STATUSES = frozenset({500, 502, 503, 504})
+    _MAX_GET_RETRIES = 6
+    _GET_RETRY_BACKOFF_SECONDS = 1.0
+    _MAX_GET_RETRY_BACKOFF_SECONDS = 30.0
+    # A single logical request may issue at most ``_MAX_GET_RETRIES + 1``
+    # physical requests total, shared across the transient-5xx, 429 rate-limit,
+    # and 401 auth-refresh layers, and must not spend more than
+    # ``_MAX_RETRY_ELAPSED_SECONDS`` retrying. One shared budget prevents the
+    # layers from multiplying into a large physical request count during an
+    # outage (e.g. interleaved 503/429 responses, or a 401 mid-sequence).
+    _MAX_RETRY_ELAPSED_SECONDS = 120.0
 
     def __init__(
         self,
@@ -121,6 +154,11 @@ def _make_request(
         clean_params = self._clean_mapping(params)
         request_method = method.upper()
 
+        budget = _RetryBudget(
+            self._MAX_GET_RETRIES + 1,
+            time.monotonic() + self._MAX_RETRY_ELAPSED_SECONDS,
+        )
+
         self._refresh_auth()
         response = self._request_with_rate_limit_retries(
             request_method,
@@ -130,8 +168,11 @@ def _make_request(
             extra_headers=extra_headers,
             timeout=timeout,
             request_kwargs=kwargs,
+            budget=budget,
         )
-        if response.status_code == 401:
+        # The auth-refresh retry draws from the same budget, so a 401 late in a
+        # transient/rate-limit sequence cannot start a fresh round of retries.
+        if response.status_code == 401 and budget.can_retry():
             self._refresh_auth()
             response = self._request_with_rate_limit_retries(
                 request_method,
@@ -141,6 +182,7 @@ def _make_request(
                 extra_headers=extra_headers,
                 timeout=timeout,
                 request_kwargs=kwargs,
+                budget=budget,
             )
         return response
 
@@ -154,10 +196,11 @@ def _request_with_rate_limit_retries(
         extra_headers: Mapping[str, str] | None,
         timeout: float,
         request_kwargs: Mapping[str, Any],
+        budget: _RetryBudget,
     ) -> requests.Response:
-        response: requests.Response | None = None
-        for attempt in range(self._MAX_RATE_LIMIT_RETRIES + 1):
-            response = self._request_with_same_origin_redirects(
+        rate_limit_round = 0
+        while True:
+            response = self._request_with_transient_retries(
                 method,
                 url,
                 params=params,
@@ -165,11 +208,105 @@ def _request_with_rate_limit_retries(
                 extra_headers=extra_headers,
                 timeout=timeout,
                 request_kwargs=request_kwargs,
+                budget=budget,
             )
-            if response.status_code != 429 or attempt == self._MAX_RATE_LIMIT_RETRIES:
+            # A 429 retry re-enters the transient layer, so it must draw from the
+            # shared budget rather than a per-round allowance.
+            if response.status_code != 429 or not budget.can_retry():
                 return response
-            self._sleep_for_rate_limit(response, attempt)
-        return response
+            self._sleep_for_rate_limit(response, rate_limit_round)
+            rate_limit_round += 1
+
+    def _request_with_transient_retries(
+        self,
+        method: str,
+        url: str,
+        *,
+        params: Mapping[str, Any] | None,
+        json_data: Any,
+        extra_headers: Mapping[str, str] | None,
+        timeout: float,
+        request_kwargs: Mapping[str, Any],
+        budget: _RetryBudget,
+    ) -> requests.Response:
+        """Retry idempotent GETs on transient 5xx and transport errors.
+
+        Mutating methods are sent once (never duplicated). Streamed GETs bypass
+        this layer so their consumer owns any stream-open retries. 429s are left
+        to the rate-limit layer, whose retries re-enter this layer with a fresh
+        backoff. ``SSLError`` raises immediately: certificate failures are
+        deterministic, so retrying only delays the report. Every physical send
+        draws from the shared ``budget`` so the transient, rate-limit, and
+        auth-refresh layers cannot multiply into a large request count. Each
+        doubling sleep is capped at ``_MAX_GET_RETRY_BACKOFF_SECONDS`` and
+        announced through ``self.logger`` (a null logger on non-verbose clients
+        built without one), so a stalled GET is bounded.
+        """
+
+        if method.upper() != "GET" or request_kwargs.get("stream"):
+            budget.consume()
+            return self._request_with_same_origin_redirects(
+                method,
+                url,
+                params=params,
+                json_data=json_data,
+                extra_headers=extra_headers,
+                timeout=timeout,
+                request_kwargs=request_kwargs,
+            )
+
+        backoff = self._GET_RETRY_BACKOFF_SECONDS
+        attempt = 0
+        while True:
+            budget.consume()
+            attempt += 1
+            try:
+                response = self._request_with_same_origin_redirects(
+                    method,
+                    url,
+                    params=params,
+                    json_data=json_data,
+                    extra_headers=extra_headers,
+                    timeout=timeout,
+                    request_kwargs=request_kwargs,
+                )
+            # Transient transport failures (reset/refused, timeout, truncated or
+            # corrupt body) can succeed on retry; other request errors surface.
+            except (
+                requests.ConnectionError,
+                requests.Timeout,
+                requests.exceptions.ChunkedEncodingError,
+                requests.exceptions.ContentDecodingError,
+            ) as exc:
+                # SSLError subclasses ConnectionError but signals a certificate
+                # or TLS configuration problem that no retry can fix.
+                if isinstance(exc, requests.exceptions.SSLError):
+                    raise
+                # Budget exhausted (attempts or deadline): surface the failure.
+                if not budget.can_retry():
+                    raise
+                self._sleep_for_transient_retry(backoff, attempt, type(exc).__name__)
+            else:
+                if response.status_code not in self._RETRYABLE_GET_STATUSES:
+                    return response
+                # Budget exhausted: return the final 5xx for raise_for_status.
+                if not budget.can_retry():
+                    return response
+                # Release the intermediate response so its connection returns to the pool.
+                try:
+                    response.close()
+                except Exception:
+                    pass
+                self._sleep_for_transient_retry(backoff, attempt, f"HTTP {response.status_code}")
+            backoff *= 2.0
+
+    def _sleep_for_transient_retry(self, backoff: float, attempt: int, reason: str) -> None:
+        delay = min(backoff, self._MAX_GET_RETRY_BACKOFF_SECONDS)
+        self.logger.warn(
+            f"transient {reason} on GET; retrying in {delay:.1f}s "
+            f"(attempt {attempt + 1}/{self._MAX_GET_RETRIES + 1})"
+        )
+        time.sleep(delay)
 
     def _sleep_for_rate_limit(self, response: requests.Response, attempt: int) -> None:
         retry_after = response.headers.get("Retry-After")
diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
index 2da0eb5..5394f72 100644
--- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
@@ -69,7 +69,9 @@ def _allow_all_hydration_for_args(args: ArgsContainer) -> bool:
     return bool(config.get("allow_all", False))
 
 
-def _api_client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str) -> LazyTangleApiClient:
+def _api_client(
+    args: ArgsContainer, *, cli_base_url: str | None, command_name: str, logger: Logger | None = None
+) -> LazyTangleApiClient:
     return LazyTangleApiClient(
         base_url=args.base_url,
         token=args.token,
@@ -77,12 +79,15 @@ def _api_client(args: ArgsContainer, *, cli_base_url: str | None, command_name:
         header=args.header,
         include_env_credentials=include_env_credentials_for_args(args, cli_base_url),
         command_name=command_name,
+        logger=logger,
     )
 
 
 def _manager(args: ArgsContainer, *, cli_base_url: str | None, logger: Logger) -> PipelineRunManager:
     return PipelineRunManager(
-        client=_api_client(args, cli_base_url=cli_base_url, command_name="pipeline-run commands"),
+        client=_api_client(
+            args, cli_base_url=cli_base_url, command_name="pipeline-run commands", logger=logger
+        ),
         hooks=PipelineRunHooks(
             logger=logger,
             trusted_python_sources=_trusted_sources_for_args(args),
@@ -117,7 +122,12 @@ def _run_annotation_action(config: str | None, cli_base_url: str | None, specs:
             raise SystemExit(str(exc)) from exc
         try:
             manager = AnnotationManager(
-                client=_api_client(args, cli_base_url=cli_base_url, command_name="pipeline-run annotation commands"),
+                client=_api_client(
+                    args,
+                    cli_base_url=cli_base_url,
+                    command_name="pipeline-run annotation commands",
+                    logger=logger,
+                ),
                 logger=logger,
             )
             print_json(fn(manager, args))
diff --git a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py
index 3a30e57..4ce9006 100644
--- a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py
@@ -232,6 +232,7 @@ def pipelines_hydrate(
                 ),
                 header=_header_entries(header, config_values),
                 include_env_credentials=include_env_credentials,
+                logger=logger,
             ),
         )
     except PipelineValidationError as exc:
diff --git a/packages/tangle-cli/src/tangle_cli/published_components_cli.py b/packages/tangle-cli/src/tangle_cli/published_components_cli.py
index de4f70a..733eb02 100644
--- a/packages/tangle-cli/src/tangle_cli/published_components_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/published_components_cli.py
@@ -24,7 +24,7 @@
     TokenOption,
 )
 from .component_publisher import ComponentPublisher, deprecate_component
-from .logger import logger_for_log_type
+from .logger import Logger, logger_for_log_type
 
 
 def _client_from_options(
@@ -35,6 +35,7 @@ def _client_from_options(
     header: list[str] | str | None = None,
     include_env_credentials: bool = True,
     command_name: str = "published-component commands",
+    logger: Logger | None = None,
 ) -> LazyTangleApiClient:
     """Create a lazy static client proxy for published-component commands.
 
@@ -49,6 +50,7 @@ def _client_from_options(
         header=header,
         include_env_credentials=include_env_credentials,
         command_name=command_name,
+        logger=logger,
     )
 
 
@@ -97,6 +99,7 @@ def published_components_search(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="published-component commands",
+                logger=logger,
             )
             if require_available := getattr(client, "require_available", None):
                 require_available()
@@ -163,6 +166,7 @@ def published_components_inspect(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="published-component commands",
+                logger=logger,
             )
             if require_available := getattr(client, "require_available", None):
                 require_available()
@@ -219,6 +223,7 @@ def published_components_library(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="published-component commands",
+                logger=logger,
             )
             if require_available := getattr(client, "require_available", None):
                 require_available()
@@ -288,6 +293,7 @@ def published_components_publish(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="published-component commands",
+                logger=logger,
             )
             publisher = ComponentPublisher(
                 dry_run=bool(args.dry_run),
@@ -358,6 +364,7 @@ def published_components_deprecate(
                 header=args.header,
                 include_env_credentials=include_env_credentials_for_args(args, base_url),
                 command_name="published-component commands",
+                logger=logger,
             )
             result = deprecate_component(
                 client,
diff --git a/packages/tangle-cli/src/tangle_cli/secrets_cli.py b/packages/tangle-cli/src/tangle_cli/secrets_cli.py
index d6c2da1..48ba9ff 100644
--- a/packages/tangle-cli/src/tangle_cli/secrets_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/secrets_cli.py
@@ -58,7 +58,9 @@
 app = App(name="secrets", help="Manage Tangle secrets.")
 
 
-def _client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str) -> LazyTangleApiClient:
+def _client(
+    args: ArgsContainer, *, cli_base_url: str | None, command_name: str, logger: Logger | None = None
+) -> LazyTangleApiClient:
     return LazyTangleApiClient(
         base_url=args.base_url,
         token=args.token,
@@ -66,6 +68,7 @@ def _client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str)
         header=args.header,
         include_env_credentials=include_env_credentials_for_args(args, cli_base_url),
         command_name=command_name,
+        logger=logger,
     )
 
 
@@ -79,7 +82,7 @@ def _run_secret_action(
     for args in load_args_or_exit(config, **specs):
         logger, finalize_logs = logger_for_log_type(getattr(args, "log_type", "console"))
         try:
-            client = _client(args, cli_base_url=cli_base_url, command_name="secret commands")
+            client = _client(args, cli_base_url=cli_base_url, command_name="secret commands", logger=logger)
             try:
                 results.append(fn(client, args, logger))
             except SecretValueError as exc:
diff --git a/tests/test_api_cli.py b/tests/test_api_cli.py
index c93a575..8993cc6 100644
--- a/tests/test_api_cli.py
+++ b/tests/test_api_cli.py
@@ -1,6 +1,7 @@
 import importlib
 import json
 import sys
+from unittest.mock import ANY
 
 import httpx
 import pytest
@@ -460,6 +461,7 @@ def fake_client_from_options(**kwargs):
         "header": ["X-Config: yes"],
         "include_env_credentials": False,
         "command_name": "published-component commands",
+        "logger": ANY,
     }
 
 
diff --git a/tests/test_artifacts_cli.py b/tests/test_artifacts_cli.py
index b7b3ec1..f635497 100644
--- a/tests/test_artifacts_cli.py
+++ b/tests/test_artifacts_cli.py
@@ -5,6 +5,7 @@
 import json
 import sys
 from typing import Any
+from unittest.mock import ANY
 
 from tangle_cli import artifacts as artifacts_module
 from tangle_cli import artifacts_cli, cli
@@ -80,6 +81,7 @@ def fake_get_artifacts(self, run_id: str, query: dict[str, Any]) -> dict[str, ob
             "header": ["X-Config: yes"],
             "include_env_credentials": False,
             "command_name": "artifact commands",
+            "logger": ANY,
         }
     ]
     assert get_calls == [
diff --git a/tests/test_client.py b/tests/test_client.py
index 5456ac0..8f21cd9 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,13 +1,46 @@
 from __future__ import annotations
 
+import json
 from types import SimpleNamespace
+from typing import Any
 from unittest.mock import MagicMock
 
+import pytest
+import requests
+
 import tangle_cli.client as client_module
 from tangle_cli.client import TangleApiClient
+from tangle_cli.logger import CaptureLogger
 from tangle_cli.models import ComponentInfo
 
 
+def _response(payload: Any = None, status_code: int = 200) -> requests.Response:
+    r = requests.Response()
+    r.status_code = status_code
+    if payload is None:
+        r._content = b""
+    else:
+        r._content = json.dumps(payload).encode("utf-8")
+        r.headers["Content-Type"] = "application/json"
+    r.request = requests.Request("GET", "https://api.test").prepare()
+    return r
+
+
+class _FakeSession:
+    def __init__(self, responses: list[requests.Response | Exception] | None = None) -> None:
+        self.calls: list[dict[str, Any]] = []
+        self.responses = responses or []
+
+    def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+        self.calls.append({"method": method, "url": url, **kwargs})
+        if self.responses:
+            next_response = self.responses.pop(0)
+            if isinstance(next_response, Exception):
+                raise next_response
+            return next_response
+        return _response({})
+
+
 def test_find_existing_components_matches_exact_names_case_insensitively() -> None:
     client = TangleApiClient("https://api.test")
     client.list_published_component_infos = MagicMock(
@@ -53,3 +86,409 @@ def fail_from_dict(*args, **kwargs):
 
     assert client.get_run_pipeline_spec("run-1") is task_spec
     client.executions_details.assert_called_once_with("root-exec-1")
+
+
+def test_get_retries_transient_5xx_then_succeeds(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    ok = _response({"ok": True})
+    session = _FakeSession([_response(status_code=503), _response(status_code=500), ok])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result is ok
+    assert len(session.calls) == 3
+    assert sleeps == [1.0, 2.0]
+
+
+def test_get_closes_intermediate_5xx_responses_before_retrying(monkeypatch) -> None:
+    events: list[str] = []
+
+    def tracking_response(status_code: int, marker: str) -> requests.Response:
+        r = _response(status_code=status_code)
+        r.close = lambda: events.append(f"close-{marker}")  # type: ignore[method-assign]
+        return r
+
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: events.append("sleep"))
+    session = _FakeSession(
+        [tracking_response(503, "1"), tracking_response(500, "2"), tracking_response(200, "3")]
+    )
+    client = TangleApiClient("https://api.test", session=session)
+
+    client._make_request("GET", "/api/test")
+
+    # Intermediate 5xx are closed before each retry; the returned one is left open.
+    assert events == ["close-1", "sleep", "close-2", "sleep"]
+
+
+def test_get_retries_transport_error_then_succeeds(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    ok = _response({"ok": True})
+    session = _FakeSession(
+        [
+            requests.ConnectionError("connection reset"),
+            requests.Timeout("read timed out"),
+            requests.exceptions.ChunkedEncodingError("incomplete chunked read"),
+            requests.exceptions.ContentDecodingError("failed to decode gzip stream"),
+            ok,
+        ]
+    )
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result is ok
+    assert len(session.calls) == 5
+    assert sleeps == [1.0, 2.0, 4.0, 8.0]
+
+
+def test_get_raises_after_exhausting_transport_retries(monkeypatch) -> None:
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+    budget = TangleApiClient._MAX_GET_RETRIES
+    final_attempt_error = requests.exceptions.ChunkedEncodingError("final permitted attempt")
+    surplus = [requests.ConnectionError("never reached") for _ in range(3)]
+    queued = [requests.ConnectionError("blip") for _ in range(budget)] + [final_attempt_error] + surplus
+    assert len(queued) > budget + 1
+    session = _FakeSession(queued)
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.exceptions.ChunkedEncodingError) as exc_info:
+        client._make_request("GET", "/api/test")
+
+    assert exc_info.value is final_attempt_error
+    assert len(session.calls) == budget + 1
+    assert len(session.responses) == len(surplus)
+
+
+def test_get_returns_final_5xx_after_exhausting_status_retries(monkeypatch) -> None:
+    closed: list[str] = []
+
+    def tracking_5xx(marker: str) -> requests.Response:
+        r = _response(status_code=503)
+        r.close = lambda: closed.append(marker)  # type: ignore[method-assign]
+        return r
+
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+    budget = TangleApiClient._MAX_GET_RETRIES
+    errors = [tracking_5xx(str(i)) for i in range(budget + 1)]
+    trailing_ok = _response({"ok": True})
+    session = _FakeSession([*errors, trailing_ok])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result is errors[budget]
+    assert len(session.calls) == budget + 1
+    assert closed == [str(i) for i in range(budget)]
+    assert session.responses == [trailing_ok]
+
+
+def test_post_is_not_retried_on_transient_5xx(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    server_error = _response(status_code=503)
+    session = _FakeSession([server_error, _response({"ok": True})])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("POST", "/api/pipeline_runs/", json_data={"a": 1})
+
+    assert result is server_error
+    assert len(session.calls) == 1
+    assert sleeps == []
+
+
+def test_streamed_get_bypasses_transient_retry(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    server_error = _response(status_code=503)
+    session = _FakeSession([server_error, _response({"ok": True})])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/logs", stream=True)
+
+    assert result is server_error
+    assert len(session.calls) == 1
+    assert sleeps == []
+
+
+def test_transient_retry_decision_is_method_case_insensitive(monkeypatch) -> None:
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+
+    def call(method: str) -> tuple[requests.Response, int]:
+        ok = _response({"ok": True})
+        session = _FakeSession([_response(status_code=503), ok])
+        client = TangleApiClient("https://api.test", session=session)
+        budget = client_module._RetryBudget(
+            client._MAX_GET_RETRIES + 1,
+            client_module.time.monotonic() + client._MAX_RETRY_ELAPSED_SECONDS,
+        )
+        result = client._request_with_transient_retries(
+            method,
+            "https://api.test/api/test",
+            params=None,
+            json_data=None,
+            extra_headers=None,
+            timeout=client.timeout,
+            request_kwargs={},
+            budget=budget,
+        )
+        return result, len(session.calls)
+
+    get_result, get_calls = call("get")
+    assert get_result.status_code == 200
+    assert get_calls == 2
+
+    post_result, post_calls = call("post")
+    assert post_result.status_code == 503
+    assert post_calls == 1
+
+
+def test_get_retries_proxy_errors(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    ok = _response({"ok": True})
+    session = _FakeSession(
+        [
+            requests.exceptions.ProxyError("proxy refused"),
+            ok,
+        ]
+    )
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result is ok
+    assert len(session.calls) == 2
+    assert sleeps == [1.0]
+
+
+def test_get_does_not_retry_ssl_errors(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    error = requests.exceptions.SSLError("certificate verify failed")
+    session = _FakeSession([error, _response({"ok": True})])
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.exceptions.SSLError) as exc_info:
+        client._make_request("GET", "/api/test")
+
+    assert exc_info.value is error
+    assert len(session.calls) == 1
+    assert sleeps == []
+
+
+def test_get_transient_and_rate_limit_retry_layers_compose(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    ok = _response({"ok": True})
+    session = _FakeSession(
+        [
+            _response(status_code=503),
+            _response(status_code=429),
+            _response(status_code=503),
+            ok,
+        ]
+    )
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result is ok
+    assert len(session.calls) == 4
+    # transient 1.0s, rate-limit 1.0s (no Retry-After), fresh transient 1.0s
+    assert sleeps == [1.0, 1.0, 1.0]
+
+
+def test_get_retry_sleeps_are_capped_and_announced_without_verbose(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    budget = TangleApiClient._MAX_GET_RETRIES
+    session = _FakeSession([_response(status_code=503) for _ in range(budget + 1)])
+    logger = CaptureLogger()
+    client = TangleApiClient("https://api.test", session=session, logger=logger)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result.status_code == 503
+    assert sleeps == [1.0, 2.0, 4.0, 8.0, 16.0, 30.0]  # final sleep capped, not 32.0
+    messages = (logger.get_logs() or "").splitlines()
+    assert len(messages) == budget
+    assert all(m.startswith("transient HTTP 503 on GET; retrying in ") for m in messages)
+    assert messages[-1] == "transient HTTP 503 on GET; retrying in 30.0s (attempt 7/7)"
+
+
+def test_get_retries_are_silent_on_default_non_verbose_client(monkeypatch, capsys) -> None:
+    # A non-verbose client built without a logger stays silent; callers that
+    # want retry announcements pass a logger (as the CLI command layer does).
+    monkeypatch.delenv("TANGLE_VERBOSE", raising=False)
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+    session = _FakeSession([_response(status_code=503), _response({"ok": True})])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result.status_code == 200
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert captured.err == ""
+
+
+def test_shared_budget_caps_total_requests_across_transient_and_rate_limit(monkeypatch) -> None:
+    # Interleaved 503/429 responses must not let the rate-limit layer hand the
+    # transient layer a fresh budget each round: the total physical request
+    # count is bounded by the single shared budget, not their product.
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+    budget = TangleApiClient._MAX_GET_RETRIES
+    # Far more responses than the budget allows, alternating retryable states.
+    session = _FakeSession([_response(status_code=503 if i % 2 == 0 else 429) for i in range(40)])
+    client = TangleApiClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    # Exactly the advertised budget of physical requests is spent, then the last
+    # response surfaces for the caller's raise_for_status (no amplification).
+    assert len(session.calls) == budget + 1
+    assert result.status_code in {503, 429}
+
+
+def test_auth_refresh_shares_transient_retry_budget(monkeypatch) -> None:
+    # A 401 that triggers an auth refresh must continue on the same budget
+    # rather than starting a fresh transient-retry round.
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+
+    class RefreshingClient(TangleApiClient):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            self.refreshes = 0
+
+        def _refresh_auth(self) -> None:
+            self.refreshes += 1
+
+    budget = TangleApiClient._MAX_GET_RETRIES
+    # One 401 (consumes a request) followed by an unbroken run of 503s.
+    session = _FakeSession(
+        [_response(status_code=401)] + [_response(status_code=503) for _ in range(budget + 5)]
+    )
+    client = RefreshingClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    assert result.status_code == 503
+    # Refresh fired once for the 401 (plus the unconditional pre-request refresh).
+    assert client.refreshes == 2
+    # The 401 request plus the post-refresh retries share one budget: the total
+    # never exceeds the shared cap (a fresh budget would allow budget+1 more).
+    assert len(session.calls) == budget + 1
+
+
+def test_shared_budget_caps_total_requests_across_transient_rate_limit_and_auth(monkeypatch) -> None:
+    # The worst case the reviewer flagged: a 401 auth refresh, 429 rate limits,
+    # and transient 503s all interleaved for one logical GET. A single shared
+    # budget must bound the total physical request count instead of letting the
+    # three layers multiply their per-layer limits together.
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+
+    class RefreshingClient(TangleApiClient):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            self.refreshes = 0
+
+        def _refresh_auth(self) -> None:
+            self.refreshes += 1
+
+    budget = TangleApiClient._MAX_GET_RETRIES
+    # Lead with a 401 (drives one refresh), then alternate 429/503 far past the
+    # budget so the cap, not the response list, is what stops the retries.
+    responses = [_response(status_code=401)]
+    responses += [_response(status_code=429 if i % 2 == 0 else 503) for i in range(40)]
+    session = _FakeSession(responses)
+    client = RefreshingClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    # Exactly the advertised shared budget of physical requests is spent.
+    assert len(session.calls) == budget + 1
+    assert result.status_code in {429, 503}
+    # Pre-request refresh plus exactly one refresh for the single 401; the 401's
+    # retry continues on the shared budget rather than opening a fresh round.
+    assert client.refreshes == 2
+
+
+def test_shared_budget_deadline_halts_composed_retries_without_wallclock(monkeypatch) -> None:
+    # The shared budget bounds retries by BOTH an attempt count and a
+    # _MAX_RETRY_ELAPSED_SECONDS wall-time deadline. This pins the deadline
+    # clause deterministically: a fake monotonic clock advances only when a
+    # request is sent (never the real wall clock), so the elapsed-time cap, not
+    # the attempt cap, is what stops the composed auth-refresh / rate-limit /
+    # transient retry sequence. Without this every other budget test would still
+    # pass on the attempt cap alone, so the deadline could be removed silently.
+    clock = SimpleNamespace(now=1_000.0)
+    step = 50.0
+    window = TangleApiClient._MAX_RETRY_ELAPSED_SECONDS
+    deadline = clock.now + window
+    monkeypatch.setattr("tangle_cli.client.time.monotonic", lambda: clock.now)
+
+    send_times: list[float] = []
+    sleep_times: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: sleep_times.append(clock.now))
+
+    class _ClockAdvancingSession(_FakeSession):
+        def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+            send_times.append(clock.now)
+            response = super().request(method, url, **kwargs)
+            # Time elapses only while a request is in flight; each send consumes
+            # a fixed slice of the deadline window.
+            clock.now += step
+            return response
+
+    class RefreshingClient(TangleApiClient):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            self.refreshes = 0
+
+        def _refresh_auth(self) -> None:
+            self.refreshes += 1
+
+    # A 401 (auth-refresh path), a 429 (rate-limit path), then unbroken 503s
+    # (transient path): all three layers draw on the one shared budget. Far more
+    # responses are queued than either cap allows, so the cap that fires first is
+    # what stops the sequence.
+    responses = [_response(status_code=401), _response(status_code=429)]
+    responses += [_response(status_code=503) for _ in range(20)]
+    session = _ClockAdvancingSession(responses)
+    client = RefreshingClient("https://api.test", session=session)
+
+    result = client._make_request("GET", "/api/test")
+
+    # The sequence genuinely retried across layers but stopped short of the
+    # attempt cap with responses still queued: the deadline, not the attempt
+    # count or the response list, is what ended it.
+    attempt_cap = TangleApiClient._MAX_GET_RETRIES + 1
+    assert 1 < len(session.calls) < attempt_cap
+    assert session.responses, "unused responses prove the queue did not stop the retries"
+    # Time actually crossed the deadline, yet no send or sleep happened at/after
+    # it: can_retry gates every physical send and every sleep across the
+    # transient and rate-limit layers.
+    assert clock.now >= deadline
+    assert all(t < deadline for t in send_times)
+    assert all(t < deadline for t in sleep_times)
+    # The composed auth path ran on the shared budget (pre-request refresh plus
+    # one for the single 401), and the final 5xx surfaces for the caller.
+    assert client.refreshes == 2
+    assert result.status_code == 503
+
+
+def test_get_5xx_exhaustion_raises_http_error_from_public_operation(monkeypatch) -> None:
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None)
+    budget = TangleApiClient._MAX_GET_RETRIES
+    session = _FakeSession([_response(status_code=503) for _ in range(budget + 1)])
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.HTTPError) as exc_info:
+        client.pipeline_runs_get("run-1")
+
+    assert exc_info.value.response is not None
+    assert exc_info.value.response.status_code == 503
+    assert len(session.calls) == budget + 1
diff --git a/tests/test_components_cli.py b/tests/test_components_cli.py
index 5764695..c52b5d8 100644
--- a/tests/test_components_cli.py
+++ b/tests/test_components_cli.py
@@ -138,6 +138,7 @@ def fake_client_from_options(**kwargs: Any) -> object:
             "header": ["X-Config: yes"],
             "include_env_credentials": False,
             "command_name": "published-component commands",
+            "logger": ANY,
         }
     ]
     assert FakePublisher.instances[0].kwargs["client"] is fake_client
@@ -271,6 +272,7 @@ def fake_client_from_options(**kwargs: Any) -> object:
             "header": None,
             "include_env_credentials": False,
             "command_name": "published-component commands",
+            "logger": ANY,
         }
     ]
     assert [publisher.kwargs for publisher in FakePublisher.instances] == [
@@ -341,6 +343,7 @@ def fake_deprecate_component(client: object, digest: str, **kwargs: Any) -> dict
             "header": ["X-Test: yes"],
             "include_env_credentials": False,
             "command_name": "published-component commands",
+            "logger": ANY,
         }
     ]
     assert deprecate_calls == [
diff --git a/tests/test_secrets_cli.py b/tests/test_secrets_cli.py
index 6072131..f61b17f 100644
--- a/tests/test_secrets_cli.py
+++ b/tests/test_secrets_cli.py
@@ -6,6 +6,7 @@
 import sys
 from types import SimpleNamespace
 from typing import Any
+from unittest.mock import ANY
 
 import pytest
 
@@ -341,6 +342,7 @@ def test_sdk_secrets_config_array_and_config_base_url_credential_isolation(
             "header": ["X-Config: yes"],
             "include_env_credentials": False,
             "command_name": "secret commands",
+            "logger": ANY,
         },
         {
             "base_url": "https://config.example",
@@ -349,6 +351,7 @@ def test_sdk_secrets_config_array_and_config_base_url_credential_isolation(
             "header": ["X-Config: yes"],
             "include_env_credentials": False,
             "command_name": "secret commands",
+            "logger": ANY,
         },
     ]
     assert [instance.calls[0]["secret_name"] for instance in FakeLazyTangleApiClient.instances] == [