Harden satellite: uplink probe in domain dialog, live e2e tests

Review follow-ups for remote domains:

- Domain dialog probes {url}/auth/api/settings as the admin types:
  reachability, rp-id match, and non-nesting (via a new remote flag in
  the settings response) — warnings, never submit blockers
- Reject conflicting settings for domains sharing one remote;
  sanitize aligns with a warning instead
- Forward profile/avatar requests (/auth/api/user/*) to the remote when
  a remote domain's auth host reaches the satellite, never write locally
- Proxied calls answer 503 on transport errors (logout then deliberately
  keeps the local replica session — the remote never saw it)
- Close forwarding httpx clients on reconcile removal and shutdown
- Clamp cache_ttl/refresh_interval server-side; syncfeed.emit is a true
  no-op without subscribers; sync pump tolerates unexpected send errors
- Rebuild listeners: no running loop warns instead of raising, tasks are
  strongly referenced and failures logged
- tests/test_satellite_e2e.py: live over-the-wire suite — real uvicorn
  remote, real replica via reconcile: snapshot, events, write-behind,
  token gate, TTL fail-closed, reconnect reconciliation

docs/RemoteProxy.md

@@ -36,7 +36,11 @@ In the stored config (`DomainConfig.remote`):
 
 - `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.
+  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
@@ -44,9 +48,13 @@ In the stored config (`DomainConfig.remote`):
   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.
+  session lifetime) for fail-open behavior. Minimum 1.
 - `refresh_interval` — seconds between reconnects; every connect starts
-  from a full snapshot, which reconciles any drift.
+  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
@@ -86,6 +94,13 @@ signing key is never replicated.
 | `/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,

frontend/src/admin/dialogs/DomainEditDialog.vue

@@ -45,6 +45,44 @@ 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.
@@ -406,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()
 })
@@ -556,6 +595,11 @@ function onRemoveOrigin(i) {
         <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" />

paskia/domains.py

@@ -299,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
@@ -317,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)
@@ -384,6 +400,14 @@ def sanitize_config(
                 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 — "
@@ -493,6 +517,7 @@ 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:
@@ -505,6 +530,34 @@ def remove_rebuild_listener(fn) -> None:
         _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:
     """Record process-global serve parameters for site URL derivation."""
     global _listen
@@ -543,12 +596,7 @@ def init_registry(config: Config) -> DomainRegistry:
     """Build and install the global registry from a combined configuration."""
     global _registry
     _registry = build(config)
-    for fn in _rebuild_listeners:
-        result = fn(_registry)
-        if asyncio.iscoroutine(result):
-            # init_registry runs within a running loop in every serving
-            # context (lifespan, tests, admin rebuild).
-            asyncio.get_running_loop().create_task(result)
+    _notify_rebuild(_registry)
     return _registry
 
 

paskia/fastapi/admin/domains.py

@@ -62,8 +62,8 @@ def _normalize_remote(
     return RemoteConfig(
         url=value["url"].rstrip("/"),
         token=token,
-        cache_ttl=int(value.get("cache_ttl") or 60),
-        refresh_interval=int(value.get("refresh_interval") or 300),
+        cache_ttl=max(1, int(value.get("cache_ttl") or 60)),
+        refresh_interval=max(30, int(value.get("refresh_interval") or 300)),
     )
 
 

paskia/fastapi/api.py

@@ -312,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"},
     )

paskia/fastapi/sync.py

@@ -92,3 +92,5 @@ async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
             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

@@ -78,7 +78,7 @@ class RemoteReplica:
         )
 
     def refresh_session(
-        self, key: str, validated, ip: str | None, user_agent: str | None
+        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)
@@ -227,6 +227,8 @@ class SatelliteManager:
         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."""
@@ -237,6 +239,7 @@ class SatelliteManager:
         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:
@@ -275,7 +278,12 @@ def store_for_host(host: str | None) -> DB:
 
 
 def refresh_session(
-    key, host: str | None, ip: str, user_agent: str, validated, ctx=None
+    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)
@@ -313,6 +321,13 @@ _HOP_BY_HOP = {
 _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.
 
@@ -327,15 +342,18 @@ async def forward_request(request: Request) -> Response | None:
     client = _clients.get(url)
     if client is None:
         client = _clients[url] = httpx.AsyncClient(base_url=url, timeout=_TIMEOUT)
-    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
-        },
-    )
+    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 = [

paskia/syncfeed.py

@@ -23,6 +23,8 @@ _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,

paskia/util/apistructs.py

@@ -185,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):

tests/test_remote.py

@@ -1,5 +1,6 @@
 """Tests for remote (satellite) domains: config, replica application, feed."""
 
+import asyncio
 import secrets
 import time
 from datetime import UTC, datetime
@@ -375,7 +376,10 @@ async def test_admin_configures_remote_domain(client, session_token, test_db):
     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}"}
+    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)
@@ -391,8 +395,11 @@ async def test_admin_configures_remote_domain(client, session_token, test_db):
     # 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}},
+        json={
+            "rp_name": "Ex",
+            "origins": origins,
+            "remote": {"url": "http://other.test", "cache_ttl": 30},
+        },
         headers=headers,
     )
     assert r.status_code == 200, r.text
@@ -425,3 +432,140 @@ async def test_admin_remote_domain_requires_auth_host(client, 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
+    )