Merge develop into local application icon discovery

AppImage/scripts/build_appimage.sh

@@ -151,7 +151,9 @@ cp "$SCRIPT_DIR/lxc_update_finalization.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/lxc_check_job.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/native_app_discovery.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/local_app_icons.py" "$APP_DIR/usr/bin/"
+cp "$SCRIPT_DIR/docker_app_discovery.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/lxc_apps.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  lxc_apps.py not found"
+cp "$SCRIPT_DIR/oci_menu_bridge.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/custom_links.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  custom_links.py not found"
 cp "$SCRIPT_DIR/recreate_docker_container.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  recreate_docker_container.py not found"
 chmod +x "$APP_DIR/usr/bin/recreate_docker_container.py" 2>/dev/null || true

AppImage/scripts/docker_app_discovery.py

@@ -0,0 +1,301 @@
+#!/usr/bin/env python3
+"""Explicit Docker endpoint discovery; no registry, updater or registration.
+
+The Docker inventory remains the owner of image updates. These candidates
+describe running containers with published TCP endpoints for user review.
+Only selected inspect fields leave the guest; Config.Env is never collected.
+"""
+import copy
+from contextlib import suppress
+import hashlib
+import ipaddress
+import json
+import os
+import re
+import selectors
+import signal
+import subprocess
+import threading
+import time
+
+import native_app_discovery as native
+
+MAX_CONTAINERS = 50
+OUTPUT_BYTES = 256 * 1024
+TOTAL_SECONDS = 20
+PROBE_SECONDS = 8
+CACHE_SECONDS = 300
+SNAPSHOT_SECONDS = 60
+# Port hints prefill only common web endpoints; other TCP ports remain evidence.
+WEB_PORTS = {80, 443, 3000, 3001, 5000, 5001, 8000, 8008, 8080, 8081, 8083, 8090, 8096, 8123, 8443, 9000, 9090, 9443}
+_cache = {}
+_locks = {}
+_generations = {}
+_mutex = threading.RLock()
+_NAME = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z')
+_ID = re.compile(r'[0-9a-f]{12,64}\Z')
+_VERSION = re.compile(r'v?\d+(?:\.\d+){1,4}(?:[-+._][A-Za-z0-9.-]+)?\Z')
+_SUPPORT = re.compile(r'(?:postgres(?:ql)?|mysql|mariadb|redis|valkey|memcached|mongo(?:db)?|rabbitmq|watchtower)(?:[_.-].*)?\Z', re.I)
+# A template is evaluated for each object even when another ID disappears.
+# Docker CLI inspect/inspect.go writes successful objects before returning an
+# error for failed IDs. Parse stdout and retain those objects as partial data.
+INSPECT_FORMAT = (
+    '{"id":{{json .Id}},"name":{{json .Name}},"image":{{json .Config.Image}},'
+    '"running":{{json .State.Running}},'
+    '"project":{{json (index .Config.Labels "com.docker.compose.project")}},'
+    '"service":{{json (index .Config.Labels "com.docker.compose.service")}},'
+    '"title":{{json (index .Config.Labels "org.opencontainers.image.title")}},'
+    '"version":{{json (index .Config.Labels "org.opencontainers.image.version")}},'
+    '"bindings":{{json .HostConfig.PortBindings}},"ports":{{json .NetworkSettings.Ports}}}'
+)
+
+
+def read(argv, deadline):
+    """Bound combined stdout/stderr during collection, including failed CLI calls."""
+    remaining = min(PROBE_SECONDS, deadline - time.monotonic())
+    if remaining <= 0:
+        raise native.ProbeError('deadline')
+    end = time.monotonic() + remaining
+    try:
+        proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                start_new_session=True)
+    except OSError:
+        raise native.ProbeError('unavailable') from None
+    outputs = [bytearray(), bytearray()]
+    try:
+        with selectors.DefaultSelector() as poll:
+            for index, stream in enumerate((proc.stdout, proc.stderr)):
+                os.set_blocking(stream.fileno(), False)
+                poll.register(stream, selectors.EVENT_READ, index)
+            while poll.get_map():
+                wait = end - time.monotonic()
+                if wait <= 0:
+                    raise native.ProbeError('timeout')
+                for key, _ in poll.select(wait):
+                    block = os.read(key.fileobj.fileno(), 4096)
+                    if not block:
+                        poll.unregister(key.fileobj)
+                        continue
+                    if sum(map(len, outputs)) + len(block) > OUTPUT_BYTES:
+                        raise native.ProbeError('output_limit')
+                    outputs[key.data].extend(block)
+            try:
+                proc.wait(timeout=max(0.001, end - time.monotonic()))
+            except subprocess.TimeoutExpired:
+                raise native.ProbeError('timeout') from None
+        return proc.returncode, outputs[0].decode('utf-8', errors='strict')
+    except UnicodeError:
+        raise native.ProbeError('invalid_data') from None
+    finally:
+        with suppress(ProcessLookupError):
+            os.killpg(proc.pid, signal.SIGKILL)
+        proc.wait()
+        proc.stdout.close()
+        proc.stderr.close()
+
+
+def _text(value, limit=128):
+    return value if isinstance(value, str) and len(value) <= limit and not any(ord(c) < 32 for c in value) else ''
+
+
+def public_ports(record):
+    """Published TCP mappings, excluding the full IPv4/IPv6 loopback ranges."""
+    bindings, live = record.get('bindings'), record.get('ports')
+    if bindings is not None and not isinstance(bindings, dict) or live is not None and not isinstance(live, dict):
+        raise ValueError('invalid ports')
+    ports = dict(bindings or {})
+    ports.update({key: value for key, value in (live or {}).items() if value})
+    result = {}
+    for endpoint, values in ports.items():
+        if not re.fullmatch(r'\d{1,5}/tcp', endpoint) or values is None:
+            continue
+        inner = int(endpoint.split('/')[0])
+        if not 1 <= inner <= 65535 or not isinstance(values, list):
+            raise ValueError('invalid binding')
+        for value in values:
+            if not isinstance(value, dict):
+                raise ValueError('invalid binding')
+            address = value.get('HostIp') or '0.0.0.0'
+            raw = value.get('HostPort')
+            # An unassigned ("") or ranged host port is not a single address.
+            if not isinstance(raw, str) or not raw.isdecimal():
+                continue
+            try:
+                ip = ipaddress.ip_address(address)
+            except (ValueError, TypeError):
+                continue
+            if ip.is_loopback or getattr(ip, 'ipv4_mapped', None) and ip.ipv4_mapped.is_loopback:
+                continue
+            outer = int(raw)
+            if not 1 <= outer <= 65535:
+                continue
+            result[outer] = {'host_port': outer, 'container_port': inner,
+                             'scheme': 'https' if inner in {443, 4443, 8443, 9443} or outer in {443, 4443, 8443, 9443} else 'http'}
+    return sorted(result.values(), key=lambda item: item['host_port'])
+
+
+def parse_record(value):
+    if not isinstance(value, dict):
+        raise ValueError('invalid record')
+    name = _text(value.get('name')).lstrip('/')
+    image = _text(value.get('image'), 512)
+    if not _ID.fullmatch(_text(value.get('id'))) or not _NAME.fullmatch(name) or not image or not isinstance(value.get('running'), bool):
+        raise ValueError('invalid identity')
+    project, service = _text(value.get('project')), _text(value.get('service'))
+    if project and not _NAME.fullmatch(project) or service and not _NAME.fullmatch(service):
+        raise ValueError('invalid compose identity')
+    version = _text(value.get('version'), 64)
+    return {'id': value['id'], 'name': name, 'image': image, 'running': value['running'],
+            'project': project, 'service': service, 'title': _text(value.get('title')),
+            'version': version if _VERSION.fullmatch(version) else None,
+            'endpoints': public_ports(value)}
+
+
+def _candidate(vmid, record):
+    base = record['image'].split('@')[0].rsplit('/')[-1].split(':')[0]
+    if not record['running'] or not record['endpoints'] or _SUPPORT.fullmatch(base) or _SUPPORT.fullmatch(record['service']):
+        return None
+    key = ':'.join([record['project'], record['service']]) if record['project'] and record['service'] else record['name']
+    identity = 'docker-' + hashlib.sha256(f'{vmid}:{key}'.encode()).hexdigest()[:24]
+    # A display name is metadata, not proof of application identity/version.
+    name = record['title'] or record['service'] or record['name']
+    evidence = [{'kind': 'docker_container', 'value': record['name']},
+                {'kind': 'docker_image', 'value': record['image']}]
+    if record['project']:
+        evidence.append({'kind': 'compose_project', 'value': record['project']})
+    evidence.extend({'kind': 'tcp_listener', 'value': str(port['host_port'])} for port in record['endpoints'])
+    return {'slug': identity, 'discovery_id': identity, 'source': 'docker_metadata',
+            'container_name': record['name'], 'name': name,
+            'installed_version': record['version'], 'evidence': evidence,
+            'default_ports': [port['host_port'] for port in record['endpoints'] if port['container_port'] in WEB_PORTS],
+            'port_details': [{'port': port['host_port'], 'scheme': port['scheme'], 'path': None,
+                              'description': '', 'logo_url': ''}
+                             for port in record['endpoints'] if port['container_port'] in WEB_PORTS] or None,
+            'tracking_suggestion': None}
+
+
+def scan(vmid, run=read, guard_run=native.bounded_run):
+    deadline = time.monotonic() + TOTAL_SECONDS
+    native._guard(vmid, deadline, guard_run)
+    with native._scan_reservation(vmid):
+        prefix = ['pct', 'exec', str(vmid), '--']
+        rc, _ = run(prefix + ['sh', '-c', 'command -v docker >/dev/null 2>&1'], deadline)
+        if rc not in (0, 1):
+            raise native.ProbeError('read_failed')
+        records, issues = [], []
+        if rc == 0:
+            rc, output = run(prefix + ['docker', 'ps', '-aq', '--no-trunc'], deadline)
+            if rc:
+                raise native.ProbeError('docker_unavailable')
+            ids = list(dict.fromkeys(output.splitlines()))
+            if any(not _ID.fullmatch(value) for value in ids):
+                raise native.ProbeError('invalid_data')
+            if len(ids) > MAX_CONTAINERS:
+                issues.append('container_limit')
+            ids = ids[:MAX_CONTAINERS]
+            if ids:
+                rc, output = run(prefix + ['docker', 'inspect', '--format', INSPECT_FORMAT, *ids], deadline)
+                if rc:
+                    issues.append('inspect_failed')
+                seen = set()
+                for line in output.splitlines():
+                    try:
+                        record = parse_record(json.loads(line))
+                        if record['id'] not in ids or record['id'] in seen:
+                            raise ValueError('unexpected record')
+                        seen.add(record['id'])
+                        records.append(record)
+                    except (ValueError, TypeError):
+                        issues.append('invalid_data')
+                if seen != set(ids):
+                    issues.append('incomplete_inspect')
+        native._guard(vmid, deadline, guard_run)
+        candidates = {}
+        for record in records:
+            candidate = _candidate(vmid, record)
+            if candidate:
+                candidates.setdefault(candidate['slug'], candidate)
+        return {'vmid': vmid, 'status': 'partial' if issues else 'complete',
+                'issues': sorted(set(issues)), 'stale': False, 'scanned_at': time.time(),
+                'candidates': list(candidates.values()), '_containers': records}
+
+
+def cached(vmid):
+    with _mutex:
+        value = copy.deepcopy(_cache.get(int(vmid)))
+    if value is None:
+        return {'vmid': int(vmid), 'status': 'not_scanned', 'candidates': [], 'stale': False, 'issues': []}
+    if time.time() - value['scanned_at'] >= CACHE_SECONDS:
+        value['stale'] = True
+    return {key: val for key, val in value.items() if not key.startswith('_')}
+
+
+def mark_stale(vmid):
+    with _mutex:
+        _generations[int(vmid)] = _generations.get(int(vmid), 0) + 1
+        if int(vmid) in _cache:
+            _cache[int(vmid)]['stale'] = True
+            _cache[int(vmid)]['_retired'] = True
+
+
+def discover(vmid):
+    vmid = int(vmid)
+    with _mutex:
+        lock = _locks.setdefault(vmid, threading.Lock())
+    if not lock.acquire(blocking=False):
+        return {**cached(vmid), 'status': 'busy', 'stale': True}
+    try:
+        with _mutex:
+            generation = _generations.get(vmid, 0)
+        try:
+            result = scan(vmid)
+        except native.ProbeError as exc:
+            result = {**cached(vmid), 'status': 'error', 'issues': [str(exc)], 'stale': True}
+            result.setdefault('scanned_at', time.time())
+        with _mutex:
+            previous = copy.deepcopy(_cache.get(vmid))
+            if generation != _generations.get(vmid, 0):
+                return {**cached(vmid), 'status': 'error', 'issues': ['guest_changed'], 'stale': True}
+            if result['status'] != 'complete' and previous:
+                known = {item['discovery_id'] for item in result['candidates']}
+                result['candidates'].extend(item for item in previous['candidates'] if item['discovery_id'] not in known)
+                result['candidates'] = result['candidates'][:MAX_CONTAINERS]
+                containers = {item['name']: item for item in previous.get('_containers', [])}
+                containers.update({item['name']: item for item in result.get('_containers', [])})
+                result['_containers'] = list(containers.values())[:MAX_CONTAINERS]
+                result['stale'] = True
+                result['scanned_at'] = previous['scanned_at']
+            result['_attempt_at'] = time.time()
+            _cache[vmid] = copy.deepcopy(result)
+        return cached(vmid)
+    finally:
+        lock.release()
+
+
+def cached_web_links(vmid, lookup):
+    """Reuse a Find snapshot in existing web-link suggestions without another scan."""
+    with _mutex:
+        value = copy.deepcopy(_cache.get(int(vmid)))
+    # Only a complete search lists every container; otherwise probe as before.
+    if (not value or value.get('_retired') or value.get('status') != 'complete'
+            or time.time() - value['_attempt_at'] >= SNAPSHOT_SECONDS):
+        return None
+    result = []
+    for record in value.get('_containers', []):
+        meta = lookup(record['service'], record['name'], record['image'])
+        for endpoint in record['endpoints']:
+            result.append({**endpoint, 'container_name': record['name'],
+                           'service_name': meta['name'], 'service_slug': meta['slug'],
+                           'image': record['image'], 'web_path': '/', 'logo_url': meta['logo_url']})
+    return sorted(result, key=lambda item: (item["service_name"].lower(), item["host_port"]))
+
+
+def filter_candidates(result, suggestions, apps):
+    """Suppress existing registrations/catalog entries; no probe or persistence."""
+    known = [item for item in [*apps, *(suggestions.get('docker_workloads') or []), *(suggestions.get('extras') or [])] if isinstance(item, dict)]
+    ids = {item.get('discovery_id') for item in known}
+    containers = {item.get('container_name') or (item.get('tracking_suggestion') or {}).get('container_name') for item in known}
+    ports = {p.get('port') for app in apps if isinstance(app, dict) for p in app.get('ports') or [] if isinstance(p, dict)}
+    return {**result, 'candidates': [item for item in result['candidates']
+            if item['discovery_id'] not in ids and item['container_name'] not in containers
+            and not (item['default_ports'] and set(item['default_ports']).issubset(ports))]}

AppImage/scripts/flask_pbs_routes.py

@@ -24,11 +24,30 @@ def platform():
 @require_auth
 def servers():
     try:
-        return jsonify(servers=pbs.public_servers(), local_available=pbs.is_pbs_host())
+        servers = pbs.public_servers()
+        return jsonify(servers=servers, local_available=pbs.local_available(), detected=_detected(servers))
     except pbs.PBSError as exc:
         return jsonify(error=str(exc)), 500
 
 
+def _detected(servers):
+    """PBS storages of this PVE that have no monitoring connection yet."""
+    if pbs.is_pbs_host():
+        return []
+    import monitor_links
+    from urllib.parse import urlsplit
+    known = {urlsplit(row.get("url", "")).hostname for row in servers}
+    found = {}
+    for row in monitor_links._storage_pbs():
+        host = row["server"]
+        if host in known or monitor_links._is_this_host(host):
+            continue
+        port = row.get("port") if str(row.get("port", "")).isdigit() else monitor_links.PBS_UI_PORT
+        url = monitor_links._host_url(host, port, "https")
+        found.setdefault(url, {"name": row["name"], "url": url, "fingerprint": row.get("fingerprint", "")})
+    return list(found.values())
+
+
 def _protected_configuration():
     local = auth_manager.load_auth_config()
     return local.get("enabled") and not local.get("declined") and local.get("password_hash")

AppImage/scripts/flask_server.py

@@ -2280,6 +2280,8 @@ def _handle_guest_lifecycle(vmid: str, vm_type: str, action: str) -> None:
     if guest_type == 'lxc':
         import native_app_discovery
         native_app_discovery.mark_stale(guest_id)
+        import docker_app_discovery
+        docker_app_discovery.mark_stale(guest_id)
         _invalidate_lxc_ip(guest_id)
     if action in ('start', 'reboot'):
         if guest_type == 'lxc':
@@ -14159,6 +14161,20 @@ def api_vm_apps_logo_cache(vmid, token):
     })
 
 
+@app.route('/api/vms/<int:vmid>/apps/docker-discovery', methods=['GET', 'POST'])
+@require_auth
+def api_vm_apps_docker_discovery(vmid):
+    """New discovery contract: explicit POST scans; GET only composes cached data."""
+    import docker_app_discovery
+    import lxc_apps
+    result = (docker_app_discovery.discover(vmid) if request.method == 'POST'
+              else docker_app_discovery.cached(vmid))
+    suggestions = _vm_cache_get(_vm_app_suggestions_cache, vmid, _VM_APP_SUGGESTIONS_TTL) or {}
+    sidecar = lxc_apps.load_sidecar(vmid) or {}
+    return jsonify(docker_app_discovery.filter_candidates(
+        result, suggestions, sidecar.get('apps') or []))
+
+
 @app.route('/api/vms/<int:vmid>/apps/adguard-setup', methods=['GET'])
 @require_auth
 def api_vm_apps_adguard_setup(vmid):
@@ -23176,6 +23192,16 @@ if __name__ == '__main__':
     import sys
     import logging
 
+    # The root terminal menu shares the Monitor's checker and sidecar cache.
+    # This local socket changes neither the web API nor its authentication.
+    try:
+        if not is_pbs_host():
+            import lxc_apps
+            import oci_menu_bridge
+            oci_menu_bridge.start(lxc_apps)
+    except Exception as _e:
+        print(f"[startup] OCI menu bridge unavailable: {_e}", file=sys.stderr)
+
     # Regenerate /etc/cron.d/proxmenux-smart from the current schedule
     # config on every Monitor start. Pre-existing installs may carry
     # a cron file with the broken script path (without /storage/),

AppImage/scripts/lxc_apps.py

@@ -1184,7 +1184,7 @@ def validate_config(payload: dict, *, vmid: Optional[int] = None) -> tuple[bool,
     # Native discovery identity is separate from a curated helper/updater.
     discovery_id = payload.get("discovery_id")
     if discovery_id is not None:
-        if not isinstance(discovery_id, str) or not re.fullmatch(r"native-[0-9a-f]{24}", discovery_id):
+        if not isinstance(discovery_id, str) or not re.fullmatch(r"(?:native|docker)-[0-9a-f]{24}", discovery_id):
             return _err("invalid discovery_id")
         conf["discovery_id"] = discovery_id
 
@@ -4340,8 +4340,87 @@ def _detect_with_alt_healing(vmid, app: dict) -> tuple:
     return installed, err, False
 
 
+_oci_check_locks: dict = {}
+_oci_check_locks_guard = threading.Lock()
+
+
 def check_app(
     vmid, app_id: str, force: bool = False, notify: bool = True,
+) -> Optional[dict]:
+    # Keep legacy detectors' locking unchanged. OCI network requests use a
+    # per-app lock and hold the shared cache lock only while reading/saving.
+    with _cache_lock:
+        sidecar = _read_sidecar(vmid)
+        app = _find_app(sidecar, app_id) if sidecar else None
+        if not app or app.get("installed_via") != "oci_image":
+            return _check_non_oci_app(vmid, app_id, force=force, notify=notify)
+    return _check_oci_app(vmid, app_id, force=force, notify=notify)
+
+
+def _check_oci_app(vmid, app_id, force=False, notify=True):
+    key = (int(vmid), app_id)
+    with _oci_check_locks_guard:
+        lock = _oci_check_locks.setdefault(key, threading.Lock())
+    with lock:
+        with _cache_lock:
+            sidecar = _read_sidecar(vmid)
+            app = _find_app(sidecar, app_id) if sidecar else None
+            if not app or app.get("installed_via") != "oci_image":
+                return sidecar
+            state = app.get("state") or _empty_state()
+            if not force and state.get("checked_at"):
+                try:
+                    checked = datetime.datetime.strptime(state["checked_at"].rstrip("Z"), "%Y-%m-%dT%H:%M:%S")
+                    age = (datetime.datetime.utcnow() - checked).total_seconds()
+                    if 0 <= age < _UPSTREAM_CACHE_TTL_SEC:
+                        return sidecar
+                except (ValueError, TypeError):
+                    pass
+            before_app = copy.deepcopy(app)
+            before_record = _read_oci_record(vmid)
+
+        # No global cache lock is held while waiting for registries. The same
+        # native resolver, credentials and digest comparisons are retained.
+        result = _oci_image_versions(
+            vmid, known=state, with_latest=not _oci_secondary_member(before_record))
+        if result.get("busy"):
+            _recheck_after_oci_operation(vmid, app_id)
+            return _read_sidecar(vmid)
+
+        with _cache_lock:
+            # Re-read rather than writing the pre-query copy: other apps,
+            # schedules and registrations may have changed while we waited.
+            sidecar = _read_sidecar(vmid)
+            app = _find_app(sidecar, app_id) if sidecar else None
+            if app != before_app or _read_oci_record(vmid) != before_record:
+                return sidecar  # Never publish a result for a replaced/edited app.
+            app["state"] = {
+                "installed_version": result.get("installed_version"),
+                "latest_version": result.get("latest_version"),
+                "latest_published_at": None,
+                "update_available": result.get("update_available"),
+                "error": result.get("error"),
+                "checked_at": _now_iso(),
+                "installed_digest": result.get("installed_digest"),
+                "installed_registry_digest": result.get("installed_registry_digest"),
+                "registry_retry": bool(result.get("registry_retry")),
+                "latest_digest": result.get("latest_digest"),
+                "image_created": result.get("image_created"),
+                "latest_image_created": result.get("latest_image_created"),
+                "image_reference": result.get("image_reference"),
+                "image_repository": result.get("image_repository"),
+            }
+            sidecar["updated_at"] = _now_iso()
+            _write_sidecar(vmid, sidecar)
+        if app["state"]["registry_retry"]:
+            _retry_oci_registry(vmid, app_id)
+        if notify and app["state"]["update_available"]:
+            _fire_update_notification(vmid, app)
+        return sidecar
+
+
+def _check_non_oci_app(
+    vmid, app_id: str, force: bool = False, notify: bool = True,
 ) -> Optional[dict]:
     with _cache_lock:
         sidecar = _read_sidecar(vmid)
@@ -4371,40 +4450,6 @@ def check_app(
             except (ValueError, TypeError):
                 pass
 
-        if app.get("installed_via") == "oci_image":
-            # A secondary container of a stack shows the version it runs; the
-            # registry is not asked, because it is not updated on its own.
-            result = _oci_image_versions(
-                vmid, known=state, with_latest=not _oci_secondary_member(_read_oci_record(vmid)))
-            if result.get("busy"):
-                _recheck_after_oci_operation(vmid, app_id)
-                return sidecar
-            app["state"] = {
-                "installed_version": result.get("installed_version"),
-                "latest_version": result.get("latest_version"),
-                "latest_published_at": None,
-                "update_available": result.get("update_available"),
-                "error": result.get("error"),
-                "checked_at": _now_iso(),
-                "installed_digest": result.get("installed_digest"),
-                "installed_registry_digest": result.get("installed_registry_digest"),
-                "registry_retry": bool(result.get("registry_retry")),
-                "latest_digest": result.get("latest_digest"),
-                "image_created": result.get("image_created"),
-                "latest_image_created": result.get("latest_image_created"),
-                "image_reference": result.get("image_reference"),
-                "image_repository": result.get("image_repository"),
-            }
-            sidecar["updated_at"] = _now_iso()
-            _write_sidecar(vmid, sidecar)
-            if app["state"]["registry_retry"]:
-                _retry_oci_registry(vmid, app_id)
-            # The payload names an image by its build date and digest when it
-            # states no version, so it decides whether there is anything to send.
-            if notify and app["state"]["update_available"]:
-                _fire_update_notification(vmid, app)
-            return sidecar
-
         installed, inst_err, _healed = _detect_with_alt_healing(vmid, app)
         # Trigger the upstream fetch when ANY upstream source is
         # configured. The dispatcher inside `fetch_latest_upstream`
@@ -5363,6 +5408,10 @@ def _probe_docker_web_links(vmid) -> list[dict]:
     Stopped containers are included from their persistent HostConfig bindings,
     so their links remain registrable before the workload is started again.
     """
+    import docker_app_discovery
+    shared = docker_app_discovery.cached_web_links(vmid, _docker_service_catalog_meta)
+    if shared is not None:
+        return shared
     key = str(vmid)
     now = time.time()
     with _docker_web_links_lock:
@@ -5418,7 +5467,12 @@ def _probe_docker_web_links(vmid) -> list[dict]:
                             if not isinstance(binding, dict):
                                 continue
                             host_ip = str(binding.get("HostIp") or "").strip()
-                            if host_ip in {"127.0.0.1", "::1"}:
+                            try:
+                                import ipaddress
+                                bind_ip = ipaddress.ip_address(host_ip or "0.0.0.0")
+                                if bind_ip.is_loopback or (getattr(bind_ip, "ipv4_mapped", None) and bind_ip.ipv4_mapped.is_loopback):
+                                    continue
+                            except ValueError:
                                 continue
                             try:
                                 host_port = int(binding.get("HostPort"))
@@ -5977,14 +6031,22 @@ def _oci_repository(reference: str) -> str:
     return repository
 
 
-def _oci_resolve(module, reference: str, architecture: str) -> dict:
+def _oci_resolve(module, reference: str, architecture: str, known=None) -> dict:
     # One retry: an anonymous registry answers the occasional request with an
     # error that the next one does not repeat.
-    try:
+    def resolve():
+        # Older installed engines remain compatible; both paths use their
+        # native resolver. No separate registry client or verdict is added.
+        metadata = getattr(module, "resolve_candidate_metadata", None)
+        if known is not None and callable(metadata):
+            return metadata(reference, architecture, known)
         return module.resolve_candidate(reference, architecture)
+
+    try:
+        return resolve()
     except RuntimeError:
         time.sleep(2)
-        return module.resolve_candidate(reference, architecture)
+        return resolve()
 
 
 # What the panel shows, translated, when the registry of an image did not answer.
@@ -6067,8 +6129,15 @@ def _oci_image_versions(vmid, known: Optional[dict] = None, with_latest: bool =
         result["installed_version"] = _oci_guest_version(vmid, meta.get("template_id"))
     if not with_latest:
         return result
+    cached_metadata = None
+    if (known.get("installed_digest") == installed_digest
+            and known.get("image_reference") == reference and known.get("latest_digest")):
+        cached_metadata = {"manifest_digest": known["latest_digest"],
+                           "version": known.get("latest_version"),
+                           "created": known.get("latest_image_created"),
+                           "architecture": architecture}
     try:
-        latest = _oci_resolve(module, reference, architecture)
+        latest = _oci_resolve(module, reference, architecture, known=cached_metadata)
     except Exception:
         # A registry that does not answer says nothing new about the image:
         # what the last check found for this same image still stands.

AppImage/scripts/monitor_links.py

@@ -107,6 +107,32 @@ def _is_private(host):
         return False
 
 
+def _own_addresses():
+    addresses = set()
+    try:
+        import psutil
+        for entries in psutil.net_if_addrs().values():
+            for entry in entries:
+                if entry.family in (socket.AF_INET, socket.AF_INET6):
+                    addresses.add(entry.address.split("%")[0])
+    except Exception:
+        pass
+    return addresses
+
+
+def _is_this_host(host):
+    """A PBS installed next to PVE is named localhost or one of the host's own addresses."""
+    try:
+        resolved = {info[4][0].split("%")[0] for info in socket.getaddrinfo(host, None)}
+    except OSError:
+        return False
+    own = _own_addresses()
+    try:
+        return bool(resolved) and all(ipaddress.ip_address(a).is_loopback or a in own for a in resolved)
+    except ValueError:
+        return False
+
+
 def _cluster_nodes():
     try:
         result = subprocess.run(["pvesh", "get", "/cluster/status", "--output-format", "json"],
@@ -135,7 +161,7 @@ def _storage_pbs():
             if current is not None:
                 servers.append(current)
             continue
-        option = re.match(r"^\s+(server|port)\s+(\S+)", line)
+        option = re.match(r"^\s+(server|port|fingerprint)\s+(\S+)", line)
         if current is not None and option:
             current[option.group(1)] = option.group(2)
     return [row for row in servers if row.get("server")]
@@ -154,8 +180,17 @@ def _pbs_servers():
                 found.setdefault(host, {"name": row.get("name") or host, "native_url": normalize_url(row["url"])})
     except Exception:
         pass
-    return [{"key": f"pbs:{host}", "name": info["name"], "product": "pbs", "group": "pbs",
-             "host": host, "native_url": info["native_url"]} for host, info in found.items()]
+    servers, local = [], None
+    for host, info in found.items():
+        if _is_this_host(host):
+            # Its web interface is on the address the browser already uses.
+            if local is None:
+                local = {"key": "pbs:local", "name": info["name"], "product": "pbs", "group": "pbs", "local": True}
+                servers.append(local)
+            continue
+        servers.append({"key": f"pbs:{host}", "name": info["name"], "product": "pbs", "group": "pbs",
+                        "host": host, "native_url": info["native_url"]})
+    return servers
 
 
 def _probe(url):
@@ -188,6 +223,8 @@ def _resolve(entry, override):
         return override["url"], "online" if _probe(override["url"]) else "offline"
     if entry.get("url"):
         return entry["url"], "online" if _probe(entry["url"]) else "offline"
+    if entry.get("local"):
+        return None, "local"
     host = entry["host"]
     if not _is_private(host):
         return None, "unknown"
@@ -218,6 +255,7 @@ def list_links(include_hidden=False):
         links.append({"key": entry["key"], "name": entry["name"], "product": entry["product"],
                       "group": entry["group"], "url": url, "status": state,
                       "native_url": entry.get("native_url"), "manual": bool(entry.get("manual")),
+                      "local": bool(entry.get("local")),
                       "hidden": bool(override.get("hidden")), "custom_url": bool(override.get("url"))})
     return {"self": {"name": platform.get("hostname") or socket.gethostname(), "product": platform.get("product")},
             "links": links}

AppImage/scripts/oci_menu_bridge.py

@@ -0,0 +1,161 @@
+"""Root-only local access to the Monitor's existing OCI update checker.
+
+The terminal menu must not run a second registry checker or write sidecars in
+another process. Requests execute in the Monitor, using its normal locks,
+registration rules, retries and /etc/proxmenux/apps results. No TCP API or
+authentication setting is changed.
+"""
+from __future__ import annotations
+
+import atexit
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import errno
+import json
+import os
+from pathlib import Path
+import socket
+import socketserver
+import stat
+import struct
+import threading
+
+SOCKET = Path('/run/proxmenux-monitor/oci-check.sock')
+_check_lock = threading.Lock()
+_MAX_REQUEST = 128 * 1024
+_CHECK_WORKERS = 16
+
+
+def _root_peer(connection):
+    if not hasattr(socket, 'SO_PEERCRED'):
+        return False
+    credentials = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize('3i'))
+    return struct.unpack('3i', credentials)[1] == 0
+
+
+def _applications(request):
+    if not isinstance(request, dict) or set(request) != {'applications'}:
+        raise ValueError('Invalid request')
+    rows = request['applications']
+    if not isinstance(rows, list) or len(rows) > 1000:
+        raise ValueError('Invalid applications')
+    seen = set()
+    for row in rows:
+        if (not isinstance(row, dict) or set(row) != {'vmid', 'installation_id'}
+                or type(row['vmid']) is not int or not 100 <= row['vmid'] <= 999999999
+                or row['vmid'] in seen or not isinstance(row['installation_id'], str)
+                or not 1 <= len(row['installation_id']) <= 128):
+            raise ValueError('Invalid application')
+        seen.add(row['vmid'])
+    return rows
+
+
+def check_applications(module, rows, report):
+    """Delegate exclusively to check_app, exactly as the Monitor's Check does."""
+    report({'completed': 0, 'total': len(rows)})
+    stopped = threading.Event()
+
+    def check_one(row):
+        if stopped.is_set():
+            return
+        vmid = row['vmid']
+        record = module._read_oci_record(vmid) or {}
+        if (record.get('installation_id') == row['installation_id']
+                and record.get('status') == 'installed'
+                and not record.get('pending_transaction')
+                and not record.get('pending_stack_transaction')
+                and not module._oci_secondary_member(record)):
+            # Honour dismissals and reuse the Monitor's registration and
+            # checker. The menu never invents registrations or notifications.
+            module.ensure_oci_registration(vmid)
+            sidecar = module.load_sidecar(vmid) or {}
+            app = next((app for app in sidecar.get('apps', [])
+                        if app.get('installed_via') == 'oci_image'), None)
+            if app:
+                module.check_app(vmid, app['id'], force=True, notify=False)
+
+    pool = ThreadPoolExecutor(max_workers=_CHECK_WORKERS, thread_name_prefix='oci-check')
+    failed = []
+    try:
+        futures = {pool.submit(check_one, row): row['vmid'] for row in rows}
+        for completed, future in enumerate(as_completed(futures), 1):
+            try:
+                future.result()
+            except Exception:
+                failed.append(futures[future])  # One failure never stops the other applications.
+            report({'completed': completed, 'total': len(rows)})
+    finally:
+        stopped.set()
+        pool.shutdown(wait=True, cancel_futures=True)
+    report({'done': True, 'failed': sorted(failed)})
+
+
+class _Handler(socketserver.StreamRequestHandler):
+    def handle(self):
+        if not _root_peer(self.request):
+            return
+        self.request.settimeout(10)
+        acquired = False
+        try:
+            line = self.rfile.readline(_MAX_REQUEST + 1)
+            if len(line) > _MAX_REQUEST or not line.endswith(b'\n'):
+                return
+            rows = _applications(json.loads(line))
+            acquired = _check_lock.acquire(blocking=False)
+            if not acquired:
+                self.report({'error': 'check_running'})
+                return
+            check_applications(self.server.apps, rows, self.report)
+        except (BrokenPipeError, ConnectionResetError, TimeoutError):
+            pass  # Closing the menu stops the rest of this request.
+        except Exception:
+            try:
+                self.report({'error': 'check_failed'})
+            except OSError:
+                pass
+        finally:
+            if acquired:
+                _check_lock.release()
+
+    def report(self, event):
+        self.wfile.write((json.dumps(event) + '\n').encode())
+        self.wfile.flush()
+
+
+class _Server(socketserver.ThreadingUnixStreamServer):
+    daemon_threads = True
+    block_on_close = False
+
+
+def start(apps, path=SOCKET):
+    """Start alongside Flask; a private socket avoids web login bypasses."""
+    path = Path(path)
+    if os.geteuid() != 0:
+        return None
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    parent = path.parent.lstat()
+    if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != 0 or parent.st_mode & 0o077:
+        raise OSError('Unsafe Monitor runtime directory')
+    if path.exists() or path.is_symlink():
+        old = path.lstat()
+        if not stat.S_ISSOCK(old.st_mode) or old.st_uid != 0:
+            raise OSError('Unsafe Monitor socket')
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
+            probe.settimeout(.2)
+            try:
+                probe.connect(str(path))
+                return None  # Already owned by a running Monitor.
+            except OSError as exc:
+                if exc.errno not in (errno.ECONNREFUSED, errno.ENOENT):
+                    raise
+        path.unlink(missing_ok=True)
+    server = _Server(str(path), _Handler)
+    server.apps = apps
+    os.chmod(path, 0o600)
+    threading.Thread(target=server.serve_forever, name='oci-menu-check', daemon=True).start()
+
+    def close():
+        server.shutdown()
+        server.server_close()
+        path.unlink(missing_ok=True)
+    atexit.register(close)
+    return server

AppImage/scripts/pbs_manager.py

@@ -11,6 +11,7 @@ from pathlib import Path
 import re
 import secrets
 import shutil
+import socket
 import ssl
 import subprocess
 import tempfile
@@ -53,6 +54,26 @@ def is_pbs_host():
     return get_platform().get("product") == "pbs"
 
 
+def local_available():
+    """This host runs PBS: as its product, or installed next to PVE."""
+    if is_pbs_host():
+        return True
+    return bool(shutil.which("proxmox-backup-manager")) and any(
+        Path(folder, "proxmox-backup-proxy.service").exists()
+        for folder in ("/lib/systemd/system", "/usr/lib/systemd/system"))
+
+
+def _local_version():
+    if is_pbs_host():
+        return get_platform().get("version")
+    try:
+        result = subprocess.run(["dpkg-query", "-W", "-f=${Version}", "proxmox-backup-server"],
+                                capture_output=True, text=True, timeout=5)
+        return result.stdout.strip() or None if result.returncode == 0 else None
+    except (OSError, subprocess.SubprocessError):
+        return None
+
+
 def _servers_path():
     return auth_manager.CONFIG_DIR / "pbs_servers.json"
 
@@ -248,9 +269,9 @@ def get_status(identifier):
             return cached[1]
     errors = {}
     if identifier == "local":
-        if not is_pbs_host():
+        if not local_available():
             raise PBSError("The local host is not a PBS server.")
-        name, version = "Local PBS", get_platform().get("version")
+        name, version = get_platform().get("hostname") or socket.gethostname(), _local_version()
         services = _local_services()
         operations = {"datastores": _local_datastores,
                       "tasks": lambda: _cli("task", "list", "--all", "true", "--limit", "20"),

AppImage/scripts/tests/fixtures/docker_discovery_inspect.json

@@ -0,0 +1,158 @@
+{
+  "normal_rc": 0,
+  "records": [
+    {
+      "id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+      "name": "/manual-app-current",
+      "image": "example/manual-app:2",
+      "running": false,
+      "project": "manual-app",
+      "service": "web-current",
+      "title": null,
+      "version": null,
+      "bindings": {
+        "8080/tcp": [
+          {
+            "HostIp": "127.0.0.1",
+            "HostPort": "8081"
+          }
+        ]
+      },
+      "ports": {}
+    },
+    {
+      "id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+      "name": "/manual-app-previous",
+      "image": "example/manual-app:1",
+      "running": true,
+      "project": "manual-app",
+      "service": "web-previous",
+      "title": null,
+      "version": null,
+      "bindings": {
+        "21115/tcp": [
+          {
+            "HostIp": "",
+            "HostPort": "21115"
+          }
+        ],
+        "21116/tcp": [
+          {
+            "HostIp": "",
+            "HostPort": "21116"
+          }
+        ],
+        "21116/udp": [
+          {
+            "HostIp": "",
+            "HostPort": "21116"
+          }
+        ],
+        "21117/tcp": [
+          {
+            "HostIp": "",
+            "HostPort": "21117"
+          }
+        ],
+        "8080/tcp": [
+          {
+            "HostIp": "",
+            "HostPort": "8080"
+          }
+        ]
+      },
+      "ports": {
+        "21115/tcp": [
+          {
+            "HostIp": "0.0.0.0",
+            "HostPort": "21115"
+          },
+          {
+            "HostIp": "::",
+            "HostPort": "21115"
+          }
+        ],
+        "21116/tcp": [
+          {
+            "HostIp": "0.0.0.0",
+            "HostPort": "21116"
+          },
+          {
+            "HostIp": "::",
+            "HostPort": "21116"
+          }
+        ],
+        "21116/udp": [
+          {
+            "HostIp": "0.0.0.0",
+            "HostPort": "21116"
+          },
+          {
+            "HostIp": "::",
+            "HostPort": "21116"
+          }
+        ],
+        "21117/tcp": [
+          {
+            "HostIp": "0.0.0.0",
+            "HostPort": "21117"
+          },
+          {
+            "HostIp": "::",
+            "HostPort": "21117"
+          }
+        ],
+        "21118/tcp": null,
+        "21119/tcp": null,
+        "8080/tcp": [
+          {
+            "HostIp": "0.0.0.0",
+            "HostPort": "8080"
+          },
+          {
+            "HostIp": "::",
+            "HostPort": "8080"
+          }
+        ],
+        "9000/tcp": null
+      }
+    },
+    {
+      "id": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+      "name": "/db-1",
+      "image": "mysql:8.4",
+      "running": true,
+      "project": "manual-app",
+      "service": "db",
+      "title": null,
+      "version": null,
+      "bindings": {},
+      "ports": {
+        "3306/tcp": null,
+        "33060/tcp": null
+      }
+    }
+  ],
+  "partial_rc": 1,
+  "partial_records": [
+    {
+      "id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+      "name": "/manual-app-current",
+      "image": "example/manual-app:2",
+      "running": false,
+      "project": "manual-app",
+      "service": "web-current",
+      "title": null,
+      "version": null,
+      "bindings": {
+        "8080/tcp": [
+          {
+            "HostIp": "127.0.0.1",
+            "HostPort": "8081"
+          }
+        ]
+      },
+      "ports": {}
+    }
+  ]
+}

AppImage/scripts/tests/test_docker_app_discovery.py

@@ -0,0 +1,336 @@
+"""Docker Find contracts; no real guests, registry, auth or app writes."""
+import ast
+import copy
+from functools import wraps
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch, Mock
+
+SCRIPTS = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(SCRIPTS))
+import docker_app_discovery as docker
+import native_app_discovery as native
+import lxc_apps
+
+ONE, TWO = 'a' * 64, 'b' * 64
+
+
+def record(**values):
+    return {'id': ONE, 'name': '/cortendesk-v1-11-0-cortendesk-1',
+            'image': 'cortendesk-v1-11-0-cortendesk', 'running': True,
+            'project': 'cortendesk-v1-11-0', 'service': 'cortendesk',
+            'title': None, 'version': None,
+            'bindings': {'8080/tcp': [{'HostIp': '0.0.0.0', 'HostPort': '8080'}]},
+            'ports': None, **values}
+
+
+class Scan(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        root = Path(self.temp.name)
+        (root / '101.conf').write_text('ostype: debian\n')
+        self.patches = [patch.object(native, 'LOCAL_CONFIG', root),
+                        patch.object(native, 'OCI_ROOT', root / 'oci'),
+                        patch.object(native, 'UPDATE_LOCK_ROOT', root)]
+        for p in self.patches:
+            p.start()
+            self.addCleanup(p.stop)
+        self.calls = []
+
+    def runner(self, records=None, ids=None, inspect_rc=0, docker_present=True):
+        def execute(argv, deadline):
+            self.calls.append(argv)
+            self.assertEqual(argv[:4], ['pct', 'exec', '101', '--'])
+            if argv[4] == 'sh':
+                self.assertEqual(argv[5:], ['-c', 'command -v docker >/dev/null 2>&1'])
+                return (0 if docker_present else 1), ''
+            if argv[4:6] == ['docker', 'ps']:
+                return 0, '\n'.join(ids if ids is not None else [ONE])
+            self.assertEqual(argv[4:8], ['docker', 'inspect', '--format', docker.INSPECT_FORMAT])
+            return inspect_rc, '\n'.join(json.dumps(r) if not isinstance(r, str) else r
+                                         for r in (records if records is not None else [record()]))
+        return execute
+
+    def scan(self, **kwargs):
+        return docker.scan(101, run=self.runner(**kwargs), guard_run=lambda argv, deadline: 'status: running')
+
+    def test_cortendesk_and_unrelated_compose_use_the_same_generic_contract(self):
+        for app in [record(), record(name='/notes-1', image='example/notes:3', project='notes', service='web')]:
+            result = self.scan(records=[app])
+            self.assertEqual(result['status'], 'complete')
+            item, = result['candidates']
+            self.assertEqual(item['source'], 'docker_metadata')
+            self.assertEqual(item['default_ports'], [8080])
+            self.assertIsNone(item['installed_version'], 'no version guessed from image/project/tag')
+            self.assertIsNone(item['tracking_suggestion'])
+            self.assertTrue(item['discovery_id'].startswith('docker-'))
+            self.assertNotIn('repo', item)
+        self.assertNotIn('.Env', docker.INSPECT_FORMAT)
+        self.assertNotIn('json .}}', docker.INSPECT_FORMAT, 'no whole inspect object')
+        self.assertTrue(all(c[4] == 'sh' or c[4:6] in (['docker', 'ps'], ['docker', 'inspect']) for c in self.calls))
+
+    def test_known_version_is_display_metadata_and_never_an_updater(self):
+        item, = self.scan(records=[record(version='1.11.0')])['candidates']
+        self.assertEqual(item['installed_version'], '1.11.0')
+        self.assertIsNone(item['tracking_suggestion'])
+
+    def test_database_stopped_and_no_published_ports_are_excluded(self):
+        for app in [record(image='mysql:8.4', service='db'), record(running=False),
+                    record(bindings={}), record(image='redis:7', service='cache')]:
+            self.assertEqual(self.scan(records=[app])['candidates'], [])
+
+    def test_all_loopback_bindings_excluded_and_live_mapping_preferred(self):
+        for address in ['127.0.0.1', '127.0.0.2', '127.255.255.254', '::1', '::ffff:127.0.0.2']:
+            value = record(bindings={'8080/tcp': [{'HostIp': address, 'HostPort': '8080'}]})
+            self.assertEqual(self.scan(records=[value])['candidates'], [], address)
+        value = record(ports={'8080/tcp': [{'HostIp': '::', 'HostPort': '9090'}]})
+        self.assertEqual(self.scan(records=[value])['candidates'][0]['default_ports'], [9090])
+
+    def test_service_tcp_ports_are_evidence_not_prefilled_web_addresses(self):
+        app = record(bindings={'21115/tcp':[{'HostIp':'0.0.0.0','HostPort':'21115'}],
+                               '8080/tcp':[{'HostIp':'0.0.0.0','HostPort':'8080'}]})
+        item, = self.scan(records=[app])['candidates']
+        self.assertEqual(item['default_ports'],[8080])
+        self.assertIn({'kind':'tcp_listener','value':'21115'},item['evidence'])
+
+    def test_unassigned_or_ranged_host_ports_are_skipped_not_invalid(self):
+        app = record(running=False, bindings={'8080/tcp': [{'HostIp': '', 'HostPort': ''}],
+                                              '9000/tcp': [{'HostIp': '', 'HostPort': '9000-9010'}],
+                                              '443/tcp': [{'HostIp': '', 'HostPort': '8444'}]})
+        result = self.scan(records=[app])
+        self.assertEqual(result['status'], 'complete')
+        endpoints = result['_containers'][0]['endpoints']
+        self.assertEqual([e['host_port'] for e in endpoints], [8444])
+
+    def test_https_container_port_prefills_its_scheme(self):
+        app = record(bindings={'443/tcp': [{'HostIp': '0.0.0.0', 'HostPort': '8444'}]})
+        item, = self.scan(records=[app])['candidates']
+        self.assertEqual(item['default_ports'], [8444])
+        self.assertEqual([(d['port'], d['scheme']) for d in item['port_details']], [(8444, 'https')])
+
+    def test_empty_engine_is_complete_and_missing_docker_never_calls_docker(self):
+        self.assertEqual(self.scan(ids=[])['status'], 'complete')
+        self.calls.clear()
+        result = self.scan(docker_present=False)
+        self.assertEqual(result['candidates'], [])
+        self.assertEqual(len(self.calls), 1)
+
+    def test_nonzero_inspect_retains_valid_stdout_and_marks_partial(self):
+        result = self.scan(ids=[ONE, TWO], inspect_rc=1)
+        self.assertEqual(result['status'], 'partial')
+        self.assertEqual(len(result['candidates']), 1)
+        self.assertIn('inspect_failed', result['issues'])
+        self.assertIn('incomplete_inspect', result['issues'])
+
+    def test_invalid_and_duplicate_records_are_partial_without_crashing(self):
+        for bad in ['{broken', [], record(id=TWO), record(bindings=[]), record(running='false')]:
+            result = self.scan(records=[record(), bad])
+            self.assertEqual(result['status'], 'partial')
+            self.assertEqual(len(result['candidates']), 1)
+
+    def test_record_limit_is_visible_and_inspect_is_bounded(self):
+        ids = [f'{i:064x}' for i in range(51)]
+        result = self.scan(ids=ids, records=[record(id=ids[0])])
+        self.assertIn('container_limit', result['issues'])
+        self.assertEqual(len(self.calls[-1][8:]), docker.MAX_CONTAINERS)
+
+    def test_stopped_missing_locked_and_oci_guests_have_zero_guest_probes(self):
+        root = Path(self.temp.name)
+        cases = [('', 'status: stopped', 'guest_stopped'),
+                 ('lock: backup\nostype: debian\n', 'status: running', 'guest_busy')]
+        for config, status, error in cases:
+            (root / '101.conf').write_text(config or 'ostype: debian\n')
+            with self.assertRaisesRegex(native.ProbeError, error):
+                docker.scan(101, run=self.runner(), guard_run=lambda a, d: status)
+        (root/'101.conf').unlink()
+        with self.assertRaisesRegex(native.ProbeError, 'not_local_lxc'):
+            self.scan()
+        (root/'101.conf').write_text('ostype: unmanaged\n')
+        (root/'oci'/'101').mkdir(parents=True)
+        with self.assertRaisesRegex(native.ProbeError, 'managed_oci'):
+            self.scan()
+        self.assertEqual(self.calls, [])
+
+    def test_guest_stopping_during_scan_never_publishes_results(self):
+        states = iter(['status: running', 'status: stopped'])
+        with self.assertRaisesRegex(native.ProbeError, 'guest_stopped'):
+            docker.scan(101, run=self.runner(), guard_run=lambda a,d: next(states))
+
+
+class Cache(unittest.TestCase):
+    def setUp(self):
+        docker._cache.clear(); docker._generations.clear(); docker._locks.clear()
+        self.addCleanup(docker._cache.clear)
+        self.addCleanup(docker._generations.clear)
+        self.value = {'vmid': 101, 'status': 'complete', 'stale': False, 'issues': [],
+                      'scanned_at': 1000, '_containers': [docker.parse_record(record())],
+                      'candidates': [docker._candidate(101, docker.parse_record(record()))]}
+
+    def test_get_never_scans_and_freshness_boundary_has_no_mutation(self):
+        with patch.object(docker, 'scan', return_value=self.value) as scan, patch.object(docker.time, 'time', return_value=1000):
+            self.assertEqual(docker.cached(101)['status'], 'not_scanned')
+            scan.assert_not_called()
+            docker.discover(101)
+        for clock, stale in [(1299.999, False), (1300, True), (1300.001, True)]:
+            with patch.object(docker.time, 'time', return_value=clock):
+                self.assertEqual(docker.cached(101)['stale'], stale)
+        self.assertFalse(docker._cache[101]['stale'])
+
+    def test_failure_partial_and_lifecycle_preserve_previous_without_new_probes(self):
+        with patch.object(docker, 'scan', return_value=self.value): docker.discover(101)
+        with patch.object(docker, 'scan', side_effect=native.ProbeError('timeout')):
+            failed = docker.discover(101)
+        self.assertEqual(failed['status'], 'error'); self.assertTrue(failed['stale'])
+        self.assertEqual(failed['candidates'], self.value['candidates'])
+        with patch.object(docker, 'scan') as scan:
+            docker.mark_stale(101)
+            self.assertTrue(docker.cached(101)['stale'])
+            scan.assert_not_called()
+
+    def test_inflight_lifecycle_rejects_late_result_and_busy_has_no_second_scan(self):
+        with patch.object(docker, 'scan', side_effect=lambda v: (docker.mark_stale(v), self.value)[1]):
+            self.assertEqual(docker.discover(101)['issues'], ['guest_changed'])
+        self.assertEqual(docker.cached(101)['status'], 'not_scanned')
+        lock = docker._locks[101]; lock.acquire()
+        try:
+            with patch.object(docker, 'scan') as scan:
+                self.assertEqual(docker.discover(101)['status'], 'busy'); scan.assert_not_called()
+        finally: lock.release()
+
+    def test_shared_web_links_avoid_duplicate_inspect_and_stop_at_snapshot_boundary(self):
+        lookup = Mock(return_value={'slug': None, 'name': 'web', 'logo_url': None})
+        with patch.object(docker, 'scan', return_value=self.value), patch.object(docker.time, 'time', return_value=1000):
+            docker.discover(101)
+        for clock, valid in [(1059.999, True), (1060, False), (1060.001, False)]:
+            with patch.object(docker.time, 'time', return_value=clock), patch.object(lxc_apps, '_pct_exec') as execute:
+                result = docker.cached_web_links(101, lookup)
+                self.assertEqual(result is not None, valid); execute.assert_not_called()
+        with patch.object(docker.time, 'time', return_value=1001), patch.object(lxc_apps, '_pct_exec') as execute:
+            with patch.object(lxc_apps, '_docker_service_catalog_meta', lookup):
+                self.assertEqual(lxc_apps._probe_docker_web_links(101)[0]['host_port'], 8080)
+            execute.assert_not_called()
+
+    def test_incomplete_search_never_replaces_the_web_link_probe(self):
+        lookup = Mock(return_value={'slug': None, 'name': 'web', 'logo_url': None})
+        partial = {**copy.deepcopy(self.value), 'status': 'partial', 'issues': ['inspect_failed']}
+        with patch.object(docker, 'scan', return_value=partial), patch.object(docker.time, 'time', return_value=1000):
+            docker.discover(101)
+        with patch.object(docker.time, 'time', return_value=1001):
+            self.assertIsNone(docker.cached_web_links(101, lookup))
+        docker._cache.clear()
+        with patch.object(docker, 'scan', side_effect=native.ProbeError('timeout')), \
+                patch.object(docker.time, 'time', return_value=1000):
+            docker.discover(101)
+        with patch.object(docker.time, 'time', return_value=1001):
+            self.assertIsNone(docker.cached_web_links(101, lookup))
+
+    def test_shared_links_keep_existing_name_then_port_order(self):
+        value = copy.deepcopy(self.value)
+        value['_containers'][0]['name'] = 'z-last'
+        second = copy.deepcopy(value['_containers'][0]); second['name'] = 'a-first'
+        second['endpoints'] = [{'host_port':9090,'container_port':8080,'scheme':'http'},
+                               {'host_port':8080,'container_port':8080,'scheme':'http'}]
+        value['_containers'].append(second)
+        lookup = lambda service,name,image: {'slug':None,'name':name,'logo_url':None}
+        with patch.object(docker,'scan',return_value=value): docker.discover(101)
+        links = docker.cached_web_links(101,lookup)
+        self.assertEqual([(x['service_name'],x['host_port']) for x in links],
+                         [('a-first',8080),('a-first',9090),('z-last',8080)])
+
+    def test_registered_container_image_app_or_existing_port_is_not_duplicated(self):
+        item = self.value['candidates'][0]
+        for app in [{'discovery_id': item['discovery_id']}, {'container_name': item['container_name']},
+                    {'ports': [{'port': 8080}]}]:
+            self.assertEqual(docker.filter_candidates(self.value, {}, [app])['candidates'], [])
+        self.assertEqual(docker.filter_candidates(self.value, {'docker_workloads': [{'container_name': item['container_name']}]}, [])['candidates'], [])
+
+    def test_same_display_name_is_not_container_identity(self):
+        item = self.value['candidates'][0]
+        other = {'name': item['name'], 'discovery_id': 'docker-'+'f'*24, 'ports':[{'port':9090}]}
+        self.assertEqual(docker.filter_candidates(self.value, {}, [other])['candidates'], [item])
+
+    def test_save_identity_accepts_docker_but_does_not_enable_update(self):
+        identity = self.value['candidates'][0]['discovery_id']
+        ok, value = lxc_apps.validate_config({'name': 'Docker endpoint', 'discovery_id': identity,
+                                            'ports': [{'port': 8080}], 'update_method': 'none'})
+        self.assertTrue(ok, value); self.assertEqual(value['discovery_id'], identity)
+        self.assertFalse(value.get('update_command')); self.assertFalse(value.get('repo'))
+        self.assertFalse(lxc_apps.validate_config({'name':'bad', 'discovery_id':'docker-bad'})[0])
+
+
+class RealContract(unittest.TestCase):
+    def test_sanitized_testhost_template_and_partial_stdout_are_parsed(self):
+        value = json.loads((SCRIPTS/'tests/fixtures/docker_discovery_inspect.json').read_text())
+        self.assertEqual(value['normal_rc'], 0)
+        self.assertEqual(value['partial_rc'], 1)
+        self.assertTrue(value['partial_records'])
+        for item in [*value['records'], *value['partial_records']]:
+            parsed = docker.parse_record(item)
+            self.assertTrue(parsed['id'])
+            self.assertNotIn('Env', json.dumps(parsed))
+            if parsed['running'] and parsed['service'] != 'db':
+                self.assertEqual(docker._candidate(101,parsed)['default_ports'],[8080])
+
+
+class Reader(unittest.TestCase):
+    def test_nonzero_exit_preserves_stdout_without_leaking_stderr(self):
+        rc, output = docker.read([sys.executable, '-c', 'import sys; print("valid"); print("private-fixture",file=sys.stderr); sys.exit(1)'], time.monotonic()+3)
+        self.assertEqual((rc, output), (1, 'valid\n'))
+
+    def test_timeout_and_output_limit_are_bounded_errors(self):
+        with patch.object(docker, 'PROBE_SECONDS', .05):
+            with self.assertRaisesRegex(native.ProbeError, 'timeout'):
+                docker.read([sys.executable, '-c', 'import time; time.sleep(2)'], time.monotonic()+3)
+        with patch.object(docker, 'OUTPUT_BYTES', 100):
+            with self.assertRaisesRegex(native.ProbeError, 'output_limit'):
+                docker.read([sys.executable, '-c', 'print("x"*1000)'], time.monotonic()+3)
+
+
+class Routes(unittest.TestCase):
+    def setUp(self):
+        from flask import Flask, request, jsonify
+        self.app = Flask(__name__)
+        def auth(fn):
+            @wraps(fn)
+            def wrapped(*args, **kwargs):
+                return fn(*args, **kwargs) if request.headers.get('X-Fixture') == 'yes' else ({'error':'unauthorized'},401)
+            return wrapped
+        source = SCRIPTS/'flask_server.py'
+        routes = [n for n in ast.parse(source.read_text()).body if isinstance(n,ast.FunctionDef) and n.name in ('api_vm_apps_docker_discovery','api_vm_apps_suggestions')]
+        self.cache = {}
+        scope = {'app':self.app, 'request':request, 'jsonify':jsonify, 'require_auth':auth,
+                 '_vm_cache_get':lambda c,v,t:c.get(v), '_vm_cache_put':lambda c,v,p:c.update({v:p}),
+                 '_vm_app_suggestions_cache':self.cache, '_VM_APP_SUGGESTIONS_TTL':60}
+        exec(compile(ast.Module(body=routes, type_ignores=[]), str(source), 'exec'),scope)
+        self.client = self.app.test_client()
+
+    def test_existing_suggestions_refresh_preserves_docker_cache_without_new_scan(self):
+        value = {'vmid':101,'status':'complete','candidates':[],'scanned_at':time.time(),'stale':False}
+        docker._cache[101] = copy.deepcopy(value)
+        self.addCleanup(docker._cache.clear)
+        with patch.object(docker,'discover') as scan, patch.object(lxc_apps,'get_suggestions',return_value={'extras':[]}) as catalog, patch.object(lxc_apps,'_write_sidecar') as write:
+            response = self.client.post('/api/vms/101/apps/suggestions',headers={'X-Fixture':'yes'})
+            self.assertEqual(response.status_code,200)
+            catalog.assert_called_once_with(101,force=True)
+            scan.assert_not_called(); write.assert_not_called()
+            self.assertEqual(docker.cached(101),value)
+
+    def test_real_route_only_explicit_post_scans_and_never_registers(self):
+        value = {'vmid':101,'status':'complete','candidates':[],'stale':False}
+        with patch.object(docker,'discover',return_value=value) as scan, patch.object(docker,'cached',return_value=value), patch.object(lxc_apps,'load_sidecar',return_value={'apps':[]}), patch.object(lxc_apps,'_write_sidecar') as write:
+            self.assertEqual(self.client.post('/api/vms/101/apps/docker-discovery').status_code,401)
+            self.assertEqual(self.client.put('/api/vms/101/apps/docker-discovery',headers={'X-Fixture':'yes'}).status_code,405)
+            self.assertEqual(self.client.get('/api/vms/101/apps/docker-discovery',headers={'X-Fixture':'yes'}).status_code,200)
+            scan.assert_not_called(); write.assert_not_called()
+            self.assertEqual(self.client.post('/api/vms/101/apps/docker-discovery',headers={'X-Fixture':'yes'}).status_code,200)
+            scan.assert_called_once_with(101); write.assert_not_called()
+
+
+if __name__ == '__main__': unittest.main()

AppImage/scripts/tests/test_monitor_links.py

@@ -62,6 +62,29 @@ class MonitorLinksTests(unittest.TestCase):
         self.assertEqual(by_key["pbs:pbs.example.com"]["native_url"], "https://pbs.example.com:8443")
         self.assertFalse(any("example.com" in url for url in probed))
 
+    def test_pbs_installed_on_this_host_opens_its_own_interface_and_is_never_probed(self):
+        links.STORAGE_CFG.write_text("pbs: PBS\n\tdatastore PBS\n\tserver localhost\n\n"
+                                     "pbs: Same\n\tdatastore x\n\tserver 10.0.0.55\n\n"
+                                     "pbs: Lab\n\tdatastore store1\n\tserver 10.0.0.49\n")
+        probed = []
+        with mock.patch.object(links, "_own_addresses", return_value={"10.0.0.55"}), \
+                mock.patch.object(links, "_probe", side_effect=lambda url: probed.append(url) or False), \
+                mock.patch.object(links, "_is_private", return_value=True):
+            result = links.list_links()
+        pbs_rows = [row for row in result["links"] if row["group"] == "pbs"]
+        self.assertEqual([row["key"] for row in pbs_rows], ["pbs:local", "pbs:10.0.0.49"])
+        local = pbs_rows[0]
+        self.assertEqual((local["name"], local["local"], local["url"], local["status"]), ("PBS", True, None, "local"))
+        self.assertIsNone(local["native_url"])
+        self.assertFalse(any("localhost" in url or "10.0.0.55" in url for url in probed))
+        self.assertFalse(pbs_rows[1]["local"])
+
+    def test_loopback_names_are_this_host(self):
+        with mock.patch.object(links, "_own_addresses", return_value=set()):
+            self.assertTrue(links._is_this_host("127.0.0.1"))
+            self.assertTrue(links._is_this_host("::1"))
+            self.assertFalse(links._is_this_host("10.0.0.49"))
+
     def test_pbs_host_lists_only_manual_entries(self):
         links.pbs_manager.get_platform.return_value = {"product": "pbs", "hostname": "pbs-lab"}
         links.add_link({"name": "amd", "url": "https://10.0.0.55:8008", "product": "pve"})

AppImage/scripts/tests/test_native_app_discovery.py

@@ -768,13 +768,19 @@ class Routes(unittest.TestCase):
             write.assert_not_called()
 
     def test_actual_lifecycle_marks_last_known_stale_without_scanning(self):
+        import docker_app_discovery as docker
+        docker._cache[101] = {'vmid':101, 'status':'complete', 'stale':False, 'candidates':[], 'scanned_at':time.time()}
+        self.addCleanup(docker._cache.clear)
         native._cache[101] = {'vmid': 101, 'status': 'complete', 'stale': False,
                               'candidates': [], 'scanned_at': time.time()}
         with patch.object(native, 'discover') as scan, \
+             patch.object(docker, 'discover') as docker_scan, \
              patch.object(lxc_apps, 'clear_schedule_reboot_required') as clear:
             for action in ('start', 'reboot', 'stop'):
                 self.globals['_handle_guest_lifecycle']('101', 'lxc', action)
                 self.assertTrue(native.cached(101)['stale'])
+                self.assertTrue(docker.cached(101)['stale'])
+            docker_scan.assert_not_called()
             scan.assert_not_called()
             self.assertEqual(clear.call_count, 2)
 

AppImage/scripts/tests/test_oci_parallel_checks.py

@@ -0,0 +1,124 @@
+"""OCI registry waits run concurrently; shared app cache changes remain atomic."""
+from concurrent.futures import ThreadPoolExecutor
+import copy
+import json
+from pathlib import Path
+import sys
+import tempfile
+import threading
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import lxc_apps
+
+DIGEST = 'sha256:' + 'a' * 64
+NEW = 'sha256:' + 'b' * 64
+
+
+class ParallelOciChecksTests(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory(); self.addCleanup(tmp.cleanup)
+        self.root = Path(tmp.name)
+        self.records = {vmid: {'vmid': vmid, 'status': 'installed', 'installation_id': f'uuid-{vmid}'}
+                        for vmid in range(110, 114)}
+        for vmid in self.records:
+            sidecar = {'vmid': vmid, 'schedule': {'enabled': False}, 'apps': [
+                {'id': 'oci', 'installed_via': 'oci_image', 'name': 'App', 'state': {}},
+                {'id': 'other', 'installed_via': 'manual', 'name': 'Other', 'state': {}}]}
+            (self.root / f'{vmid}.json').write_text(json.dumps(sidecar))
+        for item in (patch.object(lxc_apps, '_APPS_DIR', str(self.root)),
+                     patch.object(lxc_apps, '_read_oci_record', side_effect=lambda vmid: copy.deepcopy(self.records.get(int(vmid))))):
+            item.start(); self.addCleanup(item.stop)
+
+    def result(self, vmid, **kwargs):
+        return {'installed_digest': DIGEST, 'installed_registry_digest': DIGEST,
+                'latest_digest': NEW, 'update_available': True, 'installed_version': '1',
+                'latest_version': '2', 'image_reference': 'example/app:latest'}
+
+    def test_four_native_checks_enter_registry_before_any_one_can_finish(self):
+        rendezvous = threading.Barrier(4, timeout=3)
+        def versions(vmid, **kwargs):
+            rendezvous.wait()
+            return self.result(vmid)
+        with patch.object(lxc_apps, '_oci_image_versions', side_effect=versions) as query:
+            with ThreadPoolExecutor(max_workers=4) as pool:
+                results = list(pool.map(lambda vmid: lxc_apps.check_app(vmid, 'oci', force=True, notify=False), self.records))
+        self.assertEqual(query.call_count, 4)
+        self.assertTrue(all(result['apps'][0]['state']['update_available'] for result in results))
+
+    def during_query(self, edit):
+        started, release, edited = threading.Event(), threading.Event(), threading.Event()
+        def versions(vmid, **kwargs):
+            started.set()
+            if not release.wait(3): raise AssertionError('Registry wait did not end')
+            return self.result(vmid)
+        def change():
+            with lxc_apps._cache_lock:
+                sidecar = lxc_apps.load_sidecar(110)
+                edit(sidecar)
+                lxc_apps._write_sidecar(110, sidecar)
+            edited.set()
+        with patch.object(lxc_apps, '_oci_image_versions', side_effect=versions):
+            with ThreadPoolExecutor(max_workers=2) as pool:
+                future = pool.submit(lxc_apps.check_app, 110, 'oci', force=True, notify=False)
+                try:
+                    self.assertTrue(started.wait(2))
+                    writer = pool.submit(change)
+                    self.assertTrue(edited.wait(2), 'A registry call still holds the shared cache lock')
+                finally:
+                    release.set()
+                future.result(); writer.result()
+        return lxc_apps.load_sidecar(110)
+
+    def test_unrelated_apps_and_schedule_edits_survive_registry_wait(self):
+        def edit(sidecar):
+            sidecar['schedule']['enabled'] = True
+            sidecar['apps'][1]['name'] = 'Renamed while checking'
+        result = self.during_query(edit)
+        self.assertTrue(result['schedule']['enabled'])
+        self.assertEqual(result['apps'][1]['name'], 'Renamed while checking')
+        self.assertTrue(result['apps'][0]['state']['update_available'])
+
+    def test_deleted_app_is_not_recreated_by_late_registry_answer(self):
+        result = self.during_query(lambda sidecar: sidecar['apps'].pop(0))
+        self.assertEqual([app['id'] for app in result['apps']], ['other'])
+
+    def test_edited_app_is_not_overwritten_by_late_registry_answer(self):
+        result = self.during_query(lambda sidecar: sidecar['apps'][0].update(name='Edited'))
+        self.assertEqual(result['apps'][0]['name'], 'Edited')
+        self.assertEqual(result['apps'][0]['state'], {})
+
+    def test_replaced_installation_cannot_receive_previous_images_answer(self):
+        def edit(sidecar): self.records[110]['installation_id'] = 'replacement'
+        result = self.during_query(edit)
+        self.assertEqual(result['apps'][0]['state'], {})
+
+    def test_checks_for_two_apps_in_same_container_keep_both_results(self):
+        sidecar = lxc_apps.load_sidecar(110)
+        sidecar['apps'][1]['installed_via'] = 'oci_image'
+        lxc_apps._write_sidecar(110, sidecar)
+        rendezvous = threading.Barrier(2, timeout=3)
+        def versions(vmid, **kwargs):
+            rendezvous.wait()
+            return self.result(vmid)
+        with patch.object(lxc_apps, '_oci_image_versions', side_effect=versions):
+            with ThreadPoolExecutor(max_workers=2) as pool:
+                futures = [pool.submit(lxc_apps.check_app, 110, app, force=True, notify=False) for app in ('oci', 'other')]
+                for future in futures: future.result()
+        result = lxc_apps.load_sidecar(110)
+        self.assertTrue(all(app['state']['update_available'] for app in result['apps']))
+
+    def test_native_registry_failure_is_persisted_and_retried_without_blocking_others(self):
+        def versions(vmid, **kwargs):
+            return {'error': 'registry unreachable', 'registry_retry': True} if vmid == 110 else self.result(vmid)
+        with patch.object(lxc_apps, '_oci_image_versions', side_effect=versions), \
+                patch.object(lxc_apps, '_retry_oci_registry') as retry:
+            with ThreadPoolExecutor(max_workers=4) as pool:
+                results = list(pool.map(lambda vmid: lxc_apps.check_app(vmid, 'oci', force=True, notify=False), self.records))
+        self.assertEqual(results[0]['apps'][0]['state']['error'], 'registry unreachable')
+        self.assertTrue(all(result['apps'][0]['state']['update_available'] for result in results[1:]))
+        retry.assert_called_once_with(110, 'oci')
+
+
+if __name__ == '__main__': unittest.main()

AppImage/scripts/tests/test_pbs.py

@@ -166,6 +166,39 @@ class PBSManagerTests(unittest.TestCase):
         self.assertEqual(result["datastores"][0]["used"], 30)
         self.assertEqual(set(result["services"]), set(pbs.PBS_SERVICES))
 
+    def test_pbs_installed_next_to_pve_is_read_locally_with_its_own_version(self):
+        def run(command, **kwargs):
+            if command[0] == "dpkg-query":
+                return SimpleNamespace(stdout="4.2.8-1", returncode=0)
+            return SimpleNamespace(stdout="[]", returncode=0)
+        with mock.patch.object(pbs, "get_platform", return_value={"product": "pve", "version": "9.1.4", "hostname": "rimegra"}), \
+                mock.patch.object(pbs.shutil, "which", return_value="/usr/sbin/proxmox-backup-manager"), \
+                mock.patch.object(pbs.Path, "exists", return_value=True), \
+                mock.patch.object(pbs, "_local_services", return_value={key: "active" for key in pbs.PBS_SERVICES}), \
+                mock.patch.object(pbs.subprocess, "run", side_effect=run):
+            self.assertTrue(pbs.local_available())
+            result = pbs.get_status("local")
+        self.assertEqual((result["name"], result["version"]), ("rimegra", "4.2.8-1"))
+        with mock.patch.object(pbs, "get_platform", return_value={"product": "pve"}), \
+                mock.patch.object(pbs.shutil, "which", return_value=None):
+            self.assertFalse(pbs.local_available())
+
+    def test_storage_pbs_without_a_connection_is_offered_with_its_address_and_fingerprint(self):
+        import monitor_links
+        storage = self.root / "storage.cfg"
+        storage.write_text("pbs: Local\n\tserver localhost\n\tdatastore a\n\n"
+                           "pbs: Lab\n\tserver 10.0.0.49\n\tdatastore b\n\tfingerprint AA:BB\n\n"
+                           "pbs: Known\n\tserver pbs.test\n\tdatastore c\n\tport 8007\n")
+        with mock.patch.object(monitor_links, "STORAGE_CFG", storage), \
+                mock.patch.object(monitor_links, "_own_addresses", return_value=set()), \
+                mock.patch.object(pbs, "is_pbs_host", return_value=False), \
+                mock.patch.object(pbs, "local_available", return_value=False), \
+                mock.patch.object(pbs, "_remote_get", return_value=[]):
+            pbs.save_server(self.server)
+            response = self.client.get("/api/pbs/servers", headers=self.headers)
+        self.assertEqual(response.json["detected"],
+                         [{"name": "Lab", "url": "https://10.0.0.49:8007", "fingerprint": "AA:BB"}])
+
     def test_connection_api_requires_full_admin_and_does_not_return_secret(self):
         self.assertEqual(self.client.get("/api/pbs/servers").status_code, 401)
         with mock.patch.object(pbs, "_remote_get", return_value=[]):