Proxy to another Paskia (#5)

Geographically separate servers may have notable latency between them, making forward/validate request roundtrip unnecessarily slow. This adds a proxy support, where the local server can keep a local copy of a remote domain, while the auth host remains authoritative, but requests can be answered in less than 1ms from local copy.

Reviewed-on: https://git.zi.fi/LeoVasanko/paskia/pulls/5

docs/RemoteProxy.md

@@ -0,0 +1,191 @@
+# Remote Proxy / Satellite
+
+Paskia can serve a configured domain (rp-id) from a **remote** paskia
+instance instead of the local database, so latency-sensitive checks
+(`/auth/api/forward`, `/auth/api/validate`) answer in ~1 ms even when the
+auth server is on another continent. Example: `app2.example.com` runs on
+our local host and needs fast local checks, while `app1.example.com` and
+`auth.example.com` run far away — all sharing `example.com` as rp-id.
+
+Client applications that used `https://auth.example.com` as their auth
+backend (forward-auth checks) only repoint to the local satellite
+(`http://127.0.0.1:4401`); both remain usable interchangeably, and the
+satellite ultimately uses `auth.example.com`.
+
+Status: **implemented**. The feature lives in `paskia/satellite.py`
+(satellite side: replica, sync client, host dispatch, forwarding) and
+`paskia/syncfeed.py` + `paskia/fastapi/sync.py` (remote side: change
+feed and sync WebSocket). The design review comparing the rejected
+alternatives is at the end of this document.
+
+## Configuration
+
+A domain becomes remote in the admin domains UI (master admin, on the
+primary server's auth host — this configuration itself never touches a
+remote): enable *Remote instance* and set the remote URL and sync token.
+In the stored config (`DomainConfig.remote`):
+
+```json
+"remote": {
+  "url": "https://auth.example.com",
+  "token": "<sync token>",
+  "cache_ttl": 60,
+  "refresh_interval": 300
+}
+```
+
+- `url` — the remote instance's base URL. The satellite connects to
+  `{url}/auth/api/sync/ws`; the connection is server-to-server and not
+  host-dispatched, so internal addresses work. The admin dialog probes
+  `{url}/auth/api/settings` as you type: it must be reachable, serve this
+  very rp-id, and not be a satellite for it itself (nested satellites are
+  not supported — always point at the authoritative instance). Probe
+  findings are warnings; the server-side validation is authoritative.
+- `token` — bearer token for the sync channel. The **remote accepts tokens
+  via its `PASKIA_SYNC_TOKENS` environment variable** (comma-separated);
+  nothing is stored in the remote's database, and with the variable unset
+  the sync endpoint stays closed. The token is write-only over the admin
+  API (an empty field keeps the stored one).
+- `cache_ttl` — seconds the replica remains trusted after the sync channel
+  goes down; then checks fail closed (503). Set it large (up to the 24 h
+  session lifetime) for fail-open behavior. Minimum 1.
+- `refresh_interval` — seconds between reconnects; every connect starts
+  from a full snapshot, which reconciles any drift. Minimum 30.
+
+Several domains may share one remote (they share a single replica); their
+remote settings must then be identical — conflicting configurations are
+rejected at validation and aligned with a warning at load.
+
+A remote domain **must mark an auth host** (validated cross-domain and in
+the UI): the profile, admin and sign-in pages live there, so browsers and
+WebSockets go directly to the remote. Other domains on the same satellite
+remain fully local — the multi-domain config mixes both kinds freely.
+
+## How it works
+
+**Dispatch is keyed by host, and only this module knows about stores.**
+`satellite.store_for_host(host)` returns the local DB or the replica of
+the remote backing the host's domain (raising 503 `HTTPException` when
+the replica is unavailable). The session read path (`session_ctx`,
+`authz.verify`, `build_user_info`, `/check`) just passes the host it
+already has; writes dispatch likewise (`satellite.refresh_session` —
+write-behind for remote, `db.update_session` for local;
+`satellite.evict_session` on logout). `satellite.forward_request(request)`
+returns the proxied response for remote domains or `None` for local ones.
+
+**The replica** is a plain `DB` struct instance in RAM, never persisted.
+On connect the remote sends a snapshot of the replicated tables
+(permissions, orgs, roles, users, credentials, sessions), then live
+upsert/delete events emitted from the struct `store()`/`delete()` hooks
+(which also cover cascade deletes) and field-mutating operations. A
+single ordered WebSocket cannot gap; a slow subscriber is dropped and
+resyncs. The feed carries no usable secrets: sessions are keyed by
+`hash_secret` output, credentials carry public keys only, and the OIDC
+signing key is never replicated.
+
+## Endpoint behavior for remote domains
+
+| Endpoint | Handling |
+|---|---|
+| `GET /auth/api/forward`, `GET /check`, `GET /user-info`, `GET /settings` | served from the replica (sub-ms) |
+| `POST /auth/api/validate` | verified from the replica; the throttled refresh updates the replica and is written back over the sync channel; cookie renewed locally |
+| `POST /auth/api/logout` | proxied to the remote (original Host preserved) and evicted from the replica immediately |
+| `POST /auth/api/set-session`, `GET /token-info` | proxied (the exchange code/reset token lives on the remote); the session arrives via sync event |
+| `/auth/oidc/*` | proxied (signing key and OIDC sessions stay on the remote) |
+| `/auth/ws/*`, `/auth/remote-auth/*`, admin, profile | not served — the auth host requirement means these are reached on the remote directly |
+
+Proxied calls answer 503 when the remote is unreachable. `/auth/api/settings`
+reports `remote: true` for a satellite-served domain (this is what the admin
+dialog's uplink probe checks). Should a remote domain's auth host be
+misrouted to the satellite anyway, profile/avatar requests
+(`/auth/api/user/*`) are forwarded to the remote rather than answered from
+local storage — the satellite never takes writes for a remote domain.
+
+Freshness hierarchy:
+
+1. Changes made **through** the satellite: immediate (write-behind,
+   optimistic eviction).
+2. Changes made **directly on the remote**: a sync event, ~1 network RTT.
+3. Channel down: the replica stays authoritative until `cache_ttl` past
+   the disconnect (dead-peer detection is bounded by the ~10 s keepalive),
+   then 503. Every reconnect starts from a fresh snapshot.
+
+## The remote side
+
+Strictly additive and RAM-only: `syncfeed` (a subscriber set fed by the
+commit hooks) and the token-gated `/auth/api/sync/ws` endpoint serving
+snapshot + live events and accepting `session_refresh` write-backs. With
+no satellites connected, the hooks are a no-op.
+
+## Trust and caveats
+
+- The satellite host holds a full copy of the remote's auth data in RAM
+  (minus the OIDC key) — treat it as trusted as the remote.
+- Avatars are stored on the remote's disk; `user-info` from a replica
+  reports no avatar URL.
+- OIDC sessions in a replica-backed `user-info` show the client UUID
+  rather than its name (OIDC clients are not replicated).
+
+---
+
+# Design review (the rejected alternatives)
+
+## Option A — caching HTTP reverse proxy
+
+A thin proxy caching `/auth/api/forward`, `/check`, `/user-info`,
+`/settings` responses keyed by `(Host, cookie, query)` with
+`TTL = min(configured TTL, Remote-Session-Expires − now)`; everything
+else forwarded verbatim, WebSockets tunneled, `/logout` intercepted for
+eviction. **Option B** adds a remote change feed so eviction happens
+within one RTT instead of at TTL.
+
+- Remote changes: none for A; one additive endpoint for B.
+- The proxy needs no credentials — requests are authenticated by the end
+  user's cookie, forwarded on a miss.
+
+## What the read-only local state buys over the HTTP cache
+
+- **Full `SessionContext` locally.** A replays the byte-response it once
+  saw; the satellite *computes* the answer. Query combinations never seen
+  before (new `perm`/`max_age`/`public` shapes) are served locally but
+  miss A's cache. The replica holds the *domain model*, so derived
+  answers (effective permissions per host, `max_age` against
+  `credential.last_used`, `Remote-*` composition) are correct without
+  having been witnessed.
+- **One invalidation model.** A hand-builds invalidation rules per
+  endpoint (query-key mapping, cookie re-keying on renew, 401 variants).
+  Events mutate the replica (upsert/delete by table+key) and every
+  endpoint becomes consistent at once — including future ones.
+- **Degradation behaves like a real instance.** With the remote down, the
+  satellite serves a coherent auth service from the replica (expiry
+  enforced locally, bounded by `cache_ttl`); A serves unrelated cached
+  responses with gaps wherever the cache was cold.
+- **Multi-domain uniformity.** Remote backing is a property of a domain
+  in the existing registry; local and remote rp-ids coexist in one
+  instance. A is a separate component bolted in front of specific URLs.
+- **User simplicity.** Configured once in domain config; A needs
+  deployment and cache-key discipline per frontend application.
+
+## What it costs
+
+- The read path must be honest about which DB it reads: `DB.session_ctx`
+  and `/check` were rewritten to use their own tables instead of struct
+  convenience properties that reach the global database. (A first draft's
+  contextvar-dependent `db.data()` was rejected: a global accessor whose
+  meaning shifts under the caller. Dispatch is instead keyed explicitly
+  by the request host.)
+- A sync protocol (snapshot + live events + reconnect reconciliation).
+- A trusted satellite host (full data copy in RAM).
+- Additive remote code (sync endpoint + commit hooks), where A needs
+  none.
+- Replica housekeeping (expiry sweeper, write-behind, optimistic
+  eviction).
+
+## Summary
+
+A(+B) is the right tool to "make forward-auth fast in front of an
+untouched server". The satellite — implemented here — is the right tool
+when it should *be* a paskia instance for its remote domains: one
+consistency model, correct answers for un-cached query shapes, graceful
+degradation, and per-domain mixing with local rp-ids, at the price of the
+read-path cleanup, the sync protocol, and a trusted satellite host.

frontend/auth/admin/AdminApp.vue

@@ -480,6 +480,7 @@ function createDomain() {
     origins: [],
     originValidation: [],
     wellKnownCheck: null,
+    remote: null,
   })
 }
 
@@ -495,6 +496,8 @@ function openDomain(domain) {
     origins: rows.map(r => r.key),
     originValidation: rows.map(() => null),
     wellKnownCheck: null,
+    // The sync token is write-only: an empty field keeps the stored one
+    remote: domain.remote ? { ...domain.remote, token: '' } : null,
   })
 }
 
@@ -923,9 +926,19 @@ async function submitDialog() {
       }
 
       closeDialog()
+      // remote is replaced wholesale when present; null clears it, an
+      // absent key (create without remote) leaves it unset.
+      const remote = d.remote?.url?.trim()
+        ? {
+            url: d.remote.url.trim().replace(/\/+$/, ''),
+            token: d.remote.token || '',
+            cache_ttl: Number(d.remote.cache_ttl) || 60,
+            refresh_interval: Number(d.remote.refresh_interval) || 300,
+          }
+        : null
       const req = d.isNew
-        ? apiJson('/auth/api/admin/domains/', { method: 'POST', body: { rp_id, rp_name, origins } })
+        ? apiJson('/auth/api/admin/domains/', { method: 'POST', body: { rp_id, rp_name, origins, ...(remote ? { remote } : {}) } })
-        : apiJson(`/auth/api/admin/domains/${rp_id}`, { method: 'PATCH', body: { rp_name, origins } })
+        : apiJson(`/auth/api/admin/domains/${rp_id}`, { method: 'PATCH', body: { rp_name, origins, remote } })
       req
         .then(() => {
           authStore.showMessage(`Domain "${rp_id}" ${d.isNew ? 'created' : 'updated'}.`, 'success', 2500)

frontend/src/admin/AdminOverview.vue

@@ -464,6 +464,7 @@ defineExpose({ focusFirstElement })
             </div>
             <div class="perm-id-info">
               <span class="id-text">{{ domain.rp_id }}</span>
+              <span v-if="domain.remote" class="id-text" :title="`Served from remote ${domain.remote.url} (satellite mode)`">🛰 {{ domain.remote.url }}</span>
             </div>
           </td>
           <td class="domain-origins"><span v-for="(e, i) in originDisplayEntries(domain)" :key="e.key">{{ i ? ', ' : '' }}{{ e.key }}{{ e.auth ? '🔑' : '' }}{{ e.related ? '🔗' : '' }}</span></td>

frontend/src/admin/dialogs/DomainEditDialog.vue

@@ -18,6 +18,71 @@ const title = computed(() =>
 // compares against it, and hosts are case-insensitive)
 const dialogRpId = computed(() => (props.dialog.data?.rp_id || '').trim().toLowerCase())
 
+// --- Remote (satellite) backing ---
+//
+// A remote domain is served from another paskia instance: this one keeps a
+// RAM-only read replica for fast local session checks and forwards
+// mutations. The remote must accept our sync token via its
+// PASKIA_SYNC_TOKENS environment variable. An auth host (the remote's) is
+// required — profile, admin and sign-in pages live there.
+const remoteEnabled = computed({
+  get: () => !!props.dialog.data?.remote,
+  set: on => {
+    const d = props.dialog.data
+    if (!d) return
+    d.remote = on ? { url: '', token: '', cache_ttl: 60, refresh_interval: 300 } : null
+  },
+})
+
+const remoteUrlInvalid = computed(() => {
+  const url = props.dialog.data?.remote?.url?.trim()
+  if (!url) return false
+  return !/^https?:\/\/[^\s/]+/.test(url)
+})
+
+// Remote domains must mark an auth host (the server rejects the save)
+const remoteMissingAuthHost = computed(
+  () => !!props.dialog.data?.remote && !props.dialog.data?.auth_host
+)
+
+// --- Uplink verification ---
+// Fetch the remote's /auth/api/settings (CORS-open), like the origin
+// connectivity probes: proves reachability, that the peer is a paskia
+// instance, and that it serves this very domain authoritatively — not as
+// a satellite itself (nested satellites are not supported). Results are
+// warnings, never submit blockers (the remote may not be deployed yet).
+const uplinkCheck = ref(null) // null|validating|valid|unreachable|mismatch|nested
+let uplinkTimer = null
+let uplinkSeq = 0
+watch(
+  () => [props.dialog.data?.remote?.url, dialogRpId.value],
+  () => {
+    clearTimeout(uplinkTimer)
+    uplinkCheck.value = null
+    const url = props.dialog.data?.remote?.url?.trim().replace(/\/+$/, '')
+    if (!url || remoteUrlInvalid.value) return
+    uplinkTimer = setTimeout(() => checkUplink(url), 600)
+  },
+  { immediate: true }
+)
+
+async function checkUplink(url) {
+  const seq = ++uplinkSeq
+  uplinkCheck.value = 'validating'
+  try {
+    const response = await fetch(`${url}/auth/api/settings`, {
+      headers: { Accept: 'application/json' },
+    })
+    if (seq !== uplinkSeq) return // url/rp-id changed while fetching
+    if (!response.ok) throw new Error('not ok')
+    const data = await response.json()
+    uplinkCheck.value =
+      data.rp_id !== dialogRpId.value ? 'mismatch' : data.remote ? 'nested' : 'valid'
+  } catch {
+    if (seq === uplinkSeq) uplinkCheck.value = 'unreachable'
+  }
+}
+
 // Block submit on hard errors: malformed entries, an over-cap related
 // list (the server rejects the save), a save that would lock the admin
 // out of the domain they are using, or validation still in flight.
@@ -31,6 +96,8 @@ const isValidationInvalid = computed(() => {
   if (relatedEntries.value.length > 5) return true
   if (d.isNew && !isWellFormedDomain(d.rp_id || '')) return true
   if (lockoutWarning.value) return true
+  if (remoteUrlInvalid.value || remoteMissingAuthHost.value) return true
+  if (d.remote && !d.remote.url?.trim()) return true
   return false
 })
 
@@ -377,6 +444,7 @@ function onDocumentClick(e) {
 onMounted(() => document.addEventListener('click', onDocumentClick))
 onBeforeUnmount(() => {
   document.removeEventListener('click', onDocumentClick)
+  clearTimeout(uplinkTimer)
   for (const t of originValidateTimers.values()) clearTimeout(t)
   originValidateTimers.clear()
 })
@@ -515,6 +583,38 @@ function onRemoveOrigin(i) {
     <p class="small muted">
       Only the listed sites may sign in with {{ dialog.data.rp_id }} passkeys. Wildcards may be used: <strong>**.{{ dialog.data.rp_id }}</strong> allows the whole domain, <strong>*.{{ dialog.data.rp_id }}</strong> only a single subdomain level.<template v-if="relatedEntries.length"> 🔗 means related host requiring WebAuthn ROR setup.</template><template v-if="dialog.data.auth_host"> 🔑 is the dedicated Paskia host for all account management.</template>
     </p>
+
+    <div class="origin-label">
+      <label class="remote-toggle">
+        <input type="checkbox" v-model="remoteEnabled" />
+        Remote instance (satellite mode)
+      </label>
+    </div>
+    <template v-if="dialog.data.remote">
+      <label>Remote URL
+        <input v-model="dialog.data.remote.url" placeholder="https://auth.example.com" data-form-type="other" :class="{ 'input-error': remoteUrlInvalid }" />
+      </label>
+      <p v-if="remoteUrlInvalid" class="small error">Must be an http(s) URL.</p>
+      <p v-else-if="uplinkCheck === 'validating'" class="small muted">Checking the uplink…</p>
+      <p v-else-if="uplinkCheck === 'valid'" class="small">✓ Uplink verified — the remote serves this domain authoritatively.</p>
+      <p v-else-if="uplinkCheck === 'unreachable'" class="small">The remote is unreachable — the uplink cannot be verified. Save anyway only if it is not deployed yet.</p>
+      <p v-else-if="uplinkCheck === 'mismatch'" class="small error">The remote does not serve {{ dialogRpId }} — check the URL. Connect directly to the authoritative instance.</p>
+      <p v-else-if="uplinkCheck === 'nested'" class="small error">The remote is itself a satellite for this domain — nested satellites are not supported. Point directly at the authoritative instance.</p>
+      <p v-if="remoteMissingAuthHost" class="small error">A remote domain must mark an auth host above — profile, admin and sign-in pages live there (typically the remote's own site).</p>
+      <label>Sync token
+        <input v-model="dialog.data.remote.token" type="password" placeholder="Token in the remote's PASKIA_SYNC_TOKENS" autocomplete="off" data-form-type="other" />
+      </label>
+      <p class="small muted">Accepted by the remote via its PASKIA_SYNC_TOKENS environment variable.<template v-if="!dialog.data.isNew"> Leave empty to keep the stored token.</template></p>
+      <label>Staleness limit (cache TTL, seconds)
+        <input v-model.number="dialog.data.remote.cache_ttl" type="number" min="1" />
+      </label>
+      <label>Full re-sync interval (seconds)
+        <input v-model.number="dialog.data.remote.refresh_interval" type="number" min="30" />
+      </label>
+      <p class="small muted">
+        Session checks run locally against a RAM replica of the remote (sub-millisecond). If the connection is down longer than the staleness limit, checks fail closed (503).
+      </p>
+    </template>
   </AdminDialog>
 </template>
 
@@ -540,4 +640,7 @@ function onRemoveOrigin(i) {
   border-color: var(--color-error);
   background: var(--color-error-bg, rgba(239, 68, 68, 0.05));
 }
+
+.remote-toggle { display: flex; align-items: center; gap: var(--space-xs); font-weight: 600; font-size: 0.95rem; }
+.remote-toggle input { width: auto; }
 </style>

paskia/authsession.py

@@ -24,8 +24,15 @@ EXPIRES = SESSION_LIFETIME
 
 
 def session_ctx(auth: str, host: str | None = None):
-    """Get session context with normalized host."""
+    """Get session context with normalized host.
-    return db.data().session_ctx(auth, hostutil.normalize_host(host))
+
+    The store is dispatched by host: remote domains read their replica.
+    """
+    from paskia import satellite  # noqa: PLC0415  (import cycle)
+
+    return satellite.store_for_host(host).session_ctx(
+        auth, hostutil.normalize_host(host)
+    )
 
 
 def expires() -> datetime:

paskia/bootstrap.py

@@ -67,7 +67,10 @@ async def check_admin_credentials() -> bool:
         # Check first admin user for credentials on any configured domain
         admin_user = admin_users[0]
         reg = domains.registry()
-        configured = sorted(d.rp_id for d in reg.domains)
+        # Remote domains hold their credentials on the remote instance
+        configured = sorted(d.rp_id for d in reg.domains if d.remote is None)
+        if not configured:
+            return False
 
         if not any(admin_user.credential_ids_for(rp_id) for rp_id in configured):
             # Admin exists but has no credential on any domain

paskia/db/__init__.py

@@ -67,6 +67,7 @@ from paskia.db.structs import (
     DomainConfig,
     Org,
     Permission,
+    RemoteConfig,
     ResetToken,
     Role,
     Session,
@@ -90,6 +91,7 @@ __all__ = [
     "Org",
     "Permission",
     "DomainConfig",
+    "RemoteConfig",
     "ResetToken",
     "Role",
     "Session",

paskia/db/operations.py

@@ -13,7 +13,7 @@ from uuid import UUID
 
 import uuid7
 
-from paskia import oidc_notify
+from paskia import oidc_notify, syncfeed
 from paskia.config import SESSION_LIFETIME
 from paskia.db.structs import (
     DB,
@@ -23,6 +23,7 @@ from paskia.db.structs import (
     Org,
     OriginEntry,
     Permission,
+    RemoteConfig,
     ResetToken,
     Role,
     Session,
@@ -103,6 +104,7 @@ def update_permission(
         _db.permissions[uuid].scope = scope
         _db.permissions[uuid].display_name = display_name
         _db.permissions[uuid].domain = domain
+        syncfeed.emit("permissions", str(uuid), _db.permissions[uuid])
 
 
 def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
@@ -155,6 +157,7 @@ def update_org_name(
         raise ValueError(f"Organization {uuid} not found")
     with _transaction("admin:update_org_name", ctx):
         _db.orgs[uuid].display_name = display_name
+        syncfeed.emit("orgs", str(uuid), _db.orgs[uuid])
 
 
 def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
@@ -180,6 +183,9 @@ def add_permission_to_org(
 
     with _transaction("admin:add_permission_to_org", ctx):
         _db.permissions[permission_uuid].orgs[org_uuid] = True
+        syncfeed.emit(
+            "permissions", str(permission_uuid), _db.permissions[permission_uuid]
+        )
 
 
 def remove_permission_from_org(
@@ -197,6 +203,9 @@ def remove_permission_from_org(
 
     with _transaction("admin:remove_permission_from_org", ctx):
         _db.permissions[permission_uuid].orgs.pop(org_uuid, None)
+        syncfeed.emit(
+            "permissions", str(permission_uuid), _db.permissions[permission_uuid]
+        )
 
 
 def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
@@ -220,6 +229,7 @@ def update_role_name(
         raise ValueError(f"Role {uuid} not found")
     with _transaction("admin:update_role_name", ctx):
         _db.roles[uuid].display_name = display_name
+        syncfeed.emit("roles", str(uuid), _db.roles[uuid])
 
 
 def add_permission_to_role(
@@ -235,6 +245,7 @@ def add_permission_to_role(
         raise ValueError(f"Permission {permission_uuid} not found")
     with _transaction("admin:add_permission_to_role", ctx):
         _db.roles[role_uuid].permissions[permission_uuid] = True
+        syncfeed.emit("roles", str(role_uuid), _db.roles[role_uuid])
 
 
 def remove_permission_from_role(
@@ -248,6 +259,7 @@ def remove_permission_from_role(
         raise ValueError(f"Role {role_uuid} not found")
     with _transaction("admin:remove_permission_from_role", ctx):
         _db.roles[role_uuid].permissions.pop(permission_uuid, None)
+        syncfeed.emit("roles", str(role_uuid), _db.roles[role_uuid])
 
 
 def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
@@ -302,6 +314,7 @@ def update_user_display_name(
             slug = slugify_name(display_name)
             if slug and not is_username_taken(slug, exclude_uuid=uuid):
                 user.preferred_username = slug
+        syncfeed.emit("users", str(uuid), user)
 
 
 def update_user_info(
@@ -380,6 +393,7 @@ def update_user_info(
             user.preferred_username = preferred_username
         if telephone is not _UNSET:
             user.telephone = telephone
+        syncfeed.emit("users", str(uuid), user)
 
 
 def update_user_role(
@@ -395,6 +409,7 @@ def update_user_role(
         raise ValueError(f"Role {role_uuid} not found")
     with _transaction("admin:update_user_role", ctx):
         _db.users[uuid].role_uuid = role_uuid
+        syncfeed.emit("users", str(uuid), _db.users[uuid])
 
 
 def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
@@ -429,6 +444,7 @@ def update_credential_sign_count(
         _db.credentials[uuid].sign_count = sign_count
         if last_used:
             _db.credentials[uuid].last_used = last_used
+        syncfeed.emit("credentials", str(uuid), _db.credentials[uuid])
 
 
 def delete_credential(
@@ -476,6 +492,7 @@ def update_session(
             s.validated = validated
         if issuer is not None:
             s.issuer = issuer
+        syncfeed.emit("sessions", key, s)
 
 
 def delete_session(
@@ -598,6 +615,9 @@ def login(
         # Update credential
         _db.credentials[credential_uuid].sign_count = sign_count
         _db.credentials[credential_uuid].last_used = now
+        syncfeed.emit(
+            "credentials", str(credential_uuid), _db.credentials[credential_uuid]
+        )
     return token
 
 
@@ -625,6 +645,9 @@ def oidc_login(
         # Update credential
         _db.credentials[credential_uuid].sign_count = sign_count
         _db.credentials[credential_uuid].last_used = now
+        syncfeed.emit(
+            "credentials", str(credential_uuid), _db.credentials[credential_uuid]
+        )
 
 
 def create_credential_session(
@@ -714,9 +737,10 @@ def update_domain(
     *,
     rp_name: str | None,
     origins: dict[str, bool | OriginEntry],
+    remote: RemoteConfig | None = None,
     ctx: SessionContext | None = None,
 ) -> None:
-    """Replace a domain's rp_name and origins table (wholesale).
+    """Replace a domain's rp_name, origins table and remote (wholesale).
 
     The rp-id itself is immutable: credentials are stamped with it, so
     changing it would orphan them — delete and recreate the domain instead.
@@ -728,6 +752,7 @@ def update_domain(
     with _transaction("admin:update_domain", ctx):
         domain.rp_name = rp_name
         domain.origins = origins
+        domain.remote = remote
 
 
 def delete_domain(rp_id: str, *, ctx: SessionContext | None = None) -> None:

paskia/db/structs.py

@@ -9,7 +9,7 @@ from uuid import UUID
 import msgspec
 import uuid7
 
-from paskia import db
+from paskia import db, syncfeed
 from paskia.util import passphrase as passphrase_util
 from paskia.util.crypto import hash_secret
 
@@ -51,6 +51,7 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
     def store(self) -> None:
         """Store this permission in the database. Must be called inside a transaction."""
         db.data().permissions[self.uuid] = self
+        syncfeed.emit("permissions", str(self.uuid), self)
 
     def delete(self) -> None:
         """Delete this permission and remove it from all roles.
@@ -59,8 +60,10 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
         """
         _data = db.data()
         for role in _data.roles.values():
-            role.permissions.pop(self.uuid, None)
+            if role.permissions.pop(self.uuid, None) is not None:
+                syncfeed.emit("roles", str(role.uuid), role)
         del _data.permissions[self.uuid]
+        syncfeed.emit("permissions", str(self.uuid), None)
 
     @classmethod
     def create(
@@ -103,6 +106,7 @@ class Org(msgspec.Struct, dict=True):
     def store(self) -> None:
         """Store this organization in the database. Must be called inside a transaction."""
         db.data().orgs[self.uuid] = self
+        syncfeed.emit("orgs", str(self.uuid), self)
 
     def delete(self) -> None:
         """Delete this org and cascade to roles, users. Remove from permissions.
@@ -111,12 +115,16 @@ class Org(msgspec.Struct, dict=True):
         """
         _data = db.data()
         for p in _data.permissions.values():
-            p.orgs.pop(self.uuid, None)
+            if p.orgs.pop(self.uuid, None) is not None:
+                syncfeed.emit("permissions", str(p.uuid), p)
         for role in self.roles:
             for user in role.users:
                 del _data.users[user.uuid]
+                syncfeed.emit("users", str(user.uuid), None)
             del _data.roles[role.uuid]
+            syncfeed.emit("roles", str(role.uuid), None)
         del _data.orgs[self.uuid]
+        syncfeed.emit("orgs", str(self.uuid), None)
 
     @classmethod
     def create(cls, display_name: str, created_at: datetime | None = None) -> Org:
@@ -170,10 +178,12 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
     def store(self) -> None:
         """Store this role in the database. Must be called inside a transaction."""
         db.data().roles[self.uuid] = self
+        syncfeed.emit("roles", str(self.uuid), self)
 
     def delete(self) -> None:
         """Delete this role from the database. Must be called inside a transaction."""
         del db.data().roles[self.uuid]
+        syncfeed.emit("roles", str(self.uuid), None)
 
     @classmethod
     def create(
@@ -254,6 +264,7 @@ class User(msgspec.Struct, dict=True, omit_defaults=True, kw_only=True):
     def store(self) -> None:
         """Store this user in the database. Must be called inside a transaction."""
         db.data().users[self.uuid] = self
+        syncfeed.emit("users", str(self.uuid), self)
 
     def delete(self) -> None:
         """Delete this user and cascade to credentials, sessions, reset tokens.
@@ -263,11 +274,14 @@ class User(msgspec.Struct, dict=True, omit_defaults=True, kw_only=True):
         _data = db.data()
         for cred in self.credentials:
             del _data.credentials[cred.uuid]
+            syncfeed.emit("credentials", str(cred.uuid), None)
         for sess in self.sessions:
             del _data.sessions[sess.key]
+            syncfeed.emit("sessions", sess.key, None)
         for token in self.reset_tokens:
             del _data.reset_tokens[token.key]
         del _data.users[self.uuid]
+        syncfeed.emit("users", str(self.uuid), None)
 
     @classmethod
     def create(
@@ -331,6 +345,7 @@ class Credential(msgspec.Struct, dict=True):
     def store(self) -> None:
         """Store this credential in the database. Must be called inside a transaction."""
         db.data().credentials[self.uuid] = self
+        syncfeed.emit("credentials", str(self.uuid), self)
 
     def delete(self) -> None:
         """Delete this credential and all its sessions.
@@ -340,7 +355,9 @@ class Credential(msgspec.Struct, dict=True):
         _data = db.data()
         for sess in self.sessions:
             del _data.sessions[sess.key]
+            syncfeed.emit("sessions", sess.key, None)
         del _data.credentials[self.uuid]
+        syncfeed.emit("credentials", str(self.uuid), None)
 
     @classmethod
     def create(
@@ -418,10 +435,13 @@ class Session(msgspec.Struct, dict=True, omit_defaults=True):
         _data.sessions[self.key] = self
         _data.users[self.user_uuid].last_seen = last_seen
         _data.users[self.user_uuid].visits += 1
+        syncfeed.emit("sessions", self.key, self)
+        syncfeed.emit("users", str(self.user_uuid), _data.users[self.user_uuid])
 
     def delete(self) -> None:
         """Delete this session from the database. Must be called inside a transaction."""
         del db.data().sessions[self.key]
+        syncfeed.emit("sessions", self.key, None)
 
     @classmethod
     def create(
@@ -622,6 +642,21 @@ class OriginEntry(msgspec.Struct, omit_defaults=True):
     auth_host: bool = False  # This site hosts the account/admin interface
 
 
+class RemoteConfig(msgspec.Struct, omit_defaults=True):
+    """Upstream paskia instance backing a remote (satellite-served) domain.
+
+    The satellite keeps a RAM-only read replica of the remote's tables and
+    answers session-dependent reads locally; mutations are forwarded. The
+    token authenticates the sync channel (the remote reads accepted tokens
+    from its PASKIA_SYNC_TOKENS environment, never from its database).
+    """
+
+    url: str  # e.g. "https://auth.example.com"
+    token: str = ""
+    cache_ttl: int = 60  # staleness bound (seconds) while the sync channel is down
+    refresh_interval: int = 300  # full re-sync cadence (seconds)
+
+
 class DomainConfig(msgspec.Struct, omit_defaults=True):
     """Configuration for one domain (one WebAuthn rp-id).
 
@@ -641,6 +676,7 @@ class DomainConfig(msgspec.Struct, omit_defaults=True):
 
     rp_name: str | None = None
     origins: dict[str, bool | OriginEntry] = {}
+    remote: RemoteConfig | None = None
 
 
 class Config(msgspec.Struct, omit_defaults=True):
@@ -728,17 +764,21 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
         if s.host != host:
             return None
 
+        # Look up via this instance's own tables: a DB must be
+        # self-contained so that read replicas work unchanged.
         try:
-            user = s.user
+            user = self.users[s.user_uuid]
-            role = user.role
+            role = self.roles[user.role_uuid]
-            org = role.org
+            org = self.orgs[role.org_uuid]
-            credential = s.credential
+            credential = self.credentials[s.credential_uuid]
         except KeyError:
             return None
 
         # Effective permissions: role's permissions that the org can grant,
         # filtered by domain restriction
-        org_perm_uuids = {p.uuid for p in org.permissions}
+        org_perm_uuids = {
+            p.uuid for p in self.permissions.values() if org.uuid in p.orgs
+        }
 
         effective_perms = []
         for perm_uuid in role.permission_set:

paskia/domains.py

@@ -14,13 +14,14 @@ domain are in-domain, entries outside it are related.
 
 from __future__ import annotations
 
+import asyncio
 import contextvars
 import logging
 import os
 
 from fastapi_vue.hostutil import parse_endpoints
 
-from paskia.db.structs import Config, DomainConfig, OriginEntry
+from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig
 from paskia.sansio import Passkey
 from paskia.util import hostutil
 from paskia.util.constants import DEFAULT_PORT
@@ -101,6 +102,11 @@ class Domain:
     def rp_name(self) -> str:
         return self.passkey.rp_name
 
+    @property
+    def remote(self) -> RemoteConfig | None:
+        """Upstream config when this domain is served as a satellite."""
+        return self.config.remote
+
     @property
     def own_auth_host(self) -> str | None:
         """This domain's own auth host as host[:port], if configured."""
@@ -207,6 +213,10 @@ def validate_config(
 
     for rp_id, domain in config.domains.items():
         hostutil.validate_rp_id(rp_id)
+        if domain.remote is not None and not domain.remote.url.startswith(
+            ("https://", "http://")
+        ):
+            raise ValueError(f"Domain '{rp_id}': remote URL must be an http(s) URL")
 
         domain_auth_host: str | None = None
         related_count = 0
@@ -273,6 +283,11 @@ def validate_config(
                 f"Domain '{rp_id}' has {related_count} related origins "
                 f"(maximum {related_origin_cap})"
             )
+        if domain.remote is not None and domain_auth_host is None:
+            raise ValueError(
+                f"Domain '{rp_id}' is remote — it must mark an auth host "
+                "(profile, admin and sign-in pages live there)"
+            )
 
     rp_ids = set(config.domains)
     for hn, owner in auth_hosts.items():
@@ -284,6 +299,21 @@ def validate_config(
                 f"domain '{related_hosts[hn]}'"
             )
 
+    # Several domains may share one remote instance (and its replica), but
+    # then its settings must be identical — a conflict would otherwise be
+    # resolved silently and arbitrarily.
+    remotes: dict[str, tuple[str, RemoteConfig]] = {}
+    for rp_id, domain in config.domains.items():
+        if domain.remote is None:
+            continue
+        prior = remotes.setdefault(domain.remote.url, (rp_id, domain.remote))
+        if prior[1] != domain.remote:
+            raise ValueError(
+                f"Domains '{prior[0]}' and '{rp_id}' share remote "
+                f"'{domain.remote.url}' with conflicting settings — a shared "
+                "remote must have one configuration"
+            )
+
 
 def sanitize_config(
     config: Config, *, related_origin_cap: int = DEFAULT_RELATED_ORIGIN_CAP
@@ -302,6 +332,7 @@ def sanitize_config(
         warnings.append(msg)
 
     domains: dict[str, DomainConfig] = {}
+    remotes_seen: dict[str, RemoteConfig] = {}
     for rp_id, domain in config.domains.items():
         try:
             hostutil.validate_rp_id(rp_id)
@@ -363,6 +394,28 @@ def sanitize_config(
                     auth_seen = True
             origins[key] = props
 
+        if domain.remote is not None:
+            if not domain.remote.url.startswith(("https://", "http://")):
+                warn(f"Domain '{rp_id}': invalid remote URL — remote dropped")
+                remote = None
+            else:
+                remote = domain.remote
+                prior = remotes_seen.setdefault(remote.url, remote)
+                if prior != remote:
+                    warn(
+                        f"Domain '{rp_id}': remote '{remote.url}' settings "
+                        "conflict with another domain's — aligned to the "
+                        "first configuration"
+                    )
+                    remote = prior
+                if not auth_seen:
+                    warn(
+                        f"Domain '{rp_id}': remote domain without an auth host — "
+                        "profile, admin and sign-in pages have nowhere to live"
+                    )
+        else:
+            remote = None
+
         related = sorted(k for k in origins if is_related_key(rp_id, k))
         if len(related) > related_origin_cap:
             warn(
@@ -372,7 +425,9 @@ def sanitize_config(
             for key in related[related_origin_cap:]:
                 del origins[key]
 
-        domains[rp_id] = DomainConfig(rp_name=domain.rp_name, origins=origins)
+        domains[rp_id] = DomainConfig(
+            rp_name=domain.rp_name, origins=origins, remote=remote
+        )
 
     if not domains:
         raise ValueError("No servable domain in the stored configuration")
@@ -461,6 +516,46 @@ def _derive_site(
 
 _registry: DomainRegistry | None = None
 _listen: list[str] | None = None
+_rebuild_listeners: list = []
+_rebuild_tasks: set[asyncio.Task] = set()
+
+
+def add_rebuild_listener(fn) -> None:
+    """Register fn(registry), called after every init_registry rebuild."""
+    _rebuild_listeners.append(fn)
+
+
+def remove_rebuild_listener(fn) -> None:
+    if fn in _rebuild_listeners:
+        _rebuild_listeners.remove(fn)
+
+
+def _rebuild_done(task: asyncio.Task) -> None:
+    _rebuild_tasks.discard(task)
+    if not task.cancelled() and (exc := task.exception()) is not None:
+        logger.error("Registry rebuild listener failed: %s", exc)
+
+
+def _notify_rebuild(reg: DomainRegistry) -> None:
+    """Run rebuild listeners; schedule coroutines on the running loop.
+
+    Every serving context (lifespan, admin rebuild, tests) runs a loop.
+    Task references are held until completion so listeners are neither
+    garbage-collected mid-run nor silently failing.
+    """
+    for fn in _rebuild_listeners:
+        result = fn(reg)
+        if not asyncio.iscoroutine(result):
+            continue
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            result.close()
+            logger.warning("Rebuild listener %r skipped: no running event loop", fn)
+            continue
+        task = loop.create_task(result)
+        _rebuild_tasks.add(task)
+        task.add_done_callback(_rebuild_done)
 
 
 def configure(*, listen: list[str] | None = None) -> None:
@@ -501,6 +596,7 @@ def init_registry(config: Config) -> DomainRegistry:
     """Build and install the global registry from a combined configuration."""
     global _registry
     _registry = build(config)
+    _notify_rebuild(_registry)
     return _registry
 
 

paskia/fastapi/admin/domains.py

@@ -12,7 +12,7 @@ immediately.
 from fastapi import Body, FastAPI, Request
 
 from paskia import db, domains
-from paskia.db.structs import Config, DomainConfig, OriginEntry
+from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig
 from paskia.fastapi import authz
 from paskia.fastapi.admin.errors import install_error_handlers
 from paskia.fastapi.response import MsgspecResponse
@@ -27,6 +27,14 @@ install_error_handlers(app)
 
 
 def _domain_to_api(domain: domains.Domain) -> ApiDomain:
+    remote = domain.config.remote
+    if remote is not None:
+        # The sync token is a bearer secret: never echoed back
+        remote = RemoteConfig(
+            url=remote.url,
+            cache_ttl=remote.cache_ttl,
+            refresh_interval=remote.refresh_interval,
+        )
     return ApiDomain(
         rp_id=domain.rp_id,
         rp_name=domain.rp_name,
@@ -34,6 +42,28 @@ def _domain_to_api(domain: domains.Domain) -> ApiDomain:
         site_url=domain.site_url,
         auth_site_url=domain.auth_site_url,
         auth_host=domain.own_auth_host,
+        remote=remote,
+    )
+
+
+def _normalize_remote(
+    value, existing: RemoteConfig | None = None
+) -> RemoteConfig | None:
+    """Parse a remote object from the admin UI (raises on malformed).
+
+    An absent/empty token keeps the previously stored one — the token is
+    write-only over the API.
+    """
+    if value is None:
+        return None
+    if not isinstance(value, dict) or not isinstance(value.get("url"), str):
+        raise ValueError("remote must be an object with a url")
+    token = str(value.get("token") or "") or (existing.token if existing else "")
+    return RemoteConfig(
+        url=value["url"].rstrip("/"),
+        token=token,
+        cache_ttl=max(1, int(value.get("cache_ttl") or 60)),
+        refresh_interval=max(30, int(value.get("refresh_interval") or 300)),
     )
 
 
@@ -123,6 +153,7 @@ async def admin_create_domain(
     new = DomainConfig(
         rp_name=(payload.get("rp_name") or "").strip() or None,
         origins=_normalize_origins_map(payload.get("origins")),
+        remote=_normalize_remote(payload.get("remote")),
     )
 
     config = db.data().config
@@ -156,9 +187,15 @@ async def admin_update_domain(
     if rp_id not in config.domains:
         raise ValueError(f"Domain {rp_id} not found")
 
+    current_remote = config.domains[rp_id].remote
     updated = DomainConfig(
         rp_name=(payload.get("rp_name") or "").strip() or None,
         origins=_normalize_origins_map(payload.get("origins")),
+        remote=(
+            _normalize_remote(payload["remote"], existing=current_remote)
+            if "remote" in payload
+            else current_remote
+        ),
     )
     would_be = Config(
         domains={k: updated if k == rp_id else v for k, v in config.domains.items()},
@@ -171,6 +208,7 @@ async def admin_update_domain(
         rp_id,
         rp_name=updated.rp_name,
         origins=updated.origins,
+        remote=updated.remote,
         ctx=ctx,
     )
     _rebuild_registry()

paskia/fastapi/api.py

@@ -14,7 +14,7 @@ from fastapi import (
 from fastapi.responses import JSONResponse
 from fastapi.security import HTTPBearer
 
-from paskia import authcode, db
+from paskia import authcode, db, satellite
 from paskia._version import __version__
 from paskia.authsession import EXPIRES, get_reset, session_ctx
 from paskia.domains import current_domain
@@ -122,8 +122,9 @@ async def validate_token(
     if auth and renew:
         consumed = datetime.now(UTC) - ctx.session.validated
         if not timedelta(0) < consumed < _REFRESH_INTERVAL:
-            db.update_session(
+            satellite.refresh_session(
                 ctx.session.key,
+                request.headers.get("host"),
                 ip=get_client_ip(request),
                 user_agent=request.headers.get("user-agent"),
                 validated=datetime.now(UTC),
@@ -162,16 +163,16 @@ async def check_user(
 
     No session cookie is read or written. Caller authentication is not required.
     """
-    data = db.data()
+    host = hostutil.normalize_host(request.headers.get("host"))
+    data = satellite.store_for_host(host)
     try:
         u = data.users[user_uuid]
-        role = u.role
+        role = data.roles[u.role_uuid]
-        org = role.org
+        org = data.orgs[role.org_uuid]
     except KeyError:
         raise HTTPException(status_code=404, detail="User not found")
 
-    host = hostutil.normalize_host(request.headers.get("host"))
+    org_perm_uuids = {p.uuid for p in data.permissions.values() if org.uuid in p.orgs}
-    org_perm_uuids = {p.uuid for p in org.permissions}
 
     effective_perms = []
     for perm_uuid in role.permission_set:
@@ -212,7 +213,7 @@ def _remote_headers(ctx) -> dict[str, str]:
         "Remote-Session-Expires": (
             (ctx.session.validated + EXPIRES).isoformat().replace("+00:00", "Z")
         ),
-        "Remote-Credential": str(ctx.session.credential),
+        "Remote-Credential": str(ctx.credential.uuid),
     }
 
 
@@ -311,6 +312,7 @@ async def get_settings():
             auth_site_url=domain.auth_site_url,
             session_cookie=AUTH_COOKIE_NAME,
             version=__version__,
+            remote=domain.remote is not None,
         ),
         headers={"Access-Control-Allow-Origin": "*", "Vary": "Origin"},
     )
@@ -351,8 +353,10 @@ async def api_user_info(
 
 
 @app.get("/token-info")
-async def token_info(credentials=Depends(bearer_auth)):
+async def token_info(request: Request, credentials=Depends(bearer_auth)):
     """Get reset/device-add token info. Pass token via Bearer header."""
+    if (proxied := await satellite.forward_request(request)) is not None:
+        return proxied
     if not credentials or not credentials.credentials:
         raise HTTPException(401, "Bearer token required")
     token = credentials.credentials
@@ -375,6 +379,10 @@ async def token_info(credentials=Depends(bearer_auth)):
 
 @app.post("/logout")
 async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE):
+    if (proxied := await satellite.forward_request(request)) is not None:
+        if auth and proxied.status_code == 200:
+            satellite.evict_session(auth, request.headers.get("host"))
+        return proxied
     if not auth:
         return {"message": "Already logged out"}
     host = request.headers.get("host")
@@ -399,6 +407,11 @@ async def api_set_session(
     if not auth or not auth.credentials:
         raise HTTPException(400, "Bearer token required")
 
+    if (proxied := await satellite.forward_request(request)) is not None:
+        # The exchange code lives in the remote's RAM; redeem it there. The
+        # session itself reaches the replica via the sync channel.
+        return proxied
+
     host = hostutil.normalize_host(request.headers.get("host", ""))
     if not host:
         raise HTTPException(400, "Host header required")

paskia/fastapi/dispatch.py

@@ -63,6 +63,11 @@ class DispatchMiddleware:
         host = _header(scope, "host")
         host_domain = registry.resolve(host)
         if host_domain is None:
+            # The sync endpoint is server-to-server and token-gated: the
+            # satellite may reach us via an address outside our domains.
+            if scope.get("path") == "/auth/api/sync/ws" and registry.domains:
+                await self._dispatch(scope, receive, send, registry.domains[0])
+                return
             await send({"type": "websocket.close", "code": _WS_CLOSE_POLICY_VIOLATION})
             return
 

paskia/fastapi/mainapp.py

@@ -7,11 +7,11 @@ from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse, RedirectResponse
 from fastapi_vue import env
 
-from paskia import authcode, db, domains, remoteauth
+from paskia import authcode, db, domains, remoteauth, satellite
 from paskia.bootstrap import bootstrap_if_needed
 from paskia.db.background import start_background, stop_background
 from paskia.db.lifecycle import kanta
-from paskia.fastapi import admin, api, auth_host, oid, ws
+from paskia.fastapi import admin, api, auth_host, oid, sync, ws
 from paskia.fastapi.admin.adminapp import adminapp
 from paskia.fastapi.dispatch import DispatchMiddleware
 
@@ -50,6 +50,7 @@ async def lifespan(app: FastAPI):  # pragma: no cover - startup path
             domains.init_registry(db.data().config)
             await remoteauth.init()
             await authcode.start()
+            await satellite.manager.start()
         except ValueError as e:
             logging.error(f"⚠️ {e}")
             # Re-raise to fail fast
@@ -60,6 +61,7 @@ async def lifespan(app: FastAPI):  # pragma: no cover - startup path
         await start_background()
         yield
         await stop_background()
+        await satellite.manager.stop()
         await authcode.stop()
 
 
@@ -83,6 +85,7 @@ app.middleware("http")(auth_host.redirect_middleware)
 app.add_middleware(DispatchMiddleware)
 
 app.mount("/auth/api/admin/", admin.app)
+app.mount("/auth/api/sync", sync.app)
 app.mount("/auth/api/", api.app)
 app.mount("/auth/ws/", ws.app)
 app.mount("/auth/oidc/", oid.app)

paskia/fastapi/oid.py

@@ -20,7 +20,7 @@ from fastapi import Depends, FastAPI, Form, HTTPException, Request
 from fastapi.responses import JSONResponse
 from fastapi.security import HTTPBearer
 
-from paskia import authcode, db
+from paskia import authcode, db, satellite
 from paskia.db.structs import OIDC, Session
 from paskia.util import avatar, oidjwt
 from paskia.util.crypto import hash_secret
@@ -30,6 +30,14 @@ _logger = logging.getLogger(__name__)
 app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
 
 
+@app.middleware("http")
+async def proxy_remote_domain(request: Request, call_next):
+    """OIDC key material and sessions stay on the remote; proxy everything."""
+    if (proxied := await satellite.forward_request(request)) is not None:
+        return proxied
+    return await call_next(request)
+
+
 def _provider() -> OIDC:
     """Return the instance-global OIDC provider state."""
     return db.data().oidc

paskia/fastapi/sync.py

@@ -0,0 +1,96 @@
+"""Sync WebSocket endpoint: serves snapshots and live events to satellites.
+
+Token-gated via PASKIA_SYNC_TOKENS (env); closed when unset. All state is
+RAM-only (syncfeed); the database schema is untouched. Protocol: snapshot
+chunks per table, `ready`, then live upsert/delete events; the client sends
+session_refresh write-backs. Reconnects always restart from a snapshot.
+"""
+
+import asyncio
+import logging
+from datetime import datetime
+
+import msgspec
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+
+from paskia import db, syncfeed
+
+_logger = logging.getLogger(__name__)
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+
+async def _send(ws: WebSocket, message: dict) -> None:
+    await ws.send_bytes(syncfeed.encode(message))
+
+
+async def _apply_client_message(message: dict) -> None:
+    """Satellite write-behind: session refresh (validated/ip/user-agent)."""
+    if message.get("type") != "session_refresh":
+        return
+    key = message.get("key") or ""
+    session = db.data().sessions.get(key)
+    if session is None:
+        return
+    try:
+        validated = msgspec.convert(message.get("validated"), datetime)
+    except msgspec.ValidationError:
+        return
+    db.update_session(
+        key,
+        ip=message.get("ip") or None,
+        user_agent=message.get("user_agent") or None,
+        validated=validated,
+    )
+
+
+@app.websocket("/ws")
+async def sync_websocket(ws: WebSocket):
+    tokens = syncfeed.tokens_from_env()
+    auth = ws.headers.get("authorization", "")
+    if not tokens or auth.removeprefix("Bearer ").strip() not in tokens:
+        await ws.close(code=1008)
+        return
+    await ws.accept()
+
+    queue = syncfeed.subscribe()
+    try:
+        data = db.data()
+        for table in syncfeed.TABLES:
+            await _send(
+                ws,
+                {
+                    "type": "snapshot",
+                    "table": table,
+                    "items": [
+                        [str(key), msgspec.to_builtins(obj)]
+                        for key, obj in getattr(data, table).items()
+                    ],
+                },
+            )
+        await _send(ws, {"type": "ready"})
+
+        sender = asyncio.create_task(_pump(ws, queue))
+        try:
+            while True:
+                await _apply_client_message(
+                    msgspec.json.decode(await ws.receive_bytes())
+                )
+        finally:
+            sender.cancel()
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        _logger.exception("Sync WebSocket failed")
+    finally:
+        syncfeed.unsubscribe(queue)
+
+
+async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
+    try:
+        while True:
+            await _send(ws, await queue.get())
+    except WebSocketDisconnect, RuntimeError, asyncio.CancelledError:
+        pass
+    except Exception:
+        _logger.debug("Sync event pump ended", exc_info=True)

paskia/fastapi/user.py

@@ -11,7 +11,7 @@ from fastapi import (
 )
 from fastapi.responses import FileResponse, JSONResponse
 
-from paskia import db
+from paskia import db, satellite
 from paskia.authsession import (
     delete_credential,
     expires,
@@ -27,6 +27,19 @@ from paskia.util.apistructs import ApiCreateLinkResponse
 app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
 
 
+@app.middleware("http")
+async def proxy_remote_domain(request: Request, call_next):
+    """Profile and avatar data live on the remote for remote domains.
+
+    A remote domain's auth host normally resolves to the remote itself;
+    if traffic for it reaches the satellite anyway (misrouted DNS), the
+    satellite forwards rather than writing to its local database.
+    """
+    if (proxied := await satellite.forward_request(request)) is not None:
+        return proxied
+    return await call_next(request)
+
+
 def _can_manage_avatar(ctx, target_user) -> bool:
     if ctx.user.uuid == target_user.uuid:
         return True

paskia/satellite.py

@@ -0,0 +1,362 @@
+"""Satellite side of remote domains: RAM-only replicas + host dispatch.
+
+Domains configured with ``DomainConfig.remote`` are backed by a remote
+paskia instance. This module owns the whole feature: it resolves which
+store serves a request host (local DB or the remote's read replica),
+dispatches session writes (refresh write-behind, logout eviction), and
+forwards requests the satellite cannot answer (exchange-code redemption,
+OIDC, reset tokens) to the remote.
+
+A replica is a plain DB instance, never persisted, fed by a sync
+WebSocket (snapshot on connect, then live events) and swept for
+expired sessions locally. While the channel is down the replica stays trusted for the
+domain's cache_ttl, then reads fail with RemoteUnavailable (fail-closed;
+a large cache_ttl gives fail-open behavior bounded by session expiry).
+"""
+
+import asyncio
+import contextlib
+import logging
+import time
+from datetime import UTC, datetime
+from uuid import UUID
+
+import httpx
+import msgspec
+import websockets
+from fastapi import HTTPException, Request, Response
+
+from paskia import db, domains
+from paskia.config import SESSION_LIFETIME
+from paskia.db.structs import (
+    DB,
+    Credential,
+    Org,
+    Permission,
+    RemoteConfig,
+    Role,
+    Session,
+    User,
+)
+from paskia.util.crypto import hash_secret
+
+_logger = logging.getLogger(__name__)
+
+_TABLES = {
+    "permissions": (Permission, True),
+    "orgs": (Org, True),
+    "roles": (Role, True),
+    "users": (User, True),
+    "credentials": (Credential, True),
+    "sessions": (Session, False),
+}
+
+_RECONNECT_DELAY = 5
+_SWEEP_INTERVAL = 60
+
+
+class RemoteReplica:
+    """One remote instance's replica, its sync client and write-behind queue."""
+
+    def __init__(self, remote: RemoteConfig):
+        self.remote = remote
+        self.db = DB()
+        self.last_contact = 0.0  # monotonic time the feed last went down
+        self.connected = False
+        self._pending_refresh: dict[str, dict] = {}
+        self._refresh_signal = asyncio.Event()
+        self._task: asyncio.Task | None = None
+        self._sweeper: asyncio.Task | None = None
+        self._stopped = True
+
+    def available(self) -> bool:
+        """Synced, and connected now or within cache_ttl of the disconnect."""
+        if not self.last_contact:
+            return False
+        return self.connected or (
+            time.monotonic() - self.last_contact <= self.remote.cache_ttl
+        )
+
+    def refresh_session(
+        self, key: str, validated: datetime, ip: str | None, user_agent: str | None
+    ) -> None:
+        """Apply a /validate refresh locally and queue it for the remote."""
+        session = self.db.sessions.get(key)
+        if session is not None:
+            session.validated = validated
+            if ip is not None:
+                session.ip = ip
+            if user_agent is not None:
+                session.user_agent = user_agent
+        self._pending_refresh[key] = {
+            "type": "session_refresh",
+            "key": key,
+            "validated": msgspec.to_builtins(validated),
+            "ip": ip,
+            "user_agent": user_agent,
+        }
+        self._refresh_signal.set()
+
+    async def start(self) -> None:
+        self._stopped = False
+        self._task = asyncio.create_task(self._run())
+        self._sweeper = asyncio.create_task(self._sweep())
+
+    async def stop(self) -> None:
+        self._stopped = True
+        for task in (self._task, self._sweeper):
+            if task:
+                task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await task
+
+    async def _sweep(self) -> None:
+        while True:
+            await asyncio.sleep(_SWEEP_INTERVAL)
+            limit = datetime.now(UTC) - SESSION_LIFETIME
+            for key in [k for k, s in self.db.sessions.items() if s.validated < limit]:
+                del self.db.sessions[key]
+
+    async def _run(self) -> None:
+        while not self._stopped:
+            try:
+                await self._connect()
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                _logger.info("Sync to %s failed: %s", self.remote.url, e)
+            if self.connected:
+                # The TTL clock starts when the feed goes down, not at the
+                # last message — an idle connection is healthy.
+                self.connected = False
+                self.last_contact = time.monotonic()
+            if not self._stopped:
+                await asyncio.sleep(_RECONNECT_DELAY)
+
+    async def _connect(self) -> None:
+        ws_url = self.remote.url.replace("http", "ws", 1) + "/auth/api/sync/ws"
+        async with websockets.connect(
+            ws_url,
+            additional_headers={"Authorization": f"Bearer {self.remote.token}"},
+            # Prompt dead-peer detection: availability semantics count on it
+            ping_interval=5,
+            ping_timeout=5,
+        ) as ws:
+            sender = asyncio.create_task(self._send_loop(ws))
+            staging: DB | None = None
+            ready_at = 0.0
+            try:
+                while True:
+                    if ready_at:
+                        # Periodic reconnects give full-snapshot reconciliation
+                        remaining = self.remote.refresh_interval - (
+                            time.monotonic() - ready_at
+                        )
+                        if remaining <= 0:
+                            return
+                        try:
+                            message = msgspec.json.decode(
+                                await asyncio.wait_for(ws.recv(), remaining)
+                            )
+                        except TimeoutError:
+                            return  # periodic resync: reconnect for a snapshot
+                    else:
+                        message = msgspec.json.decode(await ws.recv())
+                    mtype = message.get("type")
+                    if mtype == "snapshot":
+                        staging = staging or DB()
+                        for key, fields in message["items"]:
+                            _apply(staging, message["table"], key, fields)
+                    elif mtype == "event":
+                        if staging is not None:
+                            raise ValueError("sync: event before ready")
+                        _apply(
+                            self.db,
+                            message["table"],
+                            message["key"],
+                            message.get("fields"),
+                        )
+                    elif mtype == "ready":
+                        if staging is not None:
+                            self.db = staging
+                            staging = None
+                        self.connected = True
+                        self.last_contact = ready_at = time.monotonic()
+            finally:
+                sender.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await sender
+
+    async def _send_loop(self, ws) -> None:
+        while True:
+            self._refresh_signal.clear()
+            while self._pending_refresh:
+                _, message = self._pending_refresh.popitem()
+                await ws.send(msgspec.json.encode(message))
+            await self._refresh_signal.wait()
+
+
+def _apply(replica: DB, table: str, key: str, fields: dict | None) -> None:
+    """Apply an upsert (fields given) or delete (fields None) to a replica."""
+    cls, uuid_key = _TABLES[table]
+    store = getattr(replica, table)
+    store_key = UUID(key) if uuid_key else key
+    if fields is None:
+        store.pop(store_key, None)
+        return
+    obj = msgspec.convert(fields, cls)
+    if uuid_key:
+        obj.uuid = store_key
+    else:
+        obj.key = key
+    store[store_key] = obj
+
+
+class SatelliteManager:
+    """Replicas keyed by remote URL; domains sharing a remote share one."""
+
+    def __init__(self):
+        self.replicas: dict[str, RemoteReplica] = {}
+
+    async def start(self) -> None:
+        domains.add_rebuild_listener(self.reconcile)
+        await self.reconcile(domains.registry())
+
+    async def stop(self) -> None:
+        domains.remove_rebuild_listener(self.reconcile)
+        for replica in self.replicas.values():
+            await replica.stop()
+        self.replicas.clear()
+        for url in list(_clients):
+            await _drop_client(url)
+
+    async def reconcile(self, registry: domains.DomainRegistry) -> None:
+        """Start/stop replicas to match the configured remote domains."""
+        wanted = {}
+        for domain in registry.domains:
+            if domain.remote is not None:
+                wanted.setdefault(domain.remote.url, domain.remote)
+        for url in list(self.replicas):
+            if url not in wanted:
+                await self.replicas.pop(url).stop()
+                await _drop_client(url)
+        for url, remote in wanted.items():
+            replica = self.replicas.get(url)
+            if replica is None or replica.remote != remote:
+                if replica is not None:
+                    await replica.stop()
+                replica = RemoteReplica(remote)
+                self.replicas[url] = replica
+                await replica.start()
+
+
+manager = SatelliteManager()
+
+
+# -------------------------------------------------------------------------
+# Host-keyed dispatch: the only interface the rest of the app uses
+# -------------------------------------------------------------------------
+
+
+def replica_for_host(host: str | None) -> RemoteReplica | None:
+    """The replica serving this host, or None for locally served hosts."""
+    domain = domains.registry().resolve(host)
+    if domain is None or domain.remote is None:
+        return None
+    return manager.replicas.get(domain.remote.url)
+
+
+def store_for_host(host: str | None) -> DB:
+    """The data store to read for a request host: the local database, or
+    the replica of the remote backing the host's domain."""
+    replica = replica_for_host(host)
+    if replica is None:
+        return db.data()
+    if not replica.available():
+        raise HTTPException(503, "Remote authentication service unavailable")
+    return replica.db
+
+
+def refresh_session(
+    key: str,
+    host: str | None,
+    ip: str | None,
+    user_agent: str | None,
+    validated: datetime,
+    ctx=None,
+):
+    """/validate refresh: write-behind for remote domains, else local DB."""
+    replica = replica_for_host(host)
+    if replica is not None:
+        replica.refresh_session(key, validated, ip, user_agent)
+    else:
+        db.update_session(
+            key, ip=ip, user_agent=user_agent, validated=validated, ctx=ctx
+        )
+
+
+def evict_session(auth: str, host: str | None) -> None:
+    """Drop a session from the replica (its remote deletion arrives via sync)."""
+    replica = replica_for_host(host)
+    if replica is not None:
+        replica.db.sessions.pop(hash_secret("cookie", auth), None)
+
+
+_TIMEOUT = httpx.Timeout(15.0, connect=5.0)
+
+_HOP_BY_HOP = {
+    "connection",
+    "keep-alive",
+    "proxy-authenticate",
+    "proxy-authorization",
+    "te",
+    "trailers",
+    "transfer-encoding",
+    "upgrade",
+    "content-length",
+    "accept-encoding",
+    "content-encoding",
+}
+
+_clients: dict[str, httpx.AsyncClient] = {}
+
+
+async def _drop_client(url: str) -> None:
+    """Close and forget the forwarding client for a remote URL."""
+    client = _clients.pop(url, None)
+    if client is not None:
+        await client.aclose()
+
+
+async def forward_request(request: Request) -> Response | None:
+    """Forward the request to its domain's remote, or None when local.
+
+    The original Host header is preserved so the remote dispatches to the
+    same domain (sessions are host-bound). The user's cookie authenticates
+    the forwarded call; the satellite needs no credentials of its own.
+    """
+    domain = domains.registry().resolve(request.headers.get("host"))
+    if domain is None or domain.remote is None:
+        return None
+    url = domain.remote.url
+    client = _clients.get(url)
+    if client is None:
+        client = _clients[url] = httpx.AsyncClient(base_url=url, timeout=_TIMEOUT)
+    try:
+        upstream = await client.request(
+            request.method,
+            request.url.path,
+            params=request.url.query,
+            content=await request.body(),
+            headers={
+                k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP
+            },
+        )
+    except httpx.TransportError as e:
+        raise HTTPException(503, "Remote authentication service unavailable") from e
+    response = Response(content=upstream.content, status_code=upstream.status_code)
+    # Raw headers to preserve repeated Set-Cookie
+    response.raw_headers = [
+        (k, v) for k, v in upstream.headers.raw if k.decode().lower() not in _HOP_BY_HOP
+    ]
+    return response

paskia/syncfeed.py

@@ -0,0 +1,62 @@
+"""RAM-only change feed letting satellite instances mirror this server.
+
+Nothing here touches the database file: committed mutations are pushed to
+connected satellites over the sync WebSocket (fastapi/sync.py). Satellites
+authenticate with a token from the PASKIA_SYNC_TOKENS environment variable
+(comma-separated); with the variable unset the sync endpoint stays closed.
+
+There is deliberately no replay log: snapshots are small, so a reconnecting
+satellite simply takes a fresh one.
+"""
+
+import asyncio
+import os
+
+import msgspec
+
+# Tables mirrored by satellites (reset tokens, OIDC data and domain config
+# are instance-local and never replicated).
+TABLES = ("permissions", "orgs", "roles", "users", "credentials", "sessions")
+
+_subscribers: set[asyncio.Queue] = set()
+
+
+def emit(table: str, key: str, obj) -> None:
+    """Publish an upsert (obj given) or delete (obj None) to subscribers."""
+    if not _subscribers:
+        return
+    event = {
+        "type": "event",
+        "table": table,
+        "key": key,
+        "fields": msgspec.to_builtins(obj) if obj is not None else None,
+    }
+    for queue in list(_subscribers):
+        try:
+            queue.put_nowait(event)
+        except asyncio.QueueFull:
+            # Slow consumer: drop it; the client reconnects and resyncs.
+            _subscribers.discard(queue)
+
+
+def subscribe() -> asyncio.Queue:
+    queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
+    _subscribers.add(queue)
+    return queue
+
+
+def unsubscribe(queue: asyncio.Queue) -> None:
+    _subscribers.discard(queue)
+
+
+def tokens_from_env() -> set[str]:
+    """Accepted sync tokens (PASKIA_SYNC_TOKENS, comma-separated)."""
+    return {
+        t.strip()
+        for t in os.environ.get("PASKIA_SYNC_TOKENS", "").split(",")
+        if t.strip()
+    }
+
+
+def encode(message: dict) -> bytes:
+    return msgspec.json.encode(message)

paskia/util/apistructs.py

@@ -14,7 +14,15 @@ import msgspec
 from uarite import uaparse
 
 from paskia import db
-from paskia.db.structs import Credential, Org, OriginEntry, Permission, Role, User
+from paskia.db.structs import (
+    Credential,
+    Org,
+    OriginEntry,
+    Permission,
+    RemoteConfig,
+    Role,
+    User,
+)
 
 # -------------------------------------------------------------------------
 # API structs - inherit from db structs, add uuid for serialization
@@ -177,6 +185,7 @@ class ApiSettings(msgspec.Struct):
     auth_site_url: str
     session_cookie: str
     version: str
+    remote: bool = False  # this instance satellite-serves the domain
 
 
 class ApiDomain(msgspec.Struct):
@@ -194,6 +203,7 @@ class ApiDomain(msgspec.Struct):
     site_url: str
     auth_site_url: str
     auth_host: str | None
+    remote: RemoteConfig | None = None
 
 
 class ApiTokenInfo(msgspec.Struct, omit_defaults=True):

paskia/util/userinfo.py

@@ -1,6 +1,6 @@
 """User information formatting and retrieval logic."""
 
-from paskia import aaguid, db
+from paskia import aaguid, satellite
 from paskia.db import SessionContext
 from paskia.util import avatar, hostutil
 from paskia.util.apistructs import (
@@ -43,24 +43,30 @@ async def build_user_info(
     ctx: SessionContext | None = None,
 ) -> ApiUserDetail:
     """Build user info struct for authenticated users."""
-    user = db.data().users[user_uuid]
+    data = satellite.store_for_host(request_host)
+    user = data.users[user_uuid]
     normalized_host = hostutil.normalize_host(request_host)
 
+    user_sessions = [s for s in data.sessions.values() if s.user_uuid == user_uuid]
+    user_credentials = [
+        c for c in data.credentials.values() if c.user_uuid == user_uuid
+    ]
+
     sessions = {
         s.key: ApiUserSession.from_db(
             s,
             current_key=session_key,
             normalized_host=normalized_host,
         )
-        for s in user.sessions
+        for s in user_sessions
     }
 
     return ApiUserDetail(
         user=ApiUser.from_db(user, avatar_url=avatar.avatar_browser_url(user.uuid)),
-        credentials={c.uuid: c for c in user.credentials},
+        credentials={c.uuid: c for c in user_credentials},
         aaguid_info={
             k: ApiAaguidInfo(**v)
-            for k, v in aaguid.filter(c.aaguid for c in user.credentials).items()
+            for k, v in aaguid.filter(c.aaguid for c in user_credentials).items()
         },
         sessions=sessions,
         permissions={p.uuid: ApiPermission.from_db(p) for p in ctx.permissions}

tests/test_remote.py

@@ -0,0 +1,571 @@
+"""Tests for remote (satellite) domains: config, replica application, feed."""
+
+import asyncio
+import secrets
+import time
+from datetime import UTC, datetime
+from uuid import UUID
+
+import httpx
+import msgspec
+import pytest
+import pytest_asyncio
+from fastapi import Response
+
+import paskia.db.operations as ops_db
+from paskia import domains, satellite, syncfeed
+from paskia.db.structs import (
+    DB,
+    Config,
+    Credential,
+    DomainConfig,
+    Org,
+    OriginEntry,
+    Permission,
+    RemoteConfig,
+    Role,
+    Session,
+    User,
+)
+from paskia.fastapi.mainapp import app
+from paskia.fastapi.session import AUTH_COOKIE_NAME
+from paskia.util.crypto import hash_secret
+
+from .conftest import TEST_RP_ID
+
+REMOTE_URL = "http://remote.test"
+
+
+def _remote_domain_config(**kw) -> Config:
+    return Config(
+        domains={
+            TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
+            "example.com": DomainConfig(
+                origins={
+                    "**.example.com": True,
+                    "auth.example.com": OriginEntry(auth_host=True),
+                },
+                remote=RemoteConfig(url=REMOTE_URL, token="t", **kw),
+            ),
+        }
+    )
+
+
+def test_remote_domain_valid():
+    domains.validate_config(_remote_domain_config())
+
+
+def test_remote_domain_requires_auth_host():
+    config = _remote_domain_config()
+    config.domains["example.com"].origins = {"**.example.com": True}
+    with pytest.raises(ValueError, match="auth host"):
+        domains.validate_config(config)
+
+
+def test_remote_domain_requires_http_url():
+    config = _remote_domain_config()
+    config.domains["example.com"].remote.url = "ftp://x"
+    with pytest.raises(ValueError, match="http"):
+        domains.validate_config(config)
+
+
+def test_sanitize_preserves_remote():
+    config, warnings = domains.sanitize_config(_remote_domain_config())
+    assert not warnings
+    assert config.domains["example.com"].remote.url == REMOTE_URL
+
+
+def test_apply_upsert_and_delete():
+    replica = DB()
+    user = User.create(display_name="U", role=UUID(int=1))
+    user.uuid = UUID(int=2)
+    satellite._apply(replica, "users", str(user.uuid), _builtins(user))
+    assert replica.users[user.uuid].display_name == "U"
+    satellite._apply(replica, "users", str(user.uuid), None)
+    assert not replica.users
+
+
+def _builtins(obj):
+    return msgspec.to_builtins(obj)
+
+
+def test_apply_session_roundtrip():
+    """Sessions keep their string key and datetime/UUID fields."""
+    replica = DB()
+    session = Session.create(
+        user=UUID(int=1),
+        credential=UUID(int=2),
+        key=hash_secret("cookie", "sekret"),
+        host="app2.example.com",
+        ip="127.0.0.1",
+        user_agent="ua",
+        validated=datetime.now(UTC),
+        rp_id="example.com",
+    )
+    satellite._apply(replica, "sessions", session.key, _builtins(session))
+    stored = replica.sessions[session.key]
+    assert stored.host == "app2.example.com"
+    assert stored.validated == session.validated
+    assert stored.user_uuid == UUID(int=1)
+
+
+def test_apply_credential_bytes_roundtrip():
+    """credential_id/public_key are bytes over the wire (base64 in JSON)."""
+    replica = DB()
+    cred = Credential.create(
+        credential_id=secrets.token_bytes(32),
+        user=UUID(int=1),
+        aaguid=UUID(int=0),
+        public_key=secrets.token_bytes(64),
+        sign_count=3,
+        rp_id="example.com",
+    )
+    cred.uuid = UUID(int=9)
+    # Simulate the full wire path: builtins -> JSON -> builtins
+    wire = msgspec.json.decode(msgspec.json.encode(_builtins(cred)))
+    satellite._apply(replica, "credentials", str(cred.uuid), wire)
+    stored = replica.credentials[cred.uuid]
+    assert stored.credential_id == cred.credential_id
+    assert stored.public_key == cred.public_key
+    assert stored.sign_count == 3
+
+
+def test_feed_emit_to_subscribers():
+    queue = syncfeed.subscribe()
+    try:
+        user = User.create(display_name="A", role=UUID(int=1))
+        syncfeed.emit("users", "k1", user)
+        syncfeed.emit("users", "k1", None)
+        assert queue.get_nowait()["fields"]["display_name"] == "A"
+        assert queue.get_nowait()["fields"] is None
+    finally:
+        syncfeed.unsubscribe(queue)
+
+
+def test_feed_drops_full_queue():
+    queue = syncfeed.subscribe()
+    try:
+        for i in range(1001):
+            syncfeed.emit("users", f"k{i}", None)
+        assert queue.qsize() == 1000
+        syncfeed.emit("users", "k1001", None)  # subscriber already dropped
+        assert queue.qsize() == 1000
+    finally:
+        syncfeed.unsubscribe(queue)
+
+
+@pytest.mark.asyncio
+async def test_operations_emit_events(test_db):
+    """Writes through db.operations land on the sync feed."""
+    queue = syncfeed.subscribe()
+    try:
+        user = next(iter(test_db.users.values()))
+        ops_db.update_user_display_name(user.uuid, "Renamed")
+        event = queue.get_nowait()
+        assert event["table"] == "users"
+        assert event["key"] == str(user.uuid)
+        assert event["fields"]["display_name"] == "Renamed"
+    finally:
+        syncfeed.unsubscribe(queue)
+
+
+@pytest.mark.asyncio
+async def test_replica_refresh_and_evict():
+    replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t"))
+    token = secrets.token_urlsafe(12)
+    session = Session.create(
+        user=UUID(int=1),
+        credential=UUID(int=2),
+        key=hash_secret("cookie", token),
+        host="app2.example.com",
+        ip="1.1.1.1",
+        user_agent="ua",
+        validated=datetime(2020, 1, 1, tzinfo=UTC),
+    )
+    replica.db.sessions[session.key] = session
+
+    now = datetime.now(UTC)
+    replica.refresh_session(session.key, now, "2.2.2.2", "new-ua")
+    assert replica.db.sessions[session.key].validated == now
+    queued = replica._pending_refresh[session.key]
+    assert queued["type"] == "session_refresh"
+    assert queued["ip"] == "2.2.2.2"
+
+    # Host-keyed dispatch eviction (the replica's domain is resolved by host)
+    domains.configure(listen=["localhost:4401"])
+    domains.init_registry(_remote_domain_config())
+    satellite.manager.replicas[REMOTE_URL] = replica
+    satellite.evict_session(token, "app2.example.com")
+    assert not replica.db.sessions
+    satellite.manager.replicas.pop(REMOTE_URL)
+
+
+def test_availability_gate():
+    replica = satellite.RemoteReplica(
+        RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60)
+    )
+    assert not replica.available()  # never synced
+    replica.last_contact = time.monotonic()
+    assert replica.available()
+
+
+# -------------------------------------------------------------------------
+# API-level: endpoints served from an injected replica
+# -------------------------------------------------------------------------
+
+
+def _replica_db() -> tuple[DB, str]:
+    """A replica DB holding one org/role/perm/user/credential/session."""
+    replica = DB()
+    org = Org.create(display_name="Org")
+    org.uuid = UUID(int=101)
+    replica.orgs[org.uuid] = org
+    perm = Permission.create(scope="auth:admin", display_name="Admin")
+    perm.uuid = UUID(int=102)
+    perm.orgs[org.uuid] = True
+    replica.permissions[perm.uuid] = perm
+    role = Role.create(org=org.uuid, display_name="Admins", permissions={perm.uuid})
+    role.uuid = UUID(int=103)
+    replica.roles[role.uuid] = role
+    user = User.create(display_name="Remote Admin", role=role.uuid)
+    user.uuid = UUID(int=104)
+    replica.users[user.uuid] = user
+    cred = Credential.create(
+        credential_id=b"cid",
+        user=user.uuid,
+        aaguid=UUID(int=0),
+        public_key=b"pk",
+        sign_count=0,
+        rp_id="example.com",
+    )
+    cred.uuid = UUID(int=105)
+    replica.credentials[cred.uuid] = cred
+    secret = secrets.token_urlsafe(12)
+    session = Session.create(
+        user=user.uuid,
+        credential=cred.uuid,
+        key=hash_secret("cookie", secret),
+        host="app2.example.com",
+        ip="127.0.0.1",
+        user_agent="pytest",
+        validated=datetime.now(UTC),
+        rp_id="example.com",
+    )
+    replica.sessions[session.key] = session
+    return replica, secret
+
+
+@pytest_asyncio.fixture
+async def remote_client(test_db):
+    """ASGI client with example.com as a remote domain on a warm replica."""
+    config = _remote_domain_config()
+    domains.configure(listen=["localhost:4401"])
+    domains.init_registry(config)
+    replica_db, secret = _replica_db()
+    replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t"))
+    replica.db = replica_db
+    replica.last_contact = time.monotonic()
+    replica.connected = True
+    satellite.manager.replicas[REMOTE_URL] = replica
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="http://localhost:4401"
+    ) as client:
+        yield client, secret, replica
+    satellite.manager.replicas.pop(REMOTE_URL, None)
+
+
+@pytest.mark.asyncio
+async def test_forward_served_from_replica(remote_client):
+    client, secret, _ = remote_client
+    r = await client.get(
+        "/auth/api/forward?perm=auth:admin",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 204
+    assert r.headers["remote-name"] == "Remote Admin"
+    assert r.headers["remote-groups"] == "auth:admin"
+
+
+@pytest.mark.asyncio
+async def test_forward_replica_denies_missing_perm(remote_client):
+    client, secret, _ = remote_client
+    r = await client.get(
+        "/auth/api/forward?perm=other:scope",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_validate_renews_locally_and_queues_writebehind(remote_client):
+    client, secret, replica = remote_client
+    session = next(iter(replica.db.sessions.values()))
+    session.validated = datetime(2020, 1, 1, tzinfo=UTC)  # force refresh threshold
+    r = await client.post(
+        "/auth/api/validate",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 200
+    assert r.json()["renewed"] is True
+    assert session.validated.year > 2020  # applied to the replica
+    queued = replica._pending_refresh[session.key]
+    assert queued["type"] == "session_refresh"
+
+
+@pytest.mark.asyncio
+async def test_remote_domain_503_when_replica_stale(remote_client):
+    client, secret, replica = remote_client
+    replica.connected = False
+    replica.last_contact = 0
+    r = await client.get(
+        "/auth/api/forward",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 503
+
+
+@pytest.mark.asyncio
+async def test_logout_proxied_and_evicted(remote_client, monkeypatch):
+    client, secret, replica = remote_client
+
+    async def fake_forward(request):
+        return Response(status_code=200, content=b'{"message": "Logged out"}')
+
+    monkeypatch.setattr(satellite, "forward_request", fake_forward)
+    r = await client.post(
+        "/auth/api/logout",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 200
+    assert not replica.db.sessions  # evicted optimistically
+
+
+@pytest.mark.asyncio
+async def test_admin_configures_remote_domain(client, session_token, test_db):
+    """The admin domains API stores remote config and masks the token."""
+    r = await client.post(
+        "/auth/api/admin/domains/",
+        json={
+            "rp_id": "example.com",
+            "rp_name": "Example",
+            "origins": {
+                "**.example.com": True,
+                "auth.example.com": {"auth_host": True},
+            },
+            "remote": {"url": "http://remote.test", "token": "sekret", "cache_ttl": 30},
+        },
+        headers={
+            "Host": "localhost:4401",
+            "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+        },
+    )
+    assert r.status_code == 200, r.text
+    stored = test_db.config.domains["example.com"]
+    assert stored.remote.url == "http://remote.test"
+    assert stored.remote.token == "sekret"
+
+    r = await client.get(
+        "/auth/api/admin/domains/",
+        headers={
+            "Host": "localhost:4401",
+            "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+        },
+    )
+    entry = next(d for d in r.json() if d["rp_id"] == "example.com")
+    assert entry["remote"]["url"] == "http://remote.test"
+    assert "token" not in entry["remote"]  # write-only
+
+    headers = {
+        "Host": "localhost:4401",
+        "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+    }
+    origins = {"**.example.com": True, "auth.example.com": {"auth_host": True}}
+
+    # PATCH without the remote key preserves it (and its token)
+    r = await client.patch(
+        "/auth/api/admin/domains/example.com",
+        json={"rp_name": "Ex", "origins": origins},
+        headers=headers,
+    )
+    assert r.status_code == 200, r.text
+    assert stored.remote.url == "http://remote.test"
+    assert stored.remote.token == "sekret"
+
+    # PATCH with a new URL but no token keeps the stored token
+    r = await client.patch(
+        "/auth/api/admin/domains/example.com",
+        json={
+            "rp_name": "Ex",
+            "origins": origins,
+            "remote": {"url": "http://other.test", "cache_ttl": 30},
+        },
+        headers=headers,
+    )
+    assert r.status_code == 200, r.text
+    assert stored.remote.url == "http://other.test"
+    assert stored.remote.token == "sekret"
+
+    # PATCH with remote: null clears it
+    r = await client.patch(
+        "/auth/api/admin/domains/example.com",
+        json={"rp_name": "Ex", "origins": origins, "remote": None},
+        headers=headers,
+    )
+    assert r.status_code == 200, r.text
+    assert stored.remote is None
+
+
+@pytest.mark.asyncio
+async def test_admin_remote_domain_requires_auth_host(client, session_token):
+    r = await client.post(
+        "/auth/api/admin/domains/",
+        json={
+            "rp_id": "example.com",
+            "origins": {"**.example.com": True},
+            "remote": {"url": "http://remote.test"},
+        },
+        headers={
+            "Host": "localhost:4401",
+            "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+        },
+    )
+    assert r.status_code == 400
+    assert "auth host" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_admin_remote_values_clamped(client, session_token, test_db):
+    """Nonsensical TTL/interval values are clamped to the supported floor."""
+    r = await client.post(
+        "/auth/api/admin/domains/",
+        json={
+            "rp_id": "example.com",
+            "origins": {
+                "**.example.com": True,
+                "auth.example.com": {"auth_host": True},
+            },
+            "remote": {
+                "url": "http://remote.test",
+                "token": "t",
+                "cache_ttl": -5,
+                "refresh_interval": 1,
+            },
+        },
+        headers={
+            "Host": "localhost:4401",
+            "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+        },
+    )
+    assert r.status_code == 200, r.text
+    remote = test_db.config.domains["example.com"].remote
+    assert remote.cache_ttl == 1
+    assert remote.refresh_interval == 30
+
+
+def _add_shared_remote(config: Config, token: str) -> None:
+    config.domains["sub.example.com"] = DomainConfig(
+        origins={
+            "**.sub.example.com": True,
+            "auth.sub.example.com": OriginEntry(auth_host=True),
+        },
+        remote=RemoteConfig(url=REMOTE_URL, token=token),
+    )
+
+
+def test_shared_remote_same_settings_ok():
+    config = _remote_domain_config()
+    _add_shared_remote(config, token="t")
+    domains.validate_config(config)
+
+
+def test_shared_remote_conflicting_settings_rejected():
+    config = _remote_domain_config()
+    _add_shared_remote(config, token="different")
+    with pytest.raises(ValueError, match="conflicting"):
+        domains.validate_config(config)
+
+
+def test_sanitize_aligns_conflicting_shared_remote():
+    config = _remote_domain_config()
+    _add_shared_remote(config, token="different")
+    sanitized, warnings = domains.sanitize_config(config)
+    assert any("conflict" in w for w in warnings)
+    # Aligned to the first configuration, and the result validates
+    assert sanitized.domains["sub.example.com"].remote.token == "t"
+    domains.validate_config(sanitized)
+
+
+def test_rebuild_listener_skipped_without_running_loop():
+    """A registry rebuild outside an event loop warns instead of crashing."""
+
+    async def listener(_registry):
+        raise AssertionError("must not run")  # pragma: no cover
+
+    domains.add_rebuild_listener(listener)
+    try:
+        domains.init_registry(_remote_domain_config())
+    finally:
+        domains.remove_rebuild_listener(listener)
+
+
+@pytest.mark.asyncio
+async def test_rebuild_listener_runs_with_loop():
+    ran = asyncio.Event()
+
+    async def listener(_registry):
+        ran.set()
+
+    domains.add_rebuild_listener(listener)
+    try:
+        domains.init_registry(_remote_domain_config())
+        await asyncio.wait_for(ran.wait(), 2)
+    finally:
+        domains.remove_rebuild_listener(listener)
+
+
+@pytest.mark.asyncio
+async def test_settings_reports_remote_flag(remote_client):
+    client, _, _ = remote_client
+    r = await client.get("/auth/api/settings", headers={"Host": "app2.example.com"})
+    assert r.status_code == 200
+    assert r.json()["remote"] is True
+    r = await client.get("/auth/api/settings", headers={"Host": "localhost:4401"})
+    assert r.json()["remote"] is False
+
+
+@pytest.mark.asyncio
+async def test_logout_remote_down_returns_503(remote_client, monkeypatch):
+    """A transport failure on a proxied call is a deliberate 503, not a 500."""
+    client, secret, replica = remote_client
+
+    class Down:
+        async def request(self, *args, **kwargs):
+            raise httpx.ConnectError("down")
+
+    monkeypatch.setitem(satellite._clients, REMOTE_URL, Down())
+    r = await client.post(
+        "/auth/api/logout",
+        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+    )
+    assert r.status_code == 503
+    # Not evicted: the remote never saw the logout, the session is still valid
+    assert replica.db.sessions
+
+
+@pytest.mark.asyncio
+async def test_user_api_proxied_on_remote_auth_host(remote_client, monkeypatch):
+    """Profile traffic for a remote domain's auth host is forwarded, never
+    written to the satellite's local database."""
+    client, _, _ = remote_client
+
+    async def fake_forward(request):
+        return Response(status_code=200, content=b"proxied")
+
+    monkeypatch.setattr(satellite, "forward_request", fake_forward)
+    r = await client.get(
+        f"/auth/api/user/{UUID(int=104)}/profile.webp",
+        headers={"Host": "auth.example.com"},
+    )
+    assert r.status_code == 200
+    assert r.content == b"proxied"

tests/test_satellite_e2e.py

@@ -0,0 +1,252 @@
+"""Live end-to-end tests for satellite domains over a real sync channel.
+
+A real uvicorn server plays the remote (its sync WebSocket fed from the
+process-global database), while the satellite side runs through the ASGI
+app with a real RemoteReplica started by the real SatelliteManager
+reconcile. Unlike test_remote.py (which injects a warm replica by hand),
+nothing here is stubbed: the snapshot, live events, write-behind, dead-peer
+TTL handling and reconnect reconciliation all cross a TCP socket.
+
+One in-process distortion remains: both sides share the module-level
+database and registry, so endpoints the satellite *forwards* to the remote
+(logout, set-session, OIDC) would re-forward and loop here — those paths
+are covered in test_remote.py with a mocked forward_request.
+"""
+
+import asyncio
+import socket
+import time
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import httpx
+import pytest
+import pytest_asyncio
+import uvicorn
+
+import paskia.db.operations as ops_db
+from paskia import domains, satellite
+from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig
+from paskia.fastapi.mainapp import app
+from paskia.fastapi.session import AUTH_COOKIE_NAME
+
+from .conftest import TEST_RP_ID, create_test_session
+
+REMOTE_DOMAIN = "example.com"
+APP_HOST = "app2.example.com"
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+async def wait_for(desc: str, condition, timeout: float = 10) -> None:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if condition():
+            return
+        await asyncio.sleep(0.02)
+    raise AssertionError(f"timed out waiting for {desc}")
+
+
+@pytest_asyncio.fixture
+async def live_remote(test_db, monkeypatch):
+    """A uvicorn-served 'remote' and the satellite manager wired against it.
+
+    The manager is started for real: reconcile builds the replica from the
+    domain config and its sync client connects over TCP. Yielded namespace:
+    client (ASGI, satellite side), replica, db (the remote's database),
+    url/port of the remote, start_server (for restarts), servers.
+    """
+    monkeypatch.setenv("PASKIA_SYNC_TOKENS", "live-tok")
+    servers: list[tuple[uvicorn.Server, asyncio.Task]] = []
+
+    async def start_server(port: int) -> uvicorn.Server:
+        config = uvicorn.Config(
+            app, host="127.0.0.1", port=port, log_level="error", lifespan="off"
+        )
+        server = uvicorn.Server(config)
+        task = asyncio.create_task(server.serve())
+        servers.append((server, task))
+        while not server.started:
+            await asyncio.sleep(0.02)
+        return server
+
+    port = _free_port()
+    url = f"http://127.0.0.1:{port}"
+    await start_server(port)
+
+    domains.configure(listen=["localhost:4401"])
+    domains.init_registry(
+        Config(
+            domains={
+                TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
+                REMOTE_DOMAIN: DomainConfig(
+                    origins={
+                        f"**.{REMOTE_DOMAIN}": True,
+                        f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
+                    },
+                    remote=RemoteConfig(url=url, token="live-tok"),
+                ),
+            }
+        )
+    )
+    await satellite.manager.start()
+    replica = satellite.manager.replicas[url]
+    await wait_for("initial snapshot", lambda: replica.available())
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="http://localhost:4401"
+    ) as client:
+        try:
+            yield SimpleNamespace(
+                client=client,
+                replica=replica,
+                db=test_db,
+                url=url,
+                port=port,
+                start_server=start_server,
+                servers=servers,
+            )
+        finally:
+            await satellite.manager.stop()
+            for server, task in servers:
+                server.should_exit = True
+                await task
+
+
+def _app_headers(secret: str) -> dict[str, str]:
+    return {"Host": APP_HOST, "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}
+
+
+@pytest.mark.asyncio
+async def test_live_snapshot_and_user_events(live_remote, test_user):
+    """Bootstrap data arrives with the snapshot; later mutations as events."""
+    assert test_user.uuid in live_remote.replica.db.users
+    ops_db.update_user_display_name(test_user.uuid, "Live Rename")
+    await wait_for(
+        "rename to propagate",
+        lambda: (
+            live_remote.replica.db.users[test_user.uuid].display_name == "Live Rename"
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_live_forward_and_writebehind(live_remote, test_user, test_credential):
+    """A session created on the remote authorizes locally; the /validate
+    refresh travels back over the wire into the remote's database."""
+    live = live_remote
+    key, secret = create_test_session(
+        user_uuid=test_user.uuid,
+        credential_uuid=test_credential.uuid,
+        host=APP_HOST,
+        rp_id=REMOTE_DOMAIN,
+    )
+    await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+    r = await live.client.get(
+        "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret)
+    )
+    assert r.status_code == 204
+    assert r.headers["remote-name"] == "Test Admin"
+
+    # Age the session on the remote (propagates), then validate on the satellite
+    stale = datetime.now(UTC) - timedelta(minutes=10)
+    ops_db.update_session(key, validated=stale)
+    await wait_for(
+        "staleness to propagate",
+        lambda: live.replica.db.sessions[key].validated == stale,
+    )
+    r = await live.client.post("/auth/api/validate", headers=_app_headers(secret))
+    assert r.status_code == 200
+    assert r.json()["renewed"] is True
+    await wait_for(
+        "write-behind to land on the remote",
+        lambda: live.db.sessions[key].validated > stale,
+    )
+
+
+@pytest.mark.asyncio
+async def test_live_session_delete_propagates(live_remote, test_user, test_credential):
+    live = live_remote
+    key, secret = create_test_session(
+        user_uuid=test_user.uuid,
+        credential_uuid=test_credential.uuid,
+        host=APP_HOST,
+        rp_id=REMOTE_DOMAIN,
+    )
+    await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+    ops_db.delete_session(key)
+    await wait_for(
+        "session delete to propagate", lambda: key not in live.replica.db.sessions
+    )
+    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_live_wrong_token_rejected(live_remote):
+    bad = satellite.RemoteReplica(RemoteConfig(url=live_remote.url, token="wrong"))
+    await bad.start()
+    try:
+        await asyncio.sleep(1.5)
+        assert not bad.available()
+    finally:
+        await bad.stop()
+
+
+@pytest.mark.asyncio
+async def test_live_disconnect_ttl_and_reconnect(
+    live_remote, test_user, test_credential
+):
+    """Channel down: replica keeps serving within cache_ttl, then fails
+    closed (503); reconnect starts from a snapshot and reconciles drift."""
+    live = live_remote
+    key, secret = create_test_session(
+        user_uuid=test_user.uuid,
+        credential_uuid=test_credential.uuid,
+        host=APP_HOST,
+        rp_id=REMOTE_DOMAIN,
+    )
+    await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+    server, task = live.servers[0]
+    server.should_exit = True
+    await task
+    await wait_for(
+        "dead peer detection", lambda: not live.replica.connected, timeout=15
+    )
+
+    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+    assert r.status_code == 204  # still authoritative within cache_ttl
+
+    live.replica.remote.cache_ttl = 0  # expire the trust window
+    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+    assert r.status_code == 503
+
+    # Drift while the channel is down, then let the client auto-reconnect
+    ops_db.update_user_display_name(test_user.uuid, "While Down")
+    await live.start_server(live.port)
+    await wait_for("reconnect", lambda: live.replica.available(), timeout=20)
+    assert (
+        live.replica.db.users[test_user.uuid].display_name == "While Down"
+    )  # snapshot reconciled the drift
+    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+    assert r.status_code == 204
+
+
+@pytest.mark.asyncio
+async def test_reconcile_stops_replica_when_remote_removed(live_remote):
+    """A registry rebuild without the remote config stops the replica."""
+    live = live_remote
+    domains.init_registry(
+        Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})})
+    )
+    await wait_for(
+        "replica shutdown", lambda: live.url not in satellite.manager.replicas
+    )