diff --git a/api/app.py b/api/app.py
index 6b81677..1883116 100644
--- a/api/app.py
+++ b/api/app.py
@@ -44,6 +44,9 @@ from managers import (
     caddy_manager,
     ddns_manager, service_store_manager,
     connectivity_manager,
+    service_registry,
+    service_composer,
+    account_manager,
     firewall_manager, EventType,
 )
 # Re-exports: tests do `from app import CellManager` and `from app import _resolve_peer_dns`
diff --git a/api/caddy_manager.py b/api/caddy_manager.py
index 973592a..ab9279c 100644
--- a/api/caddy_manager.py
+++ b/api/caddy_manager.py
@@ -52,11 +52,13 @@ class CaddyManager(BaseServiceManager):
     def __init__(self, config_manager=None,
                  data_dir: str = '/app/data',
                  config_dir: str = '/app/config',
-                 service_bus=None):
+                 service_bus=None,
+                 service_registry=None):
         super().__init__('caddy', data_dir, config_dir)
         self.config_manager = config_manager
         self.container_name = 'cell-caddy'
         self.caddyfile_path = LIVE_CADDYFILE
+        self._service_registry = service_registry
         # Consecutive health-check failure counter (reset on success or when
         # the caller restarts the container).
         self._health_failures = 0
@@ -187,6 +189,69 @@ class CaddyManager(BaseServiceManager):
             f"    }}"
         )
 
+    def _build_registry_service_routes(self, domain: str) -> str:
+        """Build named-matcher + handle blocks from the service registry.
+
+        Falls back to the hardcoded ``_build_core_service_routes`` when no
+        registry is wired or the registry returns nothing, so the method is
+        always safe to call even in tests that don't supply a registry.
+        """
+        routes: List[Dict] = []
+        if self._service_registry is not None:
+            try:
+                routes = self._service_registry.get_caddy_routes()
+            except Exception as exc:
+                logger.warning('_build_registry_service_routes: registry error: %s', exc)
+
+        if not routes:
+            return self._build_core_service_routes(domain)
+
+        # Pre-seed with reserved names so no registry entry can squat them.
+        seen_matchers: set = {'api', 'webui'}
+
+        blocks: List[str] = []
+        for route in routes:
+            primary_sub = route['subdomain']
+            backend = route['backend']
+            extra_subs: List[str] = route.get('extra_subdomains') or []
+            extra_backends: Dict[str, str] = route.get('extra_backends') or {}
+
+            if primary_sub in seen_matchers:
+                logger.warning('Caddy: skipping duplicate/reserved matcher %r', primary_sub)
+                continue
+            seen_matchers.add(primary_sub)
+
+            # Subdomains that share the primary backend go in one matcher block.
+            shared = [primary_sub] + [s for s in extra_subs if s not in extra_backends]
+            host_list = ' '.join(f'{s}.{domain}' for s in shared)
+            blocks.append(
+                f'    @{primary_sub} host {host_list}\n'
+                f'    handle @{primary_sub} {{\n'
+                f'        reverse_proxy {backend}\n'
+                f'    }}'
+            )
+            # Extra subdomains with their own backends each get their own block.
+            for sub, sub_backend in extra_backends.items():
+                if sub in seen_matchers:
+                    logger.warning('Caddy: skipping duplicate/reserved matcher %r', sub)
+                    continue
+                seen_matchers.add(sub)
+                blocks.append(
+                    f'    @{sub} host {sub}.{domain}\n'
+                    f'    handle @{sub} {{\n'
+                    f'        reverse_proxy {sub_backend}\n'
+                    f'    }}'
+                )
+
+        # The api subdomain is always infrastructure — not delegated to the registry.
+        blocks.append(
+            f'    @api host api.{domain}\n'
+            f'    handle @api {{\n'
+            f'        reverse_proxy cell-api:3000\n'
+            f'    }}'
+        )
+        return '\n'.join(blocks)
+
     @staticmethod
     def _indent_routes(routes: str, spaces: int = 4) -> str:
         """Indent a multi-line route block by ``spaces`` columns."""
@@ -230,7 +295,7 @@ class CaddyManager(BaseServiceManager):
                            service_routes: str, core_routes: str) -> str:
         """pic_ngo mode: wildcard DNS-01 via the pic_ngo plugin."""
         domain = f"{cell_name}.pic.ngo"
-        body = [self._build_core_service_routes(domain)]
+        body = [self._build_registry_service_routes(domain)]
         if service_routes:
             body.append(self._indent_routes(service_routes))
         body.append(core_routes)
@@ -253,7 +318,7 @@ class CaddyManager(BaseServiceManager):
     def _caddyfile_cloudflare(self, custom_domain: str,
                               service_routes: str, core_routes: str) -> str:
         """cloudflare mode: wildcard DNS-01 via the cloudflare plugin."""
-        body = [self._build_core_service_routes(custom_domain)]
+        body = [self._build_registry_service_routes(custom_domain)]
         if service_routes:
             body.append(self._indent_routes(service_routes))
         body.append(core_routes)
@@ -273,7 +338,7 @@ class CaddyManager(BaseServiceManager):
                            service_routes: str, core_routes: str) -> str:
         """duckdns mode: DNS-01 via the duckdns plugin."""
         domain = f"{cell_name}.duckdns.org"
-        body = [self._build_core_service_routes(domain)]
+        body = [self._build_registry_service_routes(domain)]
         if service_routes:
             body.append(self._indent_routes(service_routes))
         body.append(core_routes)
@@ -299,15 +364,8 @@ class CaddyManager(BaseServiceManager):
         out.append(core_routes)
         out.append("}")
 
-        # One block per core service subdomain.
-        _core_services = [
-            ('calendar', 'cell-radicale:5232'),
-            ('mail',     'cell-rainloop:8888'),
-            ('webmail',  'cell-rainloop:8888'),
-            ('files',    'cell-filegator:8080'),
-            ('webdav',   'cell-webdav:80'),
-            ('api',      'cell-api:3000'),
-        ]
+        # Build (subdomain, backend) pairs from registry when available.
+        _core_services = self._http01_service_pairs()
         for subdomain, backend in _core_services:
             out.append("")
             out.append(f"{subdomain}.{host} {{")
@@ -330,6 +388,32 @@ class CaddyManager(BaseServiceManager):
             out.append("}")
         return "\n".join(out) + "\n"
 
+    def _http01_service_pairs(self) -> List[tuple]:
+        """Return (subdomain, backend) pairs for http01 per-host blocks."""
+        pairs: List[tuple] = []
+        if self._service_registry is not None:
+            try:
+                for route in self._service_registry.get_caddy_routes():
+                    pairs.append((route['subdomain'], route['backend']))
+                    extra_subs: List[str] = route.get('extra_subdomains') or []
+                    extra_backends: Dict[str, str] = route.get('extra_backends') or {}
+                    for sub in extra_subs:
+                        backend = extra_backends.get(sub, route['backend'])
+                        pairs.append((sub, backend))
+            except Exception as exc:
+                logger.warning('_http01_service_pairs: registry error: %s', exc)
+                pairs = []
+        if not pairs:
+            pairs = [
+                ('calendar', 'cell-radicale:5232'),
+                ('mail',     'cell-rainloop:8888'),
+                ('webmail',  'cell-rainloop:8888'),
+                ('files',    'cell-filegator:8080'),
+                ('webdav',   'cell-webdav:80'),
+            ]
+        pairs.append(('api', 'cell-api:3000'))
+        return pairs
+
     # ── filesystem + admin-API operations ─────────────────────────────────
 
     def write_caddyfile(self, caddyfile_content: str) -> bool:
diff --git a/api/config_manager.py b/api/config_manager.py
index 252e954..98b39cd 100644
--- a/api/config_manager.py
+++ b/api/config_manager.py
@@ -6,6 +6,8 @@ Centralized configuration management for all services
 
 import os
 import json
+import re
+import subprocess
 import yaml
 import shutil
 import hashlib
@@ -14,6 +16,9 @@ from typing import Dict, List, Optional, Any
 from pathlib import Path
 import logging
 
+_SAFE_CONTAINER_RE = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,63}$')
+_SAFE_VOL_NAME_RE  = re.compile(r'^[a-zA-Z0-9_.-]{1,64}$')
+
 # The Caddyfile lives on a separate volume mount from the rest of config
 LIVE_CADDYFILE = os.environ.get('CADDYFILE_PATH', '/app/config-caddy/Caddyfile')
 
@@ -228,8 +233,128 @@ class ConfigManager:
             "warnings": warnings
         }
     
-    def backup_config(self) -> str:
-        """Create a backup of cell_config.json, secrets, Caddyfile, .env, Corefile, and DNS zones."""
+    @staticmethod
+    def _validate_vol_entry(service_id: str, vol: dict) -> bool:
+        """Return True if a backup volume entry is safe to use; log and return False otherwise."""
+        container = vol.get('container', '')
+        path = vol.get('path', '')
+        name = vol.get('name', '')
+        if not _SAFE_CONTAINER_RE.match(container):
+            logger.warning('Backup: unsafe container name %r for %s — skipping', container, service_id)
+            return False
+        if not path.startswith('/') or '..' in path.split('/') or '\x00' in path:
+            logger.warning('Backup: unsafe volume path %r for %s — skipping', path, service_id)
+            return False
+        if not _SAFE_VOL_NAME_RE.match(name):
+            logger.warning('Backup: unsafe volume name %r for %s — skipping', name, service_id)
+            return False
+        return True
+
+    def _backup_service_volumes(self, backup_path: Path, service_registry) -> None:
+        """Stream service data out of each container via 'docker exec tar'.
+
+        Archives are relative (created with -C <path> .) so they can be safely
+        restored with -C <path> without risk of path traversal outside the volume.
+        Writes to a .partial temp file then renames atomically on success.
+        """
+        try:
+            plan = service_registry.get_backup_plan()
+        except Exception as e:
+            logger.warning('_backup_service_volumes: could not get backup plan: %s', e)
+            return
+        for entry in plan:
+            service_id = entry['service_id']
+            volumes = entry.get('volumes') or []
+            if not volumes:
+                continue
+            svc_dir = backup_path / 'service_data' / service_id
+            svc_dir.mkdir(parents=True, exist_ok=True)
+            for vol in volumes:
+                if not self._validate_vol_entry(service_id, vol):
+                    continue
+                container = vol['container']
+                path = vol['path']
+                name = vol['name']
+                archive_path = svc_dir / f'{name}.tar.gz'
+                tmp_path = svc_dir / f'{name}.tar.gz.partial'
+                try:
+                    with open(tmp_path, 'wb') as af:
+                        result = subprocess.run(
+                            # -C path; then '.' archives the whole dir with relative entries.
+                            # '--' prevents path/container from being parsed as options.
+                            ['docker', 'exec', '--', container,
+                             'tar', '-C', path, '-czf', '-', '.'],
+                            stdout=af,
+                            stderr=subprocess.PIPE,
+                            timeout=300,
+                        )
+                    if result.returncode != 0:
+                        logger.warning(
+                            'Backup: docker exec tar failed for %s/%s: %s',
+                            service_id, name, result.stderr.decode(errors='replace'),
+                        )
+                        tmp_path.unlink(missing_ok=True)
+                    else:
+                        os.replace(tmp_path, archive_path)
+                        logger.info('Backup: archived %s/%s', service_id, name)
+                except subprocess.TimeoutExpired:
+                    logger.warning('Backup: timed out streaming %s/%s', service_id, name)
+                    tmp_path.unlink(missing_ok=True)
+                except Exception as e:
+                    logger.warning('Backup: failed to archive %s/%s: %s', service_id, name, e)
+                    tmp_path.unlink(missing_ok=True)
+
+    def _restore_service_volumes(self, backup_path: Path, service_registry) -> None:
+        """Pipe archived service data back into containers via 'docker exec -i tar'.
+
+        Extracts with -C <path>, matching how archives were created (relative paths).
+        This bounds extraction to within the declared volume directory.
+        """
+        svc_data_dir = backup_path / 'service_data'
+        if not svc_data_dir.is_dir():
+            return
+        for svc_dir in svc_data_dir.iterdir():
+            if not svc_dir.is_dir():
+                continue
+            service_id = svc_dir.name
+            svc = service_registry.get(service_id)
+            if not svc:
+                logger.warning('Restore: unknown service %s in backup, skipping', service_id)
+                continue
+            volumes = (svc.get('backup') or {}).get('volumes') or []
+            for vol in volumes:
+                if not self._validate_vol_entry(service_id, vol):
+                    continue
+                container = vol['container']
+                path = vol['path']
+                name = vol['name']
+                archive_path = svc_dir / f'{name}.tar.gz'
+                if not archive_path.exists():
+                    continue
+                try:
+                    with open(archive_path, 'rb') as af:
+                        result = subprocess.run(
+                            ['docker', 'exec', '-i', '--', container,
+                             'tar', '-C', path, '-xzf', '-'],
+                            stdin=af,
+                            stderr=subprocess.PIPE,
+                            timeout=300,
+                        )
+                    if result.returncode != 0:
+                        logger.warning(
+                            'Restore: docker exec tar failed for %s/%s: %s',
+                            service_id, name, result.stderr.decode(errors='replace'),
+                        )
+                    else:
+                        logger.info('Restore: restored %s/%s', service_id, name)
+                except subprocess.TimeoutExpired:
+                    logger.warning('Restore: timed out restoring %s/%s', service_id, name)
+                except Exception as e:
+                    logger.warning('Restore: failed to restore %s/%s: %s', service_id, name, e)
+
+    def backup_config(self, service_registry=None) -> str:
+        """Create a backup of cell_config.json, secrets, Caddyfile, .env, Corefile, DNS zones,
+        and (when service_registry is provided) live service data volumes."""
         try:
             timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
             backup_id = f"backup_{timestamp}"
@@ -278,12 +403,17 @@ class ConfigManager:
                     except (PermissionError, OSError) as e:
                         logger.warning(f"Could not back up {src.name}: {e} (skipping)")
 
+            # Live service data volumes (streamed via docker exec)
+            if service_registry is not None:
+                self._backup_service_volumes(backup_path, service_registry)
+
             services = ['identity'] + list(self.service_schemas.keys())
             manifest = {
                 "backup_id": backup_id,
                 "timestamp": datetime.now().isoformat(),
                 "services": services,
                 "files": [f.name for f in backup_path.iterdir()],
+                "includes_service_data": service_registry is not None,
             }
             with open(backup_path / 'manifest.json', 'w') as f:
                 json.dump(manifest, f, indent=2)
@@ -294,8 +424,9 @@ class ConfigManager:
         except Exception as e:
             logger.error(f"Error creating backup: {e}")
             raise
-    
-    def restore_config(self, backup_id: str, services: list = None) -> bool:
+
+    def restore_config(self, backup_id: str, services: list = None,
+                       service_registry=None) -> bool:
         """Restore from backup. If services list given, only restore those service configs (selective)."""
         try:
             backup_path = self.backup_dir / backup_id
@@ -373,6 +504,10 @@ class ConfigManager:
                     except (PermissionError, OSError) as e:
                         logger.warning(f"Could not restore {dest.name}: {e} (skipping)")
 
+            # Live service data volumes
+            if service_registry is not None:
+                self._restore_service_volumes(backup_path, service_registry)
+
             self.configs = self._load_all_configs()
             logger.info(f"Restored configuration from backup: {backup_id}")
             return True
diff --git a/api/managers.py b/api/managers.py
index eb57135..9fb6d1e 100644
--- a/api/managers.py
+++ b/api/managers.py
@@ -31,6 +31,9 @@ from setup_manager import SetupManager
 from caddy_manager import CaddyManager
 from ddns_manager import DDNSManager
 from connectivity_manager import ConnectivityManager
+from service_registry import ServiceRegistry
+from service_composer import ServiceComposer
+from account_manager import AccountManager
 
 DATA_DIR   = os.environ.get('DATA_DIR',   '/app/data')
 CONFIG_DIR = os.environ.get('CONFIG_DIR', '/app/config')
@@ -42,7 +45,13 @@ config_manager = ConfigManager(
 service_bus  = ServiceBus()
 log_manager  = LogManager(log_dir='./data/logs')
 
-network_manager    = NetworkManager(data_dir=DATA_DIR, config_dir=CONFIG_DIR)
+# ServiceRegistry depends only on config_manager; create it early so
+# NetworkManager and CaddyManager can derive subdomains from manifests
+# instead of hardcoding service names.
+service_registry = ServiceRegistry(config_manager=config_manager)
+
+network_manager    = NetworkManager(data_dir=DATA_DIR, config_dir=CONFIG_DIR,
+                                    service_registry=service_registry)
 wireguard_manager  = WireGuardManager(data_dir=DATA_DIR, config_dir=CONFIG_DIR)
 peer_registry      = PeerRegistry(data_dir=DATA_DIR, config_dir=CONFIG_DIR)
 email_manager      = EmailManager(data_dir=DATA_DIR, config_dir=CONFIG_DIR, service_bus=service_bus)
@@ -58,7 +67,8 @@ cell_link_manager  = CellLinkManager(
 )
 auth_manager = AuthManager(data_dir=DATA_DIR, config_dir=CONFIG_DIR)
 setup_manager = SetupManager(config_manager=config_manager, auth_manager=auth_manager)
-caddy_manager = CaddyManager(config_manager=config_manager, data_dir=DATA_DIR, config_dir=CONFIG_DIR, service_bus=service_bus)
+caddy_manager = CaddyManager(config_manager=config_manager, data_dir=DATA_DIR, config_dir=CONFIG_DIR,
+                              service_bus=service_bus, service_registry=service_registry)
 ddns_manager = DDNSManager(config_manager=config_manager, data_dir=DATA_DIR, config_dir=CONFIG_DIR)
 connectivity_manager = ConnectivityManager(
     config_manager=config_manager,
@@ -67,6 +77,15 @@ connectivity_manager = ConnectivityManager(
     config_dir=CONFIG_DIR,
 )
 
+service_composer = ServiceComposer(config_manager=config_manager, data_dir=DATA_DIR)
+account_manager  = AccountManager(
+    service_registry=service_registry,
+    data_dir=DATA_DIR,
+    email_manager=email_manager,
+    calendar_manager=calendar_manager,
+    file_manager=file_manager,
+)
+
 from service_store_manager import ServiceStoreManager
 service_store_manager = ServiceStoreManager(
     config_manager=config_manager,
@@ -110,6 +129,7 @@ __all__ = [
     'routing_manager', 'vault_manager', 'container_manager',
     'cell_link_manager', 'auth_manager', 'setup_manager', 'caddy_manager',
     'ddns_manager', 'service_store_manager', 'connectivity_manager',
+    'service_registry', 'service_composer', 'account_manager',
     'firewall_manager', 'EventType',
     'DATA_DIR', 'CONFIG_DIR',
 ]
diff --git a/api/network_manager.py b/api/network_manager.py
index b324400..bb1ce49 100644
--- a/api/network_manager.py
+++ b/api/network_manager.py
@@ -18,11 +18,13 @@ logger = logging.getLogger(__name__)
 class NetworkManager(BaseServiceManager):
     """Manages network services (DNS, DHCP, NTP)"""
     
-    def __init__(self, data_dir: str = '/app/data', config_dir: str = '/app/config'):
+    def __init__(self, data_dir: str = '/app/data', config_dir: str = '/app/config',
+                 service_registry=None):
         super().__init__('network', data_dir, config_dir)
         self.dns_zones_dir = os.path.join(data_dir, 'dns')
         self.dhcp_leases_file = os.path.join(data_dir, 'dhcp', 'leases')
-        
+        self._service_registry = service_registry
+
         # Ensure directories exist
         self.safe_makedirs(self.dns_zones_dir)
         self.safe_makedirs(os.path.dirname(self.dhcp_leases_file))
@@ -201,7 +203,7 @@ class NetworkManager(BaseServiceManager):
         # domain (e.g. primary_domain='pic.ngo', effective_domain='pic2.pic.ngo'),
         # bootstrap service records like 'api', 'calendar' etc. would pollute the
         # zone display and shadow the public domain.  Remove them.
-        _stale = {'api', 'webui', 'calendar', 'files', 'mail', 'webmail', 'webdav'}
+        _stale = {'api', 'webui'} | set(self._get_service_subdomains())
         if effective_domain.endswith('.' + primary_domain):
             existing = self._load_dns_records(primary_domain)
             cleaned = [r for r in existing if r.get('name', '') not in _stale]
@@ -249,6 +251,25 @@ class NetworkManager(BaseServiceManager):
             pass
         return '10.0.0.1'
 
+    _SUBDOMAIN_RE = re.compile(r'^[a-z][a-z0-9-]{0,30}$')
+
+    def _get_service_subdomains(self) -> List[str]:
+        """Return all service subdomains from the registry, or a hardcoded fallback."""
+        registry = getattr(self, "_service_registry", None)
+        if registry is not None:
+            try:
+                subs: List[str] = []
+                for route in registry.get_caddy_routes():
+                    for sub in [route['subdomain']] + list(route.get('extra_subdomains') or []):
+                        if self._SUBDOMAIN_RE.match(sub):
+                            subs.append(sub)
+                        else:
+                            logger.warning('_get_service_subdomains: skipping invalid subdomain %r', sub)
+                return subs
+            except Exception as exc:
+                logger.warning('_get_service_subdomains: registry error: %s', exc)
+        return ['calendar', 'files', 'mail', 'webmail', 'webdav']
+
     def _build_dns_records(self, cell_name: str, ip_range: str) -> List[Dict]:
         """Build the standard set of DNS A records.
 
@@ -258,16 +279,14 @@ class NetworkManager(BaseServiceManager):
         routes requests to the correct backend by Host header.
         """
         wg_ip = self._get_wg_server_ip()
-        return [
-            {'name': cell_name,  'type': 'A', 'value': wg_ip},
-            {'name': 'api',      'type': 'A', 'value': wg_ip},
-            {'name': 'webui',    'type': 'A', 'value': wg_ip},
-            {'name': 'calendar', 'type': 'A', 'value': wg_ip},
-            {'name': 'files',    'type': 'A', 'value': wg_ip},
-            {'name': 'mail',     'type': 'A', 'value': wg_ip},
-            {'name': 'webmail',  'type': 'A', 'value': wg_ip},
-            {'name': 'webdav',   'type': 'A', 'value': wg_ip},
+        records = [
+            {'name': cell_name, 'type': 'A', 'value': wg_ip},
+            {'name': 'api',     'type': 'A', 'value': wg_ip},
+            {'name': 'webui',   'type': 'A', 'value': wg_ip},
         ]
+        for sub in self._get_service_subdomains():
+            records.append({'name': sub, 'type': 'A', 'value': wg_ip})
+        return records
 
     def get_dns_records(self, zone: str = 'cell') -> List[Dict]:
         """Get all DNS records across all zones"""
@@ -595,7 +614,7 @@ class NetworkManager(BaseServiceManager):
         if not new_name:
             return {'restarted': restarted, 'warnings': warnings}
         # Exclude service names, wildcard, and apex from cell-hostname detection.
-        _service_names = {'api', 'webui', 'calendar', 'files', 'mail', 'webmail', 'webdav'}
+        _service_names = {'api', 'webui'} | set(self._get_service_subdomains())
         _reserved = _service_names | {'@', '*'}
         changed = False
         try:
diff --git a/api/routes/config.py b/api/routes/config.py
index 6a05a72..dcb6854 100644
--- a/api/routes/config.py
+++ b/api/routes/config.py
@@ -784,8 +784,8 @@ def apply_pending_config():
 @bp.route('/api/config/backup', methods=['POST'])
 def create_config_backup():
     try:
-        from app import config_manager, service_bus, EventType
-        backup_id = config_manager.backup_config()
+        from app import config_manager, service_bus, service_registry, EventType
+        backup_id = config_manager.backup_config(service_registry=service_registry)
         service_bus.publish_event(EventType.BACKUP_CREATED, 'api', {
             'backup_id': backup_id,
             'timestamp': datetime.utcnow().isoformat()
@@ -809,9 +809,14 @@ def list_config_backups():
 @bp.route('/api/config/restore/<backup_id>', methods=['POST'])
 def restore_config(backup_id):
     try:
-        from app import config_manager, service_bus, EventType
+        from app import config_manager, service_bus, service_registry, EventType
         data = request.get_json(silent=True) or {}
-        success = config_manager.restore_config(backup_id, services=data.get('services'))
+        services = data.get('services')
+        success = config_manager.restore_config(
+            backup_id,
+            services=services,
+            service_registry=service_registry if services is None else None,
+        )
         if success:
             service_bus.publish_event(EventType.RESTORE_COMPLETED, 'api', {
                 'backup_id': backup_id,
diff --git a/api/routes/peers.py b/api/routes/peers.py
index b55f838..7decbe0 100644
--- a/api/routes/peers.py
+++ b/api/routes/peers.py
@@ -125,6 +125,17 @@ def add_peer():
                 return jsonify({"error": f"Peer {peer_name} already exists"}), 400
             peer_added_to_registry = True
 
+            # Store credentials only after the peer is committed — avoids orphaned
+            # credential entries if peer_registry.add_peer rejects a duplicate name.
+            try:
+                from app import account_manager
+                _svc_names = {'email', 'calendar', 'files'}
+                for svc in provisioned:
+                    if svc in _svc_names:
+                        account_manager.store_credentials(svc, peer_name, {'password': password})
+            except Exception as _am_err:
+                logger.warning(f"Peer {peer_name}: credential storage failed (non-fatal): {_am_err}")
+
             firewall_manager.apply_peer_rules(peer_info['ip'], peer_info,
                                               wg_subnet=_wg_subnet, cell_subnets=_cell_subnets)
             firewall_applied = True
@@ -320,12 +331,46 @@ def remove_peer(peer_name):
                 _cleanup()
             except Exception:
                 pass
+        try:
+            from app import account_manager
+            account_manager.deprovision_peer(peer_name)
+        except Exception as _am_err:
+            logger.warning(f"Peer {peer_name}: account_manager cleanup failed (non-fatal): {_am_err}")
         return jsonify({"message": f"Peer {peer_name} removed successfully"})
     except Exception as e:
         logger.error(f"Error removing peer: {e}")
         return jsonify({"error": str(e)}), 500
 
 
+@bp.route('/api/peers/<peer_name>/service-credentials', methods=['GET'])
+def get_peer_service_credentials(peer_name: str):
+    """Return service credentials for a peer across all provisioned services (admin only).
+
+    Returns filled peer_config_template values for each service the peer is provisioned on.
+    Intended for an admin to view or copy credentials to share with the peer during
+    device setup. The global enforce_auth gate already restricts this to admin sessions.
+
+    Phase 2 note: a peer-self-service variant should live at /api/peer/service-credentials
+    (no path arg) and restrict to session['username'] to prevent cross-peer enumeration.
+    """
+    try:
+        from app import peer_registry, account_manager, service_registry, config_manager
+        peer = peer_registry.get_peer(peer_name)
+        if not peer:
+            return jsonify({'error': f'Peer {peer_name!r} not found'}), 404
+        raw_creds = account_manager.get_all_credentials(peer_name)
+        identity = config_manager.get_identity()
+        domain = config_manager.get_effective_domain() or identity.get('domain', '')
+        result = {}
+        for service_id, cred in raw_creds.items():
+            svc_info = service_registry.get_peer_service_info(service_id, peer_name, domain, cred)
+            result[service_id] = svc_info if svc_info is not None else cred
+        return jsonify({'peer': peer_name, 'services': result})
+    except Exception as e:
+        logger.error('get_peer_service_credentials(%s): %s', peer_name, e)
+        return jsonify({'error': str(e)}), 500
+
+
 @bp.route('/api/peers/register', methods=['POST'])
 def register_peer():
     try:
diff --git a/api/routes/services.py b/api/routes/services.py
index 13791e6..267fa95 100644
--- a/api/routes/services.py
+++ b/api/routes/services.py
@@ -6,6 +6,174 @@ from flask import Blueprint, request, jsonify
 logger = logging.getLogger('picell')
 bp = Blueprint('services', __name__)
 
+@bp.route('/api/services/catalog', methods=['GET'])
+def get_services_catalog():
+    """
+    Return all services (builtins + installed store packages) with merged config.
+    Used by the frontend to build navigation and service pages dynamically.
+    """
+    try:
+        from app import service_registry
+        return jsonify({'services': service_registry.list_all()})
+    except Exception as e:
+        logger.error('get_services_catalog: %s', e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>', methods=['GET'])
+def get_service_catalog_entry(service_id: str):
+    """Return a single service manifest+config, or 404 if unknown."""
+    try:
+        from app import service_registry
+        svc = service_registry.get(service_id)
+        if svc is None:
+            return jsonify({'error': f'Service {service_id!r} not found'}), 404
+        return jsonify(svc)
+    except Exception as e:
+        logger.error('get_service_catalog_entry(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/status', methods=['GET'])
+def get_service_container_status(service_id: str):
+    """
+    Return container status for a service.
+    Builtins query the main compose stack; store services query their own compose project.
+    """
+    try:
+        from app import service_registry, service_composer
+        svc = service_registry.get(service_id)
+        if svc is None:
+            return jsonify({'error': f'Service {service_id!r} not found'}), 404
+        result = service_composer.status_service(service_id, svc)
+        return jsonify(result)
+    except ValueError as e:
+        return jsonify({'error': str(e)}), 400
+    except Exception as e:
+        logger.error('get_service_container_status(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/restart', methods=['POST'])
+def restart_service_containers(service_id: str):
+    """
+    Restart containers for a service.
+    Builtins restart via the main compose stack; store services via their own compose project.
+    """
+    try:
+        from app import service_registry, service_composer
+        svc = service_registry.get(service_id)
+        if svc is None:
+            return jsonify({'error': f'Service {service_id!r} not found'}), 404
+        result = service_composer.restart_service(service_id, svc)
+        if result['ok']:
+            return jsonify({'message': f'Service {service_id!r} restarted', **result})
+        return jsonify({'error': result.get('stderr') or result.get('error', 'restart failed')}), 500
+    except ValueError as e:
+        return jsonify({'error': str(e)}), 400
+    except Exception as e:
+        logger.error('restart_service_containers(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/reconfigure', methods=['POST'])
+def reconfigure_service(service_id: str):
+    """
+    Re-apply the stored compose file for a store service (rolling `up -d`).
+    The compose template must already exist on disk from the original install —
+    accepting templates from the request body is deliberately not supported
+    (arbitrary compose files can mount host paths or request privileged mode).
+    """
+    try:
+        from app import service_registry, service_composer
+        svc = service_registry.get(service_id)
+        if svc is None:
+            return jsonify({'error': f'Service {service_id!r} not found'}), 404
+        if svc.get('kind') == 'builtin':
+            return jsonify({'error': 'Builtins are reconfigured via their settings routes'}), 400
+        if not service_composer.has_compose_file(service_id):
+            return jsonify({'error': f'No compose file for {service_id!r} — install it first'}), 400
+
+        result = service_composer.up(service_id)
+        if result['ok']:
+            return jsonify({'message': f'Service {service_id!r} reconfigured', **result})
+        return jsonify({'error': result.get('stderr') or result.get('error', 'reconfigure failed')}), 500
+    except ValueError as e:
+        return jsonify({'error': str(e)}), 400
+    except Exception as e:
+        logger.error('reconfigure_service(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/accounts', methods=['GET'])
+def list_service_accounts(service_id: str):
+    """Return peer usernames provisioned on a service."""
+    try:
+        from app import account_manager
+        accounts = account_manager.list_accounts(service_id)
+        return jsonify({'service_id': service_id, 'accounts': accounts})
+    except Exception as e:
+        logger.error('list_service_accounts(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/accounts', methods=['POST'])
+def provision_service_account(service_id: str):
+    """Provision a peer account on a service. Generates a password if none is given.
+
+    The generated or provided password is NOT echoed in this response — retrieve it
+    separately via GET /api/services/catalog/<id>/accounts/<username>/credentials.
+    This keeps passwords out of HTTP logs and browser network panels.
+    """
+    try:
+        from app import account_manager
+        data = request.get_json(silent=True) or {}
+        peer_username = data.get('username')
+        if not peer_username:
+            return jsonify({'error': 'username is required'}), 400
+        account_manager.provision(service_id, peer_username,
+                                  password=data.get('password'))
+        return jsonify({'service_id': service_id, 'username': peer_username,
+                        'provisioned': True}), 201
+    except ValueError as e:
+        return jsonify({'error': str(e)}), 400
+    except RuntimeError as e:
+        return jsonify({'error': str(e)}), 500
+    except Exception as e:
+        logger.error('provision_service_account(%s): %s', service_id, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/accounts/<username>', methods=['DELETE'])
+def deprovision_service_account(service_id: str, username: str):
+    """Remove a peer's account from a service."""
+    try:
+        from app import account_manager
+        ok = account_manager.deprovision(service_id, username)
+        if ok:
+            return jsonify({'message': f'{username!r} deprovisioned from {service_id!r}'})
+        return jsonify({'error': 'deprovision failed'}), 500
+    except ValueError as e:
+        return jsonify({'error': str(e)}), 400
+    except Exception as e:
+        logger.error('deprovision_service_account(%s, %s): %s', service_id, username, e)
+        return jsonify({'error': str(e)}), 500
+
+
+@bp.route('/api/services/catalog/<service_id>/accounts/<username>/credentials', methods=['GET'])
+def get_service_account_credentials(service_id: str, username: str):
+    """Return stored credentials for a peer on a service."""
+    try:
+        from app import account_manager
+        creds = account_manager.get_credentials(service_id, username)
+        if creds is None:
+            return jsonify({'error': f'{username!r} not provisioned on {service_id!r}'}), 404
+        return jsonify({'service_id': service_id, 'username': username, **creds})
+    except Exception as e:
+        logger.error('get_service_account_credentials(%s, %s): %s', service_id, username, e)
+        return jsonify({'error': str(e)}), 500
+
+
 @bp.route('/api/services/bus/status', methods=['GET'])
 def get_service_bus_status():
     try:
diff --git a/api/service_registry.py b/api/service_registry.py
new file mode 100644
index 0000000..f3267d8
--- /dev/null
+++ b/api/service_registry.py
@@ -0,0 +1,218 @@
+"""
+ServiceRegistry — single source of truth for all PIC services.
+
+Merges three layers:
+  1. Manifest defaults (config_schema.*.default)
+  2. Admin-saved config from ConfigManager (cell_config.json)
+  3. Runtime state from installed store records
+
+All consumers (CaddyManager, backup, peer services endpoint) read from here
+rather than hardcoding service names or subdomains.
+"""
+
+import json
+import logging
+import os
+import re
+from typing import Dict, List, Optional
+from urllib.parse import quote as _urlquote
+
+logger = logging.getLogger('picell')
+
+# Built-ins are baked into the container image at build time.
+# Do not bind-mount this path read-write in docker-compose.
+_BUILTINS_DIR = os.path.join(os.path.dirname(__file__), 'services', 'builtins')
+
+_SUBDOMAIN_RE = re.compile(r'^[a-z][a-z0-9-]{0,30}$')
+_BACKEND_RE   = re.compile(r'^[A-Za-z0-9._-]+:\d{1,5}$')
+_RESERVED_SUBS = frozenset({'api', 'webui', 'admin', 'www', 'ns1', 'ns2', 'git', 'registry', 'install'})
+
+
+class ServiceRegistry:
+
+    def __init__(self, config_manager):
+        self._cm = config_manager
+
+    # ── Manifest loading ──────────────────────────────────────────────────
+
+    def _load_manifest(self, path: str) -> Optional[Dict]:
+        try:
+            with open(path) as f:
+                return json.load(f)
+        except Exception as e:
+            logger.warning('ServiceRegistry: failed to load manifest %s: %s', path, e)
+            return None
+
+    def _builtin_ids(self) -> List[str]:
+        if not os.path.isdir(_BUILTINS_DIR):
+            return []
+        return sorted(
+            d for d in os.listdir(_BUILTINS_DIR)
+            if os.path.isfile(os.path.join(_BUILTINS_DIR, d, 'manifest.json'))
+        )
+
+    def _builtin_manifest(self, service_id: str) -> Optional[Dict]:
+        return self._load_manifest(
+            os.path.join(_BUILTINS_DIR, service_id, 'manifest.json')
+        )
+
+    # ── Config merging ────────────────────────────────────────────────────
+
+    _TYPE_COERCIONS = {'integer': int, 'string': str, 'boolean': bool}
+
+    def _merged_config(self, manifest: Dict) -> Dict:
+        """Return manifest defaults overridden by admin-saved values, type-coerced."""
+        svc_id = manifest.get('id', '')
+        saved = self._cm.configs.get(svc_id, {})
+        schema = manifest.get('config_schema') or {}
+        merged = {k: v['default'] for k, v in schema.items() if 'default' in v}
+        for k, spec in schema.items():
+            if k not in saved:
+                continue
+            raw = saved[k]
+            coerce = self._TYPE_COERCIONS.get(spec.get('type', ''))
+            if coerce is not None:
+                try:
+                    raw = coerce(raw)
+                except (TypeError, ValueError):
+                    raw = merged.get(k, raw)
+            merged[k] = raw
+        return merged
+
+    # ── Public API ────────────────────────────────────────────────────────
+
+    def get(self, service_id: str) -> Optional[Dict]:
+        """Return manifest + merged config for one service, or None if unknown."""
+        manifest = self._builtin_manifest(service_id)
+        if manifest is None:
+            record = self._cm.get_installed_services().get(service_id)
+            if record:
+                manifest = record.get('manifest')
+        if not manifest:
+            return None
+        return {**manifest, 'config': self._merged_config(manifest)}
+
+    def list_all(self) -> List[Dict]:
+        """
+        Return all services — builtins first, then installed store services —
+        each with merged config attached as the 'config' key.
+        """
+        results: List[Dict] = []
+        seen: set = set()
+
+        for svc_id in self._builtin_ids():
+            manifest = self._builtin_manifest(svc_id)
+            if manifest:
+                results.append({**manifest, 'config': self._merged_config(manifest)})
+                seen.add(svc_id)
+
+        for svc_id, record in self._cm.get_installed_services().items():
+            if svc_id in seen:
+                continue
+            manifest = record.get('manifest') or {}
+            if manifest.get('id'):
+                results.append({**manifest, 'config': self._merged_config(manifest)})
+
+        return results
+
+    def get_caddy_routes(self) -> List[Dict]:
+        """
+        Return routing info for all services that have a subdomain.
+        Used by CaddyManager to build service blocks without hardcoding.
+
+        Values are validated here as a chokepoint so Caddyfile/DNS builders
+        can safely interpolate them regardless of how manifests reached disk.
+        """
+        routes = []
+        for svc in self.list_all():
+            caps = svc.get('capabilities') or {}
+            if not caps.get('has_subdomain'):
+                continue
+            sub = svc.get('subdomain', '')
+            bknd = svc.get('backend', '')
+            if not sub or not bknd:
+                continue
+            svc_id = svc.get('id', '?')
+            if not _SUBDOMAIN_RE.match(sub) or sub in _RESERVED_SUBS:
+                logger.warning('ServiceRegistry: skipping %s — invalid/reserved subdomain %r', svc_id, sub)
+                continue
+            if not _BACKEND_RE.match(bknd):
+                logger.warning('ServiceRegistry: skipping %s — invalid backend %r', svc_id, bknd)
+                continue
+            extra_subs = [
+                s for s in (svc.get('extra_subdomains') or [])
+                if isinstance(s, str) and _SUBDOMAIN_RE.match(s) and s not in _RESERVED_SUBS
+            ]
+            extra_backends = {
+                k: v for k, v in (svc.get('extra_backends') or {}).items()
+                if (isinstance(k, str) and _SUBDOMAIN_RE.match(k) and k not in _RESERVED_SUBS
+                    and isinstance(v, str) and _BACKEND_RE.match(v))
+            }
+            routes.append({
+                'service_id': svc_id,
+                'subdomain': sub,
+                'backend': bknd,
+                'extra_subdomains': extra_subs,
+                'extra_backends': extra_backends,
+            })
+        return routes
+
+    def get_backup_plan(self) -> List[Dict]:
+        """
+        Return backup declarations for all services that have storage.
+        Used by the backup system instead of hardcoded file lists.
+
+        Each entry:
+          service_id   — service identifier
+          volumes      — list of {container, path, name} for docker-exec streaming
+          config_paths — host-relative paths copied directly (config files)
+        """
+        plan = []
+        for svc in self.list_all():
+            caps = svc.get('capabilities') or {}
+            if not caps.get('has_storage'):
+                continue
+            backup = svc.get('backup') or {}
+            volumes = backup.get('volumes') or []
+            config_paths = backup.get('config_paths') or []
+            if not volumes and not config_paths:
+                continue
+            plan.append({
+                'service_id': svc['id'],
+                'volumes': volumes,
+                'config_paths': config_paths,
+            })
+        return plan
+
+    def get_peer_service_info(self, service_id: str, peer_username: str,
+                              domain: str, credentials: Dict) -> Optional[Dict]:
+        """
+        Fill peer_config_template for one service+peer combination.
+        credentials: dict of {field_name: value} for that peer+service.
+        Returns None if service unknown or has no peer template.
+        """
+        svc = self.get(service_id)
+        if not svc:
+            return None
+        template = svc.get('peer_config_template')
+        if not template:
+            return None
+
+        # URL-safe peer username (safe='') — prevents path traversal in CalDAV/WebDAV URLs
+        safe_username = _urlquote(peer_username, safe='')
+
+        result = {}
+        for key, raw in template.items():
+            val = raw
+            val = val.replace('{domain}', domain)
+            val = val.replace('{peer.username}', safe_username)
+            for field, cred_val in credentials.items():
+                val = val.replace(
+                    '{peer.service_credentials.' + service_id + '.' + field + '}',
+                    str(cred_val) if cred_val is not None else '',
+                )
+            cfg = svc.get('config') or {}
+            for cfg_key, cfg_val in cfg.items():
+                val = val.replace('{config.' + cfg_key + '}', str(cfg_val) if cfg_val is not None else '')
+            result[key] = val
+        return result
diff --git a/api/service_store_manager.py b/api/service_store_manager.py
index 911b2ad..a9857fe 100644
--- a/api/service_store_manager.py
+++ b/api/service_store_manager.py
@@ -51,6 +51,8 @@ RESERVED_SUBDOMAINS = frozenset([
     'git', 'registry', 'install',
 ])
 ENV_VALUE_RE = re.compile(r'^[A-Za-z0-9._@:/+\-= ]*$')
+SUBDOMAIN_RE  = re.compile(r'^[a-z][a-z0-9-]{0,30}$')
+BACKEND_RE    = re.compile(r'^[A-Za-z0-9._-]+:\d{1,5}$')
 
 
 # ---------------------------------------------------------------------------
@@ -141,19 +143,55 @@ class ServiceStoreManager(BaseServiceManager):
                     f'iptables_rules[].proto must be tcp or udp, got: {proto}'
                 )
 
-        # Caddy route subdomain
+        # Legacy caddy_route dict subdomain (for store manifests using the old format)
         caddy_route = m.get('caddy_route') or {}
         if isinstance(caddy_route, dict):
-            subdomain = caddy_route.get('subdomain', '')
+            legacy_sub = caddy_route.get('subdomain', '')
         else:
-            subdomain = ''
-        if subdomain:
-            if subdomain in RESERVED_SUBDOMAINS:
-                errors.append(f'caddy_route.subdomain is reserved: {subdomain}')
-            elif not re.match(r'^[a-z][a-z0-9-]{0,30}$', subdomain):
+            legacy_sub = ''
+        if legacy_sub:
+            if legacy_sub in RESERVED_SUBDOMAINS:
+                errors.append(f'caddy_route.subdomain is reserved: {legacy_sub}')
+            elif not SUBDOMAIN_RE.match(legacy_sub):
                 errors.append(
                     f'caddy_route.subdomain must match ^[a-z][a-z0-9-]{{0,30}}$, '
-                    f'got: {subdomain}'
+                    f'got: {legacy_sub}'
+                )
+
+        # Top-level subdomain + backend (consumed by ServiceRegistry.get_caddy_routes)
+        subdomain = m.get('subdomain', '')
+        if subdomain:
+            if subdomain in RESERVED_SUBDOMAINS:
+                errors.append(f'subdomain is reserved: {subdomain}')
+            elif not SUBDOMAIN_RE.match(subdomain):
+                errors.append(
+                    f'subdomain must match ^[a-z][a-z0-9-]{{0,30}}$, got: {subdomain}'
+                )
+
+        backend = m.get('backend', '')
+        if backend and not BACKEND_RE.match(backend):
+            errors.append(f'backend must be host:port (e.g. cell-foo:8080), got: {backend}')
+
+        for sub in m.get('extra_subdomains') or []:
+            if not isinstance(sub, str):
+                errors.append('extra_subdomains entries must be strings')
+            elif sub in RESERVED_SUBDOMAINS:
+                errors.append(f'extra_subdomains entry is reserved: {sub}')
+            elif not SUBDOMAIN_RE.match(sub):
+                errors.append(
+                    f'extra_subdomains entry must match ^[a-z][a-z0-9-]{{0,30}}$, got: {sub}'
+                )
+
+        for sub, bknd in (m.get('extra_backends') or {}).items():
+            if not isinstance(sub, str) or not SUBDOMAIN_RE.match(sub):
+                errors.append(
+                    f'extra_backends key must match ^[a-z][a-z0-9-]{{0,30}}$, got: {sub!r}'
+                )
+            elif sub in RESERVED_SUBDOMAINS:
+                errors.append(f'extra_backends key is reserved: {sub}')
+            if not isinstance(bknd, str) or not BACKEND_RE.match(bknd):
+                errors.append(
+                    f'extra_backends[{sub!r}] value must be host:port, got: {bknd!r}'
                 )
 
         # Env value safety
diff --git a/tests/test_caddy_registry_integration.py b/tests/test_caddy_registry_integration.py
new file mode 100644
index 0000000..cc7904d
--- /dev/null
+++ b/tests/test_caddy_registry_integration.py
@@ -0,0 +1,528 @@
+"""Integration tests for registry-driven CaddyManager and NetworkManager routing.
+
+These tests cover the new registry path introduced in Step 5 of the PIC Services
+Architecture.  The no-registry (fallback) paths are already covered by
+test_caddy_manager.py and test_network_manager.py.
+"""
+
+import os
+import sys
+import shutil
+import tempfile
+import unittest
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))
+
+from caddy_manager import CaddyManager  # noqa: E402
+from network_manager import NetworkManager  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+def _mgr_with_registry(registry=None):
+    """Build a CaddyManager wired to an optional mock registry."""
+    cm = MagicMock()
+    cm.get_identity.return_value = {}
+    return CaddyManager(config_manager=cm, service_registry=registry)
+
+
+def _mock_registry():
+    """Return a mock ServiceRegistry that reproduces the 3 builtin service routes."""
+    reg = MagicMock()
+    reg.get_caddy_routes.return_value = [
+        {
+            'service_id': 'calendar',
+            'subdomain': 'calendar',
+            'backend': 'cell-radicale:5232',
+            'extra_subdomains': [],
+            'extra_backends': {},
+        },
+        {
+            'service_id': 'email',
+            'subdomain': 'mail',
+            'backend': 'cell-rainloop:8888',
+            'extra_subdomains': ['webmail'],
+            'extra_backends': {},
+        },
+        {
+            'service_id': 'files',
+            'subdomain': 'files',
+            'backend': 'cell-filegator:8080',
+            'extra_subdomains': ['webdav'],
+            'extra_backends': {'webdav': 'cell-webdav:80'},
+        },
+    ]
+    return reg
+
+
+def _nm(registry=None):
+    """Build a NetworkManager backed by temp dirs and an optional mock registry."""
+    tmpdir = tempfile.mkdtemp()
+    nm = NetworkManager(
+        data_dir=os.path.join(tmpdir, 'data'),
+        config_dir=os.path.join(tmpdir, 'config'),
+        service_registry=registry,
+    )
+    nm._tmpdir = tmpdir  # stash so the caller can clean up
+    return nm
+
+
+# ---------------------------------------------------------------------------
+# TestBuildRegistryServiceRoutes
+# ---------------------------------------------------------------------------
+
+class TestBuildRegistryServiceRoutes(unittest.TestCase):
+
+    def test_returns_hardcoded_when_no_registry(self):
+        """service_registry=None produces the same output as _build_core_service_routes."""
+        mgr = _mgr_with_registry(registry=None)
+        domain = 'alpha.pic.ngo'
+        result = mgr._build_registry_service_routes(domain)
+        expected = CaddyManager._build_core_service_routes(domain)
+        self.assertEqual(result, expected)
+
+    def test_returns_hardcoded_when_registry_empty(self):
+        """An empty route list from the registry falls back to hardcoded."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = []
+        mgr = _mgr_with_registry(registry=reg)
+        domain = 'alpha.pic.ngo'
+        result = mgr._build_registry_service_routes(domain)
+        expected = CaddyManager._build_core_service_routes(domain)
+        self.assertEqual(result, expected)
+
+    def test_registry_error_falls_back(self):
+        """When get_caddy_routes raises, output equals _build_core_service_routes."""
+        reg = MagicMock()
+        reg.get_caddy_routes.side_effect = Exception('registry unavailable')
+        mgr = _mgr_with_registry(registry=reg)
+        domain = 'alpha.pic.ngo'
+        result = mgr._build_registry_service_routes(domain)
+        expected = CaddyManager._build_core_service_routes(domain)
+        self.assertEqual(result, expected)
+
+    def test_single_service_no_extras(self):
+        """One service with no extra_subdomains produces one matcher + handle + api block."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'calendar',
+                'subdomain': 'calendar',
+                'backend': 'cell-radicale:5232',
+                'extra_subdomains': [],
+                'extra_backends': {},
+            }
+        ]
+        mgr = _mgr_with_registry(registry=reg)
+        result = mgr._build_registry_service_routes('test.cell')
+        self.assertIn('@calendar host calendar.test.cell', result)
+        self.assertIn('reverse_proxy cell-radicale:5232', result)
+        self.assertIn('@api host api.test.cell', result)
+        self.assertIn('reverse_proxy cell-api:3000', result)
+        # Only two named-matcher definition lines: @calendar and @api
+        matcher_lines = [l for l in result.splitlines() if l.strip().startswith('@') and 'host' in l]
+        self.assertEqual(len(matcher_lines), 2)
+
+    def test_extra_subdomain_same_backend(self):
+        """An extra_subdomain NOT in extra_backends shares the primary matcher host line."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'email',
+                'subdomain': 'mail',
+                'backend': 'cell-rainloop:8888',
+                'extra_subdomains': ['webmail'],
+                'extra_backends': {},  # webmail not listed → shares backend
+            }
+        ]
+        mgr = _mgr_with_registry(registry=reg)
+        result = mgr._build_registry_service_routes('test.cell')
+        # Both subdomains appear in the same host matcher line
+        self.assertIn('@mail host mail.test.cell webmail.test.cell', result)
+        # Only one reverse_proxy for cell-rainloop (shared block)
+        self.assertEqual(result.count('reverse_proxy cell-rainloop:8888'), 1)
+        # No separate @webmail block
+        self.assertNotIn('@webmail host', result)
+
+    def test_extra_subdomain_different_backend(self):
+        """An extra_subdomain listed in extra_backends gets its own matcher + handle block."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'files',
+                'subdomain': 'files',
+                'backend': 'cell-filegator:8080',
+                'extra_subdomains': ['webdav'],
+                'extra_backends': {'webdav': 'cell-webdav:80'},
+            }
+        ]
+        mgr = _mgr_with_registry(registry=reg)
+        result = mgr._build_registry_service_routes('test.cell')
+        # files gets its own block (webdav not in shared list)
+        self.assertIn('@files host files.test.cell', result)
+        self.assertIn('reverse_proxy cell-filegator:8080', result)
+        # webdav gets a separate block
+        self.assertIn('@webdav host webdav.test.cell', result)
+        self.assertIn('reverse_proxy cell-webdav:80', result)
+        # webdav must NOT appear in the @files host line
+        files_line = [l for l in result.splitlines() if '@files host' in l][0]
+        self.assertNotIn('webdav', files_line)
+
+    def test_api_always_appended(self):
+        """The @api block is always the last block even when registry has no api entry."""
+        reg = _mock_registry()
+        mgr = _mgr_with_registry(registry=reg)
+        result = mgr._build_registry_service_routes('alpha.pic.ngo')
+        self.assertIn('@api host api.alpha.pic.ngo', result)
+        self.assertIn('reverse_proxy cell-api:3000', result)
+        # api block is at the end
+        api_idx = result.rfind('@api')
+        other_matchers = ['@calendar', '@mail', '@files', '@webdav']
+        for m in other_matchers:
+            self.assertLess(result.index(m), api_idx,
+                            f'{m} should appear before @api')
+
+    def test_api_not_duplicated_when_registry_returns_api(self):
+        """Even if registry somehow returns an 'api' route, the injected api block is cell-api:3000."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'api',
+                'subdomain': 'api',
+                'backend': 'cell-other:9999',  # wrong backend — should be overridden
+                'extra_subdomains': [],
+                'extra_backends': {},
+            }
+        ]
+        mgr = _mgr_with_registry(registry=reg)
+        result = mgr._build_registry_service_routes('test.cell')
+        # The infrastructure api block is always appended with the canonical backend
+        self.assertIn('reverse_proxy cell-api:3000', result)
+        # api host matcher appears at least once (from registry AND from append)
+        self.assertGreaterEqual(result.count('@api host api.test.cell'), 1)
+
+
+# ---------------------------------------------------------------------------
+# TestHttp01ServicePairs
+# ---------------------------------------------------------------------------
+
+class TestHttp01ServicePairs(unittest.TestCase):
+
+    def test_pairs_from_registry(self):
+        """With the 3 builtins the pairs list matches expected (subdomain, backend) tuples."""
+        reg = _mock_registry()
+        mgr = _mgr_with_registry(registry=reg)
+        pairs = mgr._http01_service_pairs()
+        pairs_dict = dict(pairs)
+        self.assertEqual(pairs_dict['calendar'], 'cell-radicale:5232')
+        self.assertEqual(pairs_dict['mail'], 'cell-rainloop:8888')
+        self.assertEqual(pairs_dict['webmail'], 'cell-rainloop:8888')
+        self.assertEqual(pairs_dict['files'], 'cell-filegator:8080')
+        self.assertEqual(pairs_dict['webdav'], 'cell-webdav:80')
+        self.assertEqual(pairs_dict['api'], 'cell-api:3000')
+
+    def test_webdav_gets_own_backend(self):
+        """webdav must map to cell-webdav:80, not to cell-filegator:8080."""
+        reg = _mock_registry()
+        mgr = _mgr_with_registry(registry=reg)
+        pairs = mgr._http01_service_pairs()
+        webdav_entry = next((b for s, b in pairs if s == 'webdav'), None)
+        self.assertIsNotNone(webdav_entry)
+        self.assertEqual(webdav_entry, 'cell-webdav:80')
+        self.assertNotEqual(webdav_entry, 'cell-filegator:8080')
+
+    def test_fallback_when_no_registry(self):
+        """Without a registry the hardcoded pairs are returned, including api."""
+        mgr = _mgr_with_registry(registry=None)
+        pairs = mgr._http01_service_pairs()
+        subdomains = [s for s, _ in pairs]
+        self.assertIn('calendar', subdomains)
+        self.assertIn('mail', subdomains)
+        self.assertIn('webmail', subdomains)
+        self.assertIn('files', subdomains)
+        self.assertIn('webdav', subdomains)
+        self.assertIn('api', subdomains)
+
+    def test_fallback_when_registry_error(self):
+        """When get_caddy_routes raises, falls back to hardcoded pairs."""
+        reg = MagicMock()
+        reg.get_caddy_routes.side_effect = RuntimeError('boom')
+        mgr = _mgr_with_registry(registry=reg)
+        pairs = mgr._http01_service_pairs()
+        subdomains = [s for s, _ in pairs]
+        self.assertIn('calendar', subdomains)
+        self.assertIn('api', subdomains)
+
+
+# ---------------------------------------------------------------------------
+# TestCaddyfileWithRegistry
+# ---------------------------------------------------------------------------
+
+class TestCaddyfileWithRegistry(unittest.TestCase):
+
+    def _generate(self, domain_mode, cell_name='alpha', domain_name=None,
+                  registry=None, services=None):
+        reg = registry if registry is not None else _mock_registry()
+        mgr = _mgr_with_registry(registry=reg)
+        identity = {'cell_name': cell_name, 'domain_mode': domain_mode}
+        if domain_name:
+            identity['domain_name'] = domain_name
+        return mgr.generate_caddyfile(identity, services or [])
+
+    def test_pic_ngo_with_registry_has_correct_routes(self):
+        """pic_ngo Caddyfile has all service matchers with correct subdomains and backends."""
+        out = self._generate('pic_ngo', cell_name='alpha')
+        # calendar
+        self.assertIn('@calendar host calendar.alpha.pic.ngo', out)
+        self.assertIn('reverse_proxy cell-radicale:5232', out)
+        # mail + webmail share one matcher
+        self.assertIn('@mail host mail.alpha.pic.ngo webmail.alpha.pic.ngo', out)
+        self.assertIn('reverse_proxy cell-rainloop:8888', out)
+        # files
+        self.assertIn('@files host files.alpha.pic.ngo', out)
+        self.assertIn('reverse_proxy cell-filegator:8080', out)
+        # webdav separate block
+        self.assertIn('@webdav host webdav.alpha.pic.ngo', out)
+        self.assertIn('reverse_proxy cell-webdav:80', out)
+        # api always present
+        self.assertIn('@api host api.alpha.pic.ngo', out)
+        self.assertIn('reverse_proxy cell-api:3000', out)
+
+    def test_cloudflare_with_registry_uses_registry_routes(self):
+        """cloudflare Caddyfile routes are sourced from registry, not hardcoded."""
+        out = self._generate('cloudflare', cell_name='beta',
+                             domain_name='example.com')
+        self.assertIn('@calendar host calendar.example.com', out)
+        self.assertIn('@mail host mail.example.com webmail.example.com', out)
+        self.assertIn('@files host files.example.com', out)
+        self.assertIn('@webdav host webdav.example.com', out)
+        self.assertIn('@api host api.example.com', out)
+        # Correct DNS plugin block is still present
+        self.assertIn('dns cloudflare {$CF_API_TOKEN}', out)
+
+    def test_duckdns_with_registry_uses_registry_routes(self):
+        """duckdns Caddyfile routes are sourced from registry."""
+        out = self._generate('duckdns', cell_name='gamma')
+        self.assertIn('@calendar host calendar.gamma.duckdns.org', out)
+        self.assertIn('@api host api.gamma.duckdns.org', out)
+        self.assertIn('dns duckdns {$DUCKDNS_TOKEN}', out)
+
+    def test_http01_with_registry_has_per_host_blocks(self):
+        """http01 Caddyfile has individual per-host blocks for every service subdomain."""
+        out = self._generate('http01', cell_name='delta',
+                             domain_name='delta.noip.me')
+        self.assertIn('calendar.delta.noip.me {', out)
+        self.assertIn('mail.delta.noip.me {', out)
+        self.assertIn('webmail.delta.noip.me {', out)
+        self.assertIn('files.delta.noip.me {', out)
+        self.assertIn('webdav.delta.noip.me {', out)
+        self.assertIn('api.delta.noip.me {', out)
+        # Correct backends
+        self.assertIn('reverse_proxy cell-radicale:5232', out)
+        self.assertIn('reverse_proxy cell-rainloop:8888', out)
+        self.assertIn('reverse_proxy cell-filegator:8080', out)
+        self.assertIn('reverse_proxy cell-webdav:80', out)
+
+    def test_pic_ngo_fallback_when_registry_empty(self):
+        """pic_ngo falls back to hardcoded routes when registry returns empty list."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = []
+        out = self._generate('pic_ngo', cell_name='alpha', registry=reg)
+        # Hardcoded routes should appear
+        self.assertIn('@calendar host calendar.alpha.pic.ngo', out)
+        self.assertIn('@mail host mail.alpha.pic.ngo webmail.alpha.pic.ngo', out)
+
+
+# ---------------------------------------------------------------------------
+# TestNetworkManagerGetServiceSubdomains
+# ---------------------------------------------------------------------------
+
+class TestNetworkManagerGetServiceSubdomains(unittest.TestCase):
+
+    def setUp(self):
+        self.managers = []
+
+    def tearDown(self):
+        for nm in self.managers:
+            shutil.rmtree(nm._tmpdir, ignore_errors=True)
+
+    def _make(self, registry=None):
+        nm = _nm(registry=registry)
+        self.managers.append(nm)
+        return nm
+
+    def test_no_registry_returns_hardcoded(self):
+        """Without a registry the hardcoded service subdomain list is returned."""
+        nm = self._make(registry=None)
+        subs = nm._get_service_subdomains()
+        self.assertCountEqual(subs, ['calendar', 'files', 'mail', 'webmail', 'webdav'])
+
+    def test_registry_returns_all_subdomains(self):
+        """Primary + extra_subdomains from all routes are returned."""
+        reg = _mock_registry()
+        nm = self._make(registry=reg)
+        subs = nm._get_service_subdomains()
+        # calendar (primary), mail (primary), webmail (extra), files (primary), webdav (extra)
+        for expected in ('calendar', 'mail', 'webmail', 'files', 'webdav'):
+            self.assertIn(expected, subs)
+
+    def test_registry_error_falls_back(self):
+        """When get_caddy_routes raises, hardcoded list is returned."""
+        reg = MagicMock()
+        reg.get_caddy_routes.side_effect = Exception('broken registry')
+        nm = self._make(registry=reg)
+        subs = nm._get_service_subdomains()
+        self.assertCountEqual(subs, ['calendar', 'files', 'mail', 'webmail', 'webdav'])
+
+    def test_registry_extra_subdomains_included(self):
+        """extra_subdomains from each route are included in the returned list."""
+        reg = MagicMock()
+        reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'files',
+                'subdomain': 'files',
+                'backend': 'cell-filegator:8080',
+                'extra_subdomains': ['webdav', 'dav'],
+                'extra_backends': {},
+            }
+        ]
+        nm = self._make(registry=reg)
+        subs = nm._get_service_subdomains()
+        self.assertIn('files', subs)
+        self.assertIn('webdav', subs)
+        self.assertIn('dav', subs)
+
+    def test_build_dns_records_with_registry(self):
+        """All registry subdomains appear as A records in _build_dns_records output."""
+        reg = _mock_registry()
+        nm = self._make(registry=reg)
+        # Override WG IP lookup so we get a predictable value
+        nm._get_wg_server_ip = lambda: '10.0.0.1'
+        records = nm._build_dns_records('mycell', '172.20.0.0/16')
+        names = [r['name'] for r in records]
+        for expected in ('mycell', 'api', 'webui', 'calendar', 'mail',
+                         'webmail', 'files', 'webdav'):
+            self.assertIn(expected, names,
+                          f'{expected!r} should be in DNS records but is not')
+        # All records must point to the WG server IP
+        for r in records:
+            self.assertEqual(r['value'], '10.0.0.1')
+            self.assertEqual(r['type'], 'A')
+
+
+# ---------------------------------------------------------------------------
+# TestNetworkManagerStaleSet
+# ---------------------------------------------------------------------------
+
+class TestNetworkManagerStaleSet(unittest.TestCase):
+    """Verify that registry subdomains drive stale record cleanup in update_split_horizon_zone."""
+
+    def setUp(self):
+        self.test_dir = tempfile.mkdtemp()
+        data_dir = os.path.join(self.test_dir, 'data')
+        config_dir = os.path.join(self.test_dir, 'config')
+        os.makedirs(os.path.join(data_dir, 'dns'), exist_ok=True)
+        os.makedirs(os.path.join(config_dir, 'dns'), exist_ok=True)
+        self.reg = _mock_registry()
+        self.nm = NetworkManager(
+            data_dir=data_dir,
+            config_dir=config_dir,
+            service_registry=self.reg,
+        )
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir, ignore_errors=True)
+
+    def _write_zone(self, zone_name: str, content: str):
+        path = os.path.join(self.nm.dns_zones_dir, f'{zone_name}.zone')
+        with open(path, 'w') as f:
+            f.write(content)
+
+    def test_stale_set_includes_registry_subdomains(self):
+        """Registry subdomains (calendar, mail, webmail, files, webdav) are treated as
+        stale service records and removed from the parent zone during
+        update_split_horizon_zone."""
+        import subprocess
+        # Build a parent zone with stale service records that the registry knows about
+        stale_records = [
+            {'name': 'pic2',     'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'api',      'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'webui',    'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'calendar', 'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'mail',     'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'webmail',  'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'files',    'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'webdav',   'type': 'A', 'value': '10.0.0.1'},
+        ]
+        from unittest.mock import patch
+        with patch('subprocess.run'):
+            self.nm.update_dns_zone('pic.ngo', stale_records)
+            self.nm.update_split_horizon_zone(
+                'pic2.pic.ngo', '172.20.0.2', primary_domain='pic.ngo'
+            )
+
+        parent_zone = os.path.join(self.nm.dns_zones_dir, 'pic.ngo.zone')
+        content = open(parent_zone).read()
+
+        # All registry subdomains must be gone
+        for stale in ('api', 'webui', 'calendar', 'mail', 'webmail', 'files', 'webdav'):
+            # Check that no line *starts* with the stale name (to avoid false positives
+            # on SOA/NS lines that may contain the zone name as a suffix)
+            lines_with_stale = [
+                l for l in content.splitlines()
+                if l.startswith(stale + ' ') or l.startswith(stale + '\t')
+            ]
+            self.assertEqual(
+                lines_with_stale, [],
+                f'Stale record {stale!r} should have been removed from pic.ngo zone'
+            )
+
+    def test_stale_set_uses_registry_not_hardcoded(self):
+        """When a registry provides a custom subdomain, it is treated as stale too."""
+        custom_reg = MagicMock()
+        custom_reg.get_caddy_routes.return_value = [
+            {
+                'service_id': 'chat',
+                'subdomain': 'chat',
+                'backend': 'cell-chat:9000',
+                'extra_subdomains': ['im'],
+                'extra_backends': {},
+            }
+        ]
+        data_dir = os.path.join(self.test_dir, 'data2')
+        config_dir = os.path.join(self.test_dir, 'config2')
+        os.makedirs(os.path.join(data_dir, 'dns'), exist_ok=True)
+        os.makedirs(os.path.join(config_dir, 'dns'), exist_ok=True)
+        nm = NetworkManager(data_dir=data_dir, config_dir=config_dir,
+                            service_registry=custom_reg)
+
+        stale_records = [
+            {'name': 'pic3',  'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'chat',  'type': 'A', 'value': '10.0.0.1'},
+            {'name': 'im',    'type': 'A', 'value': '10.0.0.1'},
+        ]
+        from unittest.mock import patch
+        with patch('subprocess.run'):
+            nm.update_dns_zone('pic.ngo', stale_records)
+            nm.update_split_horizon_zone(
+                'pic3.pic.ngo', '172.20.0.2', primary_domain='pic.ngo'
+            )
+
+        parent_zone = os.path.join(nm.dns_zones_dir, 'pic.ngo.zone')
+        content = open(parent_zone).read()
+        for stale in ('chat', 'im'):
+            lines_with_stale = [
+                l for l in content.splitlines()
+                if l.startswith(stale + ' ') or l.startswith(stale + '\t')
+            ]
+            self.assertEqual(
+                lines_with_stale, [],
+                f'Custom registry subdomain {stale!r} should have been removed'
+            )
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_config_backup_restore_http.py b/tests/test_config_backup_restore_http.py
index 59cf932..e3e771d 100644
--- a/tests/test_config_backup_restore_http.py
+++ b/tests/test_config_backup_restore_http.py
@@ -119,14 +119,17 @@ class TestRestoreConfigBackup(unittest.TestCase):
             content_type='application/json',
         )
         mock_cm.restore_config.assert_called_once_with(
-            'backup_001', services=['network', 'wireguard']
+            'backup_001', services=['network', 'wireguard'], service_registry=None
         )
 
     @patch('app.config_manager')
     def test_restore_passes_none_services_when_no_body(self, mock_cm):
+        from unittest.mock import ANY
         mock_cm.restore_config.return_value = True
         self.client.post('/api/config/restore/backup_001')
-        mock_cm.restore_config.assert_called_once_with('backup_001', services=None)
+        mock_cm.restore_config.assert_called_once_with(
+            'backup_001', services=None, service_registry=ANY
+        )
 
 
 class TestExportConfig(unittest.TestCase):