diff --git a/docs/DEVICE_AUTHENTICATION.md b/docs/DEVICE_AUTHENTICATION.md
new file mode 100644
index 0000000..032fc19
--- /dev/null
+++ b/docs/DEVICE_AUTHENTICATION.md
@@ -0,0 +1,68 @@
+# Desktop Cloud authentication
+
+OpenAdapt Desktop uses one Cloud bearer format: `oai_ingest_…`. It stores the
+bearer in the operating system credential store. The token-paste path remains
+available for headless systems and recovery.
+
+## Browser sign-in
+
+`openadapt-desktop login --provider browser_pkce` starts an RFC 8252 loopback
+flow:
+
+1. Desktop binds an ephemeral `127.0.0.1` port and serves only `/callback`.
+2. Desktop creates an S256 PKCE verifier and challenge plus a random `state`.
+3. Desktop opens the OpenAdapt Cloud login page in the system browser.
+4. After sign-in, Cloud shows the user a Connect confirmation.
+5. Cloud redirects to the exact loopback URL with a five-minute, one-use
+   `oap_…` pairing code and the exact `state`.
+6. Desktop checks the state and code prefix. It claims the pairing with the
+   PKCE verifier through `/api/local-bridge/pairings/claim`.
+7. Desktop stages the returned `oai_ingest_…` bearer, validates it, promotes it
+   atomically, and confirms the connection. A failure invokes the existing
+   acknowledged rollback path.
+
+The code is not a Supabase authorization code. Desktop does not call a
+Supabase token endpoint, store a Supabase session, or add a loopback URL to a
+Supabase allowlist. `oapp_…` and `oaps_…` credentials belong only to the local
+attended-decision portal and cannot authenticate a Cloud request.
+
+Browser sign-in is unavailable on a headless machine, without a display on
+Linux, or without an unlocked secure credential store. Desktop checks the
+store before it claims the one-use code. Token paste remains available:
+
+```console
+openadapt-desktop login --provider paste
+```
+
+## Expiry and renewal
+
+New Cloud credentials expire within 90 days. Cloud enforces the deadline on
+every authenticated request. Desktop treats its local deadline as a warning,
+not as proof that a credential remains valid.
+
+Cloud reports the warning contract through `/api/needs-attention/count`.
+Desktop checks that the JSON and the `x-openadapt-credential-*` headers agree.
+Cloud sets `expiring_soon` from the exact server timestamp. The
+`expires_in_days` value is a floor-rounded display value. Thus, day 14 can
+report either state. Desktop uses the server `expiring_soon` value and does not
+recompute it from the local clock. A legacy credential can report that it has
+no expiry; the operator can rotate it to enter the 90-day policy.
+
+```console
+openadapt-desktop credential
+openadapt-desktop rotate
+```
+
+Rotation creates the replacement first. The old bearer stays valid for at
+most seven days. Desktop durably stages the one-time response before it changes
+the active keychain entries. It then authenticates one independent
+`/api/needs-attention/count` request with the replacement and checks the exact
+lifetime contract before promotion. After a process interruption, Desktop
+validates and finishes that exact stage. It does not send a second rotation
+request. Recovery requires the same stored expiry, but the remaining-day count
+can decrease while the machine is offline. A later successful login safely
+supersedes an older rejected rotation stage.
+
+If the response is lost before Desktop can stage it, the old bearer remains
+usable during the overlap. Cloud cannot replay the raw replacement, so the
+operator must sign in again. Do not retry rotation with the same old bearer.
diff --git a/engine/auth/__init__.py b/engine/auth/__init__.py
index 9cf688f..e1e2703 100644
--- a/engine/auth/__init__.py
+++ b/engine/auth/__init__.py
@@ -72,14 +72,16 @@ def login(host: str = DEFAULT_HOST, prefer: str | None = None) -> Credential:
     if prefer:
         providers = [p for p in providers if p.name == prefer] or providers
 
-    last_error: Exception | None = None
     for provider in providers:
         if not provider.is_available():
             continue
         try:
             return provider.login()
-        except Exception as exc:  # try the next provider (e.g. browser -> paste)
-            last_error = exc
-    if last_error:
-        raise RuntimeError(f"Login failed: {last_error}") from last_error
+        except Exception as exc:
+            # An available provider can fail after it receives or stages a
+            # one-use credential. Starting another provider in the same call
+            # could overwrite its retained recovery state. Token paste remains
+            # the fallback when browser login is unavailable, and the operator
+            # can select it explicitly after a completed browser failure.
+            raise RuntimeError(f"Login failed: {exc}") from exc
     raise RuntimeError("No auth provider is available on this machine.")
diff --git a/engine/auth/browser_pkce.py b/engine/auth/browser_pkce.py
index 0bcb079..a0f7640 100644
--- a/engine/auth/browser_pkce.py
+++ b/engine/auth/browser_pkce.py
@@ -1,52 +1,37 @@
-"""BrowserPkceProvider -- v1 "click Login" for interactive desktop users.
-
-Flow (spec section 3a):
-    1. Generate a PKCE verifier/challenge and bind an ephemeral loopback
-       listener on ``127.0.0.1:<port>`` at path ``/callback``.
-    2. Open the hosted login page in the SYSTEM browser (so Google / magic-link
-       "just work" with zero forked auth), passing the loopback redirect URI +
-       PKCE challenge.
-    3. The browser redirects back to the loopback with ``?code=…``; the listener
-       captures it and serves a minimal "you can close this tab" page.
-    4. Exchange the code for a Supabase session via PKCE (using the code
-       verifier), then MINT an ingest token via the hosted API.
-    5. Store a single active credential whose bearer ``token`` is the minted
-       ingest token (so the headless push / count path always has a bearer),
-       carrying the Supabase ``refresh_token`` so the session can be renewed.
-
-``is_available()`` is False on a headless server (no browser / no loopback), so
-the UI falls back to :class:`~engine.auth.paste.PasteTokenProvider`.
-
-NOT WIRED END TO END -- do not enable this provider by setting the Supabase
-environment variables alone. Steps 2-4 above each depend on a hosted contract
-that ``app.openadapt.ai`` does not implement today:
-
-* ``GET /login`` reads only ``checkout_session_id`` and ``next`` (clamped to a
-  same-origin ``/dashboard`` destination). It ignores ``redirect_to``,
-  ``code_challenge``, ``code_challenge_method`` and ``state``, so the browser
-  never navigates to the loopback listener and step 3 cannot happen.
-* The Supabase ``uri_allow_list`` is rewritten on every production deploy and
-  contains no loopback entry, so Supabase would reject a ``127.0.0.1`` redirect
-  even if the page forwarded one.
-* ``POST /api/ingest-tokens`` authenticates the browser session cookie and
-  requires a ``name`` field. It rejects a ``Authorization: Bearer <supabase
-  access token>`` with 401, so step 4 cannot mint.
-
-Because ``_exchange_code`` is only reached AFTER the loopback wait, leaving the
-provider enabled makes ``login`` open a browser tab and then block for the full
-``timeout`` before failing over to token paste. ``is_available()`` therefore
-requires the Supabase configuration up front: the provider reports unavailable
-rather than costing the operator a silent multi-minute stall. Restoring it is a
-hosted-contract change, not a client change -- see the three bullets above.
-
-The working, shipped path to a credential is the one-time pairing flow
-(:mod:`engine.auth.pairing`), which the hosted control plane does implement.
-
-Reconciliation note: the shared store (:mod:`engine.auth.store`) keys ONE active
-credential per host and ``auth_header()`` returns exactly one bearer. We
-therefore fold the Supabase session and the minted ingest token into a single
-stored ``Credential`` (``kind="ingest_token"``, ``token`` = ingest token,
-``refresh_token`` = Supabase refresh) rather than two competing entries.
+"""BrowserPkceProvider -- "click Login" for interactive desktop users.
+
+An RFC 8252 native-app login. The desktop cannot keep a client secret, so it
+uses the SYSTEM browser (where Google and magic-link sign-in already work) plus
+an ephemeral loopback listener, and binds the exchange with S256 PKCE.
+
+Flow:
+    1. Generate a PKCE verifier/challenge and bind a listener on
+       ``127.0.0.1:<ephemeral port>`` that serves ONLY ``/callback``.
+    2. Open ``{host}/login`` in the system browser, passing the loopback
+       redirect URI, the challenge, and an opaque ``state``.
+    3. The user signs in normally. OpenAdapt's own authenticated
+       ``/auth/loopback`` page then redirects to the listener with
+       ``?code=oap_...&state=...``.
+    4. The listener validates ``state`` and redeems the code through the
+       EXISTING pairing path (:func:`engine.auth.pairing.claim_pairing`),
+       sending the verifier. Cloud consumes the one-use code, checks the PKCE
+       binding, and mints a user/org-bound ingest credential.
+    5. That path stages, independently validates, atomically promotes, and
+       confirms the credential in the OS keychain, rolling back on any failure.
+
+Why the redirect comes from our own page rather than from Supabase: Supabase's
+``uri_allow_list`` is rewritten by ``scripts/setup-supabase.mjs`` on every
+deploy, and RFC 8252 section 7.3 requires the authorization server to accept an
+arbitrary loopback port -- a promise that allow-list glob grammar cannot make.
+Routing the redirect through an authenticated OpenAdapt page keeps Supabase
+configuration untouched and reuses the hardened pairing redemption whole. It
+also means browser login introduces NO new credential format: the ``code`` is
+an ``oap_`` pairing secret, and the result is the same ``oai_ingest_...``
+credential every other path produces.
+
+``is_available()`` is False on a headless box (no browser, no user at the
+keyboard) and False when no OS keychain can hold the result, so the UI falls
+back to :class:`~engine.auth.paste.PasteTokenProvider`.
 """
 
 from __future__ import annotations
@@ -60,22 +45,26 @@
 import urllib.parse
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
-import httpx
 from loguru import logger
 
+from engine.auth.pairing import (
+    PAIRING_SECRET_RE,
+    PairingError,
+    _validate_destination,
+    claim_pairing,
+)
 from engine.auth.provider import Credential
-from engine.auth.store import DEFAULT_HOST, store_credential
+from engine.auth.store import DEFAULT_HOST, load_credential, secure_store_available
 
 # Hosted login page opened in the system browser.
 LOGIN_PATH = "/login"
-# Path the loopback listener serves for the OAuth redirect.
+# The ONLY path the loopback listener serves.
 CALLBACK_PATH = "/callback"
-# Hosted endpoint that mints an ingest token from an authenticated session.
-MINT_PATH = "/api/ingest-tokens"
 
-# Supabase project config (coordinator-provisioned). Absent in CI/tests.
-SUPABASE_URL_ENV = "OPENADAPT_SUPABASE_URL"
-SUPABASE_ANON_KEY_ENV = "OPENADAPT_SUPABASE_ANON_KEY"
+# How long to wait for the browser round trip. The user can complete a magic
+# link and then explicitly choose Connect before Cloud mints the five-minute
+# pairing code. No server secret exists during this wait.
+DEFAULT_TIMEOUT_S = 300.0
 
 _CALLBACK_HTML = (
     b"<!doctype html><html><head><meta charset='utf-8'>"
@@ -87,6 +76,26 @@
     b"</body></html>"
 )
 
+_ERROR_HTML = (
+    b"<!doctype html><html><head><meta charset='utf-8'>"
+    b"<title>OpenAdapt login failed</title></head>"
+    b"<body style='font-family:system-ui;text-align:center;padding-top:3rem'>"
+    b"<h2>Sign-in did not complete.</h2>"
+    b"<p>Return to OpenAdapt for the reason.</p>"
+    b"</body></html>"
+)
+
+
+def _write_callback_body(stream, body: bytes, event: threading.Event) -> None:
+    """Signal callback completion even when the browser closes before reading."""
+    try:
+        stream.write(body)
+    finally:
+        # The callback parameters were already retained. A browser disconnect
+        # during this courtesy page must not make login wait for the full
+        # timeout or discard a valid one-use code.
+        event.set()
+
 
 def generate_pkce_pair() -> tuple[str, str]:
     """Return ``(code_verifier, code_challenge)`` for the S256 PKCE method."""
@@ -97,11 +106,17 @@ def generate_pkce_pair() -> tuple[str, str]:
 
 
 class _LoopbackReceiver:
-    """Ephemeral 127.0.0.1 listener that captures the OAuth ``code``."""
+    """Ephemeral 127.0.0.1 listener that captures one authorization code.
+
+    Binds ``127.0.0.1`` explicitly -- never ``0.0.0.0`` -- so nothing outside
+    this machine can reach it, and takes an ephemeral port because RFC 8252
+    section 7.3 requires the client to choose one at request time.
+    """
 
     def __init__(self) -> None:
         self.code: str | None = None
         self.state: str | None = None
+        self.error_code: str | None = None
         self.error: str | None = None
         self._event = threading.Event()
         parent = self
@@ -112,21 +127,58 @@ def log_message(self, *args) -> None:  # noqa: D401 - silence stdlib logging
 
             def do_GET(self) -> None:  # noqa: N802 - stdlib naming
                 parsed = urllib.parse.urlparse(self.path)
-                if parsed.path.rstrip("/") not in ("", CALLBACK_PATH.rstrip("/")):
+                # Exactly one path. A scan of the ephemeral port finds nothing
+                # else, and no other route can deliver a code.
+                if parsed.path != CALLBACK_PATH:
                     self.send_response(404)
                     self.end_headers()
                     return
-                params = urllib.parse.parse_qs(parsed.query)
-                parent.code = (params.get("code") or [None])[0]
-                parent.state = (params.get("state") or [None])[0]
-                parent.error = (params.get("error") or [None])[0]
+                params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
+
+                def single(key: str) -> str | None:
+                    values = params.get(key) or []
+                    return values[0] if len(values) == 1 else None
+
+                # First delivery wins. A later request cannot overwrite a
+                # captured code with a different one.
+                if not parent._event.is_set():
+                    parent.state = single("state")
+                    code = single("code")
+                    error_code = single("error")
+                    description = single("error_description")
+                    repeated = any(len(values) != 1 for values in params.values())
+                    success_shape = set(params) == {"code", "state"}
+                    error_shape = set(params) in (
+                        {"error", "state"},
+                        {"error", "error_description", "state"},
+                    )
+                    safe_description = description is None or (
+                        len(description) <= 256
+                        and not any(ord(char) < 32 or ord(char) == 127 for char in description)
+                    )
+                    if repeated or not safe_description:
+                        parent.error = "The login callback was malformed."
+                    elif success_shape and code is not None:
+                        parent.code = code
+                    elif error_shape and error_code is not None:
+                        parent.error_code = error_code
+                        parent.error = description or error_code
+                    else:
+                        parent.error = "The login callback was malformed."
+                failed = parent.error is not None or parent.code is None
                 self.send_response(200)
                 self.send_header("Content-Type", "text/html; charset=utf-8")
+                self.send_header("Cache-Control", "no-store")
+                self.send_header("Referrer-Policy", "no-referrer")
                 self.end_headers()
-                self.wfile.write(_CALLBACK_HTML)
-                parent._event.set()
+                _write_callback_body(
+                    self.wfile,
+                    _ERROR_HTML if failed else _CALLBACK_HTML,
+                    parent._event,
+                )
 
         self._server = HTTPServer(("127.0.0.1", 0), Handler)
+        self._thread: threading.Thread | None = None
 
     @property
     def port(self) -> int:
@@ -136,20 +188,34 @@ def port(self) -> int:
     def redirect_uri(self) -> str:
         return f"http://127.0.0.1:{self.port}{CALLBACK_PATH}"
 
+    def __enter__(self) -> "_LoopbackReceiver":
+        return self
+
+    def __exit__(self, *exc_info) -> None:
+        self.close()
+
     def serve_until_code(self, timeout: float) -> None:
-        thread = threading.Thread(target=self._server.serve_forever, daemon=True)
-        thread.start()
-        try:
-            self._event.wait(timeout=timeout)
-        finally:
-            self._server.shutdown()
-            self._server.server_close()
+        """Serve until a callback arrives or ``timeout`` elapses."""
+        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
+        self._thread.start()
+        self._event.wait(timeout=timeout)
 
     def close(self) -> None:
+        """Tear the listener down. Safe to call twice, and on any exit path."""
+        # HTTPServer.shutdown() deadlocks when serve_forever() never started,
+        # for example when the browser launcher itself raises.
+        if self._thread is not None:
+            try:
+                self._server.shutdown()
+            except Exception:  # pragma: no cover - defensive stdlib boundary
+                pass
         try:
             self._server.server_close()
-        except Exception:
+        except Exception:  # pragma: no cover - already closed
             pass
+        if self._thread is not None:
+            self._thread.join(timeout=5.0)
+            self._thread = None
 
 
 class BrowserPkceProvider:
@@ -158,8 +224,6 @@ class BrowserPkceProvider:
     Args:
         host: Hosted base URL.
         open_browser: Callable that opens a URL in the system browser.
-        supabase_url: Supabase project URL (defaults to the ``OPENADAPT_SUPABASE_URL`` env).
-        supabase_anon_key: Supabase anon key (defaults to the env var).
         timeout: Seconds to wait for the browser redirect.
     """
 
@@ -169,36 +233,49 @@ def __init__(
         self,
         host: str = DEFAULT_HOST,
         open_browser=None,
-        supabase_url: str | None = None,
-        supabase_anon_key: str | None = None,
-        timeout: float = 180.0,
+        timeout: float = DEFAULT_TIMEOUT_S,
     ) -> None:
-        self.host = host.rstrip("/")
+        raw_host = host.rstrip("/")
+        kind = self._kind_for_host(raw_host)
+        try:
+            self.host = _validate_destination(raw_host, kind)
+            self._host_is_valid = True
+        except PairingError:
+            # Retain the safe, secret-free input only so `is_available()` can
+            # refuse it. No login URL or claim uses an invalid destination.
+            self.host = raw_host
+            self._host_is_valid = False
+        self._uses_system_browser = open_browser is None
         self._open_browser = open_browser or self._default_open_browser
-        self._supabase_url = (supabase_url or os.environ.get(SUPABASE_URL_ENV, "")).rstrip("/")
-        self._supabase_anon_key = supabase_anon_key or os.environ.get(SUPABASE_ANON_KEY_ENV, "")
         self._timeout = timeout
 
     @staticmethod
     def _default_open_browser(url: str) -> None:
         import webbrowser
 
-        webbrowser.open(url)
+        if not webbrowser.open(url):
+            raise RuntimeError("OpenAdapt could not open the system browser.")
 
     def is_available(self) -> bool:
-        """False on a headless box, and False while the flow is unconfigured.
-
-        The Supabase check is deliberately first. ``_exchange_code`` cannot run
-        without a project URL and anon key, but it is only reached after the
-        loopback wait -- so an unconfigured provider would open a browser tab
-        and stall for the whole ``timeout`` before failing over to token paste.
-        Reporting unavailable up front keeps ``login`` responsive and keeps the
-        provider chain honest about what it can actually complete.
+        """False on a headless box, and False without a usable OS keychain.
+
+        The keychain check is not cosmetic: this flow spends a one-use server
+        secret, so it must refuse BEFORE claiming when the resulting credential
+        could not be stored securely.
         """
-        if not (self._supabase_url and self._supabase_anon_key):
-            return False
         if os.environ.get("OPENADAPT_HEADLESS", "").strip():
             return False
+        if not self._host_is_valid:
+            return False
+        if not secure_store_available():
+            return False
+        if self._uses_system_browser:
+            import webbrowser
+
+            try:
+                webbrowser.get()
+            except webbrowser.Error:
+                return False
         if sys.platform.startswith("linux"):
             # No X11 / Wayland display -> no system browser to drive.
             return bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"))
@@ -206,56 +283,82 @@ def is_available(self) -> bool:
         return True
 
     def login(self) -> Credential:
-        """Run the browser-PKCE flow and store the resulting credential.
+        """Run the browser login and return the stored credential.
 
         Returns:
-            The stored ``Credential`` (bearer = minted ingest token).
+            The stored ``Credential`` (bearer = the minted ingest token).
 
         Raises:
-            RuntimeError: If the flow cannot complete (headless, denied, timeout,
-                or a code/token/mint failure).
+            RuntimeError: If the flow cannot complete (headless, denied,
+                timeout, state mismatch, or a claim failure).
         """
         if not self.is_available():
-            raise RuntimeError(
-                "Browser login is unavailable on this machine; use token paste."
-            )
+            raise RuntimeError("Browser login is unavailable on this machine; use token paste.")
 
         verifier, challenge = generate_pkce_pair()
         state = secrets.token_urlsafe(24)
-        receiver = _LoopbackReceiver()
 
-        auth_url = self._build_login_url(receiver.redirect_uri, challenge, state)
-        logger.info("Opening system browser for hosted login")
-        self._open_browser(auth_url)
-
-        receiver.serve_until_code(self._timeout)
-
-        if receiver.error:
-            raise RuntimeError(f"Login was denied: {receiver.error}")
-        if not receiver.code:
-            raise RuntimeError("Timed out waiting for the browser login to complete.")
-        if receiver.state and receiver.state != state:
+        # `with` rather than a caller-side try/finally: the listener is closed
+        # on success, on failure, and on an exception raised inside the block.
+        with _LoopbackReceiver() as receiver:
+            auth_url = self._build_login_url(receiver.redirect_uri, challenge, state)
+            logger.info("Opening the system browser to sign in to {host}", host=self.host)
+            self._open_browser(auth_url)
+            receiver.serve_until_code(self._timeout)
+            code = receiver.code
+            received_state = receiver.state
+            error_code = receiver.error_code
+            error = receiver.error
+
+        # Every delivered callback, including a user refusal, must bind the
+        # exact request. A missing state never degrades into an accepted error.
+        if (code is not None or error is not None) and not secrets.compare_digest(
+            received_state or "", state
+        ):
             raise RuntimeError("Login state mismatch (possible CSRF); aborting.")
+        if error:
+            if error_code not in {"access_denied", "server_error"}:
+                raise RuntimeError("Login failed because the callback was malformed.")
+            raise RuntimeError(f"Login was denied: {error}")
+        if not code:
+            raise RuntimeError(
+                "Timed out waiting for the browser login to complete. "
+                "Finish sign-in and choose Connect, then try again. "
+                "You can also paste an ingest token."
+            )
+        # This is our one-use pairing grant, not a Supabase authorization code
+        # and not either runner-local portal credential. Refuse the confusion
+        # before any network claim can spend or disclose it.
+        if not PAIRING_SECRET_RE.fullmatch(code):
+            raise RuntimeError("Login returned an invalid OpenAdapt pairing code.")
+
+        try:
+            claim_pairing(
+                self.host,
+                code,
+                code_verifier=verifier,
+                destination_kind=self._destination_kind(),
+            )
+        except PairingError as exc:
+            raise RuntimeError(f"Login failed: {exc}") from exc
 
-        session = self._exchange_code(receiver.code, verifier, receiver.redirect_uri)
-        access_token = session["access_token"]
-        refresh_token = session.get("refresh_token")
-        expires_at = session.get("expires_at")
-
-        ingest_token, org_id = self._mint_ingest_token(access_token)
-
-        cred: Credential = {
-            "kind": "ingest_token",
-            "token": ingest_token,
-            "refresh_token": refresh_token,
-            "org_id": org_id,
-            "host": self.host,
-            "expires_at": expires_at,
-        }
-        store_credential(cred)
-        logger.info("Browser login complete; ingest token stored for {host}", host=self.host)
+        cred = load_credential(self.host)
+        if cred is None:
+            raise RuntimeError(
+                "Login completed but the credential could not be read back from the keychain."
+            )
+        logger.info("Browser login complete; credential stored for {host}", host=self.host)
         return cred
 
+    def _destination_kind(self) -> str | None:
+        """Classify the host for the pairing destination policy."""
+        return self._kind_for_host(self.host)
+
+    @staticmethod
+    def _kind_for_host(host: str) -> str | None:
+        hostname = urllib.parse.urlparse(host).hostname
+        return "local" if hostname in {"localhost", "127.0.0.1", "::1"} else None
+
     def _build_login_url(self, redirect_uri: str, challenge: str, state: str) -> str:
         """Build the hosted login URL carrying the loopback redirect + PKCE."""
         query = urllib.parse.urlencode(
@@ -267,57 +370,3 @@ def _build_login_url(self, redirect_uri: str, challenge: str, state: str) -> str
             }
         )
         return f"{self.host}{LOGIN_PATH}?{query}"
-
-    def _exchange_code(self, code: str, verifier: str, redirect_uri: str) -> dict:
-        """Exchange the auth code for a Supabase session via PKCE.
-
-        Raises:
-            RuntimeError: If Supabase is not configured or the exchange fails.
-        """
-        if not (self._supabase_url and self._supabase_anon_key):
-            raise RuntimeError(
-                "Supabase is not configured "
-                f"(set {SUPABASE_URL_ENV} / {SUPABASE_ANON_KEY_ENV}); "
-                "browser login is unavailable."
-            )
-        url = f"{self._supabase_url}/auth/v1/token"
-        try:
-            resp = httpx.post(
-                url,
-                params={"grant_type": "pkce"},
-                headers={"apikey": self._supabase_anon_key},
-                json={"auth_code": code, "code_verifier": verifier, "redirect_to": redirect_uri},
-                timeout=30.0,
-            )
-        except httpx.HTTPError as exc:
-            raise RuntimeError(f"Token exchange failed: {exc}") from exc
-        if resp.status_code >= 400:
-            raise RuntimeError(f"Token exchange rejected ({resp.status_code}).")
-        return resp.json()
-
-    def _mint_ingest_token(self, access_token: str) -> tuple[str, str | None]:
-        """Mint an ingest token from an authenticated Supabase session.
-
-        Returns:
-            ``(ingest_token, org_id)``.
-
-        Raises:
-            RuntimeError: If minting fails.
-        """
-        url = f"{self.host}{MINT_PATH}"
-        try:
-            resp = httpx.post(
-                url,
-                headers={"Authorization": f"Bearer {access_token}"},
-                json={"label": "desktop"},
-                timeout=30.0,
-            )
-        except httpx.HTTPError as exc:
-            raise RuntimeError(f"Could not mint an ingest token: {exc}") from exc
-        if resp.status_code >= 400:
-            raise RuntimeError(f"Ingest-token mint rejected ({resp.status_code}).")
-        body = resp.json()
-        token = body.get("token") or body.get("ingest_token")
-        if not token:
-            raise RuntimeError("Mint response did not include an ingest token.")
-        return token, body.get("org_id")
diff --git a/engine/auth/lifetime.py b/engine/auth/lifetime.py
new file mode 100644
index 0000000..300b879
--- /dev/null
+++ b/engine/auth/lifetime.py
@@ -0,0 +1,135 @@
+"""Strict Cloud credential lifetime and prefix contracts.
+
+Cloud exposes one credential to a local client: an ``oai_ingest_`` bearer.
+Pairing codes and runner-local portal credentials must never cross that role
+boundary.  This module also validates the additive lifetime block returned by
+Cloud, including its response headers, so the Desktop never invents a safe
+deadline from a partial or contradictory response.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from typing import Any, Mapping
+
+INGEST_TOKEN_RE = re.compile(r"^oai_ingest_[A-Za-z0-9_-]{43}$")
+WARNING_DAYS = 14
+
+_LIFETIME_FIELDS = frozenset(
+    {
+        "expires_at",
+        "expires_in_days",
+        "expiring_soon",
+        "legacy_non_expiring",
+        "warning_days",
+    }
+)
+
+
+class CredentialContractError(ValueError):
+    """Cloud returned a malformed or contradictory credential contract."""
+
+
+def valid_ingest_token(value: Any) -> bool:
+    """Return true only for the exact Cloud ingest-bearer format."""
+    return isinstance(value, str) and INGEST_TOKEN_RE.fullmatch(value) is not None
+
+
+def _header(headers: Mapping[str, str] | Any, name: str) -> str | None:
+    if headers is None:
+        return None
+    try:
+        value = headers.get(name)
+        if value is None:
+            value = headers.get(name.lower())
+    except (AttributeError, TypeError):
+        return None
+    return str(value) if value is not None else None
+
+
+def _timestamp(value: str) -> float:
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value)
+    except ValueError as exc:
+        raise CredentialContractError("credential expiry is not a valid timestamp") from exc
+    if parsed.tzinfo is None:
+        raise CredentialContractError("credential expiry does not include a timezone")
+    return parsed.timestamp()
+
+
+def parse_credential_lifetime(
+    body: Any,
+    *,
+    headers: Mapping[str, str] | Any | None = None,
+    require_headers: bool = False,
+    require_fresh: bool = False,
+    require_no_store: bool = False,
+) -> dict[str, Any]:
+    """Validate and normalize Cloud's ``credential`` block.
+
+    ``require_fresh`` applies to a newly minted credential.  Such a response
+    must report 89 or 90 remaining whole days, no expiry warning, and no legacy
+    exemption.  ``require_headers`` applies to the needs-attention endpoint,
+    where the warning and remaining-day headers are part of the contract.
+    """
+    if not isinstance(body, dict):
+        raise CredentialContractError("credential response is not an object")
+    block = body.get("credential")
+    if not isinstance(block, dict) or set(block) != _LIFETIME_FIELDS:
+        raise CredentialContractError("credential lifetime block is incomplete")
+
+    expires_at = block["expires_at"]
+    days = block["expires_in_days"]
+    expiring = block["expiring_soon"]
+    legacy = block["legacy_non_expiring"]
+    warning_days = block["warning_days"]
+
+    if type(expiring) is not bool or type(legacy) is not bool:
+        raise CredentialContractError("credential lifetime flags are invalid")
+    if type(warning_days) is not int or warning_days != WARNING_DAYS:
+        raise CredentialContractError("credential warning window is invalid")
+
+    expires_at_timestamp: float | None
+    if legacy:
+        if expires_at is not None or days is not None or expiring:
+            raise CredentialContractError("legacy credential lifetime is contradictory")
+        expires_at_timestamp = None
+    else:
+        if not isinstance(expires_at, str) or not expires_at:
+            raise CredentialContractError("credential expiry is missing")
+        if type(days) is not int or days < 0:
+            raise CredentialContractError("credential remaining days are invalid")
+        # Cloud computes the renewal trigger from the exact server-side
+        # timestamp. ``expires_in_days`` is floor-rounded display data. Thus
+        # day 14 can report either state: false above the exact 14-day
+        # boundary, then true at or below it. Other whole-day values are
+        # unambiguous. Do not recompute this control from the client clock.
+        if (days < WARNING_DAYS and not expiring) or (days > WARNING_DAYS and expiring):
+            raise CredentialContractError("credential warning state is contradictory")
+        expires_at_timestamp = _timestamp(expires_at)
+
+    if require_fresh and (
+        legacy or days not in {89, 90} or expiring or expires_at_timestamp is None
+    ):
+        raise CredentialContractError("new credential does not have a fresh 90-day lifetime")
+
+    if require_no_store and (_header(headers, "cache-control") or "").strip().lower() != "no-store":
+        raise CredentialContractError("credential response is not marked no-store")
+
+    if require_headers:
+        if _header(headers, "x-openadapt-credential-warning-days") != str(WARNING_DAYS):
+            raise CredentialContractError("credential warning header is missing")
+        expires_header = _header(headers, "x-openadapt-credential-expires-in-days")
+        expected = None if days is None else str(days)
+        if expires_header != expected:
+            raise CredentialContractError("credential expiry header disagrees with its body")
+
+    return {
+        "expires_at": expires_at,
+        "expires_at_timestamp": expires_at_timestamp,
+        "expires_in_days": days,
+        "expiring_soon": expiring,
+        "legacy_non_expiring": legacy,
+        "warning_days": warning_days,
+    }
diff --git a/engine/auth/pairing.py b/engine/auth/pairing.py
index d58fc5c..e5ece3d 100644
--- a/engine/auth/pairing.py
+++ b/engine/auth/pairing.py
@@ -4,6 +4,12 @@
 Python engine even though the Tauri shell validates it first, so neither IPC
 nor an operating-system protocol invocation can become a general command,
 browser-navigation, or arbitrary-network surface.
+
+The same claim -> validate -> promote -> confirm -> rollback path also redeems
+the authorization code delivered to the RFC 8252 loopback listener in
+:mod:`engine.auth.browser_pkce`. That code IS an ``oap_`` pairing secret, so
+browser login adds a way to OBTAIN one rather than a second protocol: see
+:func:`claim_pairing`.
 """
 
 from __future__ import annotations
@@ -16,6 +22,11 @@
 
 import httpx
 
+from engine.auth.lifetime import (
+    CredentialContractError,
+    parse_credential_lifetime,
+    valid_ingest_token,
+)
 from engine.auth.provider import Credential
 from engine.auth.store import (
     DEFAULT_HOST,
@@ -30,7 +41,8 @@
 )
 
 PAIRING_SECRET_RE = re.compile(r"^oap_[A-Za-z0-9_-]{43}$")
-INGEST_TOKEN_RE = re.compile(r"^oai_ingest_[A-Za-z0-9_-]{32,}$")
+# RFC 7636 code verifier: 43-128 unreserved characters.
+PKCE_VERIFIER_RE = re.compile(r"^[A-Za-z0-9\-._~]{43,128}$")
 ALLOWED_FIELDS = frozenset({"pairing", "host", "destination_kind"})
 ALLOWED_DESTINATIONS = frozenset({"openadapt-managed", "local"})
 API_TIMEOUT_S = 8.0
@@ -72,9 +84,7 @@ def _origin(raw: str) -> str:
 def _validate_destination(host: str, destination_kind: str | None) -> str:
     origin = _origin(host)
     managed_origin = _origin(DEFAULT_HOST)
-    kind = destination_kind or (
-        "openadapt-managed" if origin == managed_origin else None
-    )
+    kind = destination_kind or ("openadapt-managed" if origin == managed_origin else None)
     if kind not in ALLOWED_DESTINATIONS:
         raise PairingError("Connect link has an unsupported destination")
     if kind == "openadapt-managed":
@@ -125,6 +135,19 @@ def parse_connect_uri(uri: str) -> dict[str, str]:
     return result
 
 
+def credential_expires_at(body: Any) -> float | None:
+    """Read the credential deadline Cloud reports, or None when it reports none.
+
+    Cloud is the authority on when a credential dies; this value is stored so
+    the local surfaces can WARN early. It is never used to decide that a
+    credential is still good -- every request is checked server-side.
+    """
+    try:
+        return cast(float | None, parse_credential_lifetime(body)["expires_at_timestamp"])
+    except CredentialContractError:
+        return None
+
+
 def _safe_device_name() -> str:
     value = re.sub(r"[\x00-\x1f\x7f]", "", socket.gethostname()).strip()[:80]
     return value or "this computer"
@@ -158,9 +181,7 @@ def _stage_identity(stage: dict) -> tuple[str, str, str, str, Credential]:
     state = stage.get("state")
     credential = stage.get("credential")
     try:
-        canonical_pairing_id = (
-            str(UUID(pairing_id)) if isinstance(pairing_id, str) else None
-        )
+        canonical_pairing_id = str(UUID(pairing_id)) if isinstance(pairing_id, str) else None
     except ValueError:
         canonical_pairing_id = None
     if (
@@ -169,36 +190,30 @@ def _stage_identity(stage: dict) -> tuple[str, str, str, str, Credential]:
         or canonical_pairing_id != pairing_id
         or not isinstance(device_name, str)
         or not 1 <= len(device_name) <= 80
-        or state not in {
+        or state
+        not in {
             "claimed",
             "canonical_written",
             "confirm_ambiguous",
             "abort_acknowledged",
         }
         or not isinstance(credential, dict)
-        or set(credential)
-        != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"}
+        or set(credential) != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"}
         or credential.get("kind") != "ingest_token"
         or credential.get("refresh_token") is not None
         or credential.get("org_id") is not None
-        or credential.get("expires_at") is not None
-        or not isinstance(credential.get("token"), str)
-        or not INGEST_TOKEN_RE.fullmatch(credential["token"])
+        # Cloud now returns a deadline with the credential. Accept a number or
+        # None, but nothing else: a corrupt value must not reach the keychain.
+        or not isinstance(credential.get("expires_at"), (int, float, type(None)))
+        or isinstance(credential.get("expires_at"), bool)
+        or not valid_ingest_token(credential.get("token"))
         or not isinstance(credential.get("host"), str)
     ):
-        raise PairingError(
-            "Desktop found an invalid pairing recovery record in the keychain"
-        )
+        raise PairingError("Desktop found an invalid pairing recovery record in the keychain")
     host = credential["host"]
-    destination = (
-        "openadapt-managed"
-        if _origin(host) == _origin(DEFAULT_HOST)
-        else "local"
-    )
+    destination = "openadapt-managed" if _origin(host) == _origin(DEFAULT_HOST) else "local"
     if _validate_destination(host, destination) != host:
-        raise PairingError(
-            "Desktop found an invalid pairing recovery record in the keychain"
-        )
+        raise PairingError("Desktop found an invalid pairing recovery record in the keychain")
     return pairing_id, host, credential["token"], state, cast(Credential, credential)
 
 
@@ -271,7 +286,7 @@ def _fail_staged_pairing(stage: dict, message: str) -> NoReturn:
 
 
 def _finish_staged_pairing(stage: dict) -> dict[str, Any]:
-    pairing_id, host, token, state, _ = _stage_identity(stage)
+    pairing_id, host, token, state, credential = _stage_identity(stage)
     device_name = cast(str, stage["device_name"])
     if state == "abort_acknowledged":
         if restore_pairing_stage(pairing_id) and clear_pairing_stage(pairing_id):
@@ -299,6 +314,24 @@ def _finish_staged_pairing(stage: dict) -> dict[str, Any]:
             stage,
             f"Cloud could not verify the new credential ({validation.status_code}).",
         )
+    try:
+        validation_lifetime = parse_credential_lifetime(
+            validation.json(),
+            headers=validation.headers,
+            require_headers=True,
+            require_fresh=True,
+            require_no_store=True,
+        )
+    except (AttributeError, TypeError, ValueError, CredentialContractError):
+        _fail_staged_pairing(
+            stage,
+            "Cloud returned an incomplete credential lifetime contract.",
+        )
+    if validation_lifetime["expires_at_timestamp"] != credential["expires_at"]:
+        _fail_staged_pairing(
+            stage,
+            "Cloud returned a different credential deadline during validation.",
+        )
 
     if not commit_pairing_stage(pairing_id):
         _fail_staged_pairing(
@@ -342,7 +375,48 @@ def recover_pending_pairing() -> dict[str, Any] | None:
 def connect_uri(uri: str) -> dict[str, Any]:
     """Claim, stage, verify, commit, and confirm one Desktop pairing URI."""
     request = parse_connect_uri(uri)
-    host = request["host"]
+    return claim_pairing(
+        request["host"],
+        request["pairing"],
+        destination_kind=request.get("destination_kind"),
+    )
+
+
+def claim_pairing(
+    host: str,
+    pairing_secret: str,
+    code_verifier: str | None = None,
+    destination_kind: str | None = None,
+) -> dict[str, Any]:
+    """Claim, stage, verify, commit, and confirm one pairing secret.
+
+    This is the single redemption path. A deep link supplies the secret through
+    ``openadapt://connect``; the RFC 8252 loopback listener in
+    :mod:`engine.auth.browser_pkce` receives the same kind of secret as an
+    authorization ``code`` and supplies the PKCE ``code_verifier`` that Cloud
+    bound to it. Everything after the claim -- durable staging, independent
+    validation, atomic keychain promotion, idempotent confirmation, and
+    acknowledged rollback -- is identical either way.
+
+    Args:
+        host: The Cloud origin, already validated by the caller.
+        pairing_secret: A one-use ``oap_`` secret.
+        code_verifier: The PKCE verifier for a loopback grant, or None for a
+            deep-link pairing that carries no challenge.
+
+    Returns:
+        The paired result dictionary.
+
+    Raises:
+        PairingError: On any failure, with the prior connection preserved.
+    """
+    if not PAIRING_SECRET_RE.fullmatch(str(pairing_secret)):
+        raise PairingError("Pairing code is malformed")
+    if code_verifier is not None and not PKCE_VERIFIER_RE.fullmatch(code_verifier):
+        raise PairingError("Login proof is malformed")
+    # Re-validated here, not only at the caller: every entry point to this
+    # redemption path must prove its destination before a secret is spent.
+    host = _validate_destination(host, destination_kind)
     if not secure_store_available():
         raise PairingError(
             "Secure pairing needs an unlocked operating-system keychain. "
@@ -354,15 +428,19 @@ def connect_uri(uri: str) -> dict[str, Any]:
         return recovered
     previous = snapshot_pairing_canonical(host)
     if previous is None:
-        raise PairingError(
-            "Desktop could not safely snapshot the current keychain connection"
-        )
+        raise PairingError("Desktop could not safely snapshot the current keychain connection")
 
     device_name = _safe_device_name()
+    payload: dict[str, Any] = {
+        "pairing_secret": pairing_secret,
+        "device_name": device_name,
+    }
+    if code_verifier is not None:
+        payload["code_verifier"] = code_verifier
     try:
         claim = httpx.post(
             f"{host}/api/local-bridge/pairings/claim",
-            json={"pairing_secret": request["pairing"], "device_name": device_name},
+            json=payload,
             timeout=API_TIMEOUT_S,
             follow_redirects=False,
         )
@@ -372,14 +450,22 @@ def connect_uri(uri: str) -> dict[str, Any]:
         raise PairingError("Pairing code expired, was cancelled, or was already used")
     if claim.status_code >= 400:
         raise PairingError(f"Pairing failed ({claim.status_code})")
+    if claim.status_code != 201:
+        raise PairingError("Pairing response used an unexpected success status")
     try:
         body = claim.json()
-        token = str(body["ingest_token"])
-        pairing_id = str(UUID(str(body["pairing_id"])))
-    except (KeyError, TypeError, ValueError) as exc:
-        raise PairingError("Pairing response did not contain a valid credential") from exc
-    if not INGEST_TOKEN_RE.fullmatch(token):
-        raise PairingError("Pairing response contained a malformed credential")
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise PairingError("Pairing response was not valid JSON") from exc
+    try:
+        token, pairing_id, expiry = _validated_claim(body, claim.headers)
+    except PairingError:
+        rollback = _claim_abort_coordinates(body)
+        if rollback and _abort_claim(host, rollback[0], rollback[1]):
+            raise PairingError(
+                "Cloud rolled back a pairing response that did not match the "
+                "Desktop credential contract. Your previous connection is unchanged."
+            ) from None
+        raise
 
     credential: Credential = {
         "kind": "ingest_token",
@@ -387,7 +473,7 @@ def connect_uri(uri: str) -> dict[str, Any]:
         "refresh_token": None,
         "org_id": None,
         "host": host,
-        "expires_at": None,
+        "expires_at": expiry,
     }
     if not stage_pairing_credential(
         pairing_id,
@@ -412,3 +498,57 @@ def connect_uri(uri: str) -> dict[str, Any]:
             "Desktop could not recover the newly staged credential from the keychain"
         )
     return _finish_staged_pairing(stage)
+
+
+def _canonical_uuid(value: Any) -> str:
+    if not isinstance(value, str):
+        raise PairingError("Pairing response did not contain a valid credential")
+    try:
+        canonical = str(UUID(value))
+    except ValueError as exc:
+        raise PairingError("Pairing response did not contain a valid credential") from exc
+    if canonical != value:
+        raise PairingError("Pairing response did not contain a valid credential")
+    return canonical
+
+
+def _validated_claim(body: Any, headers: Any) -> tuple[str, str, float]:
+    """Validate the complete Cloud 201 response before staging its bearer."""
+    expected = {
+        "paired",
+        "pairing_id",
+        "ingest_token_id",
+        "ingest_token",
+        "credential",
+    }
+    if not isinstance(body, dict) or set(body) != expected or body.get("paired") is not True:
+        raise PairingError("Pairing response did not match the credential contract")
+    pairing_id = _canonical_uuid(body.get("pairing_id"))
+    _canonical_uuid(body.get("ingest_token_id"))
+    token = body.get("ingest_token")
+    if not valid_ingest_token(token):
+        raise PairingError("Pairing response contained a malformed credential")
+    try:
+        lifetime = parse_credential_lifetime(
+            body,
+            headers=headers,
+            require_fresh=True,
+            require_no_store=True,
+        )
+    except CredentialContractError as exc:
+        raise PairingError("Pairing response contained an invalid credential lifetime") from exc
+    expiry = lifetime["expires_at_timestamp"]
+    if not isinstance(expiry, float):
+        raise PairingError("Pairing response contained an invalid credential lifetime")
+    return token, pairing_id, expiry
+
+
+def _claim_abort_coordinates(body: Any) -> tuple[str, str] | None:
+    """Salvage only the exact values needed to revoke a malformed 201."""
+    if not isinstance(body, dict) or not valid_ingest_token(body.get("ingest_token")):
+        return None
+    try:
+        pairing_id = _canonical_uuid(body.get("pairing_id"))
+    except PairingError:
+        return None
+    return pairing_id, cast(str, body["ingest_token"])
diff --git a/engine/auth/paste.py b/engine/auth/paste.py
index 1d0cac0..4d06a98 100644
--- a/engine/auth/paste.py
+++ b/engine/auth/paste.py
@@ -21,6 +21,11 @@
 import httpx
 from loguru import logger
 
+from engine.auth.lifetime import (
+    CredentialContractError,
+    parse_credential_lifetime,
+    valid_ingest_token,
+)
 from engine.auth.provider import Credential
 from engine.auth.store import (
     DEFAULT_HOST,
@@ -92,7 +97,13 @@ def login(self, token: str | None = None) -> Credential:
         if not token:
             raise TokenValidationError("No ingest token provided.")
 
-        org_id = self._validate(token)
+        if not valid_ingest_token(token):
+            raise TokenValidationError(
+                "Expected an OpenAdapt Cloud ingest token that starts with "
+                "oai_ingest_. A pairing code or portal credential cannot be used here."
+            )
+
+        org_id, expires_at = self._validate(token)
 
         cred: Credential = {
             "kind": "ingest_token",
@@ -100,7 +111,7 @@ def login(self, token: str | None = None) -> Credential:
             "refresh_token": None,
             "org_id": org_id,
             "host": self.host,
-            "expires_at": None,
+            "expires_at": expires_at,
         }
         store_credential(cred)
         logger.info("Stored ingest token for {host}", host=self.host)
@@ -114,8 +125,8 @@ def _prompt_for_token(self) -> str:
         except (EOFError, KeyboardInterrupt):
             return ""
 
-    def _validate(self, token: str) -> str | None:
-        """Validate a token via the count endpoint; return org_id if exposed.
+    def _validate(self, token: str) -> tuple[str | None, float | None]:
+        """Validate a token and retain its Cloud-reported deadline when present.
 
         Raises:
             TokenValidationError: on any non-2xx / network failure.
@@ -130,11 +141,25 @@ def _validate(self, token: str) -> str | None:
         if resp.status_code == 401:
             raise TokenValidationError("Ingest token was rejected (401).")
         if resp.status_code >= 400:
-            raise TokenValidationError(
-                f"Token validation failed ({resp.status_code})."
-            )
+            raise TokenValidationError(f"Token validation failed ({resp.status_code}).")
         try:
             body = resp.json()
         except ValueError:
             body = {}
-        return body.get("org_id")
+        expires_at: float | None = None
+        if isinstance(body, dict) and "credential" in body:
+            try:
+                lifetime = parse_credential_lifetime(
+                    body,
+                    headers=resp.headers,
+                    require_headers=True,
+                    require_no_store=True,
+                )
+                expires_at = lifetime["expires_at_timestamp"]
+            except (AttributeError, CredentialContractError):
+                # A working bearer remains a valid fallback when an older Cloud
+                # does not yet supply the additive warning contract. Never
+                # convert a malformed warning into a claim that expiry is safe.
+                expires_at = None
+        org_id = body.get("org_id") if isinstance(body, dict) else None
+        return org_id if isinstance(org_id, str) else None, expires_at
diff --git a/engine/auth/provider.py b/engine/auth/provider.py
index 5d15cbe..f782db8 100644
--- a/engine/auth/provider.py
+++ b/engine/auth/provider.py
@@ -28,22 +28,20 @@ class Credential(TypedDict):
     """A stored-ready hosted credential.
 
     Both providers write this exact shape via
-    :func:`engine.auth.store.store_credential`. The ``kind`` discriminates
-    between a bare ingest token (the machine/push path) and a full Supabase
-    session (the interactive browser path, which additionally mints an ingest
-    token so the headless push/count path always has a bearer token).
+    :func:`engine.auth.store.store_credential`. Both browser pairing and token
+    paste produce the same Cloud ingest credential. A Supabase authorization
+    code or session is never a Desktop bearer.
 
     Attributes:
-        kind: ``"ingest_token"`` for an ``oai_ingest_…`` token, or
-            ``"supabase_session"`` for a Supabase access token.
-        token: The ingest token (``oai_ingest_…``) OR the Supabase access token.
-        refresh_token: Supabase refresh token (Supabase sessions only), else None.
+        kind: Always ``"ingest_token"``.
+        token: The ingest token (``oai_ingest_…``).
+        refresh_token: Reserved compatibility field. Always None.
         org_id: The organization the token resolves to, if known.
         host: The hosted base URL, e.g. ``https://app.openadapt.ai``.
         expires_at: POSIX timestamp when the credential expires, or None.
     """
 
-    kind: Literal["ingest_token", "supabase_session"]
+    kind: Literal["ingest_token"]
     token: str
     refresh_token: str | None
     org_id: str | None
diff --git a/engine/auth/rotation.py b/engine/auth/rotation.py
new file mode 100644
index 0000000..a1c52c9
--- /dev/null
+++ b/engine/auth/rotation.py
@@ -0,0 +1,463 @@
+"""Credential lifetime: report the deadline, and renew it without an outage.
+
+A Cloud ingest credential lives 90 days. Two things follow, and this module
+owns both:
+
+* **Warn early.** ``GET /api/needs-attention/count`` -- the one authenticated
+  call this machine already makes on a timer -- returns how long the credential
+  it just used has left. :func:`credential_status` reads it, and
+  :func:`expiry_warning` turns it into a line the operator sees.
+* **Renew without downtime.** ``POST /api/ingest-tokens/rotate`` mints the
+  replacement FIRST and only then shortens the outgoing credential to a bounded
+  overlap. So :func:`rotate_credential` can store the replacement knowing the
+  credential currently in the keychain keeps working meanwhile: if the write
+  fails, or the process dies between the two, the machine is still connected.
+
+Nothing here decides that a credential is still valid. Expiry is enforced
+server-side on every request; this is a warning and a renewal, not a check.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from uuid import UUID
+
+import httpx
+from loguru import logger
+
+from engine.auth.lifetime import (
+    CredentialContractError,
+    parse_credential_lifetime,
+    valid_ingest_token,
+)
+from engine.auth.provider import Credential
+from engine.auth.store import (
+    DEFAULT_HOST,
+    INGEST_TOKEN_ENV,
+    clear_rotation_stage,
+    clear_superseded_rotation_stage,
+    commit_rotation_stage,
+    load_credential,
+    load_rotation_stage,
+    secure_store_available,
+    snapshot_pairing_canonical,
+    stage_rotation_credential,
+)
+
+COUNT_PATH = "/api/needs-attention/count"
+ROTATE_PATH = "/api/ingest-tokens/rotate"
+API_TIMEOUT_S = 10.0
+
+# Mirrors the server's own threshold. The server sends `warning_days` with
+# every answer, so this is only the fallback when it does not.
+DEFAULT_WARNING_DAYS = 14
+
+
+class RotationError(RuntimeError):
+    """A safe, user-facing renewal failure that never contains a secret."""
+
+
+def credential_status(host: str = DEFAULT_HOST, token: str | None = None) -> dict[str, Any] | None:
+    """Ask Cloud how long the credential for ``host`` has left.
+
+    Args:
+        host: Hosted base URL.
+        token: Bearer to ask about. Defaults to the stored credential.
+
+    Returns:
+        The server's ``credential`` block, or None when there is no credential,
+        Cloud cannot be reached, or the answer is not readable. None means
+        "unknown", never "fine".
+    """
+    if token is not None and not valid_ingest_token(token):
+        return None
+    bearer = token if token is not None else _stored_token(host)
+    if not bearer:
+        return None
+    try:
+        response = httpx.get(
+            f"{host.rstrip('/')}{COUNT_PATH}",
+            headers={"Authorization": f"Bearer {bearer}"},
+            timeout=API_TIMEOUT_S,
+            follow_redirects=False,
+        )
+    except httpx.HTTPError:
+        return None
+    if not 200 <= response.status_code < 300:
+        return None
+    try:
+        lifetime = parse_credential_lifetime(
+            response.json(),
+            headers=response.headers,
+            require_headers=True,
+            require_no_store=True,
+        )
+    except (AttributeError, TypeError, ValueError, CredentialContractError):
+        return None
+    return lifetime
+
+
+def expiry_warning(status: dict[str, Any] | None) -> str | None:
+    """Render the operator-visible warning, or None when there is nothing to say.
+
+    Returns None for an unknown status too: inventing "your credential is fine"
+    from a failed request would be worse than saying nothing.
+    """
+    if not status:
+        return None
+    if status.get("legacy_non_expiring") is True:
+        return (
+            "This connection was created before connections expired, so it has "
+            "no renewal date. Run `openadapt-desktop rotate` to move it to a "
+            "90-day credential."
+        )
+    if status.get("expiring_soon") is not True:
+        return None
+    days = status.get("expires_in_days")
+    if not isinstance(days, int):
+        return "This connection expires soon. Run `openadapt-desktop rotate` to renew it."
+    if days <= 0:
+        return "This connection expires today. Run `openadapt-desktop rotate` to renew it."
+    unit = "day" if days == 1 else "days"
+    return (
+        f"This connection expires in {days} {unit}. Run `openadapt-desktop rotate` "
+        "to renew it; the current one keeps working while the new one takes over."
+    )
+
+
+def rotate_credential(host: str = DEFAULT_HOST) -> Credential:
+    """Renew the stored credential for ``host`` without interrupting this machine.
+
+    Args:
+        host: Hosted base URL.
+
+    Returns:
+        The newly stored ``Credential``.
+
+    Raises:
+        RotationError: If there is nothing to renew, Cloud refuses, or the
+            replacement could not be stored. In every failure case the existing
+            credential is left in place and keeps working.
+    """
+    host = host.rstrip("/")
+    recovered = recover_pending_rotation(host)
+    if recovered is not None:
+        return recovered
+    if not secure_store_available():
+        raise RotationError(
+            "Credential renewal needs an unlocked operating-system keychain. "
+            "Unlock it before OpenAdapt asks Cloud for a one-time replacement."
+        )
+    current = load_credential(host)
+    if current is None or not valid_ingest_token(current.get("token")):
+        raise RotationError(
+            "There is no stored connection for this workspace. "
+            "Run `openadapt-desktop login` to connect."
+        )
+    previous = snapshot_pairing_canonical(host)
+    if previous is None:
+        raise RotationError(
+            "OpenAdapt could not safely snapshot the current keychain connection. "
+            "Cloud did not receive a renewal request."
+        )
+
+    try:
+        response = httpx.post(
+            f"{host}{ROTATE_PATH}",
+            json={},
+            headers={"Authorization": f"Bearer {current['token']}"},
+            timeout=API_TIMEOUT_S,
+            follow_redirects=False,
+        )
+    except httpx.HTTPError as exc:
+        raise RotationError(
+            "OpenAdapt did not receive the renewal response. The old credential "
+            "remains valid for at most seven days if Cloud rotated it. Sign in "
+            "again; do not retry the same renewal."
+        ) from exc
+
+    if response.status_code == 401:
+        raise RotationError(
+            "This connection is no longer valid. Run `openadapt-desktop login` to connect again."
+        )
+    if response.status_code == 409:
+        raise RotationError(
+            "This connection was already renewed once, and the replacement "
+            "cannot be issued twice. Run `openadapt-desktop login` to connect again."
+        )
+    if response.status_code != 201:
+        raise RotationError(
+            f"Renewal did not complete with the required response ({response.status_code}). "
+            "Keep the current credential and sign in again if Cloud already rotated it."
+        )
+
+    try:
+        body = response.json()
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise RotationError(
+            "Cloud renewed the connection but returned an unreadable one-time response. "
+            "The old credential remains valid for at most seven days. Sign in again."
+        ) from exc
+    try:
+        token, previous_id, expires_at = _validated_rotation(body, response.headers)
+    except RotationError as exc:
+        raise RotationError(
+            "Cloud renewed the connection, but its one-time response did not "
+            "match the Desktop credential contract. The old credential remains "
+            "valid for at most seven days. Sign in again; do not retry the same renewal."
+        ) from exc
+
+    replacement: Credential = {
+        "kind": "ingest_token",
+        "token": token,
+        "refresh_token": None,
+        "org_id": current.get("org_id"),
+        "host": host,
+        "expires_at": expires_at,
+    }
+    # Retain the one-time response before changing any canonical entry. A crash
+    # after this point resumes from the exact retained replacement and does not
+    # issue an unsafe second rotation request.
+    if not stage_rotation_credential(previous_id, replacement, previous):
+        raise RotationError(
+            "Cloud renewed the connection, but OpenAdapt could not retain its "
+            "one-time response. The old credential remains valid for at most "
+            "seven days. Sign in again; do not retry the same renewal."
+        )
+    _validate_staged_replacement(previous_id, replacement, require_fresh=True)
+    if not commit_rotation_stage(previous_id):
+        raise RotationError(
+            "OpenAdapt retained the renewed credential but could not finish its "
+            "keychain update. Recovery state remains available. Restart OpenAdapt "
+            "before you sign in again."
+        )
+    if not clear_rotation_stage(previous_id):
+        raise RotationError(
+            "The renewed credential is active, but OpenAdapt could not clear its "
+            "recovery record. Restart OpenAdapt to finish recovery."
+        )
+    logger.info("Renewed the hosted credential for {host}", host=host)
+    return replacement
+
+
+def recover_pending_rotation(host: str | None = None) -> Credential | None:
+    """Promote an exact retained one-time response without another Cloud call."""
+    try:
+        stage = load_rotation_stage()
+    except RuntimeError as exc:
+        raise RotationError(
+            "OpenAdapt could not read the retained credential renewal state."
+        ) from exc
+    if stage is None:
+        return None
+    previous_id, credential = _validated_rotation_stage(stage)
+    if host is not None and credential["host"] != host.rstrip("/"):
+        raise RotationError("A credential renewal for another Cloud host needs recovery first.")
+    if clear_superseded_rotation_stage(previous_id):
+        logger.info(
+            "Cleared a stale credential renewal after a later login for {host}",
+            host=credential["host"],
+        )
+        return None
+    _validate_staged_replacement(previous_id, credential, require_fresh=False)
+    if not commit_rotation_stage(previous_id):
+        raise RotationError(
+            "OpenAdapt could not safely finish the retained credential renewal. "
+            "Recovery state remains available."
+        )
+    if not clear_rotation_stage(previous_id):
+        raise RotationError("The renewed credential is active, but its recovery record remains.")
+    logger.info("Recovered the renewed hosted credential for {host}", host=credential["host"])
+    return credential
+
+
+def _validate_staged_replacement(
+    previous_id: str,
+    credential: Credential,
+    *,
+    require_fresh: bool,
+) -> None:
+    """Prove the retained replacement bearer before canonical promotion.
+
+    The immediate check must still show a new 89/90-day credential. Recovery
+    can occur later, so it binds the exact stored expiry while permitting the
+    display day count to decrease.
+    """
+    try:
+        response = httpx.get(
+            f"{credential['host']}{COUNT_PATH}",
+            headers={"Authorization": f"Bearer {credential['token']}"},
+            timeout=API_TIMEOUT_S,
+            follow_redirects=False,
+        )
+    except httpx.HTTPError as exc:
+        raise RotationError(
+            "OpenAdapt retained the renewed credential but could not verify it. "
+            "The old credential remains active, and recovery will retry the "
+            "retained replacement without another renewal request."
+        ) from exc
+    if not 200 <= response.status_code < 300:
+        raise RotationError(
+            "OpenAdapt retained the renewed credential, but Cloud did not accept "
+            f"it during verification ({response.status_code}). The old credential "
+            "remains active. Sign in again if verification does not recover."
+        )
+    try:
+        lifetime = parse_credential_lifetime(
+            response.json(),
+            headers=response.headers,
+            require_headers=True,
+            require_fresh=require_fresh,
+            require_no_store=True,
+        )
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise RotationError(
+            "OpenAdapt retained the renewed credential, but Cloud returned an "
+            "incomplete verification contract. The old credential remains active."
+        ) from exc
+    if (
+        lifetime["legacy_non_expiring"]
+        or lifetime["expires_at_timestamp"] != credential["expires_at"]
+    ):
+        raise RotationError(
+            "OpenAdapt retained the renewed credential, but Cloud verified a "
+            "different expiry. The old credential remains active."
+        )
+    # The recovery stage identity must remain unchanged across the network
+    # check. A concurrent operation cannot swap in another retained bearer and
+    # then have this successful response authorize its promotion.
+    try:
+        current_stage = load_rotation_stage()
+    except RuntimeError as exc:
+        raise RotationError("The retained credential changed during verification.") from exc
+    if current_stage is None:
+        raise RotationError("The retained credential changed during verification.")
+    try:
+        current_previous_id, current_credential = _validated_rotation_stage(current_stage)
+    except RotationError as exc:
+        raise RotationError("The retained credential changed during verification.") from exc
+    if current_previous_id != previous_id or current_credential != credential:
+        raise RotationError("The retained credential changed during verification.")
+
+
+def _stored_token(host: str) -> str | None:
+    """Resolve the bearer this machine would actually send."""
+    import os
+
+    env_token = os.environ.get(INGEST_TOKEN_ENV, "").strip()
+    if valid_ingest_token(env_token):
+        return env_token
+    cred = load_credential(host.rstrip("/"))
+    return cred.get("token") if cred and valid_ingest_token(cred.get("token")) else None
+
+
+def _validated_rotation(body: Any, headers: Any) -> tuple[str, str, float]:
+    """Validate the exact one-time response and its seven-day overlap."""
+    expected = {"token", "record", "previous_id", "previous_expires_at", "credential"}
+    if not isinstance(body, dict) or set(body) != expected:
+        raise RotationError(
+            "Cloud renewed the connection but returned an incomplete one-time response. "
+            "The old credential remains valid for at most seven days. Sign in again."
+        )
+    token = body.get("token")
+    previous_id = body.get("previous_id")
+    record = body.get("record")
+    if not valid_ingest_token(token):
+        raise RotationError("Renewal response contained a malformed credential")
+    if not _is_canonical_uuid(previous_id):
+        raise RotationError("Renewal response contained an invalid previous credential id")
+    credential_block = body.get("credential")
+    if (
+        not isinstance(record, dict)
+        or set(record)
+        != {
+            "id",
+            "org_id",
+            "name",
+            "token_prefix",
+            "created_at",
+            "last_used_at",
+            "expires_at",
+            "revoked_at",
+            "rotated_to_id",
+        }
+        or not _is_canonical_uuid(record.get("id"))
+        or not isinstance(record.get("org_id"), str)
+        or not 1 <= len(record["org_id"]) <= 128
+        or not isinstance(record.get("name"), str)
+        or record.get("token_prefix") != token[:20]
+        or record.get("last_used_at") is not None
+        or not isinstance(credential_block, dict)
+        or record.get("expires_at") != credential_block.get("expires_at")
+        or record.get("revoked_at") is not None
+        or record.get("rotated_to_id") is not None
+    ):
+        raise RotationError("Renewal response contained a contradictory credential record")
+    try:
+        _parse_timestamp(record.get("created_at"))
+        lifetime = parse_credential_lifetime(
+            body,
+            headers=headers,
+            require_fresh=True,
+            require_no_store=True,
+        )
+        previous_expiry = _parse_timestamp(body.get("previous_expires_at"))
+    except CredentialContractError as exc:
+        raise RotationError("Renewal response contained an invalid credential lifetime") from exc
+    now = datetime.now(timezone.utc)
+    if previous_expiry > now + timedelta(days=7, minutes=5):
+        raise RotationError("Renewal response exceeded the seven-day overlap contract")
+    expiry = lifetime["expires_at_timestamp"]
+    if not isinstance(expiry, float):
+        raise RotationError("Renewal response contained an invalid credential lifetime")
+    return token, previous_id, expiry
+
+
+def _parse_timestamp(value: Any) -> datetime:
+    if not isinstance(value, str) or not value:
+        raise CredentialContractError("credential overlap expiry is missing")
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value)
+    except ValueError as exc:
+        raise CredentialContractError("credential overlap expiry is invalid") from exc
+    if parsed.tzinfo is None:
+        raise CredentialContractError("credential overlap expiry does not include a timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+def _is_canonical_uuid(value: Any) -> bool:
+    if not isinstance(value, str):
+        return False
+    try:
+        return str(UUID(value)) == value
+    except ValueError:
+        return False
+
+
+def _validated_rotation_stage(stage: Any) -> tuple[str, Credential]:
+    if not isinstance(stage, dict) or set(stage) != {
+        "version",
+        "previous_id",
+        "credential",
+        "previous",
+        "state",
+    }:
+        raise RotationError("The retained credential renewal state is malformed")
+    previous_id = stage.get("previous_id")
+    credential = stage.get("credential")
+    if (
+        stage.get("version") != 1
+        or stage.get("state") not in {"received", "canonical_written"}
+        or not _is_canonical_uuid(previous_id)
+        or not isinstance(credential, dict)
+        or set(credential) != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"}
+        or credential.get("kind") != "ingest_token"
+        or credential.get("refresh_token") is not None
+        or not valid_ingest_token(credential.get("token"))
+        or not isinstance(credential.get("host"), str)
+        or not isinstance(credential.get("expires_at"), (int, float))
+        or isinstance(credential.get("expires_at"), bool)
+    ):
+        raise RotationError("The retained credential renewal state is malformed")
+    return previous_id, credential  # type: ignore[return-value]
diff --git a/engine/auth/store.py b/engine/auth/store.py
index ae87ca6..cb4e116 100644
--- a/engine/auth/store.py
+++ b/engine/auth/store.py
@@ -43,12 +43,15 @@
 
 from loguru import logger
 
+from engine.auth.lifetime import valid_ingest_token
 from engine.auth.provider import Credential
 
 SERVICE_NAME = "ai.openadapt.desktop"
 _ACTIVE_HOST_ACCOUNT = "__active_host__"
 _PAIRING_STAGE_ACCOUNT = "__pairing_stage__"
 _PAIRING_STAGE_VERSION = 1
+_ROTATION_STAGE_ACCOUNT = "__rotation_stage__"
+_ROTATION_STAGE_VERSION = 1
 
 # Suffix for the companion account that holds the full Credential JSON. The bare
 # ``host`` account holds the RAW bearer token (what the tray reads).
@@ -425,6 +428,198 @@ def clear_pairing_stage(pairing_id: str) -> bool:
     return _apply_exact(_keyring(), _PAIRING_STAGE_ACCOUNT, None)
 
 
+def stage_rotation_credential(
+    previous_id: str,
+    credential: Credential,
+    previous: dict[str, str | None],
+) -> bool:
+    """Durably retain a one-time rotation response before canonical writes."""
+    kr = _keyring()
+    readable, current = _strict_get(kr, _ROTATION_STAGE_ACCOUNT)
+    if not readable or current is not None or previous.get("host") != credential["host"]:
+        return False
+    payload = json.dumps(
+        {
+            "version": _ROTATION_STAGE_VERSION,
+            "previous_id": previous_id,
+            "credential": credential,
+            "previous": previous,
+            "state": "received",
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return _apply_exact(kr, _ROTATION_STAGE_ACCOUNT, payload)
+
+
+def load_rotation_stage() -> dict | None:
+    """Load the one-time rotation response retained for crash recovery."""
+    readable, raw = _strict_get(_keyring(), _ROTATION_STAGE_ACCOUNT)
+    if not readable:
+        raise RuntimeError("rotation recovery keychain entry is unreadable")
+    if raw is None:
+        return None
+    try:
+        value = json.loads(raw)
+    except (json.JSONDecodeError, TypeError) as exc:
+        raise RuntimeError("rotation recovery keychain entry is invalid") from exc
+    if not isinstance(value, dict):
+        raise RuntimeError("rotation recovery keychain entry is invalid")
+    return value
+
+
+def _rotation_stage_values(
+    stage: dict,
+) -> tuple[str, tuple[str | None, ...], tuple[str, ...]] | None:
+    credential = stage.get("credential")
+    previous = stage.get("previous")
+    if (
+        stage.get("version") != _ROTATION_STAGE_VERSION
+        or stage.get("state") not in {"received", "canonical_written"}
+        or not isinstance(stage.get("previous_id"), str)
+        or not isinstance(credential, dict)
+        or not isinstance(previous, dict)
+        or not isinstance(credential.get("host"), str)
+        or previous.get("host") != credential["host"]
+        or not isinstance(credential.get("token"), str)
+    ):
+        return None
+    if any(
+        previous.get(key) is not None and not isinstance(previous.get(key), str)
+        for key in ("token", "credential", "active_host")
+    ):
+        return None
+    host = credential["host"]
+    prior = (
+        previous.get("token"),
+        previous.get("credential"),
+        previous.get("active_host"),
+    )
+    replacement = (credential["token"], json.dumps(credential), host)
+    return host, prior, replacement
+
+
+def _mark_rotation_stage(previous_id: str, state: str) -> bool:
+    if state not in {"received", "canonical_written"}:
+        return False
+    try:
+        stage = load_rotation_stage()
+    except RuntimeError:
+        return False
+    if stage is None or stage.get("previous_id") != previous_id:
+        return False
+    stage["state"] = state
+    payload = json.dumps(stage, sort_keys=True, separators=(",", ":"))
+    return _apply_exact(_keyring(), _ROTATION_STAGE_ACCOUNT, payload)
+
+
+def commit_rotation_stage(previous_id: str) -> bool:
+    """Finish or resume the canonical write for one retained replacement."""
+    try:
+        stage = load_rotation_stage()
+    except RuntimeError:
+        return False
+    if stage is None or stage.get("previous_id") != previous_id:
+        return False
+    values = _rotation_stage_values(stage)
+    if values is None:
+        return False
+    host, previous, replacement = values
+    kr = _keyring()
+    accounts = (host, host + _CRED_SUFFIX, _ACTIVE_HOST_ACCOUNT)
+    observed: list[str | None] = []
+    for account in accounts:
+        readable, value = _strict_get(kr, account)
+        if not readable:
+            return False
+        observed.append(value)
+    if tuple(observed) == replacement:
+        return _mark_rotation_stage(previous_id, "canonical_written")
+    if any(value not in {prior, new} for value, prior, new in zip(observed, previous, replacement)):
+        # Another login won after the rotation request. Preserve both the
+        # newer canonical value and this one-time response for operator review.
+        return False
+    if not all(_apply_exact(kr, account, value) for account, value in zip(accounts, replacement)):
+        # The stage remains, so the next process can finish any partial write.
+        return False
+    return _mark_rotation_stage(previous_id, "canonical_written")
+
+
+def clear_rotation_stage(previous_id: str) -> bool:
+    """Delete only the exact rotation recovery record after promotion."""
+    try:
+        stage = load_rotation_stage()
+    except RuntimeError:
+        return False
+    if stage is None:
+        return True
+    if stage.get("previous_id") != previous_id:
+        return False
+    return _apply_exact(_keyring(), _ROTATION_STAGE_ACCOUNT, None)
+
+
+def clear_superseded_rotation_stage(previous_id: str) -> bool:
+    """Clear an exact stale stage only after a coherent later login won.
+
+    A rejected replacement remains staged so transient propagation can recover.
+    If a later login writes a third, internally consistent credential for the
+    same host, that credential supersedes both the old snapshot and the staged
+    replacement. Retaining the dead stage would otherwise block every future
+    rotation of the new connection.
+    """
+    try:
+        stage = load_rotation_stage()
+    except RuntimeError:
+        return False
+    if stage is None or stage.get("previous_id") != previous_id:
+        return False
+    values = _rotation_stage_values(stage)
+    if values is None:
+        return False
+    host, previous, replacement = values
+    kr = _keyring()
+    accounts = (host, host + _CRED_SUFFIX, _ACTIVE_HOST_ACCOUNT)
+    observed: list[str | None] = []
+    for account in accounts:
+        readable, value = _strict_get(kr, account)
+        if not readable:
+            return False
+        observed.append(value)
+    current = tuple(observed)
+    if current in {previous, replacement}:
+        return False
+    if all(value in {prior, new} for value, prior, new in zip(current, previous, replacement)):
+        # A partial old/replacement write needs recovery, not deletion.
+        return False
+    raw_token, raw_credential, active = current
+    if not valid_ingest_token(raw_token) or not isinstance(raw_credential, str) or active != host:
+        return False
+    try:
+        credential = json.loads(raw_credential)
+    except (json.JSONDecodeError, TypeError):
+        return False
+    if (
+        not isinstance(credential, dict)
+        or set(credential)
+        != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"}
+        or credential.get("kind") != "ingest_token"
+        or credential.get("refresh_token") is not None
+        or credential.get("token") != raw_token
+        or credential.get("host") != host
+    ):
+        return False
+
+    # Re-read the stage after inspecting canonical state. Delete only the same
+    # stage identity; a concurrent rotation cannot be mistaken for this one.
+    try:
+        current_stage = load_rotation_stage()
+    except RuntimeError:
+        return False
+    if current_stage != stage:
+        return False
+    return clear_rotation_stage(previous_id)
+
+
 def load_credential(host: str) -> Credential | None:
     """Load a stored credential for ``host``, or None if absent/unreadable.
 
diff --git a/engine/cli.py b/engine/cli.py
index b1f3412..8028f6e 100644
--- a/engine/cli.py
+++ b/engine/cli.py
@@ -262,6 +262,43 @@ def cmd_login(args: argparse.Namespace, engine: types.SimpleNamespace) -> None:
     engine.audit.log("hosted_login", host=cred["host"], kind=cred["kind"])
 
 
+def cmd_credential(args: argparse.Namespace, engine: types.SimpleNamespace) -> None:
+    """Show the server-authoritative credential lifetime without its bearer."""
+    from engine.auth.rotation import credential_status, expiry_warning
+
+    host = getattr(args, "host", None) or engine.config.hosted_host
+    status = credential_status(host)
+    if status is None:
+        print("Credential status is unavailable. Sign in again if Cloud rejects requests.")
+        return
+    warning = expiry_warning(status)
+    if warning:
+        print(warning)
+        return
+    days = status.get("expires_in_days")
+    if isinstance(days, int):
+        print(f"This connection expires in {days} days. OpenAdapt warns at 14 days.")
+    else:
+        print("This connection has no server-reported expiry date.")
+
+
+def cmd_rotate(args: argparse.Namespace, engine: types.SimpleNamespace) -> None:
+    """Renew the stored Cloud credential through the seven-day overlap."""
+    from engine.auth.rotation import RotationError, rotate_credential
+
+    host = getattr(args, "host", None) or engine.config.hosted_host
+    try:
+        credential = rotate_credential(host)
+    except RotationError as exc:
+        print(f"Credential renewal failed: {exc}")
+        sys.exit(1)
+    print(
+        "Credential renewed. The old credential remains valid for at most "
+        "seven days while this computer uses the replacement."
+    )
+    engine.audit.log("hosted_credential_rotated", host=credential["host"])
+
+
 def cmd_push(args: argparse.Namespace, engine: types.SimpleNamespace) -> None:
     """Zip a recording/bundle directory and push it to /api/ingest."""
     from engine import hosted
@@ -711,6 +748,8 @@ def cmd_doctor(args: argparse.Namespace, engine: types.SimpleNamespace) -> None:
     "dismiss": cmd_dismiss,
     "upload": cmd_upload,
     "login": cmd_login,
+    "credential": cmd_credential,
+    "rotate": cmd_rotate,
     "push": cmd_push,
     "compile": cmd_compile,
     "replay": cmd_replay,
@@ -775,6 +814,12 @@ def main(argv: list[str] | None = None) -> None:
     p.add_argument("--provider", default=None, choices=["paste", "browser_pkce"],
                    help="Force an auth provider")
 
+    # credential lifetime / rotation
+    p = subparsers.add_parser("credential", help="Show Cloud credential lifetime")
+    p.add_argument("--host", default=None, help="Hosted base URL")
+    p = subparsers.add_parser("rotate", help="Renew the stored Cloud credential")
+    p.add_argument("--host", default=None, help="Hosted base URL")
+
     # push
     p = subparsers.add_parser("push", help="Push a recording/bundle to /api/ingest")
     p.add_argument("path", nargs="?", default=None, help="Recording/bundle dir (default: latest)")
diff --git a/src/lib/types.ts b/src/lib/types.ts
index 3a5cf67..bc48179 100644
--- a/src/lib/types.ts
+++ b/src/lib/types.ts
@@ -70,7 +70,7 @@ export type StepState =
 
 export interface AuthStatus {
   authenticated: boolean;
-  kind?: "ingest_token" | "supabase_session";
+  kind?: "ingest_token";
   host?: string;
   org_id?: string | null;
 }
diff --git a/tests/test_engine/conftest.py b/tests/test_engine/conftest.py
index 4b50819..94d97b3 100644
--- a/tests/test_engine/conftest.py
+++ b/tests/test_engine/conftest.py
@@ -64,11 +64,17 @@ def fake_keyring(monkeypatch) -> FakeKeyring:
 class FakeResponse:
     """Minimal httpx.Response stand-in for monkeypatched requests."""
 
-    def __init__(self, status_code: int = 200, json_body: dict | None = None,
-                 text: str = "") -> None:
+    def __init__(
+        self,
+        status_code: int = 200,
+        json_body: dict | None = None,
+        text: str = "",
+        headers: dict[str, str] | None = None,
+    ) -> None:
         self.status_code = status_code
         self._json = json_body if json_body is not None else {}
         self.text = text
+        self.headers = headers or {}
 
     def json(self) -> dict:
         return self._json
diff --git a/tests/test_engine/test_auth_browser_pkce.py b/tests/test_engine/test_auth_browser_pkce.py
index 4247505..93d754f 100644
--- a/tests/test_engine/test_auth_browser_pkce.py
+++ b/tests/test_engine/test_auth_browser_pkce.py
@@ -1,4 +1,4 @@
-"""Tests for BrowserPkceProvider + the loopback receiver."""
+"""End-to-end contract tests for the RFC 8252 Desktop browser login."""
 
 from __future__ import annotations
 
@@ -7,141 +7,436 @@
 import threading
 import time
 import urllib.parse
+from pathlib import Path
+from uuid import uuid4
 
 import httpx
 import pytest
 
-from engine.auth import store
+from engine.auth import pairing, store
 from engine.auth.browser_pkce import (
     BrowserPkceProvider,
     _LoopbackReceiver,
+    _write_callback_body,
     generate_pkce_pair,
 )
 
+HOST = "https://app.openadapt.ai"
+SECRET = "oap_" + "A" * 43
+TOKEN = "oai_ingest_" + "B" * 43
+EXPIRES_AT = "2026-10-26T12:00:00+00:00"
+EXPIRES_TS = 1793016000.0
+REAL_HTTPX_GET = httpx.get
+
+
+class _Response:
+    def __init__(
+        self,
+        status_code: int,
+        body: dict,
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        self.status_code = status_code
+        self._body = body
+        self.headers = headers or {}
+
+    def json(self) -> dict:
+        return self._body
+
+
+def _lifetime() -> dict:
+    return {
+        "expires_at": EXPIRES_AT,
+        "expires_in_days": 90,
+        "expiring_soon": False,
+        "legacy_non_expiring": False,
+        "warning_days": 14,
+    }
+
+
+def _claim_body(pairing_id: str) -> dict:
+    return {
+        "paired": True,
+        "pairing_id": pairing_id,
+        "ingest_token_id": str(uuid4()),
+        "ingest_token": TOKEN,
+        "credential": _lifetime(),
+    }
+
+
+def _validation() -> _Response:
+    return _Response(
+        200,
+        {"count": 0, "credential": _lifetime()},
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "90",
+        },
+    )
+
+
+def _desktop_available(monkeypatch) -> None:
+    monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
+    monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
+    monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True)
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+
+
+def _deliver_from_login_url(
+    url: str,
+    *,
+    code: str | None = SECRET,
+    state: str | None = None,
+    error: str | None = None,
+    include_state: bool = True,
+) -> dict[str, list[str]]:
+    query = urllib.parse.parse_qs(
+        urllib.parse.urlparse(url).query,
+        keep_blank_values=True,
+    )
+    redirect = query["redirect_to"][0]
+    callback_state = query["state"][0] if state is None else state
+
+    def _deliver() -> None:
+        time.sleep(0.02)
+        params: dict[str, str] = {}
+        if code is not None:
+            params["code"] = code
+        if include_state and callback_state is not None:
+            params["state"] = callback_state
+        if error is not None:
+            params["error"] = error
+            params["error_description"] = "The user refused the connection."
+        REAL_HTTPX_GET(redirect, params=params, timeout=5)
+
+    threading.Thread(target=_deliver, daemon=True).start()
+    return query
+
 
 class TestPkce:
-    def test_pair_is_s256(self) -> None:
+    def test_pair_is_s256_and_within_rfc_7636_bounds(self) -> None:
         verifier, challenge = generate_pkce_pair()
-        expected = base64.urlsafe_b64encode(
-            hashlib.sha256(verifier.encode()).digest()
-        ).rstrip(b"=").decode()
+        expected = (
+            base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
+            .rstrip(b"=")
+            .decode()
+        )
         assert challenge == expected
-        assert "=" not in challenge  # base64url, unpadded
+        assert 43 <= len(verifier) <= 128
+        assert len(challenge) == 43
+        assert "=" not in verifier + challenge
 
     def test_pairs_are_unique(self) -> None:
         assert generate_pkce_pair()[0] != generate_pkce_pair()[0]
 
 
 class TestLoopbackReceiver:
-    def test_captures_code(self) -> None:
-        receiver = _LoopbackReceiver()
-        redirect = receiver.redirect_uri
-        assert redirect.startswith("http://127.0.0.1:")
-        assert redirect.endswith("/callback")
+    def test_captures_only_the_exact_callback(self) -> None:
+        with _LoopbackReceiver() as receiver:
+            redirect = receiver.redirect_uri
+            assert redirect.startswith("http://127.0.0.1:")
+            assert redirect.endswith("/callback")
 
-        def _deliver():
-            time.sleep(0.1)
-            httpx.get(redirect, params={"code": "abc", "state": "s1"}, timeout=5)
+            def _deliver() -> None:
+                time.sleep(0.02)
+                wrong = redirect.replace("/callback", "/")
+                assert httpx.get(wrong, timeout=5).status_code == 404
+                response = httpx.get(
+                    redirect,
+                    params={"code": SECRET, "state": "state_1234567890"},
+                    timeout=5,
+                )
+                assert response.headers["cache-control"] == "no-store"
+                assert response.headers["referrer-policy"] == "no-referrer"
 
-        threading.Thread(target=_deliver, daemon=True).start()
-        receiver.serve_until_code(timeout=5)
-        assert receiver.code == "abc"
-        assert receiver.state == "s1"
+            threading.Thread(target=_deliver, daemon=True).start()
+            receiver.serve_until_code(timeout=5)
+            assert receiver.code == SECRET
+            assert receiver.state == "state_1234567890"
 
+    def test_close_before_server_start_does_not_call_blocking_shutdown(self) -> None:
+        receiver = _LoopbackReceiver()
+        receiver.close()
+        receiver.close()
+
+    def test_browser_disconnect_after_callback_still_signals_completion(self) -> None:
+        event = threading.Event()
+
+        class _DisconnectedStream:
+            def write(self, body: bytes) -> None:
+                raise BrokenPipeError("browser closed")
 
-def _configured(**kwargs) -> BrowserPkceProvider:
-    """A provider whose Supabase project is configured, as in a wired deployment."""
-    kwargs.setdefault("supabase_url", "https://project.supabase.co")
-    kwargs.setdefault("supabase_anon_key", "anon_key")
-    return BrowserPkceProvider(**kwargs)
+        with pytest.raises(BrokenPipeError):
+            _write_callback_body(_DisconnectedStream(), b"done", event)
+        assert event.is_set()
 
 
 class TestIsAvailable:
     def test_headless_env_false(self, monkeypatch) -> None:
         monkeypatch.setenv("OPENADAPT_HEADLESS", "1")
-        assert _configured().is_available() is False
+        monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True)
+        assert BrowserPkceProvider().is_available() is False
 
     def test_linux_without_display_false(self, monkeypatch) -> None:
         monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
         monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "linux")
+        monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True)
         monkeypatch.delenv("DISPLAY", raising=False)
         monkeypatch.delenv("WAYLAND_DISPLAY", raising=False)
-        assert _configured().is_available() is False
-
-    def test_macos_true_when_configured(self, monkeypatch) -> None:
-        monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
-        monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
-        assert _configured().is_available() is True
-
-    def test_unconfigured_is_unavailable_even_on_a_desktop(self, monkeypatch) -> None:
-        """An unconfigured provider must not lead the chain and stall the login.
+        assert BrowserPkceProvider().is_available() is False
 
-        Without a Supabase project the exchange cannot succeed, but it is only
-        attempted after the loopback wait -- so reporting available here would
-        open a browser tab and block for the full timeout before falling back
-        to token paste. Regression guard for that stall.
-        """
-        monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
+    def test_complete_cloud_contract_needs_no_supabase_client_configuration(
+        self, monkeypatch
+    ) -> None:
+        _desktop_available(monkeypatch)
         monkeypatch.delenv("OPENADAPT_SUPABASE_URL", raising=False)
         monkeypatch.delenv("OPENADAPT_SUPABASE_ANON_KEY", raising=False)
-        monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
-        assert BrowserPkceProvider().is_available() is False
+        assert BrowserPkceProvider(open_browser=lambda _: None).is_available() is True
+        source = Path("engine/auth/browser_pkce.py").read_text()
+        assert "/auth/v1/token" not in source
+        assert "OPENADAPT_SUPABASE_" not in source
 
-    def test_partial_configuration_is_unavailable(self, monkeypatch) -> None:
-        monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
-        monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
-        url_only = BrowserPkceProvider(
-            supabase_url="https://project.supabase.co", supabase_anon_key=""
+    def test_missing_system_browser_is_unavailable(self, monkeypatch) -> None:
+        import webbrowser
+
+        _desktop_available(monkeypatch)
+        monkeypatch.setattr(
+            webbrowser,
+            "get",
+            lambda: (_ for _ in ()).throw(webbrowser.Error("no browser")),
         )
-        key_only = BrowserPkceProvider(supabase_url="", supabase_anon_key="anon_key")
-        assert url_only.is_available() is False
-        assert key_only.is_available() is False
+        assert BrowserPkceProvider().is_available() is False
 
-    def test_unconfigured_provider_never_opens_a_browser(self, monkeypatch) -> None:
-        """The stall is user-visible as a stray tab; assert none is opened."""
+    def test_locked_keychain_refuses_before_opening_browser(self, monkeypatch) -> None:
         monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
-        monkeypatch.delenv("OPENADAPT_SUPABASE_URL", raising=False)
-        monkeypatch.delenv("OPENADAPT_SUPABASE_ANON_KEY", raising=False)
         monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
+        monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: False)
         opened: list[str] = []
-        provider = BrowserPkceProvider(open_browser=opened.append)
+        with pytest.raises(RuntimeError, match="unavailable"):
+            BrowserPkceProvider(open_browser=opened.append).login()
+        assert opened == []
+
+    def test_untrusted_cloud_origin_refuses_before_opening_browser(self, monkeypatch) -> None:
+        _desktop_available(monkeypatch)
+        opened: list[str] = []
+        provider = BrowserPkceProvider(
+            host="https://app.openadapt.ai.evil.example",
+            open_browser=opened.append,
+        )
         with pytest.raises(RuntimeError, match="unavailable"):
             provider.login()
         assert opened == []
 
 
 class TestLogin:
-    def test_full_flow(self, fake_keyring, monkeypatch) -> None:
-        monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False)
-        monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin")
+    def test_full_flow_uses_one_oap_claim_and_the_shared_durable_pipeline(
+        self, fake_keyring, monkeypatch
+    ) -> None:
+        _desktop_available(monkeypatch)
+        pairing_id = str(uuid4())
+        events: list[str] = []
+        sent_challenge: list[str] = []
+        real_snapshot = pairing.snapshot_pairing_canonical
+        real_stage = pairing.stage_pairing_credential
+        real_commit = pairing.commit_pairing_stage
+        real_clear = pairing.clear_pairing_stage
 
         def _open_browser(url: str) -> None:
-            # Extract the loopback redirect and deliver a code asynchronously.
-            qs = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
-            redirect = qs["redirect_to"][0]
-            state = qs["state"][0]
+            query = _deliver_from_login_url(url)
+            assert set(query) == {
+                "redirect_to",
+                "code_challenge",
+                "code_challenge_method",
+                "state",
+            }
+            assert all(len(values) == 1 for values in query.values())
+            assert query["code_challenge_method"] == ["S256"]
+            assert len(query["code_challenge"][0]) == 43
+            sent_challenge.append(query["code_challenge"][0])
+            assert 16 <= len(query["state"][0]) <= 128
 
-            def _deliver():
-                time.sleep(0.1)
-                httpx.get(redirect, params={"code": "auth_code_1", "state": state}, timeout=5)
+        def _post(url, *, json, headers=None, **kwargs):
+            if url.endswith("/claim"):
+                events.append("claim")
+                assert headers is None
+                assert set(json) == {"pairing_secret", "device_name", "code_verifier"}
+                assert json["pairing_secret"] == SECRET
+                assert 43 <= len(json["code_verifier"]) <= 128
+                challenge = (
+                    base64.urlsafe_b64encode(
+                        hashlib.sha256(json["code_verifier"].encode()).digest()
+                    )
+                    .rstrip(b"=")
+                    .decode()
+                )
+                # It is the verifier for the challenge sent in the login URL.
+                assert sent_challenge == [challenge]
+                return _Response(
+                    201,
+                    _claim_body(pairing_id),
+                    {"cache-control": "no-store", "referrer-policy": "no-referrer"},
+                )
+            events.append("confirm")
+            assert url.endswith("/confirm")
+            assert json == {"pairing_id": pairing_id}
+            assert headers == {"Authorization": f"Bearer {TOKEN}"}
+            return _Response(200, {"connected": True})
 
-            threading.Thread(target=_deliver, daemon=True).start()
+        def _snapshot(host):
+            events.append("snapshot")
+            return real_snapshot(host)
+
+        def _stage(*args):
+            events.append("stage")
+            return real_stage(*args)
+
+        def _validate(*args, **kwargs):
+            events.append("validate")
+            return _validation()
 
-        provider = _configured(host="https://app.openadapt.ai", open_browser=_open_browser)
-        provider._exchange_code = lambda code, verifier, redirect_uri: {  # type: ignore[assignment]
-            "access_token": "supabase_access",
-            "refresh_token": "supabase_refresh",
-            "expires_at": 1234.0,
+        def _commit(value):
+            events.append("commit")
+            return real_commit(value)
+
+        def _clear(value):
+            events.append("clear")
+            return real_clear(value)
+
+        monkeypatch.setattr(pairing, "_safe_device_name", lambda: "test-device")
+        monkeypatch.setattr(pairing, "snapshot_pairing_canonical", _snapshot)
+        monkeypatch.setattr(pairing, "stage_pairing_credential", _stage)
+        monkeypatch.setattr(pairing, "commit_pairing_stage", _commit)
+        monkeypatch.setattr(pairing, "clear_pairing_stage", _clear)
+        monkeypatch.setattr(pairing.httpx, "post", _post)
+        monkeypatch.setattr(pairing.httpx, "get", _validate)
+
+        credential = BrowserPkceProvider(host=HOST, open_browser=_open_browser).login()
+
+        assert credential == {
+            "kind": "ingest_token",
+            "token": TOKEN,
+            "refresh_token": None,
+            "org_id": None,
+            "host": HOST,
+            "expires_at": EXPIRES_TS,
         }
-        provider._mint_ingest_token = lambda access_token: ("oai_ingest_minted", "org_9")  # type: ignore[assignment]
-
-        cred = provider.login()
-        assert cred["kind"] == "ingest_token"
-        assert cred["token"] == "oai_ingest_minted"
-        assert cred["refresh_token"] == "supabase_refresh"
-        assert cred["org_id"] == "org_9"
-        # The bearer path resolves the minted ingest token.
-        assert store.auth_header() == {"Authorization": "Bearer oai_ingest_minted"}
+        assert store.auth_header() == {"Authorization": f"Bearer {TOKEN}"}
+        assert events == [
+            "snapshot",
+            "claim",
+            "stage",
+            "validate",
+            "commit",
+            "confirm",
+            "clear",
+        ]
+
+    def test_local_alias_uses_one_canonical_host_for_login_claim_and_readback(
+        self, fake_keyring, monkeypatch
+    ) -> None:
+        _desktop_available(monkeypatch)
+        pairing_id = str(uuid4())
+        login_urls: list[str] = []
+
+        def _open_browser(url: str) -> None:
+            login_urls.append(url)
+            _deliver_from_login_url(url)
+
+        def _post(url, *, json, headers=None, **kwargs):
+            if url.endswith("/claim"):
+                assert url == "http://localhost/api/local-bridge/pairings/claim"
+                return _Response(
+                    201,
+                    _claim_body(pairing_id),
+                    {"cache-control": "no-store", "referrer-policy": "no-referrer"},
+                )
+            assert url == "http://localhost/api/local-bridge/pairings/confirm"
+            return _Response(200, {"connected": True})
+
+        monkeypatch.setattr(pairing.httpx, "post", _post)
+        monkeypatch.setattr(pairing.httpx, "get", lambda *a, **k: _validation())
+
+        credential = BrowserPkceProvider(
+            host="http://LOCALHOST:80/",
+            open_browser=_open_browser,
+        ).login()
+
+        assert login_urls[0].startswith("http://localhost/login?")
+        assert credential["host"] == "http://localhost"
+        assert store.load_credential("http://localhost") == credential
+
+    @pytest.mark.parametrize(
+        "wrong_code",
+        [
+            "supabase_authorization_code",
+            "oai_ingest_" + "A" * 43,
+            "oapp_" + "A" * 43,
+            "oaps_" + "A" * 43,
+        ],
+    )
+    def test_rejects_supabase_and_other_credential_prefixes_before_claim(
+        self, wrong_code, monkeypatch
+    ) -> None:
+        _desktop_available(monkeypatch)
+        claimed = False
+
+        def _post(*args, **kwargs):
+            nonlocal claimed
+            claimed = True
+            raise AssertionError("a wrong credential role must not reach claim")
+
+        monkeypatch.setattr(pairing.httpx, "post", _post)
+        provider = BrowserPkceProvider(
+            open_browser=lambda url: _deliver_from_login_url(url, code=wrong_code)
+        )
+        with pytest.raises(RuntimeError, match="invalid OpenAdapt pairing code"):
+            provider.login()
+        assert claimed is False
+
+    @pytest.mark.parametrize("callback_state", ["", "wrong_state_123456"])
+    def test_requires_present_exact_state(self, callback_state, monkeypatch) -> None:
+        _desktop_available(monkeypatch)
+        provider = BrowserPkceProvider(
+            open_browser=lambda url: _deliver_from_login_url(
+                url,
+                state=callback_state,
+            )
+        )
+        with pytest.raises(RuntimeError, match="state mismatch"):
+            provider.login()
+
+    def test_access_denied_stops_without_a_claim(self, monkeypatch) -> None:
+        _desktop_available(monkeypatch)
+        provider = BrowserPkceProvider(
+            open_browser=lambda url: _deliver_from_login_url(
+                url,
+                code=None,
+                error="access_denied",
+            )
+        )
+        with pytest.raises(RuntimeError, match="user refused"):
+            provider.login()
+
+    def test_access_denied_without_state_is_rejected_as_a_mismatch(self, monkeypatch) -> None:
+        _desktop_available(monkeypatch)
+        provider = BrowserPkceProvider(
+            open_browser=lambda url: _deliver_from_login_url(
+                url,
+                code=None,
+                error="access_denied",
+                include_state=False,
+            )
+        )
+        with pytest.raises(RuntimeError, match="state mismatch"):
+            provider.login()
+
+    def test_pairing_410_is_a_safe_login_failure(self, monkeypatch) -> None:
+        _desktop_available(monkeypatch)
+        monkeypatch.setattr(pairing.httpx, "post", lambda *a, **k: _Response(410, {}))
+        provider = BrowserPkceProvider(open_browser=lambda url: _deliver_from_login_url(url))
+        with pytest.raises(RuntimeError, match="expired.*already used"):
+            provider.login()
 
     def test_headless_login_raises(self, monkeypatch) -> None:
         monkeypatch.setenv("OPENADAPT_HEADLESS", "1")
diff --git a/tests/test_engine/test_auth_dispatch.py b/tests/test_engine/test_auth_dispatch.py
index 9a7907e..393c5f8 100644
--- a/tests/test_engine/test_auth_dispatch.py
+++ b/tests/test_engine/test_auth_dispatch.py
@@ -69,6 +69,37 @@ def login(self):
         auth.login(prefer="paste")
         assert called == ["paste"]
 
+    def test_available_browser_failure_does_not_overwrite_recovery_with_paste(
+        self, monkeypatch
+    ) -> None:
+        called = []
+
+        class _Browser:
+            name = "browser_pkce"
+
+            def is_available(self):
+                return True
+
+            def login(self):
+                called.append("browser")
+                raise RuntimeError("retained one-use credential needs recovery")
+
+        class _Paste:
+            name = "paste"
+
+            def is_available(self):
+                return True
+
+            def login(self):
+                called.append("paste")
+                return {}
+
+        monkeypatch.setattr(auth, "available_providers", lambda host="": [_Browser(), _Paste()])
+
+        with pytest.raises(RuntimeError, match="needs recovery"):
+            auth.login()
+        assert called == ["browser"]
+
     def test_no_provider_raises(self, monkeypatch) -> None:
         class _Unavailable:
             name = "browser_pkce"
diff --git a/tests/test_engine/test_auth_pairing.py b/tests/test_engine/test_auth_pairing.py
index 96af644..3bdea5a 100644
--- a/tests/test_engine/test_auth_pairing.py
+++ b/tests/test_engine/test_auth_pairing.py
@@ -9,23 +9,67 @@
 from engine.auth import pairing, store
 
 SECRET = "oap_" + "A" * 43
-TOKEN = "oai_ingest_" + "B" * 32
-OLD_TOKEN = "oai_ingest_" + "C" * 32
-OTHER_TOKEN = "oai_ingest_" + "D" * 32
+TOKEN = "oai_ingest_" + "B" * 43
+OLD_TOKEN = "oai_ingest_" + "C" * 43
+OTHER_TOKEN = "oai_ingest_" + "D" * 43
 HOST = "https://app.openadapt.ai"
 OTHER_HOST = "https://previous.example"
 VALID_URI = f"openadapt://connect?pairing={SECRET}&host=https%3A%2F%2Fapp.openadapt.ai"
+EXPIRES_AT = "2026-10-26T12:00:00+00:00"
+EXPIRES_TS = 1793016000.0
 
 
 class _Response:
-    def __init__(self, status_code: int, body: dict) -> None:
+    def __init__(
+        self,
+        status_code: int,
+        body: dict,
+        headers: dict[str, str] | None = None,
+    ) -> None:
         self.status_code = status_code
         self._body = body
+        self.headers = headers or {}
 
     def json(self) -> dict:
         return self._body
 
 
+def _credential_lifetime() -> dict:
+    return {
+        "expires_at": EXPIRES_AT,
+        "expires_in_days": 90,
+        "expiring_soon": False,
+        "legacy_non_expiring": False,
+        "warning_days": 14,
+    }
+
+
+def _claim_response(pairing_id: str) -> _Response:
+    return _Response(
+        201,
+        {
+            "paired": True,
+            "pairing_id": pairing_id,
+            "ingest_token_id": str(uuid4()),
+            "ingest_token": TOKEN,
+            "credential": _credential_lifetime(),
+        },
+        {"cache-control": "no-store", "referrer-policy": "no-referrer"},
+    )
+
+
+def _validation_response(status: int = 200) -> _Response:
+    return _Response(
+        status,
+        {"count": 0, "credential": _credential_lifetime()},
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "90",
+        },
+    )
+
+
 def test_parser_accepts_only_the_fixed_connect_action() -> None:
     assert pairing.parse_connect_uri(VALID_URI) == {
         "pairing": SECRET,
@@ -123,7 +167,7 @@ def _post(url, *, json, headers=None, **kwargs):
             events.append("claim")
             assert json == {"pairing_secret": SECRET, "device_name": "test-device"}
             assert headers is None
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         assert not url.endswith("/abort")
         events.append("confirm")
         assert url.endswith("/confirm")
@@ -143,7 +187,7 @@ def _get(url, *, headers, **kwargs):
         events.append("validate")
         assert url == f"{HOST}/api/needs-attention/count"
         assert headers == {"Authorization": f"Bearer {TOKEN}"}
-        return _Response(200, {"count": 0})
+        return _validation_response()
 
     def _commit(value):
         events.append("commit")
@@ -167,7 +211,7 @@ def _clear(value):
     assert result["authenticated"] is True
     assert result["host"] == HOST
     assert "token" not in result
-    assert store.load_credential(HOST) == _credential(HOST, TOKEN)
+    assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS)
     assert store.load_pairing_stage() is None
     assert events == [
         "snapshot",
@@ -180,14 +224,14 @@ def _clear(value):
     ]
 
 
-def _credential(host: str, token: str) -> dict:
+def _credential(host: str, token: str, expires_at: float | None = None) -> dict:
     return {
         "kind": "ingest_token",
         "token": token,
         "refresh_token": None,
         "org_id": None,
         "host": host,
-        "expires_at": None,
+        "expires_at": expires_at,
     }
 
 
@@ -202,7 +246,7 @@ def test_rejected_verification_aborts_new_claim_without_touching_prior_state(
     def _post(url, *, json, headers=None, **kwargs):
         posts.append((url, json, headers or {}))
         if url.endswith("/claim"):
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         assert url.endswith("/abort")
         return _Response(200, {"revoked": True})
 
@@ -239,7 +283,7 @@ def test_confirm_ambiguity_preserves_new_canonical_and_recovers_idempotently(
     def _post(url, *, json, headers=None, **kwargs):
         endpoints.append(url.rsplit("/", 1)[-1])
         if url.endswith("/claim"):
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         if url.endswith("/confirm"):
             if confirm_ambiguous:
                 return _Response(503, {})
@@ -248,11 +292,11 @@ def _post(url, *, json, headers=None, **kwargs):
 
     monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
     monkeypatch.setattr(pairing.httpx, "post", _post)
-    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {}))
+    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response())
 
     with pytest.raises(pairing.PairingError, match="confirmation.*uncertain"):
         pairing.connect_uri(VALID_URI)
-    assert store.load_credential(HOST) == _credential(HOST, TOKEN)
+    assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS)
     assert store.active_host() == HOST
     assert store.load_pairing_stage()["state"] == "confirm_ambiguous"
     assert endpoints == ["claim", "confirm", "confirm"]
@@ -260,7 +304,7 @@ def _post(url, *, json, headers=None, **kwargs):
     confirm_ambiguous = False
     recovered = pairing.recover_pending_pairing()
     assert recovered is not None and recovered["authenticated"] is True
-    assert store.load_credential(HOST) == _credential(HOST, TOKEN)
+    assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS)
     assert store.load_pairing_stage() is None
     assert endpoints[-1] == "confirm"
 
@@ -274,7 +318,7 @@ def test_crash_after_canonical_write_recovers_without_a_second_claim(
     assert previous is not None
     assert store.stage_pairing_credential(
         pairing_id,
-        _credential(HOST, TOKEN),
+        _credential(HOST, TOKEN, EXPIRES_TS),
         previous,
         "crash-device",
     )
@@ -293,13 +337,13 @@ def _post(url, **kwargs):
         return _Response(200, {"connected": True})
 
     monkeypatch.setattr(pairing.httpx, "post", _post)
-    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {}))
+    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response())
 
     recovered = pairing.recover_pending_pairing()
     assert recovered is not None
     assert recovered["device_name"] == "crash-device"
     assert posts == ["confirm"]
-    assert store.load_credential(HOST) == _credential(HOST, TOKEN)
+    assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS)
     assert store.load_pairing_stage() is None
 
 
@@ -315,14 +359,14 @@ def _post(url, **kwargs):
         endpoint = url.rsplit("/", 1)[-1]
         endpoints.append(endpoint)
         if endpoint == "claim":
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         if endpoint == "confirm":
             return _Response(409, {})
         return _Response(200, {"revoked": True})
 
     monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
     monkeypatch.setattr(pairing.httpx, "post", _post)
-    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {}))
+    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response())
 
     with pytest.raises(pairing.PairingError, match="previous Desktop connection is unchanged"):
         pairing.connect_uri(VALID_URI)
@@ -341,16 +385,16 @@ def test_abort_409_never_blindly_restores_over_possibly_confirmed_token(
 
     def _post(url, **kwargs):
         if url.endswith("/claim"):
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         return _Response(409, {})
 
     monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
     monkeypatch.setattr(pairing.httpx, "post", _post)
-    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {}))
+    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response())
 
     with pytest.raises(pairing.PairingError, match="preserved.*safe recovery"):
         pairing.connect_uri(VALID_URI)
-    assert store.load_credential(HOST) == _credential(HOST, TOKEN)
+    assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS)
     assert store.active_host() == HOST
     assert store.load_pairing_stage() is not None
 
@@ -375,13 +419,13 @@ def _fail_new_companion(service, account, value):
     def _post(url, *, json, headers=None, **kwargs):
         endpoints.append(url.rsplit("/", 1)[-1])
         if url.endswith("/claim"):
-            return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id})
+            return _claim_response(pairing_id)
         assert url.endswith("/abort")
         return _Response(200, {"revoked": True})
 
     monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
     monkeypatch.setattr(pairing.httpx, "post", _post)
-    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {}))
+    monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response())
 
     with pytest.raises(pairing.PairingError, match="atomically write"):
         pairing.connect_uri(VALID_URI)
@@ -390,3 +434,116 @@ def _post(url, *, json, headers=None, **kwargs):
     assert store.active_host() == OTHER_HOST
     assert store.load_pairing_stage() is None
     assert endpoints == ["claim", "abort"]
+
+
+def test_pkce_claim_sends_the_exact_verifier_and_maps_410(monkeypatch) -> None:
+    verifier = "V" * 43
+    requests: list[dict] = []
+
+    def _post(url, *, json, headers=None, **kwargs):
+        requests.append(json)
+        assert url == f"{HOST}/api/local-bridge/pairings/claim"
+        assert headers is None
+        return _Response(410, {})
+
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+    monkeypatch.setattr(pairing, "_safe_device_name", lambda: "test-device")
+    monkeypatch.setattr(pairing.httpx, "post", _post)
+
+    with pytest.raises(pairing.PairingError, match="expired.*already used"):
+        pairing.claim_pairing(HOST, SECRET, code_verifier=verifier)
+    assert requests == [
+        {
+            "pairing_secret": SECRET,
+            "device_name": "test-device",
+            "code_verifier": verifier,
+        }
+    ]
+
+
+@pytest.mark.parametrize(
+    "wrong_role",
+    [
+        "supabase_authorization_code",
+        "oai_ingest_" + "A" * 43,
+        "oapp_" + "A" * 43,
+        "oaps_" + "A" * 43,
+    ],
+)
+def test_claim_rejects_every_non_pairing_credential_before_network(wrong_role, monkeypatch) -> None:
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+    monkeypatch.setattr(
+        pairing.httpx,
+        "post",
+        lambda *a, **k: pytest.fail("wrong credential role reached the claim route"),
+    )
+    with pytest.raises(pairing.PairingError, match="malformed"):
+        pairing.claim_pairing(HOST, wrong_role, code_verifier="V" * 43)
+
+
+def test_malformed_201_is_aborted_when_the_rollback_identity_is_available(
+    monkeypatch,
+) -> None:
+    pairing_id = str(uuid4())
+    endpoints: list[str] = []
+    malformed = _claim_response(pairing_id)._body
+    malformed.pop("ingest_token_id")
+
+    def _post(url, **kwargs):
+        endpoints.append(url.rsplit("/", 1)[-1])
+        if url.endswith("/claim"):
+            return _Response(201, malformed)
+        assert kwargs["headers"] == {"Authorization": f"Bearer {TOKEN}"}
+        return _Response(200, {"revoked": True})
+
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+    monkeypatch.setattr(pairing.httpx, "post", _post)
+    with pytest.raises(pairing.PairingError, match="rolled back.*previous connection"):
+        pairing.claim_pairing(HOST, SECRET)
+    assert endpoints == ["claim", "abort"]
+    assert store.load_credential(HOST) is None
+    assert store.load_pairing_stage() is None
+
+
+def test_claim_without_no_store_is_aborted(monkeypatch) -> None:
+    pairing_id = str(uuid4())
+    endpoints: list[str] = []
+    claim = _claim_response(pairing_id)
+    claim.headers = {}
+
+    def _post(url, **kwargs):
+        endpoints.append(url.rsplit("/", 1)[-1])
+        if url.endswith("/claim"):
+            return claim
+        return _Response(200, {"revoked": True})
+
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+    monkeypatch.setattr(pairing.httpx, "post", _post)
+    with pytest.raises(pairing.PairingError, match="rolled back"):
+        pairing.claim_pairing(HOST, SECRET)
+    assert endpoints == ["claim", "abort"]
+    assert store.load_credential(HOST) is None
+
+
+def test_validation_requires_matching_14_day_headers_and_rolls_back(monkeypatch) -> None:
+    pairing_id = str(uuid4())
+    endpoints: list[str] = []
+
+    def _post(url, **kwargs):
+        endpoints.append(url.rsplit("/", 1)[-1])
+        if url.endswith("/claim"):
+            return _claim_response(pairing_id)
+        if url.endswith("/abort"):
+            return _Response(200, {"revoked": True})
+        raise AssertionError("a malformed validation must not confirm")
+
+    invalid_headers = _validation_response()
+    invalid_headers.headers["x-openadapt-credential-warning-days"] = "30"
+    monkeypatch.setattr(pairing, "secure_store_available", lambda: True)
+    monkeypatch.setattr(pairing.httpx, "post", _post)
+    monkeypatch.setattr(pairing.httpx, "get", lambda *a, **k: invalid_headers)
+
+    with pytest.raises(pairing.PairingError, match="lifetime contract"):
+        pairing.claim_pairing(HOST, SECRET)
+    assert endpoints == ["claim", "abort"]
+    assert store.load_credential(HOST) is None
diff --git a/tests/test_engine/test_auth_paste.py b/tests/test_engine/test_auth_paste.py
index 55c9bfe..3072cd0 100644
--- a/tests/test_engine/test_auth_paste.py
+++ b/tests/test_engine/test_auth_paste.py
@@ -10,6 +10,33 @@
 
 from .conftest import FakeResponse
 
+TOKEN = "oai_ingest_" + "A" * 43
+ENV_TOKEN = "oai_ingest_" + "B" * 43
+PASTED_TOKEN = "oai_ingest_" + "C" * 43
+EXPIRES_AT = "2026-10-26T12:00:00+00:00"
+EXPIRES_TS = 1793016000.0
+
+
+def _lifetime_response(org_id: str = "org_42") -> FakeResponse:
+    return FakeResponse(
+        200,
+        {
+            "org_id": org_id,
+            "credential": {
+                "expires_at": EXPIRES_AT,
+                "expires_in_days": 90,
+                "expiring_soon": False,
+                "legacy_non_expiring": False,
+                "warning_days": 14,
+            },
+        },
+        headers={
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "90",
+        },
+    )
+
 
 class TestPasteTokenProvider:
     def test_is_available_always_true(self) -> None:
@@ -25,18 +52,19 @@ def test_settings_url(self) -> None:
     def test_login_with_explicit_token(self, fake_keyring, monkeypatch) -> None:
         monkeypatch.setattr(
             "engine.auth.paste.httpx.get",
-            lambda *a, **k: FakeResponse(200, {"org_id": "org_42"}),
+            lambda *a, **k: _lifetime_response(),
         )
         provider = PasteTokenProvider(host="https://app.openadapt.ai")
-        cred = provider.login(token="oai_ingest_xyz")
+        cred = provider.login(token=TOKEN)
         assert cred["kind"] == "ingest_token"
-        assert cred["token"] == "oai_ingest_xyz"
+        assert cred["token"] == TOKEN
         assert cred["org_id"] == "org_42"
+        assert cred["expires_at"] == EXPIRES_TS
         # Persisted + resolvable via auth_header.
-        assert store.auth_header() == {"Authorization": "Bearer oai_ingest_xyz"}
+        assert store.auth_header() == {"Authorization": f"Bearer {TOKEN}"}
 
     def test_login_reads_env_when_headless(self, fake_keyring, monkeypatch) -> None:
-        monkeypatch.setenv("OPENADAPT_INGEST_TOKEN", "oai_ingest_env")
+        monkeypatch.setenv("OPENADAPT_INGEST_TOKEN", ENV_TOKEN)
         monkeypatch.setattr(
             "engine.auth.paste.httpx.get",
             lambda *a, **k: FakeResponse(200, {"org_id": "org_env"}),
@@ -46,15 +74,27 @@ def _no_prompt(_):
             raise AssertionError("should not prompt when env is set")
 
         cred = PasteTokenProvider(prompt=_no_prompt).login()
-        assert cred["token"] == "oai_ingest_env"
+        assert cred["token"] == ENV_TOKEN
 
     def test_login_prompts_interactively(self, fake_keyring, monkeypatch) -> None:
         monkeypatch.setattr(
             "engine.auth.paste.httpx.get",
             lambda *a, **k: FakeResponse(200, {}),
         )
-        cred = PasteTokenProvider(prompt=lambda _: "  oai_ingest_pasted  ").login()
-        assert cred["token"] == "oai_ingest_pasted"
+        cred = PasteTokenProvider(prompt=lambda _: f"  {PASTED_TOKEN}  ").login()
+        assert cred["token"] == PASTED_TOKEN
+
+    def test_working_token_fallback_keeps_unknown_expiry_from_an_older_cloud(
+        self, fake_keyring, monkeypatch
+    ) -> None:
+        response = _lifetime_response()
+        response.headers.pop("cache-control")
+        monkeypatch.setattr("engine.auth.paste.httpx.get", lambda *a, **k: response)
+
+        cred = PasteTokenProvider().login(token=TOKEN)
+
+        assert cred["token"] == TOKEN
+        assert cred["expires_at"] is None
 
     def test_login_rejects_bad_token(self, fake_keyring, monkeypatch) -> None:
         monkeypatch.setattr(
@@ -62,7 +102,7 @@ def test_login_rejects_bad_token(self, fake_keyring, monkeypatch) -> None:
             lambda *a, **k: FakeResponse(401),
         )
         with pytest.raises(TokenValidationError, match="rejected"):
-            PasteTokenProvider().login(token="oai_ingest_bad")
+            PasteTokenProvider().login(token=TOKEN)
 
     def test_login_network_error(self, fake_keyring, monkeypatch) -> None:
         def _raise(*a, **k):
@@ -70,7 +110,26 @@ def _raise(*a, **k):
 
         monkeypatch.setattr("engine.auth.paste.httpx.get", _raise)
         with pytest.raises(TokenValidationError, match="Could not reach"):
-            PasteTokenProvider().login(token="oai_ingest_x")
+            PasteTokenProvider().login(token=TOKEN)
+
+    @pytest.mark.parametrize(
+        "wrong_role",
+        [
+            "supabase_authorization_code",
+            "oap_" + "A" * 43,
+            "oapp_" + "A" * 43,
+            "oaps_" + "A" * 43,
+        ],
+    )
+    def test_rejects_every_non_ingest_credential_before_network(
+        self, wrong_role, monkeypatch
+    ) -> None:
+        monkeypatch.setattr(
+            "engine.auth.paste.httpx.get",
+            lambda *a, **k: pytest.fail("wrong credential role reached Cloud validation"),
+        )
+        with pytest.raises(TokenValidationError, match="pairing code or portal"):
+            PasteTokenProvider().login(token=wrong_role)
 
     def test_login_no_token_raises(self, fake_keyring) -> None:
         with pytest.raises(TokenValidationError, match="No ingest token"):
diff --git a/tests/test_engine/test_auth_rotation.py b/tests/test_engine/test_auth_rotation.py
new file mode 100644
index 0000000..0495506
--- /dev/null
+++ b/tests/test_engine/test_auth_rotation.py
@@ -0,0 +1,514 @@
+"""Credential lifetime warnings and no-outage rotation contracts."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+import httpx
+import pytest
+
+from engine.auth import rotation, store
+
+HOST = "https://app.openadapt.ai"
+OLD_TOKEN = "oai_ingest_" + "A" * 43
+NEW_TOKEN = "oai_ingest_" + "B" * 43
+THIRD_TOKEN = "oai_ingest_" + "C" * 43
+FOURTH_TOKEN = "oai_ingest_" + "D" * 43
+PREVIOUS_ID = str(uuid4())
+REPLACEMENT_ID = str(uuid4())
+
+
+class _Response:
+    def __init__(
+        self,
+        status_code: int,
+        body: dict,
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        self.status_code = status_code
+        self._body = body
+        self.headers = headers or {}
+
+    def json(self) -> dict:
+        return self._body
+
+
+def _iso(days: int) -> str:
+    return (datetime.now(timezone.utc) + timedelta(days=days)).isoformat()
+
+
+def _lifetime(days: int = 90) -> dict:
+    return {
+        "expires_at": _iso(days),
+        "expires_in_days": days,
+        "expiring_soon": days <= 14,
+        "legacy_non_expiring": False,
+        "warning_days": 14,
+    }
+
+
+def _rotation_response(
+    *,
+    token: str = NEW_TOKEN,
+    previous_id: str = PREVIOUS_ID,
+    replacement_id: str = REPLACEMENT_ID,
+) -> _Response:
+    lifetime = _lifetime(90)
+    return _Response(
+        201,
+        {
+            "token": token,
+            "record": {
+                "id": replacement_id,
+                "org_id": "org_1",
+                "name": "Desktop",
+                "token_prefix": token[:20],
+                "created_at": _iso(0),
+                "last_used_at": None,
+                "expires_at": lifetime["expires_at"],
+                "revoked_at": None,
+                "rotated_to_id": None,
+            },
+            "previous_id": previous_id,
+            "previous_expires_at": _iso(7),
+            "credential": lifetime,
+        },
+        {"cache-control": "no-store", "referrer-policy": "no-referrer"},
+    )
+
+
+def _replacement_validation_response(
+    credential: dict,
+    *,
+    days: int | None = None,
+) -> _Response:
+    current = dict(credential)
+    if days is not None:
+        current["expires_in_days"] = days
+        current["expiring_soon"] = days < 14
+    return _Response(
+        200,
+        {"count": 0, "credential": current},
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": str(current["expires_in_days"]),
+        },
+    )
+
+
+def _credential(token: str, expires_at: float | None = None) -> dict:
+    return {
+        "kind": "ingest_token",
+        "token": token,
+        "refresh_token": None,
+        "org_id": "org_1",
+        "host": HOST,
+        "expires_at": expires_at,
+    }
+
+
+def test_14_day_warning_requires_matching_header_and_body(monkeypatch) -> None:
+    status_body = {"count": 0, "credential": _lifetime(14)}
+    response = _Response(
+        200,
+        status_body,
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "14",
+        },
+    )
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response)
+
+    status = rotation.credential_status(HOST, OLD_TOKEN)
+
+    assert status is not None
+    assert status["expires_in_days"] == 14
+    assert status["expiring_soon"] is True
+    assert "expires in 14 days" in rotation.expiry_warning(status)
+
+
+def test_status_rejects_a_wrong_credential_role_before_network(monkeypatch) -> None:
+    monkeypatch.setattr(
+        rotation.httpx,
+        "get",
+        lambda *a, **k: pytest.fail("wrong credential role reached Cloud status"),
+    )
+    assert rotation.credential_status(HOST, "oap_" + "A" * 43) is None
+
+
+def test_day_14_accepts_server_authoritative_false_before_exact_boundary(monkeypatch) -> None:
+    lifetime = _lifetime(14)
+    lifetime["expiring_soon"] = False
+    response = _Response(
+        200,
+        {"count": 0, "credential": lifetime},
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "14",
+        },
+    )
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response)
+
+    status = rotation.credential_status(HOST, OLD_TOKEN)
+
+    assert status is not None
+    assert status["expires_in_days"] == 14
+    assert status["expiring_soon"] is False
+    assert rotation.expiry_warning(status) is None
+
+
+@pytest.mark.parametrize(
+    ("days", "expiring"),
+    [(13, False), (15, True)],
+)
+def test_warning_rejects_impossible_whole_day_state(days, expiring, monkeypatch) -> None:
+    lifetime = _lifetime(days)
+    lifetime["expiring_soon"] = expiring
+    response = _Response(
+        200,
+        {"count": 0, "credential": lifetime},
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": str(days),
+        },
+    )
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response)
+    assert rotation.credential_status(HOST, OLD_TOKEN) is None
+
+
+def test_warning_rejects_a_header_body_disagreement(monkeypatch) -> None:
+    response = _Response(
+        200,
+        {"count": 0, "credential": _lifetime(14)},
+        {
+            "x-openadapt-credential-warning-days": "14",
+            "x-openadapt-credential-expires-in-days": "13",
+        },
+    )
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response)
+    assert rotation.credential_status(HOST, OLD_TOKEN) is None
+
+
+def test_legacy_non_expiring_status_has_no_expiry_header(monkeypatch) -> None:
+    response = _Response(
+        200,
+        {
+            "count": 0,
+            "credential": {
+                "expires_at": None,
+                "expires_in_days": None,
+                "expiring_soon": False,
+                "legacy_non_expiring": True,
+                "warning_days": 14,
+            },
+        },
+        {
+            "cache-control": "no-store",
+            "x-openadapt-credential-warning-days": "14",
+        },
+    )
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response)
+    status = rotation.credential_status(HOST, OLD_TOKEN)
+    assert status is not None
+    assert "no renewal date" in rotation.expiry_warning(status)
+
+
+def test_rotation_stages_then_atomically_promotes_the_one_time_replacement(
+    monkeypatch,
+) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    requests: list[tuple[str, dict, dict]] = []
+    rotated = _rotation_response()
+
+    def _post(url, *, json, headers, **kwargs):
+        requests.append((url, json, headers))
+        return rotated
+
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", _post)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "get",
+        lambda *a, **k: _replacement_validation_response(rotated.json()["credential"]),
+    )
+
+    replacement = rotation.rotate_credential(HOST)
+
+    assert replacement["token"] == NEW_TOKEN
+    assert isinstance(replacement["expires_at"], float)
+    assert store.load_credential(HOST) == replacement
+    assert store.load_rotation_stage() is None
+    assert requests == [
+        (
+            f"{HOST}/api/ingest-tokens/rotate",
+            {},
+            {"Authorization": f"Bearer {OLD_TOKEN}"},
+        )
+    ]
+
+
+def test_immediate_replacement_verification_still_requires_a_fresh_lifetime(
+    monkeypatch,
+) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    rotated = _rotation_response()
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: rotated)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "get",
+        lambda *a, **k: _replacement_validation_response(
+            rotated.json()["credential"],
+            days=1,
+        ),
+    )
+
+    with pytest.raises(rotation.RotationError, match="incomplete verification contract"):
+        rotation.rotate_credential(HOST)
+
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+    assert store.load_rotation_stage() is not None
+
+
+@pytest.mark.parametrize("field", ["previous_id", "record_id"])
+def test_rotation_response_requires_production_uuid_ids(field) -> None:
+    response = _rotation_response()
+    body = response.json()
+    if field == "previous_id":
+        body["previous_id"] = "ingest_previous_1"
+    else:
+        body["record"]["id"] = "ingest_replacement_1"
+
+    with pytest.raises(rotation.RotationError, match="credential id|credential record"):
+        rotation._validated_rotation(body, response.headers)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("token_prefix", "oai_ingest_wrong"),
+        ("last_used_at", "2026-07-28T00:00:00+00:00"),
+        ("revoked_at", "2026-07-28T00:00:00+00:00"),
+        ("rotated_to_id", str(uuid4())),
+    ],
+)
+def test_rotation_response_rejects_a_noncurrent_record(field, value) -> None:
+    response = _rotation_response()
+    response.json()["record"][field] = value
+
+    with pytest.raises(rotation.RotationError, match="credential record"):
+        rotation._validated_rotation(response.json(), response.headers)
+
+
+def test_rotation_response_rejects_extra_record_fields() -> None:
+    response = _rotation_response()
+    response.json()["record"]["token_hash"] = "must-not-cross-the-wire"
+
+    with pytest.raises(rotation.RotationError, match="credential record"):
+        rotation._validated_rotation(response.json(), response.headers)
+
+
+def test_rotation_refuses_before_network_when_keychain_is_unavailable(monkeypatch) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: False)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "post",
+        lambda *a, **k: pytest.fail("rotation must not consume a response"),
+    )
+    with pytest.raises(rotation.RotationError, match="keychain"):
+        rotation.rotate_credential(HOST)
+
+
+def test_lost_rotation_response_preserves_old_token_and_requires_reconnect(
+    monkeypatch,
+) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "post",
+        lambda *a, **k: (_ for _ in ()).throw(httpx.ReadError("lost")),
+    )
+
+    with pytest.raises(rotation.RotationError, match="did not receive.*do not retry"):
+        rotation.rotate_credential(HOST)
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+    assert store.load_rotation_stage() is None
+
+
+def test_stage_failure_preserves_old_token_and_does_not_recommend_replay(monkeypatch) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: _rotation_response())
+    monkeypatch.setattr(rotation, "stage_rotation_credential", lambda *a, **k: False)
+
+    with pytest.raises(rotation.RotationError, match="Sign in again; do not retry"):
+        rotation.rotate_credential(HOST)
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+
+
+def test_rotation_rejects_a_one_time_response_without_no_store(monkeypatch) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    response = _rotation_response()
+    response.headers = {}
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: response)
+
+    with pytest.raises(rotation.RotationError, match="Sign in again; do not retry"):
+        rotation.rotate_credential(HOST)
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+    assert store.load_rotation_stage() is None
+
+
+def test_delayed_rotation_recovery_accepts_decreased_days_without_a_second_request(
+    monkeypatch,
+) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    response = _rotation_response().json()
+    token, previous_id, expiry = rotation._validated_rotation(
+        response,
+        {"cache-control": "no-store"},
+    )
+    replacement = _credential(token, expiry)
+    previous = store.snapshot_pairing_canonical(HOST)
+    assert previous is not None
+    assert store.stage_rotation_credential(previous_id, replacement, previous)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "post",
+        lambda *a, **k: pytest.fail("recovery must not issue another rotation"),
+    )
+    monkeypatch.setattr(
+        rotation.httpx,
+        "get",
+        lambda *a, **k: _replacement_validation_response(response["credential"], days=80),
+    )
+
+    recovered = rotation.rotate_credential(HOST)
+
+    assert recovered == replacement
+    assert store.load_credential(HOST) == replacement
+    assert store.load_rotation_stage() is None
+
+
+def test_bad_replacement_stays_staged_and_recovery_validates_without_reissuing(
+    monkeypatch,
+) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    rotated = _rotation_response()
+    post_calls = 0
+
+    def _post(*args, **kwargs):
+        nonlocal post_calls
+        post_calls += 1
+        return rotated
+
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", _post)
+    monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: _Response(401, {}))
+
+    with pytest.raises(rotation.RotationError, match="did not accept.*old credential"):
+        rotation.rotate_credential(HOST)
+
+    assert post_calls == 1
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+    assert store.load_rotation_stage() is not None
+
+    monkeypatch.setattr(
+        rotation.httpx,
+        "get",
+        lambda *a, **k: _replacement_validation_response(rotated.json()["credential"]),
+    )
+    recovered = rotation.rotate_credential(HOST)
+
+    assert post_calls == 1
+    assert recovered["token"] == NEW_TOKEN
+    assert store.load_credential(HOST) == recovered
+    assert store.load_rotation_stage() is None
+
+
+def test_later_login_supersedes_rejected_stage_before_future_rotation(monkeypatch) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    rejected = _rotation_response()
+    replacement = _rotation_response(
+        token=FOURTH_TOKEN,
+        previous_id=str(uuid4()),
+        replacement_id=str(uuid4()),
+    )
+    posted_bearers: list[str] = []
+    validated_bearers: list[str] = []
+
+    def _post(url, *, headers, **kwargs):
+        posted_bearers.append(headers["Authorization"])
+        return rejected if len(posted_bearers) == 1 else replacement
+
+    def _get(url, *, headers, **kwargs):
+        bearer = headers["Authorization"]
+        validated_bearers.append(bearer)
+        if bearer == f"Bearer {NEW_TOKEN}":
+            return _Response(401, {})
+        assert bearer == f"Bearer {FOURTH_TOKEN}"
+        return _replacement_validation_response(replacement.json()["credential"])
+
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", _post)
+    monkeypatch.setattr(rotation.httpx, "get", _get)
+
+    with pytest.raises(rotation.RotationError, match="did not accept"):
+        rotation.rotate_credential(HOST)
+    assert store.load_rotation_stage() is not None
+    assert store.load_credential(HOST)["token"] == OLD_TOKEN
+
+    # A later browser/deep-link login writes a coherent third credential.
+    store.store_credential(_credential(THIRD_TOKEN))
+
+    rotated = rotation.rotate_credential(HOST)
+
+    assert posted_bearers == [
+        f"Bearer {OLD_TOKEN}",
+        f"Bearer {THIRD_TOKEN}",
+    ]
+    assert validated_bearers == [
+        f"Bearer {NEW_TOKEN}",
+        f"Bearer {FOURTH_TOKEN}",
+    ]
+    assert rotated["token"] == FOURTH_TOKEN
+    assert store.load_credential(HOST) == rotated
+    assert store.load_rotation_stage() is None
+
+
+def test_second_rotation_409_requires_a_new_login(monkeypatch) -> None:
+    store.store_credential(_credential(OLD_TOKEN))
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: _Response(409, {}))
+    with pytest.raises(rotation.RotationError, match="already renewed.*login"):
+        rotation.rotate_credential(HOST)
+
+
+@pytest.mark.parametrize(
+    "wrong_role",
+    [
+        "supabase_authorization_code",
+        "oap_" + "A" * 43,
+        "oapp_" + "A" * 43,
+        "oaps_" + "A" * 43,
+    ],
+)
+def test_rotation_rejects_every_non_ingest_credential_before_network(
+    wrong_role, monkeypatch
+) -> None:
+    store.store_credential(_credential(wrong_role))
+    monkeypatch.setattr(rotation, "secure_store_available", lambda: True)
+    monkeypatch.setattr(
+        rotation.httpx,
+        "post",
+        lambda *a, **k: pytest.fail("wrong credential role reached Cloud rotation"),
+    )
+    with pytest.raises(rotation.RotationError, match="no stored connection"):
+        rotation.rotate_credential(HOST)
diff --git a/tests/test_engine/test_cli.py b/tests/test_engine/test_cli.py
index 8d4a545..5d86dae 100644
--- a/tests/test_engine/test_cli.py
+++ b/tests/test_engine/test_cli.py
@@ -131,6 +131,45 @@ def test_login_success(self, cli_config: EngineConfig, capsys) -> None:
         assert "Logged in" in captured.out
         assert "org_5" in captured.out
 
+    def test_credential_command_surfaces_the_14_day_warning(
+        self, cli_config: EngineConfig, capsys
+    ) -> None:
+        status = {
+            "expires_at": "2026-08-10T00:00:00+00:00",
+            "expires_at_timestamp": 1786320000.0,
+            "expires_in_days": 14,
+            "expiring_soon": True,
+            "legacy_non_expiring": False,
+            "warning_days": 14,
+        }
+        with (
+            patch("engine.cli.EngineConfig", return_value=cli_config),
+            patch("engine.auth.rotation.credential_status", return_value=status),
+        ):
+            main(["credential"])
+        assert "expires in 14 days" in capsys.readouterr().out
+
+    def test_rotate_command_reports_the_overlap_without_printing_the_token(
+        self, cli_config: EngineConfig, capsys
+    ) -> None:
+        token = "oai_ingest_" + "S" * 43
+        credential = {
+            "kind": "ingest_token",
+            "token": token,
+            "refresh_token": None,
+            "org_id": "org_5",
+            "host": "https://app.openadapt.ai",
+            "expires_at": 1793016000.0,
+        }
+        with (
+            patch("engine.cli.EngineConfig", return_value=cli_config),
+            patch("engine.auth.rotation.rotate_credential", return_value=credential),
+        ):
+            main(["rotate"])
+        output = capsys.readouterr().out
+        assert "old credential remains valid for at most seven days" in output
+        assert token not in output
+
     def test_push_success(self, cli_config: EngineConfig, capsys) -> None:
         """push should print the returned workflow id + dashboard URL."""
         result = {"success": True, "workflow_id": "wf_2",