diff --git a/backend/druks/harnesses/config.py b/backend/druks/harnesses/config.py index b844fb5a..1998a266 100644 --- a/backend/druks/harnesses/config.py +++ b/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( diff --git a/backend/druks/harnesses/exceptions.py b/backend/druks/harnesses/exceptions.py index 92c31924..f82c19c4 100644 --- a/backend/druks/harnesses/exceptions.py +++ b/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: diff --git a/backend/druks/harnesses/providers.py b/backend/druks/harnesses/providers.py index d80e4c8c..777bb135 100644 --- a/backend/druks/harnesses/providers.py +++ b/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( diff --git a/backend/druks/sandbox/routes.py b/backend/druks/sandbox/routes.py index 77cbbec4..5e6a3937 100644 --- a/backend/druks/sandbox/routes.py +++ b/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) diff --git a/backend/tests/test_agent_config.py b/backend/tests/test_agent_config.py index f6904c4c..82bf0128 100644 --- a/backend/tests/test_agent_config.py +++ b/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( { diff --git a/backend/tests/test_agents.py b/backend/tests/test_agents.py index ba75c5d9..83f07186 100644 --- a/backend/tests/test_agents.py +++ b/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. diff --git a/backend/tests/test_provider_auth.py b/backend/tests/test_provider_auth.py index bdac97f3..dfe3f882 100644 --- a/backend/tests/test_provider_auth.py +++ b/backend/tests/test_provider_auth.py @@ -823,7 +823,18 @@ 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)) ) @@ -831,7 +842,7 @@ async def test_an_expired_token_with_a_failed_rotation_answers_nothing(monkeypat 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): diff --git a/backend/tests/test_sandbox_identities.py b/backend/tests/test_sandbox_identities.py index 2d1eb030..04ec0471 100644 --- a/backend/tests/test_sandbox_identities.py +++ b/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: