Retry on Unauthorized: Add example retries config block

cecli/args.py

@@ -360,7 +360,7 @@ def get_parser(default_config_files, git_root):
         metavar="RETRIES_JSON",
         help=(
             'Specify LLM retry configuration as a JSON/YAML string (e.g., \'{"retry_on_empty": '
-            "true}')"
+            'true, "retry-on-unauthorized": false}\')'
         ),
         default=None,
     )

cecli/args_formatter.py

@@ -138,6 +138,7 @@ def _format_action(self, action):
             parts.append("#  retry-timeout: 60")
             parts.append("#  retry-backoff-factor: 2.0")
             parts.append("#  retry-on-unavailable: true")
+            parts.append("#  retry-on-unauthorized: false")
             parts.append("#  retry-on-empty: false")
             parts.append("")
             return "\n".join(parts)

cecli/coders/base_coder.py

@@ -2861,6 +2861,20 @@ async def format_in_executor():
                     if ex_info.name == "ServiceUnavailableError":
                         should_retry = should_retry or retry_config["retry_on_unavailable"]
 
+                    # Opt-in retry for auth failures (retry-on-unauthorized).
+                    # Some providers (e.g. vertex_ai_beta) carry the status on
+                    # `code` or omit `status_code` entirely, so match on the
+                    # exception name first, then on status/code fields.
+                    status_code = getattr(err, "status_code", None)
+                    code = getattr(err, "code", None)
+                    is_auth_error = (
+                        ex_info.name in ("AuthenticationError", "PermissionDeniedError")
+                        or status_code in (401, 403, "401", "403")
+                        or str(code) in ("401", "403")
+                    )
+                    if is_auth_error:
+                        should_retry = should_retry or retry_config["retry_on_unauthorized"]
+
                     if should_retry:
                         retry_delay *= retry_config["retry_backoff_factor"]
                         if retry_delay > retry_config["retry_timeout"]:

cecli/models.py

@@ -133,6 +133,7 @@ class ModelSettings:
     retries: Optional[dict] = None
     retry_backoff_factor: float = 1.5
     retry_on_unavailable: bool = True
+    retry_on_unauthorized: bool = False
     retry_timeout: float = 30
     request_timeout: int = request_timeout
     debug: bool = False
@@ -566,7 +567,10 @@ def __init__(
         self._apply_reasoning_defaults()
         self.get_weak_model(weak_model)
         self.get_agent_model(agent_model)
-        self.retries = retries
+        # Keep a `retries:` block from model settings unless an explicit value
+        # was passed (main.py passes retries=None when --retries is unset).
+        if retries is not None:
+            self.retries = retries
         self.debug = debug
 
         if editor_model is False:
@@ -1454,6 +1458,7 @@ async def send_completion(
 
         retry_config = parse_retry_config(self.retries)
         self.retry_on_unavailable = retry_config["retry_on_unavailable"]
+        self.retry_on_unauthorized = retry_config["retry_on_unauthorized"]
         self.retry_backoff_factor = retry_config["retry_backoff_factor"]
         self.retry_timeout = retry_config["retry_timeout"]
 
@@ -1487,6 +1492,19 @@ async def send_completion(
                 if ex_info.name == "ServiceUnavailableError":
                     should_retry = should_retry or self.retry_on_unavailable
 
+                # Opt-in retry for 401/403 auth failures (retry-on-unauthorized).
+                # HTTP 401/403 map to AuthenticationError/PermissionDeniedError,
+                # both default to retry=False so behavior is unchanged unless enabled.
+                status_code = getattr(err, "status_code", None)
+                code = getattr(err, "code", None)
+                is_auth_error = (
+                    ex_info.name in ("AuthenticationError", "PermissionDeniedError")
+                    or status_code in (401, 403, "401", "403")
+                    or str(code) in ("401", "403")
+                )
+                if is_auth_error:
+                    should_retry = should_retry or self.retry_on_unauthorized
+
                 custom_retry_delay = self._extract_retry_delay(err)
                 if custom_retry_delay is not None:
                     retry_delay = custom_retry_delay
@@ -1604,6 +1622,18 @@ async def simple_send_with_retries(
                 if ex_info.name == "ServiceUnavailableError":
                     should_retry = should_retry or retry_on_unavailable
 
+                # Opt-in retry for auth failures (retry-on-unauthorized);
+                # see send_completion for why the name check comes first.
+                status_code = getattr(err, "status_code", None)
+                code = getattr(err, "code", None)
+                is_auth_error = (
+                    ex_info.name in ("AuthenticationError", "PermissionDeniedError")
+                    or status_code in (401, 403, "401", "403")
+                    or str(code) in ("401", "403")
+                )
+                if is_auth_error:
+                    should_retry = should_retry or retry_config["retry_on_unauthorized"]
+
                 custom_retry_delay = self._extract_retry_delay(err)
                 if custom_retry_delay is not None:
                     retry_delay = custom_retry_delay
@@ -1797,6 +1827,7 @@ def parse_retry_config(retries_input):
       retry_backoff_factor: 1.5
       retry_on_unavailable: True
       retry_on_empty: False
+      retry_on_unauthorized: False
     """
     config = dict()
     if isinstance(retries_input, str):
@@ -1821,6 +1852,7 @@ def _get(key, default):
         "retry_timeout": float(_get("retry_timeout", 30)),
         "retry_backoff_factor": float(_get("retry_backoff_factor", 1.5)),
         "retry_on_unavailable": bool(_get("retry_on_unavailable", True)),
+        "retry_on_unauthorized": bool(_get("retry_on_unauthorized", False)),
         "retry_on_empty": bool(_get("retry_on_empty", False)),
     }
 

cecli/website/docs/config/retries.md

@@ -11,6 +11,7 @@ Cecli can be configured to retry failed API calls. This is useful for handling i
 - `retry-timeout`: The timeout in seconds for each retry.
 - `retry-backoff-factor`: The backoff factor to use between retries.
 - `retry-on-unavailable`: Whether to retry on 503 Service Unavailable errors.
+- `retry-on-unauthorized`: Whether to retry on 401 Unauthorized (and 403 Forbidden) errors. Default: false.
 
 Example usage in `.cecli.conf.yml`:
 
@@ -19,18 +20,19 @@ retries:
   retry-timeout: 30
   retry-backoff-factor: 1.50
   retry-on-unavailable: true
+  retry-on-unauthorized: false
 ```
 
 This can also be set with the `--retries` command line switch, passing a JSON string:
 
 ```
-$ cecli --retries '{"retry-timeout": 30, "retry-backoff-factor": 1.50, "retry-on-unavailable": true}'
+$ cecli --retries '{"retry-timeout": 30, "retry-backoff-factor": 1.50, "retry-on-unavailable": true, "retry-on-unauthorized": false}'
 ```
 
 Or by setting the `CECLI_RETRIES` environment variable:
 
 ```
-export CECLI_RETRIES='{"retry-timeout": 30, "retry-backoff-factor": 1.50, "retry-on-unavailable": true}'
+export CECLI_RETRIES='{"retry-timeout": 30, "retry-backoff-factor": 1.50, "retry-on-unavailable": true, "retry-on-unauthorized": false}'
 ```
 
 > **Tip:**

requirements.txt

@@ -608,7 +608,7 @@ uvicorn[standard]==0.38.0
     #   -c requirements/common-constraints.txt
     #   chromadb
     #   mcp
-uvloop==0.22.1
+uvloop==0.22.1 ; platform_python_implementation != 'PyPy' and sys_platform != 'cygwin' and sys_platform != 'win32'
     # via
     #   -c requirements/common-constraints.txt
     #   uvicorn
@@ -642,6 +642,6 @@ zipp==3.23.0
     # via
     #   -c requirements/common-constraints.txt
     #   importlib-metadata
-    
+
 tree-sitter==0.23.2; python_version < "3.10"
 tree-sitter>=0.25.1; python_version >= "3.10"

requirements/common-constraints.txt

@@ -514,7 +514,7 @@ uvicorn[standard]==0.38.0
     # via
     #   chromadb
     #   mcp
-uvloop==0.22.1
+uvloop==0.22.1 ; platform_python_implementation != 'PyPy' and sys_platform != 'cygwin' and sys_platform != 'win32'
     # via uvicorn
 virtualenv==20.35.4
     # via pre-commit

tests/unit/test_retry_backoff.py

@@ -534,3 +534,158 @@ def mock_sleep(delay):
         assert result == (None, None)
 
     asyncio.run(run_test())
+
+
+def test_retry_on_unauthorized_disabled_by_default():
+    async def run_test():
+        model = Model("openai/gpt-4o")
+        model.caches_by_default = False
+
+        auth_err = litellm.AuthenticationError(
+            message="401 Unauthorized",
+            model="openai/gpt-4o",
+            llm_provider="openai",
+        )
+        auth_err.status_code = 401
+
+        slept_delays = []
+
+        async def mock_acompletion(*args, **kwargs):
+            raise auth_err
+
+        async def mock_sleep(delay):
+            slept_delays.append(delay)
+
+        with (
+            patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion),
+            patch("asyncio.sleep", side_effect=mock_sleep),
+        ):
+            _hash, resp = await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+
+        # Default (retry_on_unauthorized=False): 401 fails immediately, no retry.
+        assert model.retry_on_unauthorized is False
+        assert len(slept_delays) == 0
+        assert "Model API Response Error" in resp.choices[0].message.content
+
+    asyncio.run(run_test())
+
+
+def test_retry_on_unauthorized_enabled_retries_401():
+    async def run_test():
+        model = Model("openai/gpt-4o", retries='{"retry-on-unauthorized": true}')
+        model.caches_by_default = False
+
+        auth_err = litellm.AuthenticationError(
+            message="401 Unauthorized",
+            model="openai/gpt-4o",
+            llm_provider="openai",
+        )
+        auth_err.status_code = 401
+
+        call_count = 0
+        slept_delays = []
+
+        async def mock_acompletion(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise auth_err
+            return MagicMock(choices=[MagicMock(message=MagicMock(content="ok"))])
+
+        async def mock_sleep(delay):
+            slept_delays.append(delay)
+
+        with (
+            patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion),
+            patch("asyncio.sleep", side_effect=mock_sleep),
+        ):
+            _hash, resp = await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+
+        # Enabled: the 401 is retried with the standard exponential backoff.
+        assert model.retry_on_unauthorized is True
+        assert call_count == 2
+        assert len(slept_delays) == 1
+        assert pytest.approx(slept_delays[0]) == 0.125 * 1.5
+        assert resp.choices[0].message.content == "ok"
+
+    asyncio.run(run_test())
+
+
+def test_retry_on_unauthorized_enabled_retries_403():
+    async def run_test():
+        model = Model("openai/gpt-4o")
+        model.caches_by_default = False
+        model.retry_on_unauthorized = True
+
+        forbidden_err = litellm.PermissionDeniedError(
+            message="403 Forbidden",
+            model="openai/gpt-4o",
+            llm_provider="openai",
+        )
+        forbidden_err.status_code = 403
+
+        call_count = 0
+        slept_delays = []
+
+        async def mock_acompletion(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise forbidden_err
+            return MagicMock(choices=[MagicMock(message=MagicMock(content="ok"))])
+
+        async def mock_sleep(delay):
+            slept_delays.append(delay)
+
+        with (
+            patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion),
+            patch("asyncio.sleep", side_effect=mock_sleep),
+        ):
+            _hash, resp = await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+
+        assert call_count == 2
+        assert len(slept_delays) == 1
+        assert resp.choices[0].message.content == "ok"
+
+    asyncio.run(run_test())
+
+
+def test_retry_on_unauthorized_config_parsing():
+    async def run_test():
+        async def mock_acompletion(*args, **kwargs):
+            return MagicMock(choices=[MagicMock(message=MagicMock(content="ok"))])
+
+        # JSON string form
+        model = Model("openai/gpt-4o", retries='{"retry-on-unauthorized": true}')
+        model.caches_by_default = False
+        with patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion):
+            await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+        assert model.retry_on_unauthorized is True
+
+        # Already-parsed dict form
+        model = Model("openai/gpt-4o", retries={"retry-on-unauthorized": True})
+        model.caches_by_default = False
+        with patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion):
+            await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+        assert model.retry_on_unauthorized is True
+
+        # Default when unset
+        model = Model("openai/gpt-4o")
+        model.caches_by_default = False
+        with patch("cecli.llm.litellm.acompletion", side_effect=mock_acompletion):
+            await model.send_completion(
+                messages=[{"role": "user", "content": "hi"}], functions=None, stream=False
+            )
+        assert model.retry_on_unauthorized is False
+
+    asyncio.run(run_test())