DRU-984 -- Refuse calls with unusable subscription tokens

backend/druks/harnesses/config.py

@@ -13,8 +13,8 @@
 from druks.secrets.models import VaultSecret
 from druks.user_settings.models import InstallationSettings, SettingsOverride
 
+from . import exceptions
 from .base import Harness
-from .exceptions import AgentConfigError, HarnessNotConnectedError
 from .models import ProviderCatalog
 from .providers import get_provider, is_registered, provider_label
 from .registry import get_harness
@@ -76,24 +76,30 @@ async def check_config(
     """The harness that runs the triple; a triple no harness runs raises."""
     harness = get_harness(harness_name)
     if not harness:
-        raise AgentConfigError(f"no installed harness is named {harness_name!r}.")
+        raise exceptions.AgentConfigError(f"no installed harness is named {harness_name!r}.")
     provider_id = model.partition("/")[0]
     if is_registered(provider_id):
         provider = get_provider(provider_id)
         if not harness.has_provider(provider):
-            raise AgentConfigError(f"{harness_name} does not run {provider.label} models.")
+            raise exceptions.AgentConfigError(
+                f"{harness_name} does not run {provider.label} models."
+            )
     else:
         catalog = await session.get(ProviderCatalog, provider_id)
         if not catalog:
-            raise AgentConfigError(
+            raise exceptions.AgentConfigError(
                 f"model {model!r} names no provider; add one in Settings → Providers."
             )
         if harness.provider:
-            raise AgentConfigError(f"{harness_name} does not run {catalog.label} models.")
+            raise exceptions.AgentConfigError(
+                f"{harness_name} does not run {catalog.label} models."
+            )
         if model not in {entry["id"] for entry in catalog.models}:
-            raise AgentConfigError(f"{catalog.label} lists no model {model!r}.")
+            raise exceptions.AgentConfigError(f"{catalog.label} lists no model {model!r}.")
     if billing not in harness.billing_options:
-        raise AgentConfigError(f"{harness_name} runs on an API key only; set billing to api_key.")
+        raise exceptions.AgentConfigError(
+            f"{harness_name} runs on an API key only; set billing to api_key."
+        )
     return harness
 
 
@@ -170,11 +176,24 @@ async def _build_config(
         )
         if not provider_key:
             label = await provider_label(session, provider_id)
-            raise HarnessNotConnectedError(f"add the {label} API key in Settings → Providers.")
+            raise exceptions.HarnessNotConnectedError(
+                f"add the {label} API key in Settings → Providers."
+            )
         secrets = harness_class.get_secrets(provider_id, provider_key.secrets["value"])
     else:
         provider = get_provider(provider_id)
         subscription = await provider.get_subscription(session, account_id)
+        try:
+            await provider.issue_token(session, subscription.id)
+        except exceptions.OAuthTokenError as error:
+            if error.tag == "network" or error.tag.startswith("http_5"):
+                raise exceptions.HarnessOverloadedError(
+                    f"{provider.label} token refresh is temporarily unavailable ({error.tag})."
+                ) from error
+            raise exceptions.HarnessNotConnectedError(
+                f"the {provider.label} subscription cannot provide a token — "
+                "reconnect it in Settings → Providers."
+            ) from error
         identity = provider.get_identity(subscription)
         secret_refs = harness_class.get_secret_refs(subscription)
     return AgentConfig(

backend/druks/harnesses/exceptions.py

@@ -85,8 +85,10 @@ class OAuthTokenError(DruksError):
 
     ``tag`` is a short, stable code surfaced on the usage snapshot's
     ``error`` column: ``no_credentials`` (harness not connected),
-    ``no_token`` (subscription present, no access token), ``token_expired``
-    (past expiry; the refresh cron hasn't caught up).
+    ``no_token`` (subscription present, no access token), or ``token_expired``
+    (past expiry; the refresh cron hasn't caught up). When an attempted refresh
+    fails and leaves no usable token, it carries the refresh tag, such as
+    ``network`` or ``http_503``.
     """
 
     def __init__(self, tag: str, message: str | None = None) -> None:

backend/druks/harnesses/providers.py

@@ -237,6 +237,7 @@ async def issue_token(
         fetch and every new call until the rotation and its refresh requests end.
         Raises :class:`OAuthTokenError` when the row holds nothing valid."""
         row = await VaultSecret.reload(session, subscription_id)
+        rotation: RotationResult | None = None
         if row and cls.refresh_is_due(row):
             async with gate.shut(subscription_id) as is_idle:
                 if is_idle or cls.refresh_is_urgent(row):
@@ -253,8 +254,16 @@ async def issue_token(
                         )
             row = await VaultSecret.reload(session, subscription_id)
         if not row:
-            raise exceptions.OAuthTokenError("no_credentials", "the subscription is disconnected")
-        return cls.load_token(row)
+            raise exceptions.OAuthTokenError(
+                (rotation.error if rotation else None) or "no_credentials",
+                "the subscription is disconnected",
+            )
+        try:
+            return cls.load_token(row)
+        except exceptions.OAuthTokenError as error:
+            if rotation and rotation.error:
+                raise exceptions.OAuthTokenError(rotation.error, str(error)) from error
+            raise
 
     @classmethod
     async def rotate_token(

backend/druks/sandbox/routes.py

@@ -46,7 +46,10 @@ async def secret(
     except MissingSecretError as exc:
         logger.warning("fetch of %s for identity %s failed: %s", name, identity_id, exc)
         raise HTTPException(status_code=422, detail=str(exc)) from None
-    except OAuthTokenError:
+    except OAuthTokenError as exc:
+        logger.warning(
+            "fetch of %s for identity %s failed (%s): %s", name, identity_id, exc.tag, exc
+        )
         raise HTTPException(status_code=503) from None
     except Exception as exc:  # noqa: BLE001 — the source failed; the box gets a 503 and the exchange retries
         logger.warning("fetch of %s for identity %s failed: %s", name, identity_id, exc)

backend/tests/test_agent_config.py

@@ -1,6 +1,7 @@
-from datetime import datetime
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 
+import httpx
 import pytest
 from conftest import (
     CONFIG_PROBE,
@@ -16,14 +17,20 @@
 from druks.apps.registry import agents as agent_registry
 from druks.db import db_session
 from druks.durable.models import AgentCall
+from druks.harnesses import providers as provider_module
 from druks.harnesses.claude import ClaudeHarness
 from druks.harnesses.codex import CodexHarness
 from druks.harnesses.config import AgentConfig, check_config, get_config, get_default_config
-from druks.harnesses.exceptions import AgentConfigError, HarnessNotConnectedError
+from druks.harnesses.exceptions import (
+    AgentConfigError,
+    HarnessNotConnectedError,
+    HarnessOverloadedError,
+    Retry,
+)
 from druks.harnesses.models import ProviderCatalog
 from druks.harnesses.opencode import OpenCodeHarness
 from druks.harnesses.pi import PiHarness
-from druks.harnesses.providers import OpenAiProvider
+from druks.harnesses.providers import AnthropicProvider, OpenAiProvider
 from druks.sandbox.constants import MAX_AGENT_TIMEOUT_SECONDS
 from druks.secrets.datastructures import Audience
 from druks.secrets.enums import SecretKind
@@ -76,6 +83,18 @@ async def _key() -> VaultSecret:
     )
 
 
+async def _expired_anthropic(refresh_token: str = "") -> VaultSecret:
+    expired = datetime.now(UTC) - timedelta(minutes=1)
+    token = {
+        "accessToken": "old",
+        "scopes": ["user:profile"],
+        "expiresAt": int(expired.timestamp() * 1000),
+    }
+    if refresh_token:
+        token["refreshToken"] = refresh_token
+    return await connect_provider(AnthropicProvider, {"claudeAiOauth": token})
+
+
 _SHARED_ENTRY = Secret(
     "sk-shared",
     host="api.anthropic.com",
@@ -158,6 +177,53 @@ async def test_a_subscription_agent_runs_as_its_actor_or_the_default_account(dru
     assert (as_actor.effort, as_actor.timeout, as_actor.fast_mode) == ("high", 1800, False)
 
 
+async def test_a_subscription_agent_refuses_an_expired_token_without_a_refresh(druks_db):
+    subscription = await _expired_anthropic()
+
+    with pytest.raises(
+        HarnessNotConnectedError,
+        match="Anthropic subscription cannot provide a token.*reconnect it in Settings → Providers",
+    ):
+        await get_config(druks_db, CONFIG_PROBE.id, subscription.account_id)
+
+
+async def test_a_subscription_agent_runs_after_its_expired_token_refreshes(druks_db, monkeypatch):
+    subscription = await _expired_anthropic("R0")
+
+    async def refresh(self, url, *, json, **kwargs):
+        return httpx.Response(
+            200,
+            json={"access_token": "new", "refresh_token": "R1", "expires_in": 28_800},
+        )
+
+    monkeypatch.setattr(provider_module.httpx.AsyncClient, "post", refresh)
+
+    config = await get_config(druks_db, CONFIG_PROBE.id, subscription.account_id)
+    refreshed = await VaultSecret.reload(druks_db, subscription.id)
+
+    assert config.subscription.id == subscription.id
+    assert AnthropicProvider.load_token(refreshed).access_token == "new"
+
+
+@pytest.mark.parametrize("failure", ["network", "http_503"])
+async def test_a_subscription_agent_keeps_a_temporary_refresh_failure_transient(
+    druks_db, monkeypatch, failure
+):
+    subscription = await _expired_anthropic("R0")
+
+    async def refresh(self, url, *, json, **kwargs):
+        if failure == "network":
+            raise httpx.ConnectError("offline")
+        return httpx.Response(503, text="unavailable")
+
+    monkeypatch.setattr(provider_module.httpx.AsyncClient, "post", refresh)
+
+    with pytest.raises(HarnessOverloadedError, match=failure) as error:
+        await get_config(druks_db, CONFIG_PROBE.id, subscription.account_id)
+
+    assert error.value.retry is Retry.TRANSIENT
+
+
 async def test_a_codex_subscription_config_carries_its_login_facts_and_its_ref(druks_db):
     id_token = make_jwt(
         {

backend/tests/test_agents.py

@@ -15,6 +15,7 @@
 from druks.files import File
 from druks.harnesses.claude import ClaudeHarness
 from druks.harnesses.config import get_config
+from druks.harnesses.exceptions import HarnessNotConnectedError
 from druks.sandbox.client import sandbox_client
 from druks.sandbox.datastructures import McpServer
 from druks.sandbox.exceptions import SandboxDownloadError
@@ -246,9 +247,6 @@ async def test_run_refuses_an_agent_with_no_id():
 
 async def test_run_refuses_unconnected_harness(druks_db, tmp_path, monkeypatch, current_run):
     # The precondition fires where the harness is resolved — before any VM work.
-    from druks.accounts.models import Account
-    from druks.harnesses.exceptions import HarnessNotConnectedError
-
     await (
         await VaultSecret.lookup(
             druks_db,
@@ -266,6 +264,32 @@ async def test_run_refuses_unconnected_harness(druks_db, tmp_path, monkeypatch,
     sandbox.run_agent.assert_not_awaited()
 
 
+async def test_run_refuses_an_expired_subscription_before_the_sandbox(
+    druks_db, tmp_path, monkeypatch, current_run
+):
+    account = await Account.get_for_username(druks_db, "op@example.com")
+    subscription = await VaultSecret.lookup(
+        druks_db, SecretKind.SUBSCRIPTION, Audience.provider("anthropic"), account.id
+    )
+    expired = datetime.now(UTC) - timedelta(minutes=1)
+    await subscription.update_secrets(
+        {
+            "claudeAiOauth": {
+                "accessToken": "old",
+                "expiresAt": int(expired.timestamp() * 1000),
+            }
+        },
+        expires_at=expired,
+    )
+    sandbox = _patch_runtime(monkeypatch, tmp_path, {"ok": True})
+    _patch_ephemeral(monkeypatch, sandbox)
+
+    with pytest.raises(HarnessNotConnectedError, match="reconnect it in Settings → Providers"):
+        await DUMMY_AGENT._run(db_session(), workflow_id="wf-9")
+
+    sandbox.run_agent.assert_not_awaited()
+
+
 async def test_declaration_drives_run_agent_call(druks_db, tmp_path, monkeypatch, current_run):
     # The declaration and the operator default drive what run_agent sees
     # (model/schema/artifact dir/timeout); run() returns the validated contract.

backend/tests/test_provider_auth.py

@@ -823,15 +823,26 @@ async def test_a_failed_rotation_answers_the_live_token(monkeypatch, druks_db):
     assert token.access_token == "old"
 
 
-async def test_an_expired_token_with_a_failed_rotation_answers_nothing(monkeypatch, druks_db):
+async def test_an_expired_token_without_a_refresh_stays_token_expired(druks_db):
+    connection = await _seed_claude(
+        access="old", refresh=None, expires_at=_in(-timedelta(minutes=1))
+    )
+
+    with pytest.raises(OAuthTokenError) as error:
+        await AnthropicProvider.issue_token(druks_db, connection.id)
+
+    assert error.value.tag == "token_expired"
+
+
+async def test_an_expired_token_reports_the_failed_rotation(monkeypatch, druks_db):
     connection = await _seed_claude(
         access="old", refresh="R0", expires_at=_in(-timedelta(minutes=1))
     )
     _mock_post(monkeypatch, httpx.ConnectError("boom"))
 
     with pytest.raises(OAuthTokenError) as error:
         await AnthropicProvider.issue_token(druks_db, connection.id)
-    assert error.value.tag == "token_expired"
+    assert error.value.tag == "network"
 
 
 async def test_the_cron_requests_refreshes_after_a_rotation(monkeypatch, druks_db):
@@ -994,13 +1005,22 @@ async def test_a_failed_codex_rotation_answers_the_live_token(monkeypatch, druks
     assert token.access_token == live
 
 
-async def test_an_expired_codex_token_with_a_failed_rotation_answers_nothing(monkeypatch, druks_db):
+async def test_an_expired_codex_token_without_a_refresh_stays_token_expired(druks_db):
+    connection = await _seed_codex(access=_jwt_in(-timedelta(minutes=1)), refresh=None)
+
+    with pytest.raises(OAuthTokenError) as error:
+        await OpenAiProvider.issue_token(druks_db, connection.id)
+
+    assert error.value.tag == "token_expired"
+
+
+async def test_an_expired_codex_token_reports_the_failed_rotation(monkeypatch, druks_db):
     connection = await _seed_codex(access=_jwt_in(-timedelta(minutes=1)), refresh="R0")
     _mock_post(monkeypatch, httpx.ConnectError("boom"))
 
     with pytest.raises(OAuthTokenError) as error:
         await OpenAiProvider.issue_token(druks_db, connection.id)
-    assert error.value.tag == "token_expired"
+    assert error.value.tag == "network"
 
 
 async def test_the_rotation_tick_passes_a_setup_token_without_a_word(monkeypatch, druks_db, caplog):

backend/tests/test_sandbox_identities.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+import logging
 from datetime import UTC, datetime, timedelta
 
 import httpx
@@ -265,16 +266,19 @@ async def revoke_then_answer(self, session, subscription_id, **kwargs):
 
 
 async def test_the_issuer_answers_503_when_the_source_holds_nothing_valid(
-    druks_db, tmp_path, monkeypatch
+    druks_db, tmp_path, monkeypatch, caplog
 ):
     subscription = await _subscription(access="old", life=-timedelta(minutes=1))
     identity, bearer = await _bound_identity(subscription)
     _mock_post(monkeypatch, httpx.ConnectError("boom"))
 
-    response = await _fetch(tmp_path, identity.id, bearer)
+    with caplog.at_level(logging.WARNING, logger="druks.sandbox.routes"):
+        response = await _fetch(tmp_path, identity.id, bearer)
 
     assert response.status_code == 503
     assert "old" not in response.text
+    assert f"fetch of anthropic for identity {identity.id} failed (network)" in caplog.text
+    assert "access token expired at" in caplog.text
 
 
 async def test_lookup_finds_the_live_bound_identity_of_a_scope(druks_db) -> None: