diff --git a/src/ucode/oauth.py b/src/ucode/oauth.py
new file mode 100644
index 00000000..1ed51887
--- /dev/null
+++ b/src/ucode/oauth.py
@@ -0,0 +1,394 @@
+"""OAuth authorization-code flow for public Databricks apps."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+import secrets
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from ucode.config_io import APP_DIR, is_dry_run
+
+TOKEN_CACHE_PATH = APP_DIR / "oauth-tokens.json"
+CACHE_VERSION = 1
+
+DEFAULT_SCOPES = ("all-apis", "offline_access")
+
+# This URI must be registered on the custom app.
+DEFAULT_REDIRECT_URI = "http://localhost:8020"
+REDIRECT_URI_ENV = "UCODE_OAUTH_REDIRECT_URI"
+
+EXPIRY_BUFFER_SECONDS = 120
+_HTTP_TIMEOUT = 30
+_LOGIN_TIMEOUT_SECONDS = 300.0
+
+_ERROR_HINTS = {
+    "invalid_client": "Check that this is a public app client ID, not an integration ID.",
+    "invalid_grant": (
+        "The refresh token expired or was revoked. Run "
+        "`ug configure --oauth-client-id <client-id>` again."
+    ),
+    "invalid_scope": "Grant the app `all-apis` and `offline_access`.",
+    "unauthorized_client": "The app does not allow authorization-code login.",
+}
+
+
+@dataclass
+class TokenSet:
+    access_token: str
+    refresh_token: str | None = None
+    expires_at: float = 0.0
+    scope: str = ""
+
+    @property
+    def is_fresh(self) -> bool:
+        return bool(self.access_token) and time.time() < self.expires_at - EXPIRY_BUFFER_SECONDS
+
+
+def authorize_endpoint(host: str) -> str:
+    return f"{host.rstrip('/')}/oidc/v1/authorize"
+
+
+def token_endpoint(host: str) -> str:
+    return f"{host.rstrip('/')}/oidc/v1/token"
+
+
+def generate_pkce_pair() -> tuple[str, str]:
+    verifier = secrets.token_urlsafe(64)
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+    return verifier, challenge
+
+
+def redirect_uri() -> str:
+    return os.environ.get(REDIRECT_URI_ENV, "").strip() or DEFAULT_REDIRECT_URI
+
+
+def build_authorize_url(
+    host: str,
+    client_id: str,
+    *,
+    code_challenge: str,
+    state: str,
+    redirect: str | None = None,
+    scopes: tuple[str, ...] = DEFAULT_SCOPES,
+) -> str:
+    query = urllib.parse.urlencode(
+        {
+            "client_id": client_id,
+            "response_type": "code",
+            "redirect_uri": redirect or redirect_uri(),
+            "scope": " ".join(scopes),
+            "state": state,
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256",
+        }
+    )
+    return f"{authorize_endpoint(host)}?{query}"
+
+
+def _format_oauth_error(exc: urllib.error.HTTPError) -> str:
+    try:
+        detail = json.loads(exc.read().decode("utf-8"))
+    except (OSError, ValueError):
+        detail = {}
+    if isinstance(detail, dict) and detail.get("error"):
+        code = str(detail.get("error"))
+        description = str(detail.get("error_description") or "").strip()
+        message = f"Databricks OAuth error `{code}`"
+        if description:
+            message += f": {description}"
+        hint = _ERROR_HINTS.get(code)
+        if hint:
+            message += f"\n{hint}"
+        return message
+    return f"Databricks token endpoint returned HTTP {exc.code}."
+
+
+def _post_form(url: str, fields: dict[str, str]) -> dict:
+    body = urllib.parse.urlencode(fields).encode("utf-8")
+    request = urllib.request.Request(
+        url,
+        data=body,
+        method="POST",
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as response:
+            payload = json.loads(response.read().decode("utf-8") or "{}")
+    except urllib.error.HTTPError as exc:
+        raise RuntimeError(_format_oauth_error(exc)) from None
+    except (urllib.error.URLError, TimeoutError, OSError) as exc:
+        raise RuntimeError(f"Could not reach the Databricks token endpoint {url}: {exc}") from None
+    except json.JSONDecodeError:
+        raise RuntimeError(f"Databricks token endpoint {url} returned invalid JSON.") from None
+    if not isinstance(payload, dict):
+        raise RuntimeError(f"Databricks token endpoint {url} returned invalid JSON.")
+    return payload
+
+
+def _token_set_from_payload(payload: dict, *, keep_refresh_token: str | None = None) -> TokenSet:
+    access_token = str(payload.get("access_token") or "")
+    if not access_token:
+        raise RuntimeError("Databricks token endpoint returned no access_token.")
+    try:
+        lifetime = float(payload.get("expires_in", 3600))
+    except (TypeError, ValueError):
+        lifetime = 3600.0
+    # A refresh response is not required to rotate the refresh token; when it
+    # doesn't, keep the one we already hold or the session is lost on next call.
+    refresh_token = payload.get("refresh_token") or keep_refresh_token
+    return TokenSet(
+        access_token=access_token,
+        refresh_token=str(refresh_token) if refresh_token else None,
+        expires_at=time.time() + lifetime,
+        scope=str(payload.get("scope") or ""),
+    )
+
+
+def exchange_code(
+    host: str,
+    client_id: str,
+    *,
+    code: str,
+    code_verifier: str,
+    redirect: str | None = None,
+    scopes: tuple[str, ...] = DEFAULT_SCOPES,
+) -> TokenSet:
+    payload = _post_form(
+        token_endpoint(host),
+        {
+            "grant_type": "authorization_code",
+            "client_id": client_id,
+            "scope": " ".join(scopes),
+            "code": code,
+            "redirect_uri": redirect or redirect_uri(),
+            "code_verifier": code_verifier,
+        },
+    )
+    return _token_set_from_payload(payload)
+
+
+def refresh_access_token(
+    host: str,
+    client_id: str,
+    refresh_token: str,
+) -> TokenSet:
+    payload = _post_form(
+        token_endpoint(host),
+        {
+            "grant_type": "refresh_token",
+            "client_id": client_id,
+            "refresh_token": refresh_token,
+        },
+    )
+    return _token_set_from_payload(payload, keep_refresh_token=refresh_token)
+
+
+def cache_key(host: str, client_id: str) -> str:
+    return f"{host.rstrip('/')}::{client_id}"
+
+
+def _read_entries() -> dict:
+    try:
+        raw = json.loads(TOKEN_CACHE_PATH.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    if not isinstance(raw, dict) or raw.get("version") != CACHE_VERSION:
+        return {}
+    entries = raw.get("entries")
+    return entries if isinstance(entries, dict) else {}
+
+
+def load_cached_tokens(host: str, client_id: str) -> TokenSet | None:
+    entry = _read_entries().get(cache_key(host, client_id))
+    if not isinstance(entry, dict):
+        return None
+    access_token = entry.get("access_token")
+    refresh_token = entry.get("refresh_token")
+    if not isinstance(access_token, str) and not isinstance(refresh_token, str):
+        return None
+    try:
+        expires_at = float(entry.get("expires_at") or 0.0)
+    except (TypeError, ValueError):
+        expires_at = 0.0
+    return TokenSet(
+        access_token=access_token if isinstance(access_token, str) else "",
+        refresh_token=refresh_token if isinstance(refresh_token, str) else None,
+        expires_at=expires_at,
+        scope=str(entry.get("scope") or ""),
+    )
+
+
+def store_tokens(host: str, client_id: str, tokens: TokenSet) -> None:
+    if is_dry_run():
+        return
+    entries = dict(_read_entries())
+    entries[cache_key(host, client_id)] = {
+        "access_token": tokens.access_token,
+        "refresh_token": tokens.refresh_token,
+        "expires_at": tokens.expires_at,
+        "scope": tokens.scope,
+    }
+    payload = json.dumps({"version": CACHE_VERSION, "entries": entries}, indent=2)
+    try:
+        APP_DIR.mkdir(parents=True, exist_ok=True)
+        descriptor = os.open(TOKEN_CACHE_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+            handle.write(payload + "\n")
+        os.chmod(TOKEN_CACHE_PATH, 0o600)
+    except OSError as exc:
+        raise RuntimeError(f"Failed to write the OAuth token cache {TOKEN_CACHE_PATH}") from exc
+
+
+def not_signed_in_message(host: str, client_id: str) -> str:
+    return (
+        f"No custom-OAuth session for {host} (client id {client_id}). Run "
+        f"`ug configure --oauth-client-id {client_id}` once to sign in; after that "
+        "tokens refresh without a browser."
+    )
+
+
+def get_token(
+    host: str,
+    client_id: str,
+    *,
+    force_refresh: bool = False,
+) -> str:
+    cached = load_cached_tokens(host, client_id)
+    if cached and cached.is_fresh and not force_refresh:
+        return cached.access_token
+    if not cached or not cached.refresh_token:
+        raise RuntimeError(not_signed_in_message(host, client_id))
+    tokens = refresh_access_token(host, client_id, cached.refresh_token)
+    store_tokens(host, client_id, tokens)
+    return tokens.access_token
+
+
+def _await_callback(redirect: str, timeout: float) -> dict[str, str]:
+    parsed = urllib.parse.urlparse(redirect)
+    captured: dict[str, str] = {}
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's API
+            query = urllib.parse.urlparse(self.path).query
+            params = {key: values[0] for key, values in urllib.parse.parse_qs(query).items()}
+            if "code" not in params and "error" not in params:
+                self.send_response(404)
+                self.end_headers()
+                return
+            captured.update(params)
+            succeeded = "code" in params
+            note = (
+                "Signed in. You can close this tab and return to your terminal."
+                if succeeded
+                else "Authorization failed: "
+                + (params.get("error_description") or params.get("error") or "unknown error")
+            )
+            body = (
+                '<html><body style="font-family:system-ui;padding:2rem">'
+                f"<p>{note}</p></body></html>"
+            ).encode()
+            self.send_response(200 if succeeded else 400)
+            self.send_header("Content-Type", "text/html; charset=utf-8")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, format: str, *args: object) -> None:  # noqa: A002
+            pass
+
+    address = (parsed.hostname or "localhost", parsed.port or 80)
+    try:
+        server = ThreadingHTTPServer(address, Handler)
+    except OSError as exc:
+        raise RuntimeError(
+            f"Could not listen on {address[0]}:{address[1]} to receive the OAuth "
+            f"redirect ({exc}). Free that port, or set {REDIRECT_URI_ENV} to a "
+            "redirect URI registered on the app integration."
+        ) from None
+    deadline = time.time() + timeout
+    with server:
+        server.timeout = 1.0
+        while not captured and time.time() < deadline:
+            server.handle_request()
+    return captured
+
+
+def login(
+    host: str,
+    client_id: str,
+    *,
+    scopes: tuple[str, ...] = DEFAULT_SCOPES,
+    open_browser: bool = True,
+    timeout: float = _LOGIN_TIMEOUT_SECONDS,
+) -> TokenSet:
+    import webbrowser
+
+    from ucode.ui import print_note, print_section, print_success
+
+    verifier, challenge = generate_pkce_pair()
+    expected_state = secrets.token_urlsafe(16)
+    redirect = redirect_uri()
+    url = build_authorize_url(
+        host,
+        client_id,
+        code_challenge=challenge,
+        state=expected_state,
+        redirect=redirect,
+        scopes=scopes,
+    )
+
+    print_section("Databricks Custom OAuth Login")
+    print_note(f"Opening a browser to sign in to {host}.")
+    print_note(f"If it doesn't open, visit:\n{url}")
+    if open_browser:
+        try:
+            webbrowser.open(url)
+        except webbrowser.Error:
+            pass
+
+    captured = _await_callback(redirect, timeout)
+    if not captured:
+        raise RuntimeError(
+            f"Timed out after {int(timeout)}s waiting for the OAuth redirect to {redirect}. "
+            "Confirm that URI is registered on the app integration."
+        )
+    if captured.get("error"):
+        detail = captured.get("error_description") or captured["error"]
+        raise RuntimeError(f"Databricks refused the authorization request: {detail}")
+    if captured.get("state") != expected_state:
+        raise RuntimeError(
+            "OAuth state mismatch — the redirect did not come from this login attempt. "
+            "Nothing was saved; run the command again."
+        )
+    code = captured.get("code") or ""
+    if not code:
+        raise RuntimeError("The OAuth redirect carried no authorization code.")
+
+    tokens = exchange_code(
+        host,
+        client_id,
+        code=code,
+        code_verifier=verifier,
+        redirect=redirect,
+        scopes=scopes,
+    )
+    if not tokens.refresh_token:
+        raise RuntimeError(
+            "Databricks issued an access token but no refresh token, so the session "
+            "cannot outlive it. Add `offline_access` to the app integration's scopes."
+        )
+    store_tokens(host, client_id, tokens)
+    print_success(f"Custom OAuth session saved for {host}")
+    return tokens
diff --git a/tests/conftest.py b/tests/conftest.py
index ce36378b..885a43f9 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -27,6 +27,7 @@ def _isolate_ucode_state(tmp_path, monkeypatch):
     import ucode.config_io as config_io_mod
     import ucode.databricks as databricks_mod
     import ucode.managed_files as managed_files_mod
+    import ucode.oauth as oauth_mod
     import ucode.state as state_mod
     from ucode.agents import codex as codex_mod
 
@@ -34,6 +35,8 @@ def _isolate_ucode_state(tmp_path, monkeypatch):
     state_dir.mkdir()
     monkeypatch.setattr(state_mod, "STATE_PATH", state_dir / "state.json")
     monkeypatch.setattr(config_io_mod, "APP_DIR", state_dir)
+    # TOKEN_CACHE_PATH is resolved at import time.
+    monkeypatch.setattr(oauth_mod, "TOKEN_CACHE_PATH", state_dir / "oauth-tokens.json")
     backup_dir = state_dir / "managed-backups"
     monkeypatch.setattr(managed_files_mod, "MANAGED_BACKUP_DIR", backup_dir)
     monkeypatch.setattr(
diff --git a/tests/test_oauth.py b/tests/test_oauth.py
new file mode 100644
index 00000000..13a9a08e
--- /dev/null
+++ b/tests/test_oauth.py
@@ -0,0 +1,314 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+import json
+import stat
+import time
+import urllib.error
+import urllib.parse
+from unittest.mock import patch
+
+import pytest
+
+from ucode import oauth
+
+HOST = "https://dbc-test.cloud.databricks.com"
+CLIENT_ID = "abc-client-id"
+
+
+class _FakeResponse:
+    def __init__(self, payload: dict):
+        self._body = json.dumps(payload).encode("utf-8")
+
+    def read(self) -> bytes:
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args) -> bool:
+        return False
+
+
+def _http_error(status: int, payload: dict) -> urllib.error.HTTPError:
+    return urllib.error.HTTPError(
+        "https://example.invalid/oidc/v1/token",
+        status,
+        "error",
+        {},  # type: ignore[arg-type]
+        io.BytesIO(json.dumps(payload).encode("utf-8")),
+    )
+
+
+def _posted_fields(mock_urlopen) -> dict[str, str]:
+    request = mock_urlopen.call_args[0][0]
+    return {
+        key: values[0]
+        for key, values in urllib.parse.parse_qs(request.data.decode("utf-8")).items()
+    }
+
+
+class TestGeneratePkcePair:
+    def test_challenge_is_unpadded_base64url_sha256_of_verifier(self):
+        verifier, challenge = oauth.generate_pkce_pair()
+        expected = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
+        assert challenge == expected.rstrip(b"=").decode()
+
+    def test_verifier_length_is_within_rfc7636_range(self):
+        verifier, _ = oauth.generate_pkce_pair()
+        assert 43 <= len(verifier) <= 128
+
+    def test_each_call_is_unique(self):
+        assert oauth.generate_pkce_pair()[0] != oauth.generate_pkce_pair()[0]
+
+
+class TestEndpoints:
+    def test_workspace_endpoints(self):
+        assert oauth.token_endpoint(HOST) == f"{HOST}/oidc/v1/token"
+        assert oauth.authorize_endpoint(HOST) == f"{HOST}/oidc/v1/authorize"
+
+    def test_trailing_slash_does_not_double_up(self):
+        assert oauth.token_endpoint(f"{HOST}/") == f"{HOST}/oidc/v1/token"
+
+
+class TestBuildAuthorizeUrl:
+    def _params(self, **kwargs) -> dict[str, str]:
+        url = oauth.build_authorize_url(
+            HOST, CLIENT_ID, code_challenge="chal", state="st", **kwargs
+        )
+        return {
+            key: values[0]
+            for key, values in urllib.parse.parse_qs(urllib.parse.urlparse(url).query).items()
+        }
+
+    def test_requests_s256_pkce_and_a_code(self):
+        params = self._params()
+        assert params["code_challenge_method"] == "S256"
+        assert params["code_challenge"] == "chal"
+        assert params["response_type"] == "code"
+        assert params["client_id"] == CLIENT_ID
+        assert params["state"] == "st"
+
+    def test_requests_offline_access_so_a_refresh_token_comes_back(self):
+        assert "offline_access" in self._params()["scope"].split()
+
+    def test_sends_no_client_secret(self):
+        assert "client_secret" not in self._params()
+
+    def test_scopes_are_space_joined(self):
+        assert self._params(scopes=("all-apis", "offline_access"))["scope"] == (
+            "all-apis offline_access"
+        )
+
+
+class TestExchangeCode:
+    def test_posts_authorization_code_grant_without_a_secret(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse(
+                {"access_token": "at", "refresh_token": "rt", "expires_in": 3600}
+            )
+            tokens = oauth.exchange_code(
+                HOST, CLIENT_ID, code="the-code", code_verifier="the-verifier"
+            )
+        fields = _posted_fields(mock_urlopen)
+        assert fields["grant_type"] == "authorization_code"
+        assert fields["code"] == "the-code"
+        assert fields["code_verifier"] == "the-verifier"
+        assert fields["client_id"] == CLIENT_ID
+        assert fields["scope"] == "all-apis offline_access"
+        assert "client_secret" not in fields
+        assert tokens.access_token == "at"
+        assert tokens.refresh_token == "rt"
+
+    def test_posts_form_encoded_with_no_authorization_header(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"access_token": "at", "expires_in": 60})
+            oauth.exchange_code(HOST, CLIENT_ID, code="c", code_verifier="v")
+        request = mock_urlopen.call_args[0][0]
+        assert request.headers["Content-type"] == "application/x-www-form-urlencoded"
+        assert "Authorization" not in request.headers
+
+    def test_missing_access_token_is_an_error(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"expires_in": 60})
+            with pytest.raises(RuntimeError, match="no access_token"):
+                oauth.exchange_code(HOST, CLIENT_ID, code="c", code_verifier="v")
+
+
+class TestRefreshAccessToken:
+    def test_posts_refresh_token_grant_without_a_secret(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"access_token": "new", "expires_in": 3600})
+            oauth.refresh_access_token(HOST, CLIENT_ID, "old-rt")
+        fields = _posted_fields(mock_urlopen)
+        assert fields == {
+            "grant_type": "refresh_token",
+            "client_id": CLIENT_ID,
+            "refresh_token": "old-rt",
+        }
+
+    def test_keeps_the_existing_refresh_token_when_the_response_omits_one(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"access_token": "new", "expires_in": 3600})
+            tokens = oauth.refresh_access_token(HOST, CLIENT_ID, "old-rt")
+        assert tokens.refresh_token == "old-rt"
+
+    def test_uses_a_rotated_refresh_token_when_one_is_returned(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse(
+                {"access_token": "new", "refresh_token": "rotated", "expires_in": 3600}
+            )
+            tokens = oauth.refresh_access_token(HOST, CLIENT_ID, "old-rt")
+        assert tokens.refresh_token == "rotated"
+
+
+class TestOauthErrorMessages:
+    def test_invalid_client_explains_the_public_app_requirement(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = _http_error(
+                401,
+                {"error": "invalid_client", "error_description": "Client authentication failed"},
+            )
+            with pytest.raises(RuntimeError) as excinfo:
+                oauth.refresh_access_token(HOST, CLIENT_ID, "rt")
+        message = str(excinfo.value)
+        assert "invalid_client" in message
+        assert "Client authentication failed" in message
+        assert "public app" in message
+
+    def test_invalid_grant_tells_the_user_how_to_sign_in_again(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = _http_error(
+                401, {"error": "invalid_grant", "error_description": "Refresh token is invalid"}
+            )
+            with pytest.raises(RuntimeError, match="--oauth-client-id"):
+                oauth.refresh_access_token(HOST, CLIENT_ID, "rt")
+
+    def test_unreachable_endpoint_is_reported_with_the_url(self):
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.side_effect = urllib.error.URLError("no route to host")
+            with pytest.raises(RuntimeError, match="Could not reach"):
+                oauth.refresh_access_token(HOST, CLIENT_ID, "rt")
+
+
+class TestTokenCache:
+    def test_round_trips_a_token_set(self):
+        tokens = oauth.TokenSet("at", "rt", time.time() + 3600, "all-apis offline_access")
+        oauth.store_tokens(HOST, CLIENT_ID, tokens)
+        loaded = oauth.load_cached_tokens(HOST, CLIENT_ID)
+        assert loaded is not None
+        assert (loaded.access_token, loaded.refresh_token) == ("at", "rt")
+        assert loaded.scope == "all-apis offline_access"
+
+    def test_cache_file_is_owner_only(self):
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("at", "rt", time.time() + 60))
+        mode = stat.S_IMODE(oauth.TOKEN_CACHE_PATH.stat().st_mode)
+        assert mode == 0o600, f"refresh token cache is mode {oct(mode)}"
+
+    def test_entries_are_keyed_by_host_and_client_id(self):
+        oauth.store_tokens(HOST, "client-a", oauth.TokenSet("at-a", "rt-a", time.time() + 60))
+        oauth.store_tokens(HOST, "client-b", oauth.TokenSet("at-b", "rt-b", time.time() + 60))
+        a = oauth.load_cached_tokens(HOST, "client-a")
+        b = oauth.load_cached_tokens(HOST, "client-b")
+        assert a is not None and b is not None
+        assert a.access_token == "at-a"
+        assert b.access_token == "at-b"
+
+    def test_unknown_client_id_is_a_cache_miss(self):
+        oauth.store_tokens(HOST, "client-a", oauth.TokenSet("at-a", "rt-a", time.time() + 60))
+        assert oauth.load_cached_tokens(HOST, "other-client") is None
+
+    def test_missing_cache_file_is_a_miss_not_an_error(self):
+        assert oauth.load_cached_tokens(HOST, CLIENT_ID) is None
+
+    def test_corrupt_cache_file_is_a_miss_not_an_error(self):
+        oauth.TOKEN_CACHE_PATH.write_text("{ not json", encoding="utf-8")
+        assert oauth.load_cached_tokens(HOST, CLIENT_ID) is None
+
+    def test_dry_run_writes_nothing(self, monkeypatch):
+        monkeypatch.setattr(oauth, "is_dry_run", lambda: True)
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("at", "rt", time.time() + 60))
+        assert not oauth.TOKEN_CACHE_PATH.exists()
+
+
+class TestTokenSetFreshness:
+    def test_token_expiring_inside_the_buffer_is_not_fresh(self):
+        tokens = oauth.TokenSet("at", "rt", time.time() + oauth.EXPIRY_BUFFER_SECONDS - 5)
+        assert not tokens.is_fresh
+
+    def test_token_well_inside_its_lifetime_is_fresh(self):
+        assert oauth.TokenSet("at", "rt", time.time() + 3600).is_fresh
+
+    def test_empty_access_token_is_never_fresh(self):
+        assert not oauth.TokenSet("", "rt", time.time() + 3600).is_fresh
+
+
+class TestGetToken:
+    def test_fresh_cached_token_is_served_without_any_request(self):
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("cached", "rt", time.time() + 3600))
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            assert oauth.get_token(HOST, CLIENT_ID) == "cached"
+        mock_urlopen.assert_not_called()
+
+    def test_expired_token_is_refreshed_and_the_new_one_cached(self):
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("stale", "rt", time.time() - 10))
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse(
+                {"access_token": "refreshed", "expires_in": 3600}
+            )
+            assert oauth.get_token(HOST, CLIENT_ID) == "refreshed"
+        cached = oauth.load_cached_tokens(HOST, CLIENT_ID)
+        assert cached is not None and cached.access_token == "refreshed"
+
+    def test_force_refresh_ignores_a_still_fresh_token(self):
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("cached", "rt", time.time() + 3600))
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"access_token": "fresh", "expires_in": 3600})
+            assert oauth.get_token(HOST, CLIENT_ID, force_refresh=True) == "fresh"
+
+    def test_no_cached_session_raises_an_actionable_error(self):
+        with pytest.raises(RuntimeError, match="ug configure --oauth-client-id"):
+            oauth.get_token(HOST, CLIENT_ID)
+
+    def test_never_opens_a_browser(self):
+        oauth.store_tokens(HOST, CLIENT_ID, oauth.TokenSet("cached", "rt", time.time() + 3600))
+        with patch("webbrowser.open") as mock_open:
+            oauth.get_token(HOST, CLIENT_ID)
+        mock_open.assert_not_called()
+
+
+class TestLogin:
+    def _login(self, captured: dict, monkeypatch):
+        monkeypatch.setattr(oauth, "_await_callback", lambda *_a, **_k: captured)
+        monkeypatch.setattr("webbrowser.open", lambda *_a, **_k: True)
+        return oauth.login(HOST, CLIENT_ID, open_browser=False)
+
+    def test_state_mismatch_is_rejected_and_nothing_is_cached(self, monkeypatch):
+        with pytest.raises(RuntimeError, match="state mismatch"):
+            self._login({"code": "c", "state": "not-the-state-we-sent"}, monkeypatch)
+        assert oauth.load_cached_tokens(HOST, CLIENT_ID) is None
+
+    def test_timeout_with_no_redirect_is_reported(self, monkeypatch):
+        with pytest.raises(RuntimeError, match="Timed out"):
+            self._login({}, monkeypatch)
+
+    def test_authorization_error_is_surfaced(self, monkeypatch):
+        with pytest.raises(RuntimeError, match="access_denied"):
+            self._login({"error": "access_denied"}, monkeypatch)
+
+    def test_a_response_without_a_refresh_token_is_rejected(self, monkeypatch):
+        def fake_await(redirect, timeout):
+            return {"code": "c", "state": fake_await.state}
+
+        def capture_state(*args, **kwargs):
+            fake_await.state = kwargs["state"]
+            return "https://example.invalid/authorize"
+
+        monkeypatch.setattr(oauth, "_await_callback", fake_await)
+        monkeypatch.setattr(oauth, "build_authorize_url", capture_state)
+        with patch("ucode.oauth.urllib.request.urlopen") as mock_urlopen:
+            mock_urlopen.return_value = _FakeResponse({"access_token": "at", "expires_in": 3600})
+            with pytest.raises(RuntimeError, match="offline_access"):
+                oauth.login(HOST, CLIENT_ID, open_browser=False)