feat(auth): hand the free-tier guest to the account after sign-in (#143234)

* feat(auth): hand the free tier's apps to the account that replaces it

When a Nous account is saved over a free-tier identity, Hermes now moves
that identity's connected apps into the account with one call to the Nous
Portal claim endpoint, authorized by the account's own token. Nothing asks
first; the result (account email, what moved, what needs reconnecting) is
shown afterwards in `hermes auth status nous` and exposed for the desktop.

Sign-in code does not know about it. A hook at the two lowest write seams
(`_store_provider_state`, `_write_shared_nous_state`) writes a pending
record under the existing locks and does no network. The claim runs
outside every lock: after a shared-store write, when a `deferred()` block
exits (the desktop poller and `run_sign_in` save under their own lock),
after a free-tier reconcile, and on each Nous keepalive tick, which
resumes retries from an earlier process.

Records sit beside the shared Nous store, keyed by a hash of the free-tier
credential and bound to the destination account, so profiles sharing one
free-tier identity share one record and one claim. A claim lock spans
read, claim and write. A Claimed record is never reopened; a lost success
followed by an unknown-credential answer reads as "probably moved". The
account token is refreshed after the move only when the cohort carried.
The same path runs `settle_after_upgrade` once per replacement, so every
replacing writer settles the config.

The legacy promotion sign-in already moves the apps on the server, so it
records the handoff as done and sends no claim.

* fix(auth): the free-tier handoff only ever claims into its bound account

Review round 1 on the guest handoff.

P1, Bearer not proven to be the destination: the runner checked the
profile's account against the record's destination, then resolved a
Bearer separately. A sign-in landing between the two could move the apps
into that other account and mark the record Claimed. The Bearer is now
resolved first and its JWT subject is compared to the record's
destination inside the reservation, so only a token proven to belong to
the destination is ever sent.

P1, stale reservation: the destination check and the attempt increment
were separate record reads, so a sibling could rebind the destination in
between. `_reserve` now does one locked read-modify-write: still active,
due (`next_attempt_at`, so a waiting runner honours a Retry-After another
runner saved), under `MAX_ATTEMPTS` (a crash after the last send now
closes the record as Failed instead of sending again), and bound to the
Bearer's subject. A record without a destination subject is never written.

P2, config completion in the runner: `drain` no longer runs
`settle_after_upgrade`. Settling stays with the sign-in that saved the
account (run_sign_in, the desktop poller), as before this PR. This
deletes the thread-local queue and the batch's settle fallback.

P2, keepalive: a handoff failure is logged and token renewal carries on.
An unparseable records file is moved aside to .corrupt on the next write
instead of blocking every Nous account save.

P2, budget: the remaining sync budget is checked before each record,
token resolve and request, and is passed as their timeout, so a refusal
at the edge starts no refresh or second claim.

P2, status: an active move bound to an account the profile no longer
holds reads as paused, with copy that says so.

* fix(auth): free-tier handoff retries reach every profile and every long-lived process

Review round 2 on the guest handoff.

P1, queued moves had no worker: retries rode the token keepalive, which
does not start in a process that boots with a guest (a Desktop backend
before sign-in). A move left by a CLI sign-in during an outage then
waited for a restart. Retries now run on a dedicated worker that
`start_nous_auth_keepalive` starts in every long-lived Nous process,
guest or not; the keepalive tick no longer drains.

P1, named profiles never retried: the keepalive drained only the launch
profile's records. Each worker tick now runs every due record under its
owning profile's home and secret scope (the way the cron ticker binds a
profile), so a named profile's move retries from a process launched as
another profile. The bound-account check and the Bearer proof in the
reservation still decide whether anything is sent.

P2, cached guest token: the resolver's 5 s memo could still hold the
guest's token from a connector call just before the save; the
reservation then refused its subject and the move waited for a later
run. The handoff now drops that memo before reading the account token.

P2, budget through locks: the token read waited the resolver's own
15 s floors for the auth and shared locks. The handoff now takes both
locks itself within the remaining budget (same order), which makes the
resolver's acquisitions reentrant, and re-checks the budget before
reserving or sending.

* fix(auth): free-tier handoff never gives up on a transient failure before the reap

Compatibility audit, items B1, B4 and B5.

B1: a handoff that reached Failed was never retried, so the free tier's
apps were lost when the idle free-tier identity was reaped after 14
days. Transient failures (5xx, the 503 breaker with no Retry-After,
timeouts, busy, unknown codes) became final after 8 attempts, about 81
minutes. Now only codes no later attempt can change are final
(client_not_allowed, account_locked, invalid_request). Every other
outcome retries, the slowest rung (1 h) repeating, until 13 days after
the record was made. The attempt reservation then closes it as Failed
`retry_window_ended`. One predicate (`_retryable`) picks the work for
every run, so a record an older build failed on a transient code re-arms
on the next sign-in, `hermes auth status nous`, `hermes auth upgrade`, or
retry-worker tick, with no new state. The attempt cap is deleted.

B4: failure copy printed raw server codes. Final codes map to a plain
sentence, and anything else shows only the generic sentence.

B5: a claim answer without a `connectors` field (an older service) read
as "There were no free-tier apps to move", and so did Gone on a first
attempt, which follows a released ticket that already moved the apps.
The first now reads as unknown with neutral copy, and the second as
"Nothing left to move from the free tier."

* fix(auth): a free tier with no apps reads as nothing moved after sign-in

A free tier that only read its connectors gets a vendor id row on the
service, so the claim answers kind `moved` (or `swapped`) with an empty
`moved_toolkits`. The status copy keyed on the kind and printed "Your
apps moved to <email>." for an account that received nothing.

One rule: an empty moved list reads as nothing moved, for moved and
swapped alike. The service now refuses a move it could not read, so an
empty list is the truth. The promotion sign-in's record, which reports
no list at all, now carries kind `unknown` and keeps its neutral copy
instead of falling under the rule.

hermes_cli/anon_auth.py

@@ -708,6 +708,9 @@ def ensure_portal_identity(
         raise err from exc
     if state is not None:
         _clear_mint_failure()
+        # Adopting a sibling profile's sign-in replaced this profile's guest under the locks above.
+        from hermes_cli.guest_handoff import drain
+        drain()
     return state
 
 

hermes_cli/anon_sign_in.py

@@ -440,10 +440,15 @@ def run_sign_in(
             yield Superseded()
             return
         guard = persist_guard or _default_persist_guard(post_promotion_cancelled)
+        from hermes_cli.guest_handoff import deferred, note_promoted
         with open_scope():
-            with guard() as may_persist:
-                if may_persist:
-                    _core.persist_nous_credentials(account_state)
+            # The handoff runs once the surface's lock is released, never inside it.
+            with deferred():
+                with guard() as may_persist:
+                    if may_persist:
+                        # The completed promotion already moved the apps: nothing left to claim.
+                        note_promoted(anon_token, account_email=str(outcome.get("account_email") or "") or None)
+                        _core.persist_nous_credentials(account_state)
             if may_persist:
                 settled = _core.settle_after_upgrade(account_state)
     except Exception as exc:

hermes_cli/anon_sign_in_cli.py

@@ -76,6 +76,8 @@ def upgrade_guest(args) -> int:
     """
     from hermes_cli import anon_auth as _core
     from hermes_cli.auth_device_flow import _is_remote_session
+    from hermes_cli.guest_handoff import drain
+    drain()   # a move an earlier sign-in left unfinished gets its next try first
     timeout_seconds = float(getattr(args, "timeout", None) or 15.0)
     open_browser = not getattr(args, "no_browser", False) and not _is_remote_session()
     return _core.render_sign_in_cli(

hermes_cli/auth.py

@@ -82,6 +82,7 @@ from hermes_cli.auth_codex import (
     _refresh_expired_codex_probe_token, _save_codex_tokens, clear_codex_pool_quota_cooldowns,
     refresh_codex_oauth_pure, resolve_codex_runtime_credentials)
 from hermes_cli.auth_openrouter import _openrouter_pkce_login
+from hermes_cli.guest_handoff import note_store_write
 from hermes_cli.auth_qwen import (
     _qwen_access_token_is_expiring, _qwen_cli_auth_path, _read_qwen_cli_tokens,
     _refresh_qwen_cli_tokens, _save_qwen_cli_tokens, get_qwen_auth_status,
@@ -861,9 +862,8 @@ def _provider_state_transaction(
             yield auth_store, _provider_state_in(_load_auth_store(source_path), provider_id), source_path
 
 
-def _store_provider_state(
-    auth_store: dict[str, Any], provider_id: str, state: dict[str, Any], *, set_active: bool = True,
-) -> None:
+def _store_provider_state(auth_store: dict[str, Any], provider_id: str, state: dict[str, Any], *, set_active: bool = True) -> None:
+    note_store_write(auth_store, provider_id, state)
     _store_section(auth_store, "providers")[provider_id] = state
     if set_active:
         auth_store["active_provider"] = provider_id

hermes_cli/auth_commands.py

@@ -702,6 +702,11 @@ def auth_status_command(args) -> None:
         value = status.get(key)
         if value:
             print(f"  {key}: {value}")
+    if provider == "nous":
+        from hermes_cli.guest_handoff import drain, handoff_status, status_lines
+        drain()   # a move left retrying (or failed transiently by an older build) gets its next try now
+        for line in status_lines(handoff_status()):
+            print(f"  {line}")
 
 
 def auth_logout_command(args) -> None:

hermes_cli/auth_nous.py

@@ -25,6 +25,7 @@ from hermes_cli.auth_constants import (
     DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS, NOUS_AUTH_PATH_INVOKE_JWT, NOUS_BILLING_MANAGE_SCOPE,
     NOUS_DEVICE_CODE_SOURCE, NOUS_INFERENCE_INVOKE_SCOPE, NOUS_INVOKE_JWT_MIN_TTL_SECONDS,
     WELCOME_CLAIM_FLAG, _nous_err, httpx)
+from hermes_cli.guest_handoff import drain, note_replacement
 
 if TYPE_CHECKING:  # annotation-only; the runtime import would be a cycle
     from hermes_cli.auth import ProviderConfig
@@ -466,12 +467,15 @@ def _write_shared_nous_state(state: dict[str, Any]) -> None:
     try:
         with _nous_shared_store_lock():
             path = _nous_shared_store_path()
+            note_replacement(_read_shared_nous_state(), state)
             _save_private_json(path, shared, sort_keys=True)
         _oauth_trace(
             "nous_shared_store_written", path=str(path),
             refresh_token_fp=_token_fingerprint(refresh_token))
     except Exception as exc:
         logger.debug("Failed to write shared Nous auth store: %s", exc)
+    # The account is saved everywhere now: hand the replaced guest over (a no-op under a lock).
+    drain()
 
 
 def _read_shared_nous_state() -> Optional[dict[str, Any]]:

hermes_cli/guest_handoff.py

@@ -0,0 +1,624 @@
+"""Guest handoff: after a Nous account replaces the free tier, move the free tier's apps into it.
+
+The account service moves a free-tier identity's connected apps (and its preview cohort) into a
+signed-in account with one call, ``POST /api/anonymous/claim {"token": "<anon_...>"}``, authorized by
+the account's own ``hermes-cli`` Bearer token. Hermes never asks first: the move just happens after
+the account is saved, and the result (the account email, what moved, what needs reconnecting) is
+shown afterwards. Sign-in code knows nothing about this; two halves keep it that way:
+
+- :func:`note_replacement` runs at the lowest write seams (``auth._store_provider_state`` and
+  ``auth_nous._write_shared_nous_state``), under the caller's auth-store locks, whenever a non-guest
+  Nous state replaces a guest. It writes a pending record and does no network. Records sit beside
+  the shared Nous store, keyed by a hash of the ``anon_`` credential and bound to the destination
+  account, so profiles that share one free-tier identity share one record and one claim.
+- :func:`drain` runs the claim outside every lock: when a shared-store write finishes, when a
+  :func:`deferred` block exits, after a free-tier reconcile, and on each Nous keepalive tick (a
+  retry left by an earlier process). The synchronous run is bounded (``SYNC_BUDGET_SECONDS``) so a
+  CLI or the desktop sees a result; what is still retrying resumes on a later run.
+
+The apps only ever go to the record's destination: every attempt is reserved in one locked
+read-modify-write that also proves the Bearer about to be sent belongs to that destination.
+
+User copy never says guest / anonymous / claim.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import threading
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Iterator, Optional
+
+logger = logging.getLogger("hermes_cli.auth")
+
+IDLE, PENDING, RETRYING, CLAIMED, GONE, FAILED = "idle", "pending", "retrying", "claimed", "gone", "failed"
+# Read side only: an active record bound to an account this profile no longer holds.
+PAUSED = "paused"
+_ACTIVE = frozenset({PENDING, RETRYING})
+HANDOFF_FILENAME = "guest_handoff.json"
+SYNC_BUDGET_SECONDS = 10.0
+CLAIM_TIMEOUT_SECONDS = 10.0
+# The service reaps an idle free-tier identity after 14 days, apps included. Transient failures keep
+# retrying (the slowest rung repeats) until this long after the record was made, safely before that.
+RETRY_WINDOW_SECONDS = 13 * 24 * 3600
+_BACKOFF_SECONDS = (2.0, 5.0, 15.0, 60.0, 300.0, 900.0, 3600.0)
+# ``_is_expiring`` treats any token as expiring inside this skew, so the account token refreshes now.
+_FORCE_REFRESH_SKEW = 10**9
+
+# Outcome of one claim response.
+_CLAIMED, _GONE, _RETRY, _FAIL, _REAUTH = "claimed", "gone", "retry", "fail", "reauth"
+# Codes no later attempt can change. Every other refusal, unknown codes included, retries: a wrong
+# "final" loses the apps for good, a wrong retry costs one request an hour.
+_FINAL_CODES = frozenset({"client_not_allowed", "account_locked", "invalid_request", "retry_window_ended"})
+# A plain reason per final code; any other code shows only the generic sentence, never the code.
+_FAILURE_COPY = {
+    "client_not_allowed": "This sign-in can't receive them. Sign in again with `hermes auth add nous`.",
+    "account_locked": "This Nous account is locked.",
+    "retry_window_ended": "Hermes kept trying for nearly two weeks, then stopped.",
+}
+# The service's refusal codes; anything unlisted retries (``_classify``).
+_REFUSALS = {
+    "unknown_token": _GONE, "already_retired": _GONE,
+    **{code: _FAIL for code in _FINAL_CODES},
+}
+
+_records_lock_holder = threading.local()
+_claim_lock_holder = threading.local()
+_local = threading.local()
+_sleep = time.sleep
+
+
+@dataclass
+class HandoffBatch:
+    """What one :func:`drain` did: the public view of each record it ran."""
+
+    results: list[dict[str, Any]] = field(default_factory=list)
+
+    @property
+    def account_email(self) -> Optional[str]:
+        return next((r["account_email"] for r in self.results if r.get("account_email")), None)
+
+
+def _now() -> float:
+    return time.time()
+
+
+def _iso(epoch: float) -> str:
+    return datetime.fromtimestamp(epoch, tz=UTC).isoformat()
+
+
+def _token_key(anon_token: str) -> str:
+    return hashlib.sha256(anon_token.encode("utf-8")).hexdigest()[:32]
+
+
+def _home_key() -> str:
+    from hermes_constants import hermes_home_key
+    return hermes_home_key()
+
+
+def _records_path() -> Path:
+    from hermes_cli.auth_nous import _nous_shared_store_path
+    return _nous_shared_store_path().with_name(HANDOFF_FILENAME)
+
+
+@contextmanager
+def _lock(path: Path, holder: threading.local, timeout_seconds: float) -> Iterator[None]:
+    from hermes_cli.auth import _file_lock
+    with _file_lock(path, holder, timeout_seconds, f"Timed out waiting for {path.name}"):
+        yield
+
+
+def _load(path: Path, *, heal: bool = False) -> dict[str, Any]:
+    """The records file. An unparseable one reads as empty; *heal* (writers, under the records lock)
+    moves it aside to ``.corrupt`` first, so a bad file neither blocks a sign-in nor loses its data."""
+    try:
+        data = json.loads(path.read_text(encoding="utf-8-sig"))
+    except FileNotFoundError:
+        data = {}
+    except ValueError:
+        logger.warning("Free-tier handoff records at %s are unreadable%s", path,
+                       "; moved aside to .corrupt" if heal else "")
+        if heal:
+            path.replace(path.with_suffix(".json.corrupt"))
+        data = {}
+    records = data.get("records") if isinstance(data, dict) else None
+    return {"_schema": 1, "records": records if isinstance(records, dict) else {}}
+
+
+def _save(path: Path, data: dict[str, Any]) -> None:
+    from hermes_cli.auth import _save_private_json
+    _save_private_json(path, data, sort_keys=True)
+
+
+def _update(path: Path, key: str, change) -> Optional[dict[str, Any]]:
+    """Apply *change(record) -> record | None* to one record under the records lock and return what
+    was written (None: nothing written). The innermost lock: the hook takes it under the auth-store
+    locks, and it never holds another lock itself."""
+    with _lock(path.with_suffix(".lock"), _records_lock_holder, 15.0):
+        data = _load(path, heal=True)
+        record = change(data["records"].get(key))
+        if record is not None:
+            record["updated_at"] = _iso(_now())
+            data["records"][key] = record
+            _save(path, data)
+        return record
+
+
+def _subject(access_token: Any) -> Optional[str]:
+    from hermes_cli.auth_constants import _decode_jwt_claims
+    sub = _decode_jwt_claims(access_token).get("sub")
+    return str(sub) if sub else None
+
+
+# --- The hook: no network, runs under the writer's locks ------------------------------------------
+
+def note_store_write(auth_store: dict[str, Any], provider_id: str, state: Any) -> None:
+    """``auth._store_provider_state``'s hook: the Nous state the store holds now is the one replaced
+    (the profile's own, else the root store's it falls back to)."""
+    if provider_id == "nous":
+        from hermes_cli.auth import _load_provider_state
+        note_replacement(_load_provider_state(auth_store, "nous"), state)
+
+
+def note_replacement(previous: Any, new: Any) -> None:
+    """Record that a non-guest Nous state replaced the guest *previous*. Called from the lowest write
+    seams; idempotent per ``anon_`` credential. A Claimed or Gone record is never reopened; a record
+    not yet attempted follows the newest destination, one already attempted keeps its own (a lost
+    answer may have moved the apps there)."""
+    from hermes_cli.anon_auth import is_guest_state
+    if not is_guest_state(previous) or is_guest_state(new) or not isinstance(new, dict):
+        return
+    anon_token, destination = previous.get("anon_token"), _subject(new.get("access_token"))
+    if not isinstance(anon_token, str) or not anon_token or not destination:
+        return  # no account subject: nothing a claim could ever be proven to land in
+    try:
+        path = _records_path()
+    except RuntimeError:
+        return  # no Hermes home yet (pre-setup): there is no shared store either
+    home = _home_key()
+
+    def change(record: Optional[dict[str, Any]]) -> dict[str, Any]:
+        if record is None:
+            record = {"state": PENDING, "anon_token": anon_token, "destination": destination,
+                      "attempts": 0, "profiles": [], "created_at": _iso(_now())}
+        elif record["state"] in _ACTIVE and not record.get("attempts"):
+            record["destination"] = destination
+        if home not in record["profiles"]:
+            record["profiles"].append(home)
+        return record
+
+    record = _update(path, _token_key(anon_token), change)
+    logger.info("Nous sign-in replaced the free tier; handoff %s (%s)", record["state"],
+                _token_key(anon_token)[:8])
+
+
+def note_promoted(anon_token: str, *, account_email: Optional[str]) -> None:
+    """The promotion sign-in (``anon_sign_in.run_sign_in``) already moved this identity's apps on the
+    service: record the handoff as done so the save that follows sends no claim."""
+    if not anon_token:
+        return
+    home = _home_key()
+
+    def change(record: Optional[dict[str, Any]]) -> dict[str, Any]:
+        record = record or {"attempts": 0, "profiles": [], "created_at": _iso(_now())}
+        record.pop("anon_token", None)
+        record.update(state=CLAIMED, error=None, result={
+            # The promotion reports no list of what moved: neutral copy, not "nothing moved".
+            "account_email": account_email, "connectors_kind": "unknown", "moved_toolkits": [],
+            "dropped_toolkits": []})
+        if home not in record["profiles"]:
+            record["profiles"].append(home)
+        return record
+
+    _update(_records_path(), _token_key(anon_token), change)
+
+
+# --- The runner: outside every lock ---------------------------------------------------------------
+
+def _holds_auth_locks() -> bool:
+    from hermes_cli import auth, auth_nous
+    if getattr(auth_nous._nous_shared_lock_holder, "depth", 0):
+        return True
+    with auth._auth_target_lock_holders_guard:
+        holders = list(auth._auth_target_lock_holders.values())
+    return any(getattr(holder, "depth", 0) for holder in holders)
+
+
+@contextmanager
+def deferred() -> Iterator[HandoffBatch]:
+    """Hold every drain on this thread until the block exits, then drain once (outside the caller's
+    lock): a surface that saves a sign-in under its own lock wraps the save in this."""
+    batch = HandoffBatch()
+    _local.holds = getattr(_local, "holds", 0) + 1
+    try:
+        yield batch
+    finally:
+        _local.holds -= 1
+    ran = drain()
+    if ran is not None:
+        batch.results = ran.results
+
+
+def drain() -> Optional[HandoffBatch]:
+    """Run this profile's due records now. None when a lock or a :func:`deferred` block on this
+    thread holds it back."""
+    if getattr(_local, "holds", 0) or getattr(_local, "running", False) or _holds_auth_locks():
+        return None
+    _local.running = True
+    try:
+        return HandoffBatch(results=run_due(budget_seconds=SYNC_BUDGET_SECONDS))
+    finally:
+        _local.running = False
+
+
+def _remaining(deadline: float) -> float:
+    return deadline - time.monotonic()
+
+
+def _retryable(record: dict[str, Any]) -> bool:
+    """A record a run may still attempt: active, or Failed on a code that is not final (an older
+    build closed transient failures after 8 attempts; those re-arm here, on any run: a sign-in,
+    ``hermes auth status``, ``hermes auth upgrade``, the retry worker). The retry window is checked
+    when the attempt is reserved, so a record past it is closed, not silently skipped."""
+    state = record.get("state")
+    return state in _ACTIVE or (state == FAILED and record.get("error") not in _FINAL_CODES
+                                and bool(record.get("anon_token")))
+
+
+def _due(record: dict[str, Any], now: float) -> bool:
+    return _retryable(record) and float(record.get("next_attempt_at") or 0) <= now
+
+
+def _window_ended(record: dict[str, Any], now: float) -> bool:
+    try:
+        created = datetime.fromisoformat(str(record.get("created_at"))).timestamp()
+    except ValueError:
+        return True
+    return now >= created + RETRY_WINDOW_SECONDS
+
+
+# --- The worker: retries in a long-lived process, for every profile -------------------------------
+#
+# A move queued by a short process (a CLI sign-in during an outage) is finished by whichever long-lived
+# Nous process runs next (Desktop backend, gateway, CLI agent), whatever its own profile or identity.
+_WORKER_FIRST_TICK_SECONDS = 15.0
+_WORKER_INTERVAL_SECONDS = 60.0
+_worker_lock = threading.Lock()
+_worker_stop = threading.Event()
+_worker_thread: Optional[threading.Thread] = None
+
+
+def start_worker() -> threading.Thread:
+    """Start the process-wide retry worker (idempotent)."""
+    global _worker_thread
+    with _worker_lock:
+        if _worker_thread is not None and _worker_thread.is_alive():
+            return _worker_thread
+        _worker_stop.clear()
+        # health: allow HX012 -- process-wide worker that binds each record's own profile scope per run; it must not inherit the starter's
+        _worker_thread = threading.Thread(target=_worker_loop, daemon=True, name="nous-guest-handoff")
+        _worker_thread.start()
+        return _worker_thread
+
+
+def stop_worker(timeout: float = 5.0) -> None:
+    global _worker_thread
+    with _worker_lock:
+        thread = _worker_thread
+        _worker_stop.set()
+    if thread is not None and thread.is_alive():
+        thread.join(timeout=timeout)
+    with _worker_lock:
+        if _worker_thread is thread:
+            _worker_thread = None
+
+
+def _worker_loop() -> None:
+    wait = _WORKER_FIRST_TICK_SECONDS
+    while not _worker_stop.wait(wait):
+        try:
+            retry_all_profiles()
+        except Exception:
+            logger.warning("Free-tier handoff retry tick failed; the records stay for the next one", exc_info=True)
+        wait = _WORKER_INTERVAL_SECONDS
+
+
+def retry_all_profiles() -> list[dict[str, Any]]:
+    """Run every due record under its owning profile's scope (home + secrets), so a named profile's
+    move retries from a process launched as another profile. The bound-account check and the Bearer
+    proof in ``_reserve`` still decide whether anything is sent."""
+    try:
+        records = _load(_records_path())["records"].values()
+    except RuntimeError:
+        return []
+    now = _now()
+    homes = sorted({home for record in records if _due(record, now) for home in record.get("profiles", ())})
+    results: list[dict[str, Any]] = []
+    for home in homes:
+        if not Path(home).is_dir():
+            continue  # a deleted profile: nothing there holds the destination account
+        with _profile_scope(home):
+            batch = drain()
+        results.extend(batch.results if batch else [])
+    return results
+
+
+@contextmanager
+def _profile_scope(home: str) -> Iterator[None]:
+    """The owning profile's home and secrets for one drain (the cron ticker binds a profile the same
+    way); no terminal scope: nothing here runs a command."""
+    from agent.secret_scope import build_profile_secret_scope, reset_secret_scope, set_secret_scope
+    from hermes_constants import get_routing_process_hermes_home, hermes_home_key
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+    foreign = hermes_home_key(get_routing_process_hermes_home()) != home
+    home_token = set_hermes_home_override(home)
+    secret_token = set_secret_scope(build_profile_secret_scope(Path(home)), profile_home=home if foreign else None)
+    try:
+        yield
+    finally:
+        reset_secret_scope(secret_token)
+        reset_hermes_home_override(home_token)
+
+
+def run_due(*, budget_seconds: float = SYNC_BUDGET_SECONDS) -> list[dict[str, Any]]:
+    """Attempt every active, due record of this profile within *budget_seconds*. Returns their public
+    views; a record the budget did not reach stays as it is for a later run."""
+    try:
+        path = _records_path()
+    except RuntimeError:
+        return []
+    home, now = _home_key(), _now()
+    due = [key for key, record in _load(path)["records"].items()
+           if home in record.get("profiles", ()) and _due(record, now)]
+    deadline = time.monotonic() + budget_seconds
+    results = []
+    for key in due:
+        if _remaining(deadline) <= 0:
+            break
+        if (view := _run_one(path, key, deadline)) is not None:
+            results.append(view)
+    return results
+
+
+def _account_bound_to(record: dict[str, Any]) -> Optional[dict[str, Any]]:
+    """The profile's account when it is the record's destination; None while the profile still holds
+    a guest (the save never landed) or holds another account (retries bind to the destination)."""
+    from hermes_cli.anon_auth import is_guest_state
+    from hermes_cli.auth import get_provider_auth_state
+    account = get_provider_auth_state("nous")
+    if not isinstance(account, dict) or is_guest_state(account) or not account.get("access_token"):
+        return None
+    destination = record.get("destination")
+    return account if destination and _subject(account["access_token"]) == destination else None
+
+
+def _run_one(path: Path, key: str, deadline: float) -> Optional[dict[str, Any]]:
+    record = _load(path)["records"].get(key)
+    if record is None or _account_bound_to(record) is None:
+        return None  # a cheap skip; the binding that counts is proven in ``_reserve``
+    # The claim lock spans read -> claim -> write so two processes never claim one record twice.
+    # Order: claim lock, then the auth-store locks (token refresh), then the records lock. A lock
+    # timeout anywhere in here is contention, not a verdict: the record stays as it is for a later run.
+    try:
+        with _lock(path.with_name(HANDOFF_FILENAME + ".claim.lock"), _claim_lock_holder,
+                   max(0.1, _remaining(deadline))):
+            return _claim_locked(path, key, deadline)
+    except TimeoutError:
+        return None
+
+
+def _reserve(path: Path, key: str, bearer_subject: Optional[str]) -> Optional[tuple[dict[str, Any], int]]:
+    """Reserve one attempt in a single locked read-modify-write: the record is still retryable and
+    due, inside its retry window, and bound to the very account whose Bearer is about to be sent.
+    Returns ``(record, attempts before this one)``, or None when nothing may be sent now. A record
+    past its window is closed as Failed instead; one an older build failed is re-armed."""
+    reserved: list[int] = []
+
+    def change(record: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]:
+        now = _now()
+        if (record is None or not _due(record, now)
+                or not bearer_subject or record.get("destination") != bearer_subject):
+            return None
+        if _window_ended(record, now):
+            return {**record, "state": FAILED, "error": "retry_window_ended"}
+        prior = int(record.get("attempts") or 0)
+        reserved.append(prior)
+        # Counted before the request leaves: an answer lost to a crash still counts as an attempt.
+        return {**record, "state": RETRYING if record["state"] == FAILED else record["state"],
+                "attempts": prior + 1, "last_attempt_at": _iso(now)}
+
+    record = _update(path, key, change)
+    return (record, reserved[0]) if reserved else None
+
+
+def _claim_locked(path: Path, key: str, deadline: float) -> Optional[dict[str, Any]]:
+    from hermes_cli.auth import get_provider_auth_state
+    from hermes_cli.auth_constants import AuthError
+    force = False
+    while True:
+        current = _load(path)["records"].get(key)
+        if current is None or _remaining(deadline) <= 0:
+            return public_view(current) if current else None
+        try:
+            bearer = _bearer(force=force, timeout=_remaining(deadline))
+        except AuthError as exc:
+            logger.info("Free-tier handoff (%s): account token unavailable (%s)", key[:8], exc.code)
+            return public_view(current)
+        if _remaining(deadline) <= 0:
+            return public_view(current)
+        reservation = _reserve(path, key, _subject(bearer))
+        if reservation is None:
+            return public_view(_load(path)["records"].get(key) or current)
+        record, prior = reservation
+        outcome, payload, retry_after = _post(
+            record["anon_token"], bearer, get_provider_auth_state("nous") or {},
+            timeout=min(CLAIM_TIMEOUT_SECONDS, max(0.1, _remaining(deadline))))
+        logger.info("Free-tier handoff attempt %d (%s): %s", prior + 1, key[:8], payload.get("error") or outcome)
+        if outcome == _REAUTH and not force:
+            # A refused bearer never reached the claim, so it is not an attempt.
+            _update(path, key, lambda r: {**r, "attempts": prior} if r.get("attempts") == prior + 1 else None)
+            force = True
+            continue
+        record = _update(path, key, lambda r: _apply(r, outcome, payload, retry_after, prior))
+        if outcome == _CLAIMED and payload.get("cohort_carried") is True:
+            _refresh_account_token(timeout=max(1.0, _remaining(deadline)))
+        wait = max(0.0, float(record.get("next_attempt_at") or 0) - _now())
+        if record["state"] != RETRYING or wait >= _remaining(deadline):
+            return public_view(record)
+        _sleep(wait)
+
+
+def _bearer(*, force: bool, timeout: float) -> str:
+    from hermes_cli import auth
+    from hermes_cli.auth_nous import _nous_shared_store_lock
+    # The resolver's 5 s memo can still hold the guest's token from just before the save.
+    with auth._RESOLVE_TOKEN_CACHE_LOCK:
+        auth._RESOLVE_TOKEN_CACHE.pop(_home_key(), None)
+    # The resolver waits its own, longer floors for these locks. Taking them first within the
+    # budget (same order: auth, then shared) makes its acquisitions reentrant, so a held lock costs
+    # the budget, not 15 s; a timeout leaves the record for a later run.
+    with auth._auth_store_lock(timeout), _nous_shared_store_lock(timeout_seconds=timeout):
+        if force:
+            return auth.resolve_nous_access_token(timeout_seconds=timeout, refresh_skew_seconds=_FORCE_REFRESH_SKEW)
+        return auth.resolve_nous_access_token(timeout_seconds=timeout)
+
+
+def _refresh_account_token(*, timeout: float) -> None:
+    """The account's token picks up the carried cohort only on its next mint, so mint it now."""
+    from hermes_cli.auth_constants import AuthError
+    try:
+        _bearer(force=True, timeout=timeout)
+    except AuthError as exc:
+        logger.info("Free-tier handoff: account token refresh after the move failed: %s", exc)
+
+
+def _post(anon_token: str, bearer: str, account: dict[str, Any], *,
+          timeout: float) -> tuple[str, dict[str, Any], Optional[float]]:
+    from agent.retry_utils import parse_retry_after_seconds
+    from hermes_cli.anon_auth import _anon_headers
+    from hermes_cli.auth import _nous_portal_base_url, _resolve_verify
+    from hermes_cli.auth_constants import httpx
+    from hermes_cli.auth_nous import _nous_http_client
+    verify = _resolve_verify(insecure=None, ca_bundle=None, auth_state=account)
+    try:
+        with _nous_http_client(timeout, verify) as client:
+            response = client.post(
+                f"{_nous_portal_base_url(account).rstrip('/')}/api/anonymous/claim",
+                headers={**_anon_headers(), "authorization": f"Bearer {bearer}"}, json={"token": anon_token})
+    except httpx.HTTPError as exc:
+        return _RETRY, {"error": "unreachable", "detail": type(exc).__name__}, None
+    try:
+        payload = response.json()
+    except ValueError:
+        payload = {}
+    payload = payload if isinstance(payload, dict) else {}
+    return _classify(response.status_code, payload), payload, parse_retry_after_seconds(response.headers)
+
+
+def _classify(status: int, payload: dict[str, Any]) -> str:
+    if status == 200:
+        return _CLAIMED if payload.get("status") == "claimed" else _RETRY
+    if status == 401:
+        return _REAUTH
+    return _REFUSALS.get(str(payload.get("error") or ""), _RETRY)
+
+
+def _apply(record: dict[str, Any], outcome: str, payload: dict[str, Any], retry_after: Optional[float],
+           prior_attempts: int) -> dict[str, Any]:
+    if record["state"] == CLAIMED:
+        return record  # never overwritten, whatever a racing attempt heard
+    record = {**record, "error": payload.get("error"), "next_attempt_at": None}
+    if outcome == _CLAIMED:
+        connectors = payload.get("connectors") if isinstance(payload.get("connectors"), dict) else None
+        record.update(state=CLAIMED, error=None, result={
+            "account_email": payload.get("account_email"),
+            "destination_org_id": payload.get("destination_org_id"),
+            # A service that does not report what moved (an older one) reads as unknown, never "none".
+            "connectors_kind": (connectors or {}).get("kind") or "unknown",
+            "moved_toolkits": list((connectors or {}).get("moved_toolkits") or []),
+            "dropped_toolkits": list((connectors or {}).get("dropped_toolkits") or []),
+            "cohort_carried": payload.get("cohort_carried") is True})
+    elif outcome == _GONE:
+        # Gone after an earlier attempt is most likely that attempt's lost success.
+        record.update(state=GONE, probably_moved=payload.get("error") == "already_retired" or prior_attempts > 0)
+    elif outcome == _FAIL:
+        record.update(state=FAILED)
+    else:
+        ladder = _BACKOFF_SECONDS[min(prior_attempts, len(_BACKOFF_SECONDS) - 1)]
+        record.update(state=RETRYING, next_attempt_at=_now() + (retry_after if retry_after else ladder))
+    if record["state"] in (CLAIMED, GONE):
+        record.pop("anon_token", None)  # retired credential: nothing left to retry with
+    return record
+
+
+# --- Read side: status for the surfaces -----------------------------------------------------------
+
+def public_view(record: dict[str, Any]) -> dict[str, Any]:
+    """The wire shape the surfaces read; never the credential."""
+    result = record.get("result") or {}
+    next_at = record.get("next_attempt_at")
+    return {
+        "state": record.get("state", IDLE), "account_email": result.get("account_email"),
+        "connectors_kind": result.get("connectors_kind"),
+        "moved_toolkits": result.get("moved_toolkits", []), "dropped_toolkits": result.get("dropped_toolkits", []),
+        "probably_moved": bool(record.get("probably_moved")), "error": record.get("error"),
+        "retry_after": max(0, int(float(next_at) - _now())) if next_at else 0,
+        "updated_at": record.get("updated_at")}
+
+
+def handoff_status() -> dict[str, Any]:
+    """This profile's latest handoff, or ``{"state": "idle"}``. A pure read (no lock, no network). An
+    active move bound to an account this profile no longer holds reads as ``paused``."""
+    try:
+        records = _load(_records_path())["records"].values()
+    except RuntimeError:
+        return {"state": IDLE}
+    home = _home_key()
+    mine = [r for r in records if home in r.get("profiles", ())]
+    if not mine:
+        return {"state": IDLE}
+    latest = max(mine, key=lambda r: str(r.get("updated_at") or ""))
+    view = public_view(latest)
+    if _retryable(latest):
+        view["state"] = RETRYING if _account_bound_to(latest) else PAUSED
+    return view
+
+
+def status_lines(status: dict[str, Any]) -> list[str]:
+    """User copy for a non-idle handoff (``hermes auth status nous``)."""
+    state = status.get("state")
+    where = status.get("account_email") or "your account"
+    moved, dropped = status.get("moved_toolkits") or [], status.get("dropped_toolkits") or []
+    if state in _ACTIVE:
+        return ["Moving your free-tier apps to this account; Hermes keeps trying in the background."]
+    if state == PAUSED:
+        return ["Your free-tier apps are set to move to another Nous account. Sign in to that account "
+                "to finish moving them."]
+    if state == FAILED:
+        reason = _FAILURE_COPY.get(str(status.get("error") or ""))
+        return ["Your free-tier apps could not be moved." + (f" {reason}" if reason else "")]
+    if state == GONE:
+        return ["Your free-tier apps were probably moved already." if status.get("probably_moved")
+                else "Nothing left to move from the free tier."]
+    if state != CLAIMED:
+        return []
+    kind = status.get("connectors_kind")
+    # An empty moved list is the truth: the service refuses a move it could not read, and gives a
+    # free tier that only read its connectors a vendor row, so kind alone says nothing.
+    if kind in ("moved", "swapped") and moved:
+        lines = [f"Your apps moved to {where}: {', '.join(moved)}."]
+    elif kind == "conflict":
+        lines = [f"{where} already had its own apps, so the free-tier apps stayed behind."]
+    elif kind in ("none", "moved", "swapped"):
+        lines = [f"Signed in as {where}. There were no free-tier apps to move."]
+    else:
+        lines = [f"Signed in as {where}. Any free-tier apps are in this account now."]
+    if dropped:
+        lines.append(f"Reconnect in {where}: {', '.join(dropped)}.")
+    return lines

hermes_cli/nous_auth_keepalive.py

@@ -210,6 +210,10 @@ def start_nous_auth_keepalive(
     min_key_ttl_seconds: int = NOUS_INVOKE_JWT_MIN_TTL_SECONDS, timeout_seconds: Optional[float] = None,
 ) -> Optional[threading.Thread]:
     """Start the process-wide Nous auth keepalive thread (idempotent; None when disabled)."""
+    # Free-tier handoff retries run in every long-lived Nous process, guest or not and keepalive or
+    # not: a CLI sign-in in another process may have left a move this one has to finish.
+    from hermes_cli.guest_handoff import start_worker
+    start_worker()
     interval_seconds = _interval_seconds(interval_seconds)
     if interval_seconds <= 0:
         return None
@@ -243,6 +247,8 @@ def start_nous_auth_keepalive(
 
 def stop_nous_auth_keepalive(timeout: float = 5.0) -> None:
     """Stop the keepalive thread. Intended for graceful shutdown/tests."""
+    from hermes_cli.guest_handoff import stop_worker
+    stop_worker(timeout)
     global _keepalive_thread
     with _keepalive_lock:
         thread = _keepalive_thread

hermes_cli/web_server_oauth.py

@@ -302,7 +302,7 @@ def _nous_plain_poller(session_id: str, sess: dict[str, Any]) -> None:
     """
     from hermes_cli.web_server_profiles import _profile_scope
     from hermes_cli.auth import _poll_for_token, persist_nous_credentials, refresh_nous_oauth_from_state
-    from hermes_cli import anon_auth
+    from hermes_cli import anon_auth, guest_handoff
     import httpx
     portal_base_url, client_id = sess["portal_base_url"], sess["client_id"]
 
@@ -346,16 +346,17 @@ def _nous_plain_poller(session_id: str, sess: dict[str, Any]) -> None:
     with _profile_scope(sess.get("profile")):
         full_state = refresh_nous_oauth_from_state(auth_state, timeout_seconds=15.0, force_refresh=False)
         # The final cancellation check and the save share the session lock, so a cancel cannot
-        # land between them.
-        with _oauth_sessions_lock:
-            if sess.get("cancelled"):
-                sess["status"] = "cancelled"
-                return
-            persist_nous_credentials(full_state)
+        # land between them. The guest handoff runs after that lock is released.
+        with guest_handoff.deferred() as handoff:
+            with _oauth_sessions_lock:
+                if sess.get("cancelled"):
+                    sess["status"] = "cancelled"
+                    return
+                persist_nous_credentials(full_state)
         # A config left on the free tier's route by a retired identity still has to move.
         settled = anon_auth.settle_after_upgrade(full_state)
     with _oauth_sessions_lock:
-        sess["account_email"] = None
+        sess["account_email"] = handoff.account_email
         sess["model"] = settled.get("model") or None
 
 

tests/hermes_cli/test_anon_surfaces.py

@@ -165,14 +165,17 @@ def test_keepalive_does_not_start_for_free_tier(isolated_store, monkeypatch):
             pass
     monkeypatch.setattr(nous_auth_keepalive.threading, "Thread", _Thread)
     monkeypatch.setattr(nous_auth_keepalive, "_keepalive_thread", None)
+    from hermes_cli import guest_handoff
+    monkeypatch.setattr(guest_handoff, "_worker_thread", None)
 
     _write_auth(_guest_state())
     assert nous_auth_keepalive.start_nous_auth_keepalive(interval_seconds=900) is None
-    assert started == []
+    # The free-tier handoff retry worker runs in a guest process too; the token keepalive does not.
+    assert started == ["nous-guest-handoff"]
 
     _write_auth(_account_state())
     thread = nous_auth_keepalive.start_nous_auth_keepalive(interval_seconds=900)
-    assert thread is not None and started == ["nous-auth-keepalive"]
+    assert thread is not None and started == ["nous-guest-handoff", "nous-auth-keepalive"]
     monkeypatch.setattr(nous_auth_keepalive, "_keepalive_thread", None)
 
 

tests/hermes_cli/test_anon_upgrade.py

@@ -174,6 +174,14 @@ class TestUpgrade:
         assert shared.get("refresh_token") == REFRESH_TOKEN
         assert "anon_token" not in shared
 
+    def test_completed_promotion_is_the_handoff_and_sends_no_claim(self, portal):
+        from hermes_cli import guest_handoff
+        anon_auth.ensure_portal_identity(explicit=True)
+        assert anon_auth.upgrade_guest(_args()) == 0
+        assert ("POST", "/api/anonymous/claim") not in portal.calls
+        status = guest_handoff.handoff_status()
+        assert status["state"] == "claimed" and status["account_email"] == EMAIL
+
 
 FREE_PICK = "upstage/solar-pro4:free"
 

tests/hermes_cli/test_guest_handoff.py

@@ -0,0 +1,588 @@
+"""The guest handoff step: a Nous account replacing the free tier moves the free tier's apps into it.
+
+Driven end to end through Hermes' real writers, stores and HTTP client against a fake account
+service (``anon_portal.FakePortal`` plus the claim and token-refresh routes).
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from tests.hermes_cli.anon_portal import PORTAL, FakePortal, install_portal, make_jwt
+
+DESTINATION = "user_acct_1"
+
+
+def _account(sub: str = DESTINATION, refresh: str = "rt_1") -> dict:
+    expires = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
+    return {
+        "auth_method": "oauth_device_code", "portal_base_url": PORTAL, "client_id": "hermes-cli",
+        "inference_base_url": "https://inference-api.nousresearch.com/v1", "scope": "inference:invoke",
+        "token_type": "Bearer", "access_token": make_jwt(sub=sub, client_id="hermes-cli", account_tier="standard"),
+        "refresh_token": refresh, "obtained_at": datetime.now(UTC).isoformat(), "expires_at": expires,
+        "expires_in": 3600}
+
+
+CLAIMED = {"status": "claimed", "destination_org_id": "org_1", "account_email": "sid@example.com",
+           "connectors": {"kind": "moved", "moved_toolkits": ["gmail"], "dropped_toolkits": []},
+           "cohort_carried": False}
+
+
+class ClaimPortal(FakePortal):
+    """FakePortal plus ``POST /api/anonymous/claim`` and the OAuth refresh route."""
+
+    def __init__(self):
+        super().__init__()
+        self.claim_answers: list = []          # each: httpx.Response, or an exception to raise
+        self.claims: list[dict] = []
+        self.refreshes = 0
+        self.locks_held_during_claim: list[bool] = []
+        self.on_claim = None                    # side effect run as each claim arrives
+
+    def handler(self, request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/api/anonymous/claim":
+            from hermes_cli import guest_handoff
+            self.locks_held_during_claim.append(guest_handoff._holds_auth_locks())
+            self.claims.append({"body": json.loads(request.content),
+                                "authorization": request.headers.get("authorization")})
+            if self.on_claim:
+                self.on_claim()
+            answer = self.claim_answers.pop(0) if self.claim_answers else httpx.Response(200, json=CLAIMED)
+            if isinstance(answer, Exception):
+                raise answer
+            return answer
+        if request.url.path == "/api/oauth/token":
+            self.refreshes += 1
+            return httpx.Response(200, json={
+                "access_token": make_jwt(sub=DESTINATION, client_id="hermes-cli", account_tier="standard",
+                                         rotation=self.refreshes),
+                "refresh_token": f"rt_{self.refreshes + 1}", "expires_in": 3600, "token_type": "Bearer"})
+        return super().handler(request)
+
+
+@pytest.fixture
+def portal(monkeypatch, tmp_path):
+    from hermes_cli import guest_handoff
+    fake = install_portal(monkeypatch, tmp_path, ClaimPortal())
+    # A runner's backoff sleep moves the handoff's own clock instead of the wall clock.
+    skew = [0.0]
+    monkeypatch.setattr(guest_handoff, "_now", lambda: time.time() + skew[0])
+    monkeypatch.setattr(guest_handoff, "_sleep", lambda s: skew.__setitem__(0, skew[0] + s))
+    monkeypatch.setattr(guest_handoff, "_local", threading.local())
+    # The settle reads the recommended default from the Portal; that is not under test here.
+    monkeypatch.setattr("hermes_cli.anon_auth.settle_after_upgrade", lambda _s: {"model": "", "changed": False})
+    return fake
+
+
+def _mint_guest() -> str:
+    from hermes_cli import anon_auth
+    state = anon_auth.ensure_portal_identity(explicit=True)
+    assert anon_auth.is_guest_state(state)
+    return state["anon_token"]
+
+
+def _records() -> dict:
+    from hermes_cli import guest_handoff
+    return guest_handoff._load(guest_handoff._records_path())["records"]
+
+
+def _only_record() -> dict:
+    records = _records()
+    assert len(records) == 1, records
+    return next(iter(records.values()))
+
+
+# --- Writer coverage: every way an account replaces the guest fires the hook ----------------------
+
+def _persist(account):
+    from hermes_cli.auth import persist_nous_credentials
+    persist_nous_credentials(account)
+
+
+def _step_up(account, monkeypatch):
+    from hermes_cli import auth
+    monkeypatch.setattr(auth, "_nous_device_code_login", lambda **_kw: account)
+    auth.step_up_nous_billing_scope(open_browser=False)
+
+
+def _login_nous(account, monkeypatch):
+    from hermes_cli import auth, auth_nous
+    monkeypatch.setattr(auth, "_nous_device_code_login", lambda **_kw: account)
+    monkeypatch.setattr(auth_nous, "_pick_nous_model_after_login", lambda *_a: None)
+    auth_nous._login_nous(SimpleNamespace(), auth.PROVIDER_REGISTRY["nous"])
+
+
+def _subscription_login(account, monkeypatch):
+    from hermes_cli import auth, nous_subscription
+    monkeypatch.setattr(auth, "_nous_device_code_login", lambda **_kw: account)
+    monkeypatch.setattr(nous_subscription, "_confirm", lambda _p: True)
+    assert nous_subscription._run_nous_portal_login_only(capability="Web search")
+
+
+def _sibling_adoption(account, monkeypatch, tmp_path):
+    """Another profile signed in (account in the shared store); this one adopts it on next use."""
+    import os
+    from hermes_cli import anon_auth, auth_nous, guest_handoff
+    mine, sibling = os.environ["HERMES_HOME"], tmp_path / "sibling-profile"
+    sibling.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(sibling))
+    monkeypatch.setattr(auth_nous, "drain", lambda: None)   # the sibling's own run is not this test
+    auth_nous._write_shared_nous_state(account)
+    monkeypatch.setattr(auth_nous, "drain", guest_handoff.drain)
+    monkeypatch.setenv("HERMES_HOME", mine)
+    anon_auth.ensure_portal_identity(explicit=True)
+
+
+WRITERS = {
+    "persist_nous_credentials": lambda a, mp, tp: _persist(a),
+    "step_up_nous_billing_scope": lambda a, mp, tp: _step_up(a, mp),
+    "_login_nous": lambda a, mp, tp: _login_nous(a, mp),
+    "nous_subscription_login": lambda a, mp, tp: _subscription_login(a, mp),
+    "shared_store_adoption": _sibling_adoption,
+}
+
+
+@pytest.mark.parametrize("writer", sorted(WRITERS))
+def test_every_writer_that_replaces_the_guest_hands_it_over(portal, monkeypatch, tmp_path, writer):
+    anon = _mint_guest()
+    WRITERS[writer](_account(), monkeypatch, tmp_path)
+
+    assert [c["body"] for c in portal.claims] == [{"token": anon}]
+    assert portal.claims[0]["authorization"].startswith("Bearer ")
+    assert portal.locks_held_during_claim == [False]
+    record = _only_record()
+    assert record["state"] == "claimed" and record["destination"] == DESTINATION
+    assert "anon_token" not in record
+    from hermes_cli import guest_handoff
+    assert guest_handoff.handoff_status()["moved_toolkits"] == ["gmail"]
+
+
+def test_an_account_replacing_an_account_is_not_a_handoff(portal):
+    _persist(_account(refresh="rt_a"))
+    _persist(_account(sub="user_acct_2", refresh="rt_b"))
+    assert portal.claims == [] and _records() == {}
+
+
+# --- Each answer of the claim endpoint ------------------------------------------------------------
+
+def _err(status, code, **headers):
+    return httpx.Response(status, json={"error": code, "error_description": "x"}, headers=headers)
+
+
+@pytest.mark.parametrize("answer, state, probably_moved", [
+    (httpx.Response(200, json=CLAIMED), "claimed", False),
+    (_err(404, "unknown_token"), "gone", False),
+    (_err(410, "already_retired"), "gone", True),
+    (_err(409, "account_busy"), "retrying", False),
+    (_err(429, "rate_limited", **{"Retry-After": "120"}), "retrying", False),
+    (_err(404, "not_found"), "retrying", False),
+    (_err(503, "temporarily_disabled"), "retrying", False),
+    (_err(503, "temporarily_unavailable", **{"Retry-After": "5"}), "retrying", False),
+    (_err(500, "internal_error"), "retrying", False),
+    (_err(409, "no_destination"), "retrying", False),
+    (_err(409, "unexpected_account_state"), "retrying", False),
+    (_err(422, "a_code_this_version_does_not_know"), "retrying", False),
+    (_err(403, "client_not_allowed"), "failed", False),
+    (_err(403, "account_locked"), "failed", False),
+], ids=lambda v: v if isinstance(v, str) else None)
+def test_each_claim_answer_lands_in_its_state(portal, monkeypatch, answer, state, probably_moved):
+    from hermes_cli import guest_handoff
+    monkeypatch.setattr(guest_handoff, "SYNC_BUDGET_SECONDS", 3.0)   # one attempt: every retry waits longer
+    monkeypatch.setattr(guest_handoff, "_BACKOFF_SECONDS", (60.0,))
+    anon = _mint_guest()
+    portal.claim_answers = [answer]
+    _persist(_account())
+
+    record = _only_record()
+    assert record["state"] == state
+    assert bool(record.get("probably_moved")) is probably_moved
+    # A retrying or failed record keeps the credential; a finished one drops it.
+    assert (record.get("anon_token") == anon) is (state in ("retrying", "failed"))
+    if state == "retrying":   # honours Retry-After, else the first backoff step
+        wait = float(record["next_attempt_at"]) - time.time()
+        assert wait >= float(answer.headers.get("Retry-After", 1)) - 1
+
+
+def test_a_refused_account_token_is_refreshed_once_then_retried(portal):
+    portal.claim_answers = [_err(401, "invalid_token"), httpx.Response(200, json=CLAIMED)]
+    _mint_guest()
+    _persist(_account())
+    assert len(portal.claims) == 2 and portal.refreshes == 1
+    assert portal.claims[0]["authorization"] != portal.claims[1]["authorization"]
+    assert _only_record()["state"] == "claimed" and _only_record()["attempts"] == 1
+
+
+def test_lost_success_then_unknown_token_reads_as_probably_moved(portal):
+    portal.claim_answers = [httpx.ReadTimeout("lost"), _err(404, "unknown_token")]
+    _mint_guest()
+    _persist(_account())
+    record = _only_record()
+    assert len(portal.claims) == 2
+    assert record["state"] == "gone" and record["probably_moved"] is True and record["attempts"] == 2
+
+
+@pytest.mark.parametrize("carried, refreshes", [(True, 1), (False, 0)])
+def test_account_token_refreshes_only_when_the_cohort_carried(portal, carried, refreshes):
+    portal.claim_answers = [httpx.Response(200, json={**CLAIMED, "cohort_carried": carried})]
+    _mint_guest()
+    _persist(_account())
+    assert _only_record()["state"] == "claimed"
+    assert portal.refreshes == refreshes
+
+
+def test_transient_failures_keep_retrying_past_the_old_attempt_cap(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    portal.claim_answers = [_err(503, "temporarily_disabled")] * 12   # the ops breaker: no Retry-After
+    monkeypatch.setattr(guest_handoff, "_BACKOFF_SECONDS", (0.0,))
+    _mint_guest()
+    _persist(_account())
+    assert len(portal.claims) == 13 and _only_record()["state"] == "claimed"
+
+
+def test_a_long_outage_retries_on_the_slowest_rung(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    _pending_for(monkeypatch, _account())
+    guest_handoff._update(guest_handoff._records_path(), next(iter(_records())),
+                          lambda r: {**r, "state": "retrying", "attempts": 40})
+    portal.claim_answers = [_err(503, "temporarily_disabled")]
+    guest_handoff.drain()
+    record = _only_record()
+    assert record["state"] == "retrying" and record["anon_token"]
+    assert float(record["next_attempt_at"]) - guest_handoff._now() >= guest_handoff._BACKOFF_SECONDS[-1] - 5
+
+
+# --- Binding, sharing, durability -----------------------------------------------------------------
+
+def test_a_retry_only_claims_into_the_destination_account(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    monkeypatch.setattr(guest_handoff, "_BACKOFF_SECONDS", (60.0,))
+    portal.claim_answers = [_err(409, "account_busy")]
+    _mint_guest()
+    _persist(_account())
+    assert _only_record()["state"] == "retrying"
+
+    # The profile now holds another account: the record is not claimed into it.
+    _persist(_account(sub="someone_else", refresh="rt_other"))
+    monkeypatch.setattr(guest_handoff, "_now", lambda: time.time() + 10_000)
+    guest_handoff.drain()
+    assert len(portal.claims) == 1 and _only_record()["destination"] == DESTINATION
+
+    _persist(_account(refresh="rt_back"))
+    guest_handoff.drain()
+    assert len(portal.claims) == 2 and _only_record()["state"] == "claimed"
+
+
+def test_two_profiles_sharing_one_guest_claim_once(portal, monkeypatch, tmp_path):
+    from hermes_cli import anon_auth, guest_handoff
+    import os
+    first = os.environ["HERMES_HOME"]
+    second = tmp_path / "second-profile"
+    second.mkdir()
+    anon = _mint_guest()
+    monkeypatch.setenv("HERMES_HOME", str(second))
+    assert anon_auth.ensure_portal_identity(explicit=True)["anon_token"] == anon   # adopts the guest
+
+    monkeypatch.setenv("HERMES_HOME", first)
+    _persist(_account())
+    monkeypatch.setenv("HERMES_HOME", str(second))
+    anon_auth.ensure_portal_identity(explicit=True)      # adopts the account, replacing the guest
+
+    assert len(portal.claims) == 1
+    assert guest_handoff.handoff_status()["state"] == "claimed"
+    assert len(_only_record()["profiles"]) == 2
+
+
+def test_a_crash_between_save_and_claim_is_claimed_on_next_boot(portal, monkeypatch):
+    from hermes_cli import auth_nous, guest_handoff
+    anon = _mint_guest()
+    monkeypatch.setattr(auth_nous, "drain", lambda: None)     # the process dies before the claim
+    _persist(_account())
+    assert portal.claims == [] and _only_record()["state"] == "pending"
+
+    monkeypatch.setattr(auth_nous, "drain", guest_handoff.drain)
+    monkeypatch.setattr(guest_handoff, "_local", threading.local())   # a fresh process
+    guest_handoff.retry_all_profiles()                                 # its retry worker's first tick
+
+    assert [c["body"] for c in portal.claims] == [{"token": anon}]
+    assert _only_record()["state"] == "claimed"
+
+
+def test_a_claimed_record_is_never_reopened(portal):
+    from hermes_cli import guest_handoff
+    _mint_guest()
+    _persist(_account())
+    record = _only_record()
+    guest_handoff.note_replacement({"auth_method": "anonymous", "anon_token": "anon_0001"}, _account())
+    guest_handoff.drain()
+    assert len(portal.claims) == 1 and _only_record()["state"] == record["state"] == "claimed"
+
+
+def test_a_surface_lock_defers_the_claim_until_it_is_released(portal):
+    from hermes_cli import guest_handoff
+    _mint_guest()
+    with guest_handoff.deferred() as batch:
+        _persist(_account())
+        assert portal.claims == [] and _only_record()["state"] == "pending"
+    assert len(portal.claims) == 1
+    assert batch.account_email == "sid@example.com"
+
+
+# --- Review round 1: the apps never land in an account the record is not bound to ----------------
+
+def _pending_for(monkeypatch, account) -> None:
+    """A saved account and its pending record, the claim not yet run."""
+    _mint_guest()
+    _pending_for_current_guest(monkeypatch, account)
+
+
+def _pending_for_current_guest(monkeypatch, account) -> None:
+    from hermes_cli import auth_nous, guest_handoff
+    monkeypatch.setattr(auth_nous, "drain", lambda: None)
+    _persist(account)
+    monkeypatch.setattr(auth_nous, "drain", guest_handoff.drain)
+    assert _only_record()["state"] == "pending"
+
+
+def test_a_sign_in_racing_the_claim_never_moves_apps_into_it(portal, monkeypatch):
+    from hermes_cli import auth_nous, guest_handoff
+    _pending_for(monkeypatch, _account())
+    real_bearer = guest_handoff._bearer
+
+    def bearer_after_another_sign_in(**kw):
+        monkeypatch.setattr(auth_nous, "drain", lambda: None)
+        _persist(_account(sub="intruder", refresh="rt_intruder"))   # lands after the bound check
+        return real_bearer(**kw)
+
+    monkeypatch.setattr(guest_handoff, "_bearer", bearer_after_another_sign_in)
+    guest_handoff.drain()
+    assert portal.claims == []
+    assert _only_record()["state"] == "pending" and _only_record()["attempts"] == 0
+
+
+def test_a_destination_rebound_before_the_attempt_is_not_claimed_into_the_old_account(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    _pending_for(monkeypatch, _account())
+    real_update, fired = guest_handoff._update, []
+
+    def sibling_wins_the_records_lock_first(path, key, change):
+        if not fired:   # the runner's first locked write; a sibling's rebind lands just before it
+            fired.append(True)
+            guest_handoff.note_replacement(
+                {"auth_method": "anonymous", "anon_token": "anon_0001"}, _account(sub="sibling_acct"))
+        return real_update(path, key, change)
+
+    monkeypatch.setattr(guest_handoff, "_update", sibling_wins_the_records_lock_first)
+    guest_handoff.drain()
+    assert portal.claims == []
+    assert _only_record()["destination"] == "sibling_acct" and _only_record()["state"] == "pending"
+
+
+def _age_record(days: float, **fields) -> None:
+    from hermes_cli import guest_handoff
+    created = guest_handoff._iso(guest_handoff._now() - days * 86400)
+    guest_handoff._update(guest_handoff._records_path(), next(iter(_records())),
+                          lambda r: {**r, "created_at": created, **fields})
+
+
+def test_retries_stop_safely_before_the_free_tier_is_reaped(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    _pending_for(monkeypatch, _account())
+    _age_record(13.5, state="retrying", attempts=300)
+    guest_handoff.drain()
+    assert portal.claims == []
+    assert _only_record()["state"] == "failed" and _only_record()["error"] == "retry_window_ended"
+
+
+@pytest.mark.parametrize("trigger", ["auth_status", "auth_upgrade", "sign_in"])
+def test_a_record_an_older_build_failed_on_a_transient_error_is_rearmed(portal, monkeypatch, capsys, trigger):
+    from hermes_cli import anon_auth, auth_commands
+    anon = _mint_guest()
+    _pending_for_current_guest(monkeypatch, _account())
+    _age_record(2, state="failed", error="retries_exhausted", attempts=8)
+    if trigger == "auth_status":
+        auth_commands.auth_status_command(SimpleNamespace(provider="nous"))
+    elif trigger == "auth_upgrade":
+        anon_auth.upgrade_guest(SimpleNamespace(no_browser=True, timeout=None))
+    else:
+        _persist(_account(refresh="rt_again"))    # the account saved again (re-login)
+    assert [c["body"] for c in portal.claims] == [{"token": anon}]
+    assert _only_record()["state"] == "claimed"
+
+
+def test_a_final_refusal_is_not_rearmed(portal, monkeypatch):
+    from hermes_cli import auth_commands
+    _pending_for(monkeypatch, _account())
+    _age_record(1, state="failed", error="account_locked", attempts=1)
+    auth_commands.auth_status_command(SimpleNamespace(provider="nous"))
+    assert portal.claims == [] and _only_record()["state"] == "failed"
+
+
+@pytest.mark.parametrize("code", ["client_not_allowed", "account_locked", "invalid_request",
+                                  "retry_window_ended", "some_new_code", None])
+def test_failure_copy_never_prints_a_server_code(code):
+    from hermes_cli import guest_handoff
+    lines = guest_handoff.status_lines({"state": "failed", "error": code})
+    assert lines and lines[0].startswith("Your free-tier apps could not be moved.")
+    assert code is None or code not in " ".join(lines)
+
+
+def test_a_claim_answer_without_connectors_reads_as_neutral(portal):
+    from hermes_cli import guest_handoff
+    portal.claim_answers = [httpx.Response(200, json={
+        "status": "claimed", "destination_org_id": "org_1", "account_email": "sid@example.com"})]
+    _mint_guest()
+    _persist(_account())
+    text = " ".join(guest_handoff.status_lines(guest_handoff.handoff_status()))
+    assert "sid@example.com" in text and "no free-tier apps" not in text
+
+
+def test_gone_on_a_first_attempt_reads_as_neutral(portal):
+    from hermes_cli import guest_handoff
+    portal.claim_answers = [_err(404, "unknown_token")]
+    _mint_guest()
+    _persist(_account())
+    assert guest_handoff.status_lines(guest_handoff.handoff_status()) == ["Nothing left to move from the free tier."]
+
+
+def test_a_slow_refusal_at_the_budget_edge_starts_no_refresh_or_second_claim(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    clock = [1000.0]
+    monkeypatch.setattr(guest_handoff.time, "monotonic", lambda: clock[0])
+    portal.claim_answers = [_err(401, "invalid_token")]
+    portal.on_claim = lambda: clock.__setitem__(0, clock[0] + guest_handoff.SYNC_BUDGET_SECONDS + 5)
+    _mint_guest()
+    _persist(_account())
+    assert len(portal.claims) == 1 and portal.refreshes == 0
+    assert _only_record()["state"] == "pending" and _only_record()["attempts"] == 0
+
+
+def test_corrupt_records_neither_block_a_sign_in_nor_stop_the_retry_worker(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    path = guest_handoff._records_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("{not json", encoding="utf-8")
+    assert guest_handoff.retry_all_profiles() == []
+
+    _mint_guest()
+    _persist(_account())
+    assert _only_record()["state"] == "claimed"
+    assert path.with_suffix(".json.corrupt").read_text(encoding="utf-8") == "{not json"
+
+
+def test_a_move_bound_to_another_account_reads_as_paused(portal, monkeypatch):
+    from hermes_cli import guest_handoff
+    portal.claim_answers = [_err(409, "account_busy")]
+    monkeypatch.setattr(guest_handoff, "_BACKOFF_SECONDS", (60.0,))
+    _mint_guest()
+    _persist(_account())
+    assert guest_handoff.handoff_status()["state"] == "retrying"
+    _persist(_account(sub="someone_else", refresh="rt_other"))
+    status = guest_handoff.handoff_status()
+    assert status["state"] == "paused"
+    assert "another Nous account" in guest_handoff.status_lines(status)[0]
+
+
+# --- Review round 2: retries reach every profile; the first attempt is not lost -------------------
+
+def test_a_long_lived_process_with_a_guest_still_runs_the_handoff_worker(portal, monkeypatch):
+    from hermes_cli import guest_handoff, nous_auth_keepalive
+    monkeypatch.setattr(nous_auth_keepalive, "get_provider_auth_state",
+                        lambda _p: {"auth_method": "anonymous", "anon_token": "anon_x"})
+    try:
+        nous_auth_keepalive.start_nous_auth_keepalive()
+        assert guest_handoff._worker_thread is not None and guest_handoff._worker_thread.is_alive()
+    finally:
+        nous_auth_keepalive.stop_nous_auth_keepalive()
+    assert guest_handoff._worker_thread is None
+
+
+def test_the_worker_retries_a_named_profiles_move_from_the_launch_profile(portal, monkeypatch, tmp_path):
+    import os
+    from hermes_cli import guest_handoff
+    anon = _mint_guest()
+    _pending_for_current_guest(monkeypatch, _account())
+    launch = tmp_path / "launch-profile"
+    launch.mkdir()
+    named_home = os.environ["HERMES_HOME"]
+    monkeypatch.setenv("HERMES_HOME", str(launch))      # the long-lived process runs as another profile
+    guest_handoff.retry_all_profiles()
+    assert [c["body"] for c in portal.claims] == [{"token": anon}]
+    monkeypatch.setenv("HERMES_HOME", named_home)
+    assert _only_record()["state"] == "claimed"
+
+
+def test_a_guest_token_still_cached_from_a_connector_call_does_not_skip_the_move(portal):
+    from hermes_cli import auth
+    from hermes_constants import hermes_home_key
+    _mint_guest()
+    auth._RESOLVE_TOKEN_CACHE[hermes_home_key()] = (time.monotonic(), make_jwt())   # the guest's own token
+    _persist(_account())
+    assert len(portal.claims) == 1 and _only_record()["state"] == "claimed"
+
+
+def test_a_held_auth_lock_cannot_stretch_the_sync_budget(portal, monkeypatch):
+    from hermes_cli import auth, guest_handoff
+    _pending_for(monkeypatch, _account())
+    monkeypatch.setattr(guest_handoff, "SYNC_BUDGET_SECONDS", 1.0)
+    held, release = threading.Event(), threading.Event()
+
+    def hold_auth_lock():
+        with auth._auth_store_lock():
+            held.set()
+            release.wait(30)
+
+    holder = threading.Thread(target=hold_auth_lock, daemon=True)
+    holder.start()
+    held.wait(5)
+    started = time.monotonic()
+    try:
+        guest_handoff.drain()
+    finally:
+        release.set()
+        holder.join(5)
+    assert time.monotonic() - started < 6
+    assert portal.claims == [] and _only_record()["state"] == "pending" and _only_record()["attempts"] == 0
+
+
+def test_auth_status_shows_what_moved(portal, capsys):
+    from hermes_cli import auth_commands
+    _mint_guest()
+    _persist(_account())
+    capsys.readouterr()
+    auth_commands.auth_status_command(SimpleNamespace(provider="nous"))
+    out = capsys.readouterr().out
+    assert "nous: logged in" in out and "Your apps moved to sid@example.com: gmail." in out
+
+
+@pytest.mark.parametrize("kind", ["moved", "swapped"])
+def test_a_move_with_no_named_apps_reads_as_nothing_moved(portal, kind):
+    """A free tier that only read its connectors gets a vendor id row on the service, so the claim
+    answers ``moved``/``swapped`` with an empty list; that empty list is the truth."""
+    from hermes_cli import guest_handoff
+    portal.claim_answers = [httpx.Response(200, json={
+        **CLAIMED, "connectors": {"kind": kind, "moved_toolkits": [], "dropped_toolkits": []}})]
+    _mint_guest()
+    _persist(_account())
+    assert guest_handoff.status_lines(guest_handoff.handoff_status()) == [
+        "Signed in as sid@example.com. There were no free-tier apps to move."]
+
+
+def test_status_copy_never_names_the_mechanism(portal):
+    from hermes_cli import guest_handoff
+    for status in (
+        {"state": "claimed", "connectors_kind": "moved", "moved_toolkits": [], "account_email": "a@b.c"},
+        {"state": "claimed", "connectors_kind": "conflict", "dropped_toolkits": ["notion"]},
+        {"state": "gone", "probably_moved": True}, {"state": "retrying"}, {"state": "failed", "error": "x"},
+    ):
+        text = " ".join(guest_handoff.status_lines(status)).lower()
+        assert text and not any(word in text for word in ("guest", "anonymous", "claim"))
+    moved = guest_handoff.status_lines(
+        {"state": "claimed", "connectors_kind": "moved", "moved_toolkits": ["gmail"], "account_email": "a@b.c"})
+    assert moved == ["Your apps moved to a@b.c: gmail."]

website/docs/developer-guide/cli-internals.md

@@ -141,6 +141,43 @@ promotion poller are renderers over it; a surface that needs the cancel check an
 atomic passes `persist_guard`. The desktop's plain "connect another Nous account" device-code login
 is a separate path (`_nous_plain_poller`) and must stay one.
 
+### Guest handoff after a sign-in
+
+When any writer saves a Nous account over a free-tier identity, `hermes_cli/guest_handoff.py` hands
+that identity's connected apps to the account: `POST /api/anonymous/claim {"token": "<anon_...>"}`
+with the account's own Bearer token. Nothing asks the user first; the result is shown afterwards
+(`hermes auth status nous`, `guest_handoff.handoff_status()` for the desktop). Sign-in code does not
+call it:
+
+- `note_replacement` runs at the two lowest write seams (`auth._store_provider_state` and
+  `auth_nous._write_shared_nous_state`) under the caller's auth-store locks and only writes a pending
+  record. Records live in `guest_handoff.json` beside the shared Nous store, keyed by a hash of the
+  `anon_` credential and bound to the destination account, so profiles sharing one free-tier
+  identity share one record and one claim.
+- `drain` runs the claim outside every lock, bounded to about 10 s (each record, lock wait, token
+  read and request counts against the budget): after a shared-store write, when a
+  `guest_handoff.deferred()` block exits (a surface that saves under its own lock, such as the desktop
+  poller, wraps the save in it), and after a free-tier reconcile. It only moves apps; config
+  completion (`settle_after_upgrade`) stays with the sign-in that saved the account.
+- Retries left by any process are finished by a retry worker that every long-lived Nous process
+  starts (`start_nous_auth_keepalive` starts it, guest or not). Each tick runs every due record
+  under its owning profile's home and secrets, so a named profile's move retries from a process
+  launched as another profile.
+- Every attempt is reserved in one locked read-modify-write that checks the record is still due,
+  inside its retry window, and bound to the account whose Bearer is about to be sent (the token's
+  `sub` must equal the record's destination). Apps never go to any other account.
+
+Answers: `claimed` records what moved (and refreshes the account token only when
+`cohort_carried`; an answer without `connectors` reads as unknown, never "nothing moved");
+`unknown_token` is Gone, read as "probably moved" after an earlier attempt (a lost success);
+`already_retired` is Gone too; a 401 refreshes the account token once. Only codes no later attempt
+can change are final (`client_not_allowed`, `account_locked`, `invalid_request`). Every other refusal,
+unknown codes included, and every transport failure retries with backoff (Retry-After honoured), the
+slowest rung repeating, until 13 days after the record was made: the service reaps an idle free-tier
+identity, apps included, after 14. A record an older build failed on a transient code re-arms on the
+next run (a sign-in, `hermes auth status`, `hermes auth upgrade`, the retry worker). A Claimed record
+is never reopened, and user copy never shows a server code.
+
 ## Process identity: never infer it from argv substrings
 
 The bug class behind ~10 fleet-update issues (#90778, #87594, #78089, #76129, #91964, ...):