Tests for the link model + lint/format pass

test_remote.py: link URL parsing, RAM-only offer mint/spend/expiry,
pushed-domain application (accept/idempotence/removal/local conflict/
auth-host collision), admin link lifecycle (offer, scope, revoke) and
uplink list/unlink, remote-write rejection on the domains API, live feed
scope mutation.

test_satellite_e2e.py: real registration handshake over TCP (key material
identical on both sides, duplicate and bad-URL rejections), established
link reconnect, revocation unlinking the satellite, live scope shrink
stopping the feed, domain conflicts reported upstream.

Also: ruff format drift fixes in startupbox/test_cli.

paskia/__main__.py

@@ -201,8 +201,8 @@ def cmd_satellite(args: argparse.Namespace) -> None:
     and the pushed domains are stored. Afterwards 'paskia' serves the
     linked domains.
     """
-    from paskia import satellite
-    from paskia.db.lifecycle import kanta
+    from paskia import satellite  # noqa: PLC0415 (heavy deps: fastapi, websockets)
+    from paskia.db.lifecycle import kanta  # noqa: PLC0415 (binds the db path)
 
     db_path = db_file_path()
     fresh = not db_path.exists()
@@ -231,7 +231,10 @@ def cmd_satellite(args: argparse.Namespace) -> None:
     except (ValueError, RuntimeError) as e:
         raise SystemExit(f"Registration failed: {e}") from e
 
-    domains_list = ", ".join(info["domains"]) or "(none yet — the upstream can scope the link later)"
+    domains_list = (
+        ", ".join(info["domains"])
+        or "(none yet — the upstream can scope the link later)"
+    )
     print(f"🛰  Linked to {info['url']} — serving: {domains_list}")
     for rp_id, reason in (info.get("rejected") or {}).items():
         print(f"⚠️  Domain {rp_id} rejected: {reason}")
@@ -310,7 +313,7 @@ def main():
         "--save",
         action="store_true",
         help="Save --listen to the database for future runs. "
-        "Use --listen \"\" to clear the stored endpoints.",
+        'Use --listen "" to clear the stored endpoints.',
     )
 
     init_parser = argparse.ArgumentParser(

paskia/db/operations.py

@@ -33,7 +33,6 @@ from paskia.db.structs import (
 )
 from paskia.util.crypto import hash_secret
 from paskia.util.nameutil import slugify_name
-from paskia.util.passphrase import generate as generate_passphrase
 
 _logger = logging.getLogger(__name__)
 

paskia/fastapi/sync.py

@@ -162,9 +162,7 @@ async def drop_domain_from_links(rp_id: str) -> None:
     """A domain was deleted upstream: shrink every link's scope and push."""
     for link_id, link in db.data().satellite_links.items():
         if rp_id in link.rp_ids:
-            db.set_satellite_link_scope(
-                link_id, [r for r in link.rp_ids if r != rp_id]
-            )
+            db.set_satellite_link_scope(link_id, [r for r in link.rp_ids if r != rp_id])
             await push_link_domains(link_id)
 
 
@@ -300,7 +298,7 @@ async def sync_websocket(ws: WebSocket):
                 )
         finally:
             sender.cancel()
-    except (WebSocketDisconnect, RuntimeError):
+    except WebSocketDisconnect, RuntimeError:
         pass
     except Exception:
         _logger.exception("Sync WebSocket failed")
@@ -314,7 +312,7 @@ async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
     try:
         while True:
             await _send(ws, await queue.get())
-    except (WebSocketDisconnect, RuntimeError, asyncio.CancelledError):
+    except WebSocketDisconnect, RuntimeError, asyncio.CancelledError:
         pass
     except Exception:
         _logger.debug("Sync event pump ended", exc_info=True)

paskia/satellite.py

@@ -43,7 +43,7 @@ import msgspec
 import websockets
 from fastapi import HTTPException, Request, Response
 
-from paskia import db, domains
+from paskia import db, domains, syncfeed
 from paskia.config import SESSION_LIFETIME
 from paskia.db.structs import (
     DB,
@@ -60,7 +60,6 @@ from paskia.db.structs import (
     Uplink,
     User,
 )
-from paskia import syncfeed
 from paskia.util.apistructs import ApiSessionContext
 from paskia.util.crypto import hash_secret
 
@@ -75,7 +74,7 @@ _RECONNECT_DELAY = 5
 _SWEEP_INTERVAL = 60
 
 
-class LinkRevoked(Exception):
+class LinkRevokedError(Exception):
     """The upstream closed the handshake with a policy violation: the link
     is revoked (or the key desynced) — the satellite unlinks its side."""
 
@@ -370,7 +369,7 @@ class UplinkReplica:
         for task in (self._task, self._sweeper):
             if task:
                 task.cancel()
-                with contextlib.suppress(asyncio.CancelledError, LinkRevoked):
+                with contextlib.suppress(asyncio.CancelledError, LinkRevokedError):
                     await task
 
     async def _sweep(self) -> None:
@@ -387,7 +386,7 @@ class UplinkReplica:
                 await self._connect()
             except asyncio.CancelledError:
                 raise
-            except LinkRevoked:
+            except LinkRevokedError:
                 raise  # the manager unlinks via the done-callback
             except Exception as e:
                 _logger.info("Sync to %s failed: %s", self.uplink.url, e)
@@ -404,7 +403,7 @@ class UplinkReplica:
             return msgspec.json.decode(await ws.recv())
         except websockets.exceptions.ConnectionClosedError as e:
             if e.code == 1008:
-                raise LinkRevoked(self.uplink.url) from e
+                raise LinkRevokedError(self.uplink.url) from e
             raise
 
     async def _connect(self) -> None:
@@ -476,7 +475,9 @@ class UplinkReplica:
                             replica_db=self.db,
                         )
                         await ws.send(
-                            syncfeed.encode({"type": "domains_ack", "rejected": rejected})
+                            syncfeed.encode(
+                                {"type": "domains_ack", "rejected": rejected}
+                            )
                         )
                     elif mtype == "ready":
                         if staging is not None:
@@ -504,7 +505,7 @@ def _replica_done(replica: UplinkReplica, task: asyncio.Task) -> None:
     if task.cancelled():
         return
     exc = task.exception()
-    if not isinstance(exc, LinkRevoked):
+    if not isinstance(exc, LinkRevokedError):
         return
     _logger.info("Uplink %s revoked the link — unlinking", replica.uplink.url)
     try:

paskia/util/startupbox.py

@@ -70,9 +70,9 @@ def _covered_by_wildcard(key: str, pattern: str) -> bool:
         matched = hostutil.is_subdomain(hostname, base)
     else:
         # '*.base' covers exactly one subdomain level
-        matched = hostname.endswith(f".{base}") and "." not in hostname[
-            : -len(base) - 1
-        ]
+        matched = (
+            hostname.endswith(f".{base}") and "." not in hostname[: -len(base) - 1]
+        )
     if not matched:
         return False
     if hostutil.is_subdomain(base, "localhost"):

tests/test_cli.py

@@ -324,7 +324,9 @@ def test_migrate_merges_kantadb_into_existing_database(run_cli, tmp_path):
 def test_migrate_merge_label_combines_rp_ids(run_cli, tmp_path):
     """A multi-domain source merges in one transaction, rp-ids slash-joined."""
     run_cli("init", "company.com")
-    src = write_kantadb(tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb")
+    src = write_kantadb(
+        tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb"
+    )
 
     run_cli("migrate", str(src))
 

tests/test_remote.py

@@ -1,4 +1,4 @@
-"""Tests for remote (satellite) domains: config, replica application, feed."""
+"""Tests for remote (satellite) domains: config, links, replica, feed."""
 
 import asyncio
 import secrets
@@ -21,8 +21,10 @@ from paskia.db.structs import (
     OriginEntry,
     RemoteConfig,
     Session,
+    Uplink,
     User,
 )
+from paskia.fastapi import sync
 from paskia.fastapi.mainapp import app
 from paskia.fastapi.session import AUTH_COOKIE_NAME
 from paskia.util.apistructs import (
@@ -36,10 +38,11 @@ from paskia.util.crypto import hash_secret
 from .conftest import TEST_RP_ID, create_test_session
 
 REMOTE_URL = "http://remote.test"
+UPLINK_ID = "up1"
 
 
 def _api_ctx(display_name="Remote Admin", permissions=("auth:admin",)):
-    """A master-pushed per-session context, as embedded in session events."""
+    """An upstream-pushed per-session context, as embedded in session events."""
     return ApiSessionContext(
         user=ApiUserContext(uuid=UUID(int=104), display_name=display_name),
         org=ApiOrgContext(uuid=UUID(int=101), display_name="Org"),
@@ -48,6 +51,12 @@ def _api_ctx(display_name="Remote Admin", permissions=("auth:admin",)):
     )
 
 
+def _uplink(**kw) -> Uplink:
+    uplink = Uplink(url=REMOTE_URL, created=datetime.now(UTC), key=b"k", **kw)
+    uplink.uplink_id = UPLINK_ID
+    return uplink
+
+
 def _remote_domain_config(**kw) -> Config:
     return Config(
         domains={
@@ -57,7 +66,7 @@ def _remote_domain_config(**kw) -> Config:
                     "**.example.com": True,
                     "auth.example.com": OriginEntry(auth_host=True),
                 },
-                remote=RemoteConfig(url=REMOTE_URL, token="t", **kw),
+                remote=RemoteConfig(link=UPLINK_ID, **kw),
             ),
         }
     )
@@ -74,17 +83,21 @@ def test_remote_domain_requires_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
+    assert config.domains["example.com"].remote.link == UPLINK_ID
+
+
+def test_sanitize_drops_remote_without_link():
+    """A remote config pointing nowhere (e.g. a v1 leftover) degrades to a
+    plain local domain instead of failing to serve."""
+    config = _remote_domain_config()
+    config.domains["example.com"].remote = RemoteConfig()
+    sanitized, warnings = domains.sanitize_config(config)
+    assert any("remote dropped" in w for w in warnings)
+    assert sanitized.domains["example.com"].remote is None
+    domains.validate_config(sanitized)
 
 
 def test_apply_upsert_and_delete():
@@ -209,9 +222,7 @@ async def test_operations_repush_session_contexts(test_db, test_user, test_crede
 
 @pytest.mark.asyncio
 async def test_replica_refresh_and_evict():
-    replica = satellite.RemoteReplica(
-        RemoteConfig(url=REMOTE_URL, token="t"), "example.com"
-    )
+    replica = satellite.UplinkReplica(_uplink())
     token = secrets.token_urlsafe(12)
     session = Session.create(
         user=UUID(int=1),
@@ -231,24 +242,120 @@ async def test_replica_refresh_and_evict():
     assert queued["type"] == "session_refresh"
     assert queued["ip"] == "2.2.2.2"
 
-    # Host-keyed dispatch eviction (the replica's domain is resolved by host)
+    # Host-keyed dispatch eviction (the replica's uplink is resolved by host)
     domains.configure(listen=["localhost:4401"])
     domains.init_registry(_remote_domain_config())
-    satellite.manager.replicas["example.com"] = replica
+    satellite.manager.replicas[UPLINK_ID] = replica
     satellite.evict_session(token, "app2.example.com")
     assert not replica.db.sessions
-    satellite.manager.replicas.pop("example.com")
+    satellite.manager.replicas.pop(UPLINK_ID)
 
 
 def test_availability_gate():
-    replica = satellite.RemoteReplica(
-        RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60), "example.com"
-    )
+    replica = satellite.UplinkReplica(_uplink(cache_ttl=60))
     assert not replica.available()  # never synced
     replica.last_contact = time.monotonic()
     assert replica.available()
 
 
+# -------------------------------------------------------------------------
+# Link URLs and registration offers
+# -------------------------------------------------------------------------
+
+
+def test_parse_link_url():
+    base, key = satellite.parse_link_url(
+        "wss://alpha.bravo.charlie.delta.echo@auth.example.com"
+    )
+    assert base == "https://auth.example.com"
+    assert key == "alpha.bravo.charlie.delta.echo"
+    base, key = satellite.parse_link_url("ws://k3y@127.0.0.1:4401")
+    assert base == "http://127.0.0.1:4401"
+    assert key == "k3y"
+    with pytest.raises(ValueError, match="wss"):
+        satellite.parse_link_url("https://alpha.bravo@auth.example.com")
+    with pytest.raises(ValueError, match="key"):
+        satellite.parse_link_url("wss://auth.example.com")
+
+
+def test_offer_mint_and_spend():
+    """Offers are wss URLs with a RAM-only, single-use, expiring key."""
+    url = sync.create_offer(["example.com"], "https://auth.example.com")
+    assert url.startswith("wss://")
+    assert url.endswith("@auth.example.com")
+    key = url.split("://", 1)[1].split("@", 1)[0]
+    offer = sync._pop_offer(key)
+    assert offer is not None and offer.rp_ids == ["example.com"]
+    assert sync._pop_offer(key) is None  # single-use
+    assert sync._pop_offer("unknown.key.here") is None
+
+    expired = sync.create_offer([], "http://127.0.0.1:4401")
+    assert expired.startswith("ws://")
+    key = expired.split("://", 1)[1].split("@", 1)[0]
+    sync._offers[key].created -= sync.OFFER_TTL + 1
+    assert sync._pop_offer(key) is None  # expired
+
+
+# -------------------------------------------------------------------------
+# Pushed-domain application (satellite side)
+# -------------------------------------------------------------------------
+
+_PUSHED = {
+    "rp_id": "example.com",
+    "rp_name": "Example",
+    "origins": {"**.example.com": True, "auth.example.com": {"auth_host": True}},
+}
+
+
+@pytest.mark.asyncio
+async def test_apply_domains_push_accept_and_remove(test_db, domain_registry):
+    """A push persists the accepted set (marked remote via the link); a
+    later push without the domain removes it."""
+    rejected = satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
+    assert not rejected
+    domain = test_db.config.domains["example.com"]
+    assert domain.remote.link == UPLINK_ID
+    assert domain.rp_name == "Example"
+    assert domains.registry().get("example.com").remote is not None
+
+    # Idempotent: the same push again changes nothing
+    assert not satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
+
+    # The uplink's scope shrank: the domain leaves the local config
+    rejected = satellite.apply_domains_push(UPLINK_ID, [])
+    assert not rejected
+    assert "example.com" not in test_db.config.domains
+    assert domains.registry().get("example.com") is None
+
+
+@pytest.mark.asyncio
+async def test_apply_domains_push_rejects_local_conflict(test_db, domain_registry):
+    """A pushed rp-id that this instance serves locally is rejected."""
+    rejected = satellite.apply_domains_push(
+        UPLINK_ID, [{**_PUSHED, "rp_id": TEST_RP_ID}]
+    )
+    assert TEST_RP_ID in rejected
+    assert test_db.config.domains[TEST_RP_ID].remote is None
+
+
+@pytest.mark.asyncio
+async def test_apply_domains_push_rejects_auth_host_collision(test_db, domain_registry):
+    """A pushed auth host that collides with a local rp-id is rejected
+    (the merged config must stay servable)."""
+    rejected = satellite.apply_domains_push(
+        UPLINK_ID,
+        [
+            {
+                "rp_id": "example.com",
+                "rp_name": "Example",
+                "origins": {"**.example.com": True, TEST_RP_ID: {"auth_host": True}},
+            }
+        ],
+    )
+    assert "example.com" in rejected
+    assert "example.com" not in test_db.config.domains
+
+
 # -------------------------------------------------------------------------
 # API-level: endpoints served from an injected replica
 # -------------------------------------------------------------------------
@@ -289,20 +396,19 @@ async def remote_client(test_db):
     config = _remote_domain_config()
     domains.configure(listen=["localhost:4401"])
     domains.init_registry(config)
+    ops_db.create_uplink(UPLINK_ID, _uplink(), {})
     replica_db, secret = _replica_db()
-    replica = satellite.RemoteReplica(
-        RemoteConfig(url=REMOTE_URL, token="t"), "example.com"
-    )
+    replica = satellite.UplinkReplica(_uplink())
     replica.db = replica_db
     replica.last_contact = time.monotonic()
     replica.connected = True
-    satellite.manager.replicas["example.com"] = replica
+    satellite.manager.replicas[UPLINK_ID] = 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("example.com", None)
+    satellite.manager.replicas.pop(UPLINK_ID, None)
 
 
 @pytest.mark.asyncio
@@ -373,178 +479,58 @@ async def test_logout_proxied_and_evicted(remote_client, monkeypatch):
 
 
 @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
-
+async def test_admin_rejects_remote_writes(client, session_token, test_db):
+    """Remote domains are uplink-managed: the domains API refuses to
+    create, edit or delete them."""
     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}",
+            "origins": {
+                "**.example.com": True,
+                "auth.example.com": {"auth_host": True},
+            },
+            "remote": {"link": "x"},
         },
+        headers=headers,
     )
     assert r.status_code == 400
-    assert "auth host" in r.json()["detail"]
+    assert "uplink" 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}",
+    # An existing remote domain (as an uplink push would have stored it)
+    ops_db.apply_uplink_domains(
+        UPLINK_ID,
+        {
+            "example.com": DomainConfig(
+                rp_name="Example",
+                origins={
+                    "**.example.com": True,
+                    "auth.example.com": OriginEntry(auth_host=True),
+                },
+                remote=RemoteConfig(link=UPLINK_ID),
+            )
         },
     )
-    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
-
-
-@pytest.mark.asyncio
-async def test_admin_remote_url_userinfo_token(client, session_token, test_db):
-    """The sync URL minted on an upstream carries the token as userinfo;
-    it is split off before storing (and never echoed back)."""
-    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": "https://alpha.bravo.charlie.delta.echo@remote.test"},
-        },
-        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.url == "https://remote.test"
-    assert remote.token == "alpha.bravo.charlie.delta.echo"
-
-    # Saving without userinfo keeps the stored token
     r = await client.patch(
         "/auth/api/admin/domains/example.com",
-        json={
-            "rp_name": "Ex",
-            "origins": {
-                "**.example.com": True,
-                "auth.example.com": {"auth_host": True},
-            },
-            "remote": {"url": "https://remote.test"},
-        },
-        headers={
-            "Host": "localhost:4401",
-            "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
-        },
-    )
-    assert r.status_code == 200, r.text
-    assert (
-        test_db.config.domains["example.com"].remote.token
-        == "alpha.bravo.charlie.delta.echo"
+        json={"rp_name": "Ex", "origins": {"**.example.com": True}},
+        headers=headers,
     )
+    assert r.status_code == 400
+    assert "uplink" in r.json()["detail"]
+    r = await client.delete("/auth/api/admin/domains/example.com", headers=headers)
+    assert r.status_code == 400
+    assert "unlink" in r.json()["detail"]
 
 
 def test_feed_scoped_subscription_filters():
-    """A scoped subscriber gets only its rp-id's sessions/credentials
+    """A scoped subscriber gets only its rp-ids' sessions/credentials
     (upserts filtered, deletes passed); the user graph is never published."""
-    scoped = syncfeed.subscribe("example.com")
+    scoped = syncfeed.subscribe({"example.com"})
     try:
         in_scope = Session.create(
             user=UUID(int=1),
@@ -575,6 +561,31 @@ def test_feed_scoped_subscription_filters():
         syncfeed.unsubscribe(scoped)
 
 
+def test_feed_scope_updates_live():
+    """The subscriber scope is mutable in place: a link's scope adjustment
+    takes effect without a reconnect."""
+    scope = {"example.com"}
+    queue = syncfeed.subscribe(scope)
+    try:
+        session = Session.create(
+            user=UUID(int=1),
+            credential=UUID(int=2),
+            key="k1",
+            host="localhost",
+            ip="1.1.1.1",
+            user_agent="ua",
+            validated=datetime.now(UTC),
+            rp_id="localhost",
+        )
+        syncfeed.emit("sessions", "k1", session)
+        assert not _drain(queue)
+        scope.add("localhost")  # scope widened live
+        syncfeed.emit("sessions", "k1", session)
+        assert [e["key"] for e in _drain(queue)] == ["k1"]
+    finally:
+        syncfeed.unsubscribe(queue)
+
+
 def _drain(queue):
     items = []
     while not queue.empty():
@@ -658,18 +669,17 @@ async def test_user_api_proxied_on_remote_auth_host(remote_client, monkeypatch):
 
 
 # -------------------------------------------------------------------------
-# Sync tokens (this instance as the upstream)
+# Satellite links and uplinks (admin API)
 # -------------------------------------------------------------------------
 
 
 @pytest.mark.asyncio
-async def test_sync_token_lifecycle(client, session_token, test_db):
-    """Mint (auth host required), list, revoke — the raw token appears once."""
+async def test_admin_satellite_link_lifecycle(client, session_token, test_db):
+    """Offer minting (auth host required), scope adjustment, revocation."""
     headers = {
         "Host": "localhost:4401",
         "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
     }
-
     r = await client.post(
         "/auth/api/admin/domains/",
         json={
@@ -681,14 +691,13 @@ async def test_sync_token_lifecycle(client, session_token, test_db):
     )
     assert r.status_code == 200, r.text
 
-    # No auth host yet — refused
+    # No auth host yet — the domain cannot be scoped into a link
     r = await client.post(
-        "/auth/api/admin/domains/example.com/sync-token", headers=headers
+        "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers
     )
     assert r.status_code == 400
     assert "auth host" in r.json()["detail"]
 
-    # Mark an auth host, then mint
     r = await client.patch(
         "/auth/api/admin/domains/example.com",
         json={
@@ -701,55 +710,155 @@ async def test_sync_token_lifecycle(client, session_token, test_db):
         headers=headers,
     )
     assert r.status_code == 200, r.text
+
+    # Mint an offer: a wss URL carrying the RAM-only initial key
     r = await client.post(
-        "/auth/api/admin/domains/example.com/sync-token", headers=headers
+        "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers
     )
     assert r.status_code == 200, r.text
     body = r.json()
-    assert body["url"].startswith("https://")
+    assert body["url"].startswith("wss://")
     assert "@auth.example.com" in body["url"]
-    raw_token = body["url"].split("://", 1)[1].split("@", 1)[0]
-    assert raw_token.count(".") == 4  # five-word passphrase
+    key = body["url"].split("://", 1)[1].split("@", 1)[0]
+    assert key.count(".") == 4  # five-word passphrase
+    assert body["expires_in"] > 0
 
-    r = await client.get(
-        "/auth/api/admin/domains/example.com/sync-tokens", headers=headers
+    # Unknown domains and nested scopes are refused
+    r = await client.post(
+        "/auth/api/admin/satellites/",
+        json={"rp_ids": ["nope.example"]},
+        headers=headers,
     )
-    assert [t["id"] for t in r.json()] == [body["id"]]
-    assert "alpha" not in r.text and raw_token not in r.text  # hashes only
+    assert r.status_code == 400
 
-    r = await client.delete(
-        f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers
+    # Register a link directly (the handshake itself is covered live),
+    # then list, adjust scope, revoke
+    ops_db.create_satellite_link("ln1", ["example.com"], b"key", label="sat-1")
+    r = await client.get("/auth/api/admin/satellites/", headers=headers)
+    assert r.status_code == 200
+    (entry,) = r.json()
+    assert entry["id"] == "ln1"
+    assert entry["label"] == "sat-1"
+    assert entry["rp_ids"] == ["example.com"]
+    assert entry["connected"] is False
+
+    r = await client.patch(
+        "/auth/api/admin/satellites/ln1", json={"rp_ids": []}, headers=headers
     )
     assert r.status_code == 200
-    assert not test_db.sync_tokens
-    r = await client.delete(
-        f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers
-    )
+    assert test_db.satellite_links["ln1"].rp_ids == []
+
+    r = await client.delete("/auth/api/admin/satellites/ln1", headers=headers)
+    assert r.status_code == 200
+    assert not test_db.satellite_links
+    r = await client.delete("/auth/api/admin/satellites/ln1", headers=headers)
     assert r.status_code == 400
 
 
 @pytest.mark.asyncio
-async def test_sync_token_refused_for_remote_domain(client, session_token):
-    """A satellite-served domain cannot be an upstream: no nesting."""
+async def test_admin_offer_scope_rejects_remote_domain(client, session_token, test_db):
+    """A satellite-served domain cannot be scoped into a link: no nesting."""
     headers = {
         "Host": "localhost:4401",
         "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
     }
-    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": "https://alpha.bravo.charlie.delta.echo@remote.test"},
+    ops_db.apply_uplink_domains(
+        UPLINK_ID,
+        {
+            "example.com": DomainConfig(
+                origins={
+                    "**.example.com": True,
+                    "auth.example.com": OriginEntry(auth_host=True),
+                },
+                remote=RemoteConfig(link=UPLINK_ID),
+            )
         },
-        headers=headers,
     )
-    assert r.status_code == 200, r.text
     r = await client.post(
-        "/auth/api/admin/domains/example.com/sync-token", headers=headers
+        "/auth/api/admin/satellites/", json={"rp_ids": ["example.com"]}, headers=headers
     )
     assert r.status_code == 400
     assert "nested" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_admin_uplinks_list_and_unlink(client, session_token, test_db):
+    """Uplinks list shows served domains and replica state; unlink removes
+    the uplink and its domains."""
+    headers = {
+        "Host": "localhost:4401",
+        "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+    }
+    ops_db.create_uplink(UPLINK_ID, _uplink(), {})
+    ops_db.apply_uplink_domains(
+        UPLINK_ID,
+        {
+            "example.com": DomainConfig(
+                origins={
+                    "**.example.com": True,
+                    "auth.example.com": OriginEntry(auth_host=True),
+                },
+                remote=RemoteConfig(link=UPLINK_ID),
+            )
+        },
+    )
+    r = await client.get("/auth/api/admin/uplinks/", headers=headers)
+    assert r.status_code == 200
+    (entry,) = r.json()
+    assert entry["id"] == UPLINK_ID
+    assert entry["url"] == REMOTE_URL
+    assert entry["domains"] == ["example.com"]
+    assert entry["connected"] is False
+
+    r = await client.delete(f"/auth/api/admin/uplinks/{UPLINK_ID}", headers=headers)
+    assert r.status_code == 200
+    assert not test_db.uplinks
+    assert "example.com" not in test_db.config.domains
+    r = await client.get("/auth/api/admin/uplinks/", headers=headers)
+    assert r.json() == []
+
+
+@pytest.mark.asyncio
+async def test_admin_register_uplink_endpoint(client, session_token, monkeypatch):
+    """The registration endpoint runs the handshake synchronously (the
+    handshake itself is covered by the live tests) and reports failures."""
+    headers = {
+        "Host": "localhost:4401",
+        "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+    }
+
+    async def fake_register(url):
+        return {
+            "link": "ln1",
+            "url": "https://remote.test",
+            "domains": ["example.com"],
+            "rejected": {},
+        }
+
+    monkeypatch.setattr(satellite, "register_uplink", fake_register)
+    r = await client.post(
+        "/auth/api/admin/uplinks/",
+        json={"url": "wss://alpha.bravo.charlie.delta.echo@remote.test"},
+        headers=headers,
+    )
+    assert r.status_code == 200
+    assert r.json()["domains"] == ["example.com"]
+
+    async def failing_register(url):
+        raise RuntimeError("the upstream rejected the link URL")
+
+    monkeypatch.setattr(satellite, "register_uplink", failing_register)
+    r = await client.post(
+        "/auth/api/admin/uplinks/", json={"url": "wss://x@remote.test"}, headers=headers
+    )
+    assert r.status_code == 400
+    assert "rejected" in r.json()["detail"]
+
+    async def malformed(url):
+        raise ValueError("A link URL is a wss:// URL")
+
+    monkeypatch.setattr(satellite, "register_uplink", malformed)
+    r = await client.post(
+        "/auth/api/admin/uplinks/", json={"url": "nope"}, headers=headers
+    )
+    assert r.status_code == 400

tests/test_satellite_e2e.py

@@ -1,16 +1,20 @@
 """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
+A real uvicorn server plays the upstream (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.
+app with a real UplinkReplica started by the real SatelliteManager. Unlike
+test_remote.py (which injects a warm replica by hand), nothing here is
+stubbed: the 3-way handshake, pushed domains, snapshot, live events,
+write-behind, dead-peer TTL handling, reconnect reconciliation, live scope
+changes and revocation 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.
+database and registry, so the satellite's pushed-domain application is
+arranged to be a no-op (the link's domains are absent from the shared
+config — the satellite registry is synthetic instead), and endpoints the
+satellite *forwards* to the upstream (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
@@ -22,6 +26,7 @@ from types import SimpleNamespace
 from uuid import UUID
 
 import httpx
+import msgspec.structs
 import pytest
 import pytest_asyncio
 import uvicorn
@@ -34,7 +39,9 @@ from paskia.db.structs import (
     DomainConfig,
     OriginEntry,
     RemoteConfig,
+    Uplink,
 )
+from paskia.fastapi import sync
 from paskia.fastapi.mainapp import app
 from paskia.fastapi.session import AUTH_COOKIE_NAME
 
@@ -42,6 +49,8 @@ from .conftest import TEST_RP_ID, create_test_session
 
 REMOTE_DOMAIN = "example.com"
 APP_HOST = "app2.example.com"
+LINK_ID = "live-link"
+LINK_KEY = b"e2e-link-key"
 
 
 def _free_port() -> int:
@@ -60,15 +69,12 @@ async def wait_for(desc: str, condition, timeout: float = 10) -> None:
 
 
 @pytest_asyncio.fixture
-async def live_remote(test_db):
-    """A uvicorn-served 'remote' and the satellite manager wired against it.
+async def live_server(test_db):
+    """A uvicorn-served upstream over a real TCP socket.
 
-    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.
+    Yields a namespace: db, url, port, start_server (for restarts),
+    servers.
     """
-    _, live_token = ops_db.create_sync_token(REMOTE_DOMAIN)
     servers: list[tuple[uvicorn.Server, asyncio.Task]] = []
 
     async def start_server(port: int) -> uvicorn.Server:
@@ -84,7 +90,34 @@ async def live_remote(test_db):
 
     port = _free_port()
     url = f"http://127.0.0.1:{port}"
+    domains.configure(listen=["localhost:4401"])
+    domains.init_registry(test_db.config)
     await start_server(port)
+    try:
+        yield SimpleNamespace(
+            db=test_db, url=url, port=port, start_server=start_server, servers=servers
+        )
+    finally:
+        for server, task in servers:
+            server.should_exit = True
+            await task
+
+
+@pytest_asyncio.fixture
+async def live_remote(live_server):
+    """An established link against the live upstream, and the satellite
+    manager started for real: the replica connects over TCP and runs the
+    established-link handshake (hello {link, key} → welcome → ready).
+
+    Both sides' link rows are stored directly (registration itself is
+    covered by the dedicated tests); the satellite's remote domain lives
+    only in the registry, not the shared config, so pushed-domain
+    application stays a no-op.
+    """
+    live = live_server
+    ops_db.create_satellite_link(LINK_ID, [REMOTE_DOMAIN], LINK_KEY, label="e2e")
+    uplink = Uplink(url=live.url, created=datetime.now(UTC), key=LINK_KEY)
+    ops_db.create_uplink(LINK_ID, uplink, {})
 
     domains.configure(listen=["localhost:4401"])
     domains.init_registry(
@@ -96,13 +129,13 @@ async def live_remote(test_db):
                         f"**.{REMOTE_DOMAIN}": True,
                         f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
                     },
-                    remote=RemoteConfig(url=url, token=live_token),
+                    remote=RemoteConfig(link=LINK_ID),
                 ),
             }
         )
     )
     await satellite.manager.start()
-    replica = satellite.manager.replicas[REMOTE_DOMAIN]
+    replica = satellite.manager.replicas[LINK_ID]
     await wait_for("initial snapshot", lambda: replica.available())
 
     transport = httpx.ASGITransport(app=app)
@@ -110,21 +143,11 @@ async def live_remote(test_db):
         transport=transport, base_url="http://localhost:4401"
     ) as client:
         try:
-            yield SimpleNamespace(
-                client=client,
-                replica=replica,
-                db=test_db,
-                url=url,
-                port=port,
-                token=live_token,
-                start_server=start_server,
-                servers=servers,
-            )
+            live.client = client
+            live.replica = replica
+            yield live
         finally:
             await satellite.manager.stop()
-            for server, task in servers:
-                server.should_exit = True
-                await task
 
 
 def _app_headers(secret: str) -> dict[str, str]:
@@ -134,7 +157,7 @@ def _app_headers(secret: str) -> dict[str, str]:
 @pytest_asyncio.fixture
 async def remote_credential(test_db, test_user):
     """A credential registered on the remote domain. The feed is rp-id
-    scoped: only the domain's own credentials replicate to its satellites."""
+    scoped: only the link's own domains' credentials replicate."""
     credential = Credential.create(
         credential_id=os.urandom(32),
         user=test_user.uuid,
@@ -147,11 +170,72 @@ async def remote_credential(test_db, test_user):
     return credential
 
 
+# -------------------------------------------------------------------------
+# Registration (the real 3-way handshake, crossing TCP)
+# -------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_live_registration(live_server):
+    """A RAM-only offer is spent by the handshake; both sides store the
+    link with identical key material, and the established-link handshake
+    works right after."""
+    live = live_server
+    offer_url = sync.create_offer([], live.url)
+    assert offer_url.startswith("ws://")
+
+    result = await satellite.register_uplink(offer_url)
+    link_id = result["link"]
+    assert result["domains"] == []  # empty scope: register first, scope later
+
+    uplink = live.db.uplinks[link_id]
+    assert uplink.url == live.url
+    link = live.db.satellite_links[link_id]
+    assert uplink.key == link.key and len(link.key) == 64  # both parts
+    assert link.label  # the satellite's self-reported name
+
+    # Re-registering to the same upstream is refused (the offer itself is
+    # single-use — covered in test_remote.py's offer test)
+    with pytest.raises(ValueError, match="Already linked"):
+        await satellite.register_uplink(offer_url)
+
+    # The established-link handshake: the replica connects and syncs
+    await satellite.manager.start()
+    try:
+        replica = satellite.manager.replicas[link_id]
+        await wait_for("established link sync", lambda: replica.available())
+    finally:
+        await satellite.manager.stop()
+
+
+@pytest.mark.asyncio
+async def test_live_registration_rejections(live_server):
+    """Malformed URLs, unknown keys and unreachable hosts all fail the
+    registration without storing anything."""
+    with pytest.raises(ValueError, match="wss"):
+        await satellite.register_uplink("https://key@auth.example.com")
+    with pytest.raises(ValueError, match="key"):
+        await satellite.register_uplink("wss://auth.example.com")
+    with pytest.raises(RuntimeError, match="rejected"):
+        await satellite.register_uplink(
+            f"ws://unknown.key@127.0.0.1:{live_server.port}"
+        )
+    with pytest.raises(RuntimeError, match="unreachable"):
+        await satellite.register_uplink("ws://some.key@127.0.0.1:1")
+    assert not live_server.db.uplinks
+    assert not live_server.db.satellite_links
+
+
+# -------------------------------------------------------------------------
+# Established link: sync behavior
+# -------------------------------------------------------------------------
+
+
 @pytest.mark.asyncio
 async def test_live_snapshot_and_context_repush(
     live_remote, test_user, remote_credential
 ):
-    """Sessions arrive with their master-computed context; user-graph
+    """Sessions arrive with their upstream-computed context; user-graph
     changes eagerly re-push the affected sessions' contexts."""
     live = live_remote
     key, _ = create_test_session(
@@ -207,8 +291,8 @@ async def test_live_permission_revocation_propagates(
 
 @pytest.mark.asyncio
 async def test_live_forward_and_writebehind(live_remote, test_user, remote_credential):
-    """A session created on the remote authorizes locally; the /validate
-    refresh travels back over the wire into the remote's database."""
+    """A session created on the upstream authorizes locally; the /validate
+    refresh travels back over the wire into the upstream's database."""
     live = live_remote
     key, secret = create_test_session(
         user_uuid=test_user.uuid,
@@ -224,7 +308,7 @@ async def test_live_forward_and_writebehind(live_remote, test_user, remote_crede
     assert r.status_code == 204
     assert r.headers["remote-name"] == "Test Admin"
 
-    # Age the session on the remote (propagates), then validate on the satellite
+    # Age the session on the upstream (propagates), then validate on the satellite
     stale = datetime.now(UTC) - timedelta(minutes=10)
     ops_db.update_session(key, validated=stale)
     await wait_for(
@@ -235,7 +319,7 @@ async def test_live_forward_and_writebehind(live_remote, test_user, remote_crede
     assert r.status_code == 200
     assert r.json()["renewed"] is True
     await wait_for(
-        "write-behind to land on the remote",
+        "write-behind to land on the upstream",
         lambda: live.db.sessions[key].validated > stale,
     )
 
@@ -262,10 +346,11 @@ async def test_live_session_delete_propagates(
 
 
 @pytest.mark.asyncio
-async def test_live_wrong_token_rejected(live_remote):
-    bad = satellite.RemoteReplica(
-        RemoteConfig(url=live_remote.url, token="wrong"), REMOTE_DOMAIN
-    )
+async def test_live_wrong_key_rejected(live_remote):
+    """An uplink whose key does not match the stored link's never syncs."""
+    uplink = Uplink(url=live_remote.url, created=datetime.now(UTC), key=b"not-the-key")
+    uplink.uplink_id = "bad-link"
+    bad = satellite.UplinkReplica(uplink)
     await bad.start()
     try:
         await asyncio.sleep(1.5)
@@ -299,7 +384,10 @@ async def test_live_disconnect_ttl_and_reconnect(
     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
+    # Expire the trust window (a copy: the db-owned struct is not mutated)
+    uplink_copy = msgspec.structs.replace(live.replica.uplink, cache_ttl=0)
+    uplink_copy.uplink_id = LINK_ID
+    live.replica.uplink = uplink_copy
     r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
     assert r.status_code == 503
 
@@ -315,21 +403,32 @@ async def test_live_disconnect_ttl_and_reconnect(
 
 
 @pytest.mark.asyncio
-async def test_reconcile_stops_replica_when_remote_removed(live_remote):
-    """A registry rebuild without the remote config stops the replica."""
-    domains.init_registry(
-        Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})})
-    )
+async def test_unlink_stops_replica(live_remote):
+    """Unlinking drops the uplink, its domains and the replica."""
+    live = live_remote
+    await satellite.manager.unlink(LINK_ID)
+    assert LINK_ID not in satellite.manager.replicas
+    assert LINK_ID not in live.db.uplinks
+
+
+@pytest.mark.asyncio
+async def test_live_revocation_unlinks_satellite(live_remote):
+    """Revoking the link upstream drops the connection; the satellite's
+    handshake retry gets a policy close and unlinks its side."""
+    live = live_remote
+    ops_db.delete_satellite_link(LINK_ID)
+    await sync.close_link(LINK_ID)
     await wait_for(
-        "replica shutdown", lambda: REMOTE_DOMAIN not in satellite.manager.replicas
+        "satellite to unlink", lambda: LINK_ID not in live.db.uplinks, timeout=25
     )
+    assert LINK_ID not in satellite.manager.replicas
 
 
 @pytest.mark.asyncio
 async def test_live_out_of_scope_session_not_replicated(
     live_remote, test_user, test_credential, remote_credential
 ):
-    """The feed is scoped to the token's rp-id: sessions of other domains
+    """The feed is scoped to the link's rp-ids: sessions of other domains
     (here the satellite's own) never cross the wire."""
     live = live_remote
     local_key, _ = create_test_session(
@@ -356,8 +455,8 @@ async def test_live_out_of_scope_session_not_replicated(
 async def test_live_out_of_scope_writebehind_ignored(
     live_remote, test_user, test_credential, remote_credential
 ):
-    """A session_refresh for a session outside the token's scope is ignored
-    by the remote (fail-closed), while an in-scope one lands."""
+    """A session_refresh for a session outside the link's scope is ignored
+    by the upstream (fail-closed), while an in-scope one lands."""
     live = live_remote
     stale = datetime.now(UTC) - timedelta(minutes=10)
     local_key, _ = create_test_session(
@@ -386,15 +485,57 @@ async def test_live_out_of_scope_writebehind_ignored(
 
 
 @pytest.mark.asyncio
-async def test_live_scope_mismatch_rejected(live_remote):
-    """A valid token scoped to another domain is useless: the ready message
-    names its scope and the replica refuses to go live on the mismatch."""
-    bad = satellite.RemoteReplica(
-        RemoteConfig(url=live_remote.url, token=live_remote.token), "other.org"
+async def test_live_scope_shrink_stops_feed(live_remote, test_user, remote_credential):
+    """Narrowing the link's scope on the upstream takes effect on the live
+    connection: sessions of removed domains stop flowing."""
+    live = live_remote
+    ops_db.set_satellite_link_scope(LINK_ID, [])
+    await sync.push_link_domains(LINK_ID)
+
+    removed_key, _ = create_test_session(
+        user_uuid=test_user.uuid,
+        credential_uuid=remote_credential.uuid,
+        host=APP_HOST,
+        rp_id=REMOTE_DOMAIN,
     )
-    await bad.start()
-    try:
-        await asyncio.sleep(1.5)
-        assert not bad.available()
-    finally:
-        await bad.stop()
+    # Positive control: widen the scope again; a session created now does
+    # arrive, so the earlier one demonstrably passed a closed pipe.
+    ops_db.set_satellite_link_scope(LINK_ID, [REMOTE_DOMAIN])
+    await sync.push_link_domains(LINK_ID)
+    remote_key, _ = create_test_session(
+        user_uuid=test_user.uuid,
+        credential_uuid=remote_credential.uuid,
+        host=APP_HOST,
+        rp_id=REMOTE_DOMAIN,
+    )
+    await wait_for(
+        "re-scoped session sync", lambda: remote_key in live.replica.db.sessions
+    )
+    assert removed_key not in live.replica.db.sessions
+
+
+@pytest.mark.asyncio
+async def test_live_domain_conflict_reported_upstream(live_remote, test_db):
+    """A pushed domain the satellite serves locally is rejected, and the
+    rejection is reported into the upstream's link state."""
+    # The shared config now holds the domain authoritatively — from the
+    # satellite's view this is the local-conflict case.
+    ops_db.create_domain(
+        REMOTE_DOMAIN,
+        DomainConfig(
+            rp_name="Example",
+            origins={
+                f"**.{REMOTE_DOMAIN}": True,
+                f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
+            },
+        ),
+    )
+    await sync.push_link_domains(LINK_ID)
+    await wait_for(
+        "rejection to arrive",
+        lambda: (
+            REMOTE_DOMAIN in sync.links_status().get(LINK_ID, {}).get("rejected", {})
+        ),
+    )
+    # The satellite kept its (synthetic) registry — no local overwrite
+    assert test_db.config.domains[REMOTE_DOMAIN].remote is None