diff --git a/.github/scripts/run_offline_qualified.py b/.github/scripts/run_offline_qualified.py
index 9bd13abc..fe6c26d4 100644
--- a/.github/scripts/run_offline_qualified.py
+++ b/.github/scripts/run_offline_qualified.py
@@ -13,6 +13,7 @@ import sys
 import tempfile
 
 PYTHON_FILES = (
+    "AppImage/scripts/tests/test_native_app_discovery.py",
     "tests/test_audit_presentation.py",
     "tests/test_audit_safety_wording.py",
     "tests/test_audit_policy.py",
@@ -24,6 +25,7 @@ PYTHON_FILES = (
     "AppImage/scripts/tests/test_health_persistence_recurrence.py",
 )
 NODE_FILES = (
+    "tests/lxc_updates/test_native_app_discovery_ui.cjs",
     "tests/lxc_updates/test_docker_delegated_ui.cjs",
     "tests/test_audit_diagnostic_document.cjs",
     "tests/test_audit_policy.cjs",
diff --git a/.github/workflows/test-offline-qualified.yml b/.github/workflows/test-offline-qualified.yml
index 29af7416..a562c6b3 100644
--- a/.github/workflows/test-offline-qualified.yml
+++ b/.github/workflows/test-offline-qualified.yml
@@ -12,6 +12,7 @@ on:
       - 'AppImage/package-lock.json'
       - 'AppImage/scripts/*.py'
       - 'AppImage/scripts/tests/test_health_persistence_recurrence.py'
+      - 'AppImage/scripts/tests/test_native_app_discovery.py'
       - 'AppImage/app/**'
       - 'AppImage/components/**'
       - 'AppImage/hooks/**'
@@ -32,6 +33,7 @@ on:
       - 'AppImage/package-lock.json'
       - 'AppImage/scripts/*.py'
       - 'AppImage/scripts/tests/test_health_persistence_recurrence.py'
+      - 'AppImage/scripts/tests/test_native_app_discovery.py'
       - 'AppImage/app/**'
       - 'AppImage/components/**'
       - 'AppImage/hooks/**'
@@ -56,7 +58,9 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: '3.11'
-      - name: Run nine qualified Python files
+      - name: Install native discovery route fixture dependency
+        run: python3 -m pip install 'Flask>=3,<4'
+      - name: Run qualified Python files
         env:
           PYTHONDONTWRITEBYTECODE: '1'
         run: python3 -I -B .github/scripts/run_offline_qualified.py --lane python
@@ -82,7 +86,7 @@ jobs:
         run: |
           npm ci --prefix .github/ci/offline-node --legacy-peer-deps --ignore-scripts --no-audit --no-fund
           ln -s ../.github/ci/offline-node/node_modules AppImage/node_modules
-      - name: Run eleven qualified Node files
+      - name: Run qualified Node files
         env:
           NODE_PATH: ${{ github.workspace }}/AppImage/node_modules
           PYTHONDONTWRITEBYTECODE: '1'
diff --git a/AppImage/components/lxc-app-panel.tsx b/AppImage/components/lxc-app-panel.tsx
index 59939f0b..22154ef6 100644
--- a/AppImage/components/lxc-app-panel.tsx
+++ b/AppImage/components/lxc-app-panel.tsx
@@ -19,7 +19,7 @@
  * Secure Gateway.
  */
 
-import { useCallback, useEffect, useMemo, useState } from "react"
+import { useCallback, useEffect, useMemo, useRef, useState } from "react"
 import {
   Loader2, Save, RefreshCw, Trash2, Package, ExternalLink,
   AlertTriangle, Info, PlusCircle, Pencil, ChevronDown, ChevronRight, EyeOff,
@@ -93,6 +93,7 @@ interface AppConfig {
   health_path?: string
   logo_url?: string
   helper_slug?: string
+  discovery_id?: string
   // Set when the application runs inside Docker: registered for its identity
   // and installed version, updated by the image it comes from.
   update_via?: string
@@ -121,7 +122,18 @@ interface PortDetail {
   logo_url: string
 }
 
+interface NativeDiscovery {
+  vmid: number
+  status: "not_scanned" | "complete" | "partial" | "error" | "busy"
+  stale: boolean
+  candidates: DetectedApp[]
+}
+
 interface DetectedApp {
+  discovery_id?: string
+  source?: "native_metadata"
+  installed_version?: string | null
+  evidence?: Array<{ kind: string; value: string }>
   slug: string
   name: string
   logo_url?: string | null
@@ -462,12 +474,37 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
   const [loading, setLoading] = useState(!seed)
   const [sidecar, setSidecar] = useState<SidecarResponse | null>(seed?.sidecar ?? null)
   const [suggestions, setSuggestions] = useState<Suggestions | null>(seed?.suggestions ?? null)
+  const [nativeDiscovery, setNativeDiscovery] = useState<NativeDiscovery | null>(null)
+  const nativeRequest = useRef(0)
+  const targetRef = useRef(vmid)
+  targetRef.current = vmid
+  const targetGeneration = useRef(0)
+  const previousTarget = useRef(vmid)
+  if (previousTarget.current !== vmid) {
+    previousTarget.current = vmid
+    targetGeneration.current += 1
+  }
+  useEffect(() => {
+    const generation = targetGeneration.current
+    const request = ++nativeRequest.current
+    setNativeDiscovery(null)
+    setEditing(null)
+    setDetectionNotice(null)
+    setSearchingApplications(false)
+    setSaving(false)
+    if (managed) return
+    fetchApi<NativeDiscovery>(`/api/vms/${vmid}/apps/native-discovery`).then(result => {
+      if (targetRef.current === vmid && targetGeneration.current === generation
+          && nativeRequest.current === request) setNativeDiscovery(result)
+    }).catch(() => { /* cached metadata is optional; never initiate a scan */ })
+    return () => { targetGeneration.current += 1 }
+  }, [vmid, managed])
   const [adguardSetupAvailable, setAdguardSetupAvailable] = useState(false)
   const [error, setError] = useState<string | null>(null)
   const [searchingApplications, setSearchingApplications] = useState(false)
   const [detectionNotice, setDetectionNotice] = useState<{ found: boolean; text: string } | null>(null)
   // Editor state
-  const [editing, setEditing] = useState<{ appId: string | null; draft: AppConfig } | null>(null)
+  const [editing, setEditing] = useState<{ appId: string | null; draft: AppConfig; targetVmid: number } | null>(null)
   const [binaryArgsInput, setBinaryArgsInput] = useState("")
   const [commandArgvInput, setCommandArgvInput] = useState("")
   const [saving, setSaving] = useState(false)
@@ -522,6 +559,8 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
     // `initialData` or the shared cache, the sidecar is already
     // populated and the user shouldn't see a flash of "Loading…"
     // while we revalidate.
+    const generation = targetGeneration.current
+    const isCurrent = () => targetRef.current === vmid && targetGeneration.current === generation
     if (!sidecar) setLoading(true)
     setError(null)
     try {
@@ -533,14 +572,14 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
       // the "Loading applications…" flash that used to show while a
       // second, racing fetch caught up.
       const bundle = await fetchLxcApps(vmid)
-      if (bundle) {
+      if (bundle && isCurrent()) {
         setSidecar(bundle.sidecar)
         setSuggestions(bundle.suggestions)
       }
     } catch (e: any) {
-      setError(e?.message || t("vmLxc.appEditor.loadFailed"))
+      if (isCurrent()) setError(e?.message || t("vmLxc.appEditor.loadFailed"))
     } finally {
-      setLoading(false)
+      if (isCurrent()) setLoading(false)
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [vmid, managed])
@@ -617,7 +656,7 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
     return msg
   }
 
-  const editorOpen = !!editing
+  const editorOpen = !!editing && editing.targetVmid === vmid
 
   // The picker catalog is only needed after the user opens the editor.
   useEffect(() => {
@@ -653,8 +692,8 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
   // (`extras[]` from the backend). Both use the same DetectedApp
   // shape so the empty state renders them uniformly.
   const detectedList: DetectedApp[] = useMemo(() => {
-    if (!suggestions) return []
-    const out: DetectedApp[] = []
+    const out: DetectedApp[] = [...(nativeDiscovery?.vmid === vmid ? nativeDiscovery.candidates : [])]
+    if (!suggestions) return out
     // A ProxMenux OCI install has no helper slug — nothing was installed by a
     // community script — so its template id identifies it instead.
     const primarySlug = suggestions.helper_slug || suggestions.oci_instance?.template_id
@@ -679,12 +718,12 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
       }
     }
     return out
-  }, [suggestions])
+  }, [suggestions, nativeDiscovery, vmid])
 
   // Registered slugs — used to filter the detection list down to
   // what the user hasn't already registered on this CT.
   const registeredSlugs = useMemo(
-    () => new Set(apps.map((a) => a.helper_slug).filter(Boolean) as string[]),
+    () => new Set(apps.flatMap((a) => [a.helper_slug, a.discovery_id]).filter(Boolean) as string[]),
     [apps],
   )
   // Dismissed slugs — persisted in the sidecar. Chips the user
@@ -750,14 +789,32 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
   }
 
   const searchInstalledApplications = async () => {
+    const generation = targetGeneration.current
+    const request = ++nativeRequest.current
+    const isCurrent = () => targetRef.current === vmid && targetGeneration.current === generation
+      && nativeRequest.current === request
     const before = new Set([...visibleDetected, ...visibleWorkloads].map((item) => item.slug))
     setSearchingApplications(true)
     setDetectionNotice(null)
     setError(null)
     try {
+      let native = await fetchApi<NativeDiscovery>(`/api/vms/${vmid}/apps/native-discovery`, {
+        method: "POST",
+      })
+      if (!isCurrent()) return
+      setNativeDiscovery(native)
+      if (native.status === "error" || native.status === "busy") {
+        setDetectionNotice({ found: false, text: t("vmLxc.appEditor.nativeScanFailed") })
+        return
+      }
       const result: Suggestions = await fetchApi(`/api/vms/${vmid}/apps/suggestions`, {
         method: "POST",
       })
+      if (!isCurrent()) return
+      // Recompose against the refreshed catalog without another native scan.
+      native = await fetchApi<NativeDiscovery>(`/api/vms/${vmid}/apps/native-discovery`)
+      if (!isCurrent()) return
+      setNativeDiscovery(native)
       setSuggestions(result)
       if (sidecar) setLxcAppsCached(vmid, sidecar, result)
 
@@ -765,11 +822,14 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
       if (result.helper_slug) detected.add(result.helper_slug)
       for (const item of result.extras || []) detected.add(item.slug)
       for (const item of result.docker_workloads || []) detected.add(item.slug)
+      for (const item of native.candidates) detected.add(item.slug)
       const visible = [...detected].filter(
         (slug) => !registeredSlugs.has(slug) && !dismissedSlugs.has(slug),
       )
       const newCount = visible.filter((slug) => !before.has(slug)).length
-      if (newCount === 1) {
+      if (native.status === "partial") {
+        setDetectionNotice({ found: visible.length > 0, text: t("vmLxc.appEditor.nativeScanPartial") })
+      } else if (newCount === 1) {
         setDetectionNotice({ found: true, text: t("vmLxc.appEditor.oneNewApplicationDetected") })
       } else if (newCount > 1) {
         setDetectionNotice({
@@ -785,9 +845,9 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
         })
       }
     } catch (e: any) {
-      setError(e?.message || t("vmLxc.appEditor.detectionFailed"))
+      if (isCurrent()) setError(e?.message || t("vmLxc.appEditor.detectionFailed"))
     } finally {
-      setSearchingApplications(false)
+      if (isCurrent()) setSearchingApplications(false)
     }
   }
 
@@ -804,6 +864,7 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
     existing?: AppEntry,
     opts?: { withTracking?: boolean, preset?: DetectedApp },
   ) => {
+    const generation = targetGeneration.current
     let seed: AppConfig
     if (existing) {
       seed = {
@@ -832,6 +893,7 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
         health_path: existing.health_path || "",
         logo_url: existing.logo_url || "",
         helper_slug: existing.helper_slug || "",
+        ...(existing.discovery_id ? { discovery_id: existing.discovery_id } : {}),
         // Preserved explicitly: saving replaces the whole record, so dropping
         // it here would silently un-delegate the app the first time someone
         // edits one of its ports.
@@ -862,7 +924,11 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
         const p = opts.preset
         seed.name = p.name
         seed.logo_url = p.logo_url || ""
-        seed.helper_slug = p.slug
+        if (p.discovery_id) {
+          seed.discovery_id = p.discovery_id
+          seed.update_method = "none"
+          seed.update_command = ""
+        } else seed.helper_slug = p.slug
         // Docker endpoints come from the real published host-port mappings
         // listed under Web links.  Do not pre-save a catalog default such as
         // 9000; the user explicitly chooses which workload links to add.
@@ -943,9 +1009,10 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
         setShowAdvanced(false)
       }
     }
+    if (targetRef.current !== vmid || targetGeneration.current !== generation) return
     setBinaryArgsInput((seed.binary_args || []).join(", "))
     setCommandArgvInput((seed.command_argv || []).join(", "))
-    setEditing({ appId: existing?.id || null, draft: seed })
+    setEditing({ appId: existing?.id || null, draft: seed, targetVmid: vmid })
     setDetectorTest(null)
     setError(null)
   }, [suggestions, vmid, sidecar, isOciAdguard, adguardSetupAvailable])
@@ -959,7 +1026,9 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
   }
 
   const save = async () => {
-    if (!editing) return
+    if (!editing || editing.targetVmid !== vmid || targetRef.current !== vmid) return
+    const generation = targetGeneration.current
+    const isCurrent = () => targetRef.current === vmid && targetGeneration.current === generation
     setSaving(true)
     setError(null)
     try {
@@ -973,6 +1042,7 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
         body: JSON.stringify(editing.draft),
       })
       if ((r as any).error) throw new Error((r as any).error)
+      if (!isCurrent()) return
       setSidecar(r)
       setLxcAppsCached(vmid, r, suggestions)
       setEditing(null)
@@ -980,9 +1050,9 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
       setCommandArgvInput("")
       onChange?.()
     } catch (e: any) {
-      setError(e?.message || t("vmLxc.appEditor.saveFailed"))
+      if (isCurrent()) setError(e?.message || t("vmLxc.appEditor.saveFailed"))
     } finally {
-      setSaving(false)
+      if (isCurrent()) setSaving(false)
     }
   }
 
@@ -1283,7 +1353,7 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
   }
 
   // ── Editor ─────────────────────────────────────────────────────
-  if (editing) {
+  if (editorOpen && editing) {
     const draft = editing.draft
     const method = draft.installed_via || ""
     const isPackaged = method === "dpkg" || method === "apk"
@@ -2452,6 +2522,24 @@ export function LxcAppPanel({ vmid, ctIp, onChange, managed, initialData, oci }:
                 ? t("vmLxc.appEditor.versionDetected", { version: d.tracking_suggestion.detected_version || "" })
                 : t("vmLxc.appEditor.detectedInContainer")}
             </div>
+            {d.source === "native_metadata" && (
+              <div className="text-xs text-muted-foreground mt-1 break-all">
+                <div>{d.installed_version
+                  ? t("vmLxc.appEditor.versionDetected", { version: d.installed_version })
+                  : t("vmLxc.appEditor.nativeVersionUnknown")}</div>
+                {(nativeDiscovery?.stale || nativeDiscovery?.status === "partial") && (
+                  <div className="text-amber-400">{t("vmLxc.appEditor.nativeEvidenceStale")}</div>
+                )}
+                {d.evidence?.map((e, index) => <div key={index}>
+                  {t(e.kind === "tcp_listener" ? "vmLxc.appEditor.nativeEvidencePort"
+                    : e.kind === "installation_file" ? "vmLxc.appEditor.nativeEvidenceFile"
+                    : e.kind === "process_executable" ? "vmLxc.appEditor.nativeEvidenceProcess"
+                    : e.kind === "working_directory" ? "vmLxc.appEditor.nativeEvidenceDirectory"
+                    : "vmLxc.appEditor.nativeEvidenceService")}: {e.value}
+                </div>)}
+                <div>{t("vmLxc.appEditor.nativeRegisterOnly")}</div>
+              </div>
+            )}
             {d.tracking_suggestion?.detector_source === "legacy_fallback" && (
               <div className="text-[10px] text-amber-400/90 mt-0.5">
                 {t("vmLxc.appEditor.legacyDetectorUsed")}
diff --git a/AppImage/messages/en/common.json b/AppImage/messages/en/common.json
index cc8d31f7..884db22e 100644
--- a/AppImage/messages/en/common.json
+++ b/AppImage/messages/en/common.json
@@ -1740,6 +1740,16 @@
       "searchApplications": "Find applications",
       "searchingApplications": "Searching applications…",
       "noApplicationsDetected": "No applications were detected.",
+      "nativeEvidencePort": "Listening TCP port",
+      "nativeEvidenceFile": "Installed file",
+      "nativeEvidenceProcess": "Running executable",
+      "nativeEvidenceDirectory": "Working directory",
+      "nativeEvidenceService": "Service definition",
+      "nativeScanFailed": "Application search could not finish. Previous evidence may be outdated; try again.",
+      "nativeScanPartial": "Application search is incomplete. Review the available evidence and try again.",
+      "nativeVersionUnknown": "Installed version unknown",
+      "nativeEvidenceStale": "Evidence may be outdated or incomplete.",
+      "nativeRegisterOnly": "Confirm registration in the editor. A known version is a manual value; no updater is configured.",
       "noNewApplicationsDetected": "No new applications were detected.",
       "dockerDetectedWithWorkloads": "Docker detected with {count} containerised application(s)",
       "dockerWorkloadsHeading": "Running inside Docker",
diff --git a/AppImage/messages/es/common.json b/AppImage/messages/es/common.json
index a9754ca6..02a4260b 100644
--- a/AppImage/messages/es/common.json
+++ b/AppImage/messages/es/common.json
@@ -1723,6 +1723,16 @@
       "searchApplications": "Buscar aplicaciones",
       "searchingApplications": "Buscando aplicaciones…",
       "noApplicationsDetected": "No se han detectado aplicaciones.",
+      "nativeEvidencePort": "Puerto TCP a la escucha",
+      "nativeEvidenceFile": "Archivo instalado",
+      "nativeEvidenceProcess": "Ejecutable en marcha",
+      "nativeEvidenceDirectory": "Directorio de trabajo",
+      "nativeEvidenceService": "Definición del servicio",
+      "nativeScanFailed": "La búsqueda de aplicaciones no ha podido terminar. Lo detectado antes puede estar desactualizado; vuelve a intentarlo.",
+      "nativeScanPartial": "La búsqueda de aplicaciones está incompleta. Revisa lo que se ha encontrado y vuelve a intentarlo.",
+      "nativeVersionUnknown": "Versión instalada desconocida",
+      "nativeEvidenceStale": "Lo detectado puede estar desactualizado o incompleto.",
+      "nativeRegisterOnly": "Confirma el registro en el editor. Una versión conocida se guarda como valor manual; no se configura ningún actualizador.",
       "noNewApplicationsDetected": "No se han detectado aplicaciones nuevas.",
       "oneNewApplicationDetected": "Se ha detectado una aplicación nueva.",
       "newApplicationsDetected": "Se han detectado {count} aplicaciones nuevas.",
diff --git a/AppImage/messages/sk/common.json b/AppImage/messages/sk/common.json
index 4b91a2cd..f699d7dd 100644
--- a/AppImage/messages/sk/common.json
+++ b/AppImage/messages/sk/common.json
@@ -1684,6 +1684,16 @@
       "searchApplications": "Hľadať aplikácie",
       "searchingApplications": "Hľadajú sa aplikácie…",
       "noApplicationsDetected": "Neboli zistené žiadne aplikácie.",
+      "nativeEvidencePort": "Otvorený TCP port",
+      "nativeEvidenceFile": "Nainštalovaný súbor",
+      "nativeEvidenceProcess": "Spustený program",
+      "nativeEvidenceDirectory": "Pracovný priečinok",
+      "nativeEvidenceService": "Definícia služby",
+      "nativeScanFailed": "Vyhľadávanie aplikácií sa nepodarilo dokončiť. Predchádzajúce dôkazy môžu byť zastarané; skús to znova.",
+      "nativeScanPartial": "Vyhľadávanie aplikácií je neúplné. Skontroluj dostupné dôkazy a skús to znova.",
+      "nativeVersionUnknown": "Nainštalovaná verzia nie je známa",
+      "nativeEvidenceStale": "Dôkazy môžu byť zastarané alebo neúplné.",
+      "nativeRegisterOnly": "Registráciu potvrď v editore. Známa verzia je ručný údaj; aktualizátor nie je nastavený.",
       "noNewApplicationsDetected": "Neboli zistené žiadne nové aplikácie.",
       "oneNewApplicationDetected": "Bola zistená jedna nová aplikácia.",
       "newApplicationsDetected": "Boli zistené {count} nové aplikácie.",
diff --git a/AppImage/scripts/build_appimage.sh b/AppImage/scripts/build_appimage.sh
index 783978f6..4bee9953 100755
--- a/AppImage/scripts/build_appimage.sh
+++ b/AppImage/scripts/build_appimage.sh
@@ -149,6 +149,7 @@ cp "$SCRIPT_DIR/managed_installs.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "
 cp "$SCRIPT_DIR/lxc_update_observation.py" "$APP_DIR/usr/bin/"
 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/lxc_apps.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  lxc_apps.py not found"
 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"
diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py
index ca789d38..d1980c36 100644
--- a/AppImage/scripts/flask_server.py
+++ b/AppImage/scripts/flask_server.py
@@ -2278,6 +2278,8 @@ def _handle_guest_lifecycle(vmid: str, vm_type: str, action: str) -> None:
     # or stops; drop the cached IP so the next /api/vms poll re-reads
     # it via `lxc-info`. QEMU guests do not touch this cache.
     if guest_type == 'lxc':
+        import native_app_discovery
+        native_app_discovery.mark_stale(guest_id)
         _invalidate_lxc_ip(guest_id)
     if action in ('start', 'reboot'):
         if guest_type == 'lxc':
@@ -14100,6 +14102,20 @@ def api_vm_apps_suggestions(vmid):
         return jsonify({'error': str(e)}), 500
 
 
+@app.route('/api/vms/<int:vmid>/apps/native-discovery', methods=['GET', 'POST'])
+@require_auth
+def api_vm_apps_native_discovery(vmid):
+    """Only the Find applications button requests POST. GET never probes."""
+    import native_app_discovery
+    import lxc_apps
+    result = (native_app_discovery.discover(vmid) if request.method == 'POST'
+              else native_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(native_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):
diff --git a/AppImage/scripts/lxc_apps.py b/AppImage/scripts/lxc_apps.py
index 494c9db5..4b6a5a2e 100644
--- a/AppImage/scripts/lxc_apps.py
+++ b/AppImage/scripts/lxc_apps.py
@@ -1179,6 +1179,13 @@ def validate_config(payload: dict) -> tuple[bool, Any]:
             return _err("logo_url must be an http(s) URL (max 512 chars)")
         conf["logo_url"] = logo
 
+    # 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):
+            return _err("invalid discovery_id")
+        conf["discovery_id"] = discovery_id
+
     # Optional helper_slug — set by the frontend when the user picks
     # an auto-detected app (primary or extra). Persisted so we can
     # filter the "also detected" chip list against apps already
diff --git a/AppImage/scripts/native_app_discovery.py b/AppImage/scripts/native_app_discovery.py
new file mode 100644
index 00000000..3ba83ec5
--- /dev/null
+++ b/AppImage/scripts/native_app_discovery.py
@@ -0,0 +1,750 @@
+#!/usr/bin/env python3
+# ==========================================================
+# ProxMenux - A menu-driven script for Proxmox VE management
+# Author      : VAIO73
+# GitHub      : github.com/Vaso73
+# Maintainer  : MacRimi
+# Copyright   : (c) 2026 MacRimi & contributors
+# License     : (GPL-3.0) (https://github.com/MacRimi/ProxMenux/blob/main/LICENSE)
+# Version     : 1.0
+# Last Updated: 14/10/2026
+# ==========================================================
+# Description:
+# On-demand native LXC discovery from service and installation metadata.
+# Reads bounded data only; never executes discovered programs, starts guests,
+# registers applications or infers their update mechanism.
+# ==========================================================
+"""Separate from catalog detection: callers must explicitly request a scan."""
+
+import copy
+from contextlib import contextmanager
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import selectors
+import signal
+import stat
+import subprocess
+import threading
+import time
+
+TOTAL_SECONDS = 20
+PROBE_SECONDS = 3
+OUTPUT_BYTES = 64 * 1024
+MAX_RECORDS = 256
+MAX_CANDIDATES = 50
+CACHE_SECONDS = 300
+LOCAL_CONFIG = Path('/etc/pve/local/lxc')
+OCI_ROOT = Path('/usr/local/share/proxmenux/oci/instances')
+UPDATE_LOCK_ROOT = Path('/run/lock')
+_cache = {}
+_locks = {}
+_generations = {}
+_mutex = threading.Lock()
+_PATH = re.compile(r'/[A-Za-z0-9_.+/@-]{1,510}\Z')
+_NAME = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.+@-]{0,63}\Z')
+_VERSION = re.compile(r'v?\d+(?:\.\d+){1,4}(?:[-+._][A-Za-z0-9.-]+)?\Z')
+_SYSTEM = {
+    'ssh', 'sshd', 'cron', 'crond', 'postfix', 'rsyslog', 'syslog', 'syslog-ng',
+    # Alpine/OpenRC system logging and service supervision are not apps.
+    'klogd', 'syslogd', 'openrc-user', 'runit', 'runsvdir', 'runsv', 'supervise-daemon',
+    'dbus', 'dbus-broker', 'systemd', 'udev', 'networking', 'network', 'networkmanager',
+    'dhcpcd', 'dhclient', 'chrony', 'chronyd', 'ntp', 'ntpd', 'openntpd', 'agetty',
+    'getty', 'login', 'polkit', 'polkitd', 'accounts-daemon', 'acpid', 'anacron',
+    'docker', 'containerd', 'containerd-shim', 'podman', 'qemu-guest-agent', 'rngd',
+    'auditd', 'logrotate', 'local', 'bootmisc', 'hostname', 'hwclock', 'modules', 'master',
+    'proxmox-boot-sync', 'pve-container', 'pve-firewall', 'pvefw-logger',
+    'pvedaemon', 'pvestatd', 'pveproxy', 'pvescheduler', 'spiceproxy', 'lxc',
+    'mount-ro', 'mdev', 'sysctl', 'sysfs', 'devfs', 'cgroups', 'rc', 'rcs', 'swap',
+    'sleep', 'true', 'false', 'echo', 'cat', 'head', 'tail', 'tee', 'logger',
+    'apt', 'apt-get', 'dpkg', 'dpkg-query', 'apk', 'unattended-upgrades',
+    'fstrim', 'fsck', 'fsckd', 'e2scrub', 'e2scrub-all', 'e2scrub-reap',
+    'irqbalance', 'smartd', 'smartmontools', 'rpcbind', 'nfs-common', 'nfs-server',
+    'rpc-statd', 'rpc-statd-notify', 'rpc-gssd', 'rpc-svcgssd', 'nftables', 'iptables',
+    'ip6tables', 'firewalld', 'ufw', 'resolvconf', 'wpa-supplicant', 'modemmanager',
+    'console-setup', 'keyboard-setup', 'plymouth', 'halt', 'reboot', 'shutdown',
+    'rc-local', 'ntpsec', 'networkd-dispatcher', 'busybox', 'atd', 'uuidd',
+    # Common packaged daemons are infrastructure, not a user-managed app.
+    'nginx', 'apache', 'apache2', 'httpd', 'lighttpd', 'haproxy', 'traefik',
+    'redis', 'redis-server', 'memcached', 'mysql', 'mysqld', 'mariadb', 'mariadbd',
+    'postgres', 'postgresql', 'mongod', 'mongodb', 'rabbitmq-server',
+    'elasticsearch', 'opensearch', 'fail2ban', 'named', 'bind9', 'dnsmasq',
+    'avahi-daemon', 'smbd', 'nmbd', 'samba', 'vsftpd', 'proftpd', 'cups', 'tor',
+}
+_RUNTIMES = re.compile(r'(?:python[0-9.]*|node(?:js)?|npm|pnpm|yarn|bun|sh|bash|dash|env|java|ruby|perl)\Z')
+_PACKAGE_ROOT_RUNTIMES = {'node', 'nodejs', 'npm', 'pnpm', 'yarn', 'bun'}
+_HELPER_ROLES = {
+    'backup', 'bootstrap', 'bridge', 'cleanup', 'housekeeping', 'maintenance',
+    'migrate', 'migration', 'restore', 'rotate', 'update', 'updater', 'upgrade',
+    'watch', 'watchdog', 'worker',
+}
+_HESTIA_NGINX = '/usr/local/hestia/nginx/sbin/hestia-nginx'
+_HESTIA_ROOT = '/usr/local/hestia'
+_HESTIA_PANEL_PORT = 8083
+
+
+class ProbeError(Exception):
+    """Safe error code only; guest output must never become an API error."""
+
+
+def bounded_run(argv, deadline):
+    """Cap combined stdout/stderr while reading, and kill/reap the process group."""
+    remaining = min(PROBE_SECONDS, deadline - time.monotonic())
+    if remaining <= 0:
+        raise ProbeError('deadline')
+    end = time.monotonic() + remaining
+    try:
+        proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                start_new_session=True)
+    except OSError:
+        raise 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 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 ProbeError('output_limit')
+                    outputs[key.data].extend(block)
+            try:
+                proc.wait(timeout=max(0.001, end - time.monotonic()))
+            except subprocess.TimeoutExpired:
+                raise ProbeError('timeout') from None
+        if proc.returncode:
+            raise ProbeError('read_failed')
+        return outputs[0].decode('utf-8', errors='strict')
+    except UnicodeError:
+        raise ProbeError('invalid_data') from None
+    finally:
+        # Also clean up descendants that outlive a successfully exited wrapper.
+        try:
+            os.killpg(proc.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+        proc.wait()
+        proc.stdout.close()
+        proc.stderr.close()
+
+
+# These are fixed read-only scripts, not guest-supplied shell or init scripts.
+# awk emits only executable/script paths and WorkingDirectory, never arguments
+# or environment values. OpenRC files are parsed as data, never sourced.
+SERVICE_SCRIPT = r'''
+command -v awk >/dev/null && command -v head >/dev/null || exit 1
+n=0
+for dir in /etc/systemd/system /usr/lib/systemd/system /lib/systemd/system /etc/init.d; do
+  # With a merged /usr, /lib is /usr/lib: the same units must not count twice.
+  [ "$dir" = /lib/systemd/system ] && [ /lib/systemd/system -ef /usr/lib/systemd/system ] && continue
+  [ ! -d "$dir" ] || [ -r "$dir" ] || exit 1
+  for f in "$dir"/*; do
+    [ -f "$f" ] || continue
+    case "$dir:$f" in /etc/init.d:*) kind=openrc;; *:*.service) kind=systemd;; *) continue;; esac
+    n=$((n+1)); [ "$n" -le 256 ] || { printf 'LIMIT\n'; exit 0; }
+    [ -r "$f" ] || exit 1
+    content=$(timeout -s KILL 1 head -c 16385 "$f") || exit 1
+    [ "${#content}" -le 16384 ] || { printf 'LIMIT\n'; exit 0; }
+    printf '%s\n' "$content" | awk -v kind="$kind" -v file="$f" '
+      function clean(v) {gsub(/^[ \t]+|[ \t]+$/, "", v); gsub(/^["\047]|["\047]$/, "", v); return v}
+      /^ExecStart=/ {v=substr($0,11); sub(/^[-+!:@]+/, "", v); split(v,a,/[ \t]+/); exe=clean(a[1]); script=clean(a[2]); if(script !~ /^\//) script=""}
+      /^WorkingDirectory=/ {dir=clean(substr($0,18))}
+      /^command=/ {exe=clean(substr($0,9))}
+      /^directory=/ {dir=clean(substr($0,11))}
+      /^command_args=/ {v=clean(substr($0,14)); split(v,a,/[ \t]+/); script=clean(a[1]); if(script !~ /^\//) script=""}
+      END {printf "S\t%s\t%s\t%s\t%s\t%s\n", file, kind, exe, script, dir}'
+  done
+done
+printf 'END\n'
+'''
+PROCESS_SCRIPT = r'''
+command -v readlink >/dev/null && command -v head >/dev/null || exit 1
+n=0
+for dir in /proc/[0-9]*; do
+  [ -d "$dir" ] || continue
+  n=$((n+1)); [ "$n" -le "$1" ] || { printf 'LIMIT\n'; exit 0; }
+  # Never inspect Docker/OCI child workloads or read cmdline/environ.
+  [ -d "$dir" ] || continue
+  [ -r "$dir/cgroup" ] || exit 1
+  group=$(timeout -s KILL 1 head -c 8193 "$dir/cgroup") || exit 1
+  [ "${#group}" -le 8192 ] || { printf 'LIMIT\n'; exit 0; }
+  case "$group" in *docker*|*libpod*|*kubepods*|*containerd*) continue;; esac
+  # Kernel threads and vanished processes have no application executable.
+  [ -L "$dir/exe" ] || continue
+  exe=$(timeout -s KILL 1 readlink "$dir/exe") || continue
+  cwd=$(timeout -s KILL 1 readlink "$dir/cwd") || continue
+  printf 'P\t%s\t%s\t%s\n' "${dir##*/}" "$exe" "$cwd"
+done
+printf 'END\n'
+'''
+PORT_SCRIPT = r'''
+command -v awk >/dev/null || exit 1
+if command -v ss >/dev/null; then
+  (timeout -s KILL 1 ss -H -lntp || printf 'FAILED\n') | awk -v cap="$1" '
+    /^FAILED$/ {failed=1; exit}
+    NR > cap {limited=1; exit}
+    {port=$4; sub(/^.*:/,"",port); rest=$0
+     # More than one owner is ambiguous; never choose the first arbitrarily.
+     if (gsub(/pid=/,"pid=",rest)!=1) next
+     if (match(rest,/pid=[0-9]+/)) {pid=substr(rest,RSTART+4,RLENGTH-4); printf "L\t%s\t%s\n",pid,port}}
+    END {print failed ? "FAILED" : (limited ? "LIMIT" : "END")}'
+elif command -v netstat >/dev/null; then
+  (timeout -s KILL 1 netstat -lntp || printf 'FAILED\n') | awk -v cap="$1" '
+    /^FAILED$/ {failed=1; exit}
+    /^tcp/ {n++; if(n>cap) {limited=1; exit}; port=$4; sub(/^.*:/,"",port)
+      pid=$7; sub(/\/.*$/,"",pid); if(pid ~ /^[0-9]+$/) printf "L\t%s\t%s\n",pid,port}
+    END {print failed ? "FAILED" : (limited ? "LIMIT" : "END")}'
+else
+  # Without ss or netstat there is no listener evidence; a retry would not change it.
+  printf 'END\n'
+fi
+'''
+HESTIA_PANEL_SCRIPT = r'''
+# The generic process collector intentionally stops after a bounded number of
+# records and ignores shared listener ownership.  Hestia's master/worker pair
+# can therefore be outside that generic result.  This narrow fallback never
+# accepts a name alone: it requires Hestia's exact executable and its panel
+# listener on the fixed local port.
+[ -x /usr/local/hestia/nginx/sbin/hestia-nginx ] || { printf 'END\n'; exit 0; }
+command -v pgrep >/dev/null && command -v readlink >/dev/null && command -v ss >/dev/null && command -v head >/dev/null || exit 1
+n=0
+for pid in $(pgrep -x hestia-nginx 2>/dev/null | head -n 9); do
+  case "$pid" in *[!0-9]*|'') continue;; esac
+  n=$((n+1)); [ "$n" -le 8 ] || { printf 'LIMIT\n'; exit 0; }
+  exe=$(timeout -s KILL 1 readlink "/proc/$pid/exe") || continue
+  [ "$exe" = /usr/local/hestia/nginx/sbin/hestia-nginx ] || continue
+  cwd=$(timeout -s KILL 1 readlink "/proc/$pid/cwd") || continue
+  if timeout -s KILL 1 ss -H -lntp | awk -v pid="$pid" '
+    {port=$4; sub(/^.*:/,"",port)
+     if (port == 8083 && index($0, "pid=" pid ",")) found=1}
+    END {exit(found ? 0 : 1)}'; then
+    printf 'H\t%s\t%s\t%s\t8083\n' "$pid" "$exe" "$cwd"
+  fi
+done
+printf 'END\n'
+'''
+FOOTPRINT_SCRIPT = r'''
+[ -f "$1" ] && [ -r "$1" ] || exit 1
+printf 'FILE\n'
+if [ -f "$2/package.json" ]; then
+  [ -r "$2/package.json" ] || exit 1
+  timeout -s KILL 1 head -c 8192 "$2/package.json"
+fi
+'''
+MANIFEST_FOOTPRINT_SCRIPT = r'''
+command -v head >/dev/null && command -v readlink >/dev/null || exit 1
+# Resolve only the operator-owned working directory and its declared parent.
+# A release symlink may change the concrete version directory, but it must not
+# escape the bounded custom application root supplied by the host.
+workdir=$(timeout -s KILL 1 readlink -f "$1") || exit 1
+anchor=$(timeout -s KILL 1 readlink -f "$2") || exit 1
+parent=$(timeout -s KILL 1 readlink -f "$3") || exit 1
+case "$anchor" in "$parent"/*) ;; *) printf 'OUTSIDE\n'; exit 0;; esac
+case "$workdir" in "$anchor"|"$anchor"/*) ;; *) printf 'OUTSIDE\n'; exit 0;; esac
+for relative in package.json server/package.json; do
+  candidate="$workdir/$relative"
+  [ -f "$candidate" ] && [ -r "$candidate" ] || continue
+  resolved=$(timeout -s KILL 1 readlink -f "$candidate") || exit 1
+  case "$resolved" in "$workdir"/*) ;; *) printf 'OUTSIDE\n'; exit 0;; esac
+  printf 'MANIFEST\t%s\n' "$relative"
+  timeout -s KILL 1 head -c 8192 "$resolved"
+  exit 0
+done
+printf 'MISSING\n'
+'''
+
+
+def _path(value):
+    return bool(isinstance(value, str) and _PATH.fullmatch(value)
+                and '..' not in value.split('/'))
+
+
+def _system(name):
+    raw = name.lower().removesuffix('.service')
+    stem = raw.split('@')[0]
+    return (raw.startswith('user@') or stem in _SYSTEM
+            or (raw.startswith('php') and raw.endswith('-fpm'))
+            or stem.startswith((
+        'systemd-', 'dbus-', 'getty', 'containerd-', 'apt-', 'dpkg-', 'e2scrub',
+        'unattended-', 'ifup', 'ifdown', 'user@', 'btrfs-', 'lvm2-', 'lxc-')))
+
+
+def _helper(name):
+    """Reject service roles that support an application but are not one themselves."""
+    raw = name.lower().removesuffix('.service')
+    if raw.endswith('@'):
+        return True  # A systemd template is not a concrete running application.
+    return any(part in _HELPER_ROLES for part in re.split(r'[._-]+', raw) if part)
+
+
+def _hestia_panel_process(executable, ports):
+    """Recognize only Hestia's own bound control-panel process.
+
+    Hestia's generated init unit does not expose an ExecStart that the generic
+    service parser may treat as an application.  Its panel process and port are
+    stable, local runtime metadata, so this stays narrower than accepting
+    arbitrary vendor services or a generic web server.
+    """
+    return executable == _HESTIA_NGINX and _HESTIA_PANEL_PORT in ports
+
+
+def _custom(path):
+    if not _path(path):
+        return False
+    if path.startswith(('/opt/', '/srv/', '/usr/local/')):
+        return True
+    # Native apps commonly keep their runtime below /home/<service-user>/<app>.
+    # Do not accept a home directory itself: a second path component is required.
+    parts = path.split('/')
+    return len(parts) >= 4 and parts[1] == 'home' and bool(parts[2]) and bool(parts[3])
+
+
+def _manifest_bounds(executable, directory):
+    """Return a narrow, operator-owned boundary for a runtime manifest probe."""
+    try:
+        shared = os.path.commonpath((executable, directory))
+    except ValueError:
+        shared = ''
+    anchor = shared if _custom(shared) else directory
+    parts = anchor.split('/')
+    if anchor.startswith(('/opt/', '/srv/')):
+        parent = '/' + parts[1]
+    elif anchor.startswith('/usr/local/'):
+        parent = '/usr/local'
+    elif len(parts) >= 4 and parts[1] == 'home' and parts[2]:
+        parent = f'/home/{parts[2]}'
+    else:
+        return None
+    return anchor, parent
+
+
+def _manifest_footprint(record, run, deadline):
+    """Read one of two fixed Node manifest locations without following an escape."""
+    output = run(['pct', 'exec', str(record['vmid']), '--', 'timeout', '-s', 'KILL', '2',
+                  'sh', '-c', MANIFEST_FOOTPRINT_SCRIPT, 'discovery',
+                  record['root'], record['manifest_anchor'], record['manifest_parent']], deadline)
+    header, separator, metadata = output.partition('\n')
+    if header == 'MISSING':
+        if metadata:
+            raise ProbeError('invalid_data')
+        return None, None
+    if header == 'OUTSIDE':
+        if metadata:
+            raise ProbeError('invalid_data')
+        raise ProbeError('unsafe_manifest_root')
+    prefix = 'MANIFEST\t'
+    relative = header.removeprefix(prefix)
+    if not header.startswith(prefix) or relative not in ('package.json', 'server/package.json'):
+        raise ProbeError('invalid_data')
+    if not separator:
+        raise ProbeError('invalid_data')
+    candidate_record = dict(record)
+    candidate_record['path'] = str(Path(record['root']) / relative)
+    return candidate_record, 'FILE\n' + metadata
+
+
+def parse_records(text, kind):
+    lines = text.splitlines()
+    if not lines or lines[-1] not in ('END', 'LIMIT'):
+        raise ProbeError('incomplete')
+    if len(lines) - 1 > MAX_RECORDS:
+        raise ProbeError('metadata_limit')
+    records = []
+    seen_services = set()
+    for line in lines[:-1]:
+        parts = line.split('\t')
+        if kind == 'S' and len(parts) == 6 and parts[0] == 'S':
+            _, source, manager, exe, script, directory = parts
+            if not _path(source) or manager not in ('systemd', 'openrc'):
+                raise ProbeError('invalid_data')
+            service = Path(source).name.removesuffix('.service')
+            if service in seen_services:
+                continue  # /etc overrides vendor units; no duplicate installations.
+            seen_services.add(service)
+            if (not _NAME.fullmatch(service) or _system(service) or _helper(service)
+                    or not _path(exe)):
+                continue
+            runtime_name = Path(exe).name
+            runtime = bool(_RUNTIMES.fullmatch(runtime_name))
+            manifest_bounds = None
+            if runtime and not _path(script):
+                # A Node/npm wrapper may keep its entry point relative to an
+                # operator-owned working directory.  Only two fixed manifest
+                # locations can prove that root; no argv, environment, recursive
+                # search or guest-provided path is ever used.
+                if runtime_name not in _PACKAGE_ROOT_RUNTIMES or not _custom(directory):
+                    continue
+                install = str(Path(directory) / 'package.json')
+                manifest_bounds = _manifest_bounds(exe, directory)
+                if manifest_bounds is None:
+                    continue
+            else:
+                install = script if runtime else exe
+            if (not _path(install) or _system(Path(install).name)
+                    or install.startswith(('/usr/lib/systemd/', '/lib/systemd/',
+                                           '/usr/lib/udev/', '/lib/udev/', '/usr/lib/apt/'))):
+                continue
+            # /etc units are operator-owned.  Do not infer an application from an
+            # arbitrary packaged systemd unit: without a package-db probe it is
+            # impossible to distinguish vendor infrastructure from a user app.
+            if (manager == 'systemd'
+                    and source.startswith(('/usr/lib/systemd/system/', '/lib/systemd/system/'))
+                    and not _custom(install)):
+                continue
+            root = directory if _custom(directory) else str(Path(install).parent)
+            # Generic runtime without an absolute application script is ambiguous.
+            if runtime and not _custom(install):
+                continue
+            record = {'name': service, 'path': install, 'root': root,
+                      'service': source, 'manager': manager, 'executable': exe}
+            if manifest_bounds:
+                record['manifest_anchor'], record['manifest_parent'] = manifest_bounds
+            records.append(record)
+        elif kind == 'P' and len(parts) == 4 and parts[0] == 'P':
+            _, pid, exe, directory = parts
+            # A process rooted at / is normally too ambiguous for generic
+            # discovery, but it is valid process metadata for the exact Hestia
+            # panel signature below.  It never becomes a generic candidate.
+            if not pid.isdigit() or not _path(exe) or (directory != '/' and not _path(directory)):
+                continue
+            records.append({'pid': pid, 'executable': exe, 'root': directory})
+        elif kind == 'L' and len(parts) == 3 and parts[0] == 'L':
+            _, pid, port = parts
+            if pid.isdigit() and port.isdigit() and 0 < int(port) < 65536:
+                records.append({'pid': pid, 'port': int(port)})
+            else:
+                raise ProbeError('invalid_data')
+        elif kind == 'H' and len(parts) == 5 and parts[0] == 'H':
+            _, pid, exe, directory, port = parts
+            if (pid.isdigit() and exe == _HESTIA_NGINX
+                    and (directory == '/' or _path(directory))
+                    and port == str(_HESTIA_PANEL_PORT)):
+                records.append({'pid': pid, 'executable': exe, 'root': directory,
+                                'port': _HESTIA_PANEL_PORT})
+            else:
+                raise ProbeError('invalid_data')
+        else:
+            raise ProbeError('invalid_data')
+    return records, lines[-1] == 'LIMIT', len(lines) - 1
+
+
+def candidate(vmid, record, footprint):
+    if not footprint.startswith('FILE\n'):
+        raise ProbeError('invalid_data')
+    version, metadata_issue = None, None
+    metadata = footprint[5:].strip()
+    if metadata:
+        try:
+            manifest = json.loads(metadata)
+        except (ValueError, TypeError):
+            manifest = None
+        # Only the whitelisted version field leaves this function. Never retain
+        # manifest scripts, repository guesses, config or arbitrary extra fields.
+        if not isinstance(manifest, dict):
+            metadata_issue = 'invalid_metadata'
+            manifest = {}
+        value = manifest.get('version')
+        if (isinstance(value, str) and len(value) <= 64 and _VERSION.fullmatch(value)
+                and record['path'].startswith(record['root'].rstrip('/') + '/')):
+            version = value
+    identity = 'native-' + hashlib.sha256(
+        f'{int(vmid)}:{record["path"]}:{record["root"]}'.encode()).hexdigest()[:24]
+    evidence = [{'kind': 'installation_file', 'value': record['path']}]
+    if record.get('service'):
+        evidence.append({'kind': record['manager'], 'value': record['service']})
+    else:
+        evidence.extend([{'kind': 'process_executable', 'value': record['process']},
+                         {'kind': 'working_directory',
+                          'value': record.get('process_root', record['root'])}])
+    ports = sorted(set(record.get('listening_ports') or []))
+    # Hestia's control panel is a narrow, verified exception to generic port
+    # inference: its own executable serves the panel on fixed HTTPS port 8083.
+    # Keep this declaration on the exact runtime signature so an unrelated
+    # application merely listening on 8083 never receives an HTTPS guess.
+    hestia_panel = (record['name'] == 'Hestia'
+                    and record['path'] == _HESTIA_NGINX
+                    and record['root'] == _HESTIA_ROOT
+                    and _HESTIA_PANEL_PORT in ports)
+    evidence.extend({'kind': 'tcp_listener', 'value': str(port)} for port in ports)
+    return {'slug': identity, 'discovery_id': identity, 'source': 'native_metadata',
+            'name': record['name'], 'installed_version': version, 'evidence': evidence,
+            'installation_path': record['path'],
+            'default_ports': [_HESTIA_PANEL_PORT] if hestia_panel else [],
+            'listening_ports': ports,
+            **({'scheme': 'https'} if hestia_panel else {}),
+            'metadata_issue': metadata_issue,
+            'tracking_suggestion': ({'installed_via': 'manual', 'installed_version': version,
+                                     'detected_version': version, 'detector_verified': True}
+                                    if version else None)}
+
+
+def _guard(vmid, deadline, run):
+    # The local node path rejects QEMU, missing and remote CTs before pct exec.
+    path = LOCAL_CONFIG / f'{vmid}.conf'
+    if not path.is_file():
+        raise ProbeError('not_local_lxc')
+    # Presence, including a damaged/in-progress record, excludes generic probes.
+    try:
+        os.lstat(OCI_ROOT / str(vmid))
+    except FileNotFoundError:
+        pass
+    except OSError:
+        raise ProbeError('oci_unknown') from None
+    else:
+        raise ProbeError('managed_oci')
+    try:
+        with path.open('rb') as handle:
+            raw = handle.read(OUTPUT_BYTES + 1)
+        if len(raw) > OUTPUT_BYTES:
+            raise ProbeError('config_limit')
+        config = raw.decode('utf-8')
+    except (OSError, UnicodeError):
+        raise ProbeError('config_unavailable') from None
+    if re.search(r'^lock:', config, re.M):
+        raise ProbeError('guest_busy')
+    update_lock = UPDATE_LOCK_ROOT / f'proxmenux-lxc-update-{vmid}.lock'
+    try:
+        fd = os.open(update_lock, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        with os.fdopen(fd, 'rb') as handle:
+            info = os.fstat(handle.fileno())
+            if not stat.S_ISREG(info.st_mode):
+                raise ProbeError('lock_unavailable')
+            if info.st_size:
+                raise ProbeError('guest_busy')
+            try:
+                fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB)
+            except BlockingIOError:
+                raise ProbeError('guest_busy') from None
+            finally:
+                fcntl.flock(handle, fcntl.LOCK_UN)
+    except FileNotFoundError:
+        pass
+    except OSError:
+        raise ProbeError('lock_unavailable') from None
+    if re.search(r'^ostype:\s*unmanaged\s*$', config, re.M):
+        raise ProbeError('unsupported_guest')
+    if not re.search(r'^ostype:\s*(?:debian|ubuntu|alpine|archlinux|centos|fedora|opensuse|gentoo)\s*$', config, re.M):
+        raise ProbeError('unsupported_guest')
+    status = run(['pct', 'status', str(vmid)], deadline).strip()
+    if status != 'status: running':
+        raise ProbeError('guest_stopped' if status == 'status: stopped' else 'status_unknown')
+
+
+@contextmanager
+def _scan_reservation(vmid):
+    """Exclude cooperating updates/control tasks for this bounded scan only.
+
+    Keep an empty lock inode after first use, as the update runner does.
+    Never truncate a pending control-task reservation or unlink its inode.
+    """
+    path = UPDATE_LOCK_ROOT / f'proxmenux-lxc-update-{vmid}.lock'
+    fd = None
+    try:
+        fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            raise ProbeError('lock_unavailable')
+        try:
+            fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise ProbeError('guest_busy') from None
+        if os.fstat(fd).st_size:
+            raise ProbeError('guest_busy')
+        yield
+    except OSError:
+        raise ProbeError('lock_unavailable') from None
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
+def scan(vmid, run=bounded_run):
+    deadline = time.monotonic() + TOTAL_SECONDS
+    _guard(vmid, deadline, run)
+    with _scan_reservation(vmid):
+        result = _scan_metadata(vmid, deadline, run)
+        # External PVE lifecycle operations do not share our flock. Fail closed
+        # if the guest stopped, changed ownership or became locked meanwhile.
+        _guard(vmid, deadline, run)
+        return result
+
+
+def _scan_metadata(vmid, deadline, run):
+    records, issues, consumed, processes, listeners, hestia_panels = [], [], 0, [], [], []
+    for kind, script in (('S', SERVICE_SCRIPT), ('P', PROCESS_SCRIPT), ('L', PORT_SCRIPT)):
+        try:
+            if consumed >= MAX_RECORDS:
+                issues.append('metadata_limit')
+                break
+            parsed, limited, count = parse_records(run(
+                ['pct', 'exec', str(vmid), '--', 'timeout', '-s', 'KILL', '2',
+                 'sh', '-c', script, 'discovery', str(MAX_RECORDS - consumed)], deadline), kind)
+            if consumed + count > MAX_RECORDS:
+                issues.append('metadata_limit')
+                parsed = parsed[:MAX_RECORDS - consumed]
+            consumed = min(MAX_RECORDS, consumed + count)
+            # The entire collector, not each metadata source, has 256 records.
+            capacity = MAX_RECORDS - len(records)
+            if kind == 'S':
+                records.extend(parsed[:capacity])
+            elif kind == 'P':
+                processes = parsed[:capacity]
+            else:
+                listeners = parsed[:capacity]
+            if limited or len(parsed) > capacity:
+                issues.append('metadata_limit')
+        except ProbeError as exc:
+            issues.append(str(exc))
+    try:
+        # This is outside the generic record budget: it returns only an exact,
+        # fixed Hestia signature, not additional arbitrary process metadata.
+        hestia_panels, limited, _ = parse_records(run(
+            ['pct', 'exec', str(vmid), '--', 'timeout', '-s', 'KILL', '2',
+             'sh', '-c', HESTIA_PANEL_SCRIPT, 'discovery'], deadline), 'H')
+        if limited:
+            issues.append('metadata_limit')
+    except ProbeError as exc:
+        issues.append(str(exc))
+    for process in processes:
+        exe, directory = process['executable'], process['root']
+        related = [record for record in records if record['executable'] == exe and (
+            not _RUNTIMES.fullmatch(Path(exe).name) or record['root'] == directory)]
+        ports = [entry['port'] for entry in listeners if entry['pid'] == process['pid']]
+        if len(related) == 1:
+            related[0].setdefault('listening_ports', []).extend(ports)
+        elif not related and _hestia_panel_process(exe, ports):
+            records.append({'name': 'Hestia', 'path': exe, 'root': _HESTIA_ROOT,
+                            'process': exe, 'process_root': directory, 'executable': exe,
+                            'listening_ports': ports})
+        elif not related and (_custom(exe) and _custom(directory)
+                              and not _system(Path(exe).name)
+                              and not _helper(Path(exe).name)
+                              and not _RUNTIMES.fullmatch(Path(exe).name)
+                              and exe.startswith(directory.rstrip('/') + '/')
+                              and _NAME.fullmatch(Path(exe).name)):
+            records.append({'name': Path(exe).name, 'path': exe, 'root': directory,
+                            'process': exe, 'executable': exe, 'listening_ports': ports})
+    for panel in hestia_panels:
+        records.append({'name': 'Hestia', 'path': panel['executable'], 'root': _HESTIA_ROOT,
+                        'process': panel['executable'], 'process_root': panel['root'],
+                        'executable': panel['executable'],
+                        'listening_ports': [panel['port']]})
+    items, seen = [], set()
+    for record in records:
+        key = (record['path'], record['root'])
+        if key in seen:
+            continue
+        seen.add(key)
+        if len(items) >= MAX_CANDIDATES:
+            issues.append('candidate_limit')
+            break
+        try:
+            probe_record = dict(record)
+            probe_record['vmid'] = vmid
+            if 'manifest_anchor' in record:
+                probe_record, footprint = _manifest_footprint(probe_record, run, deadline)
+                if probe_record is None:
+                    continue
+            else:
+                footprint = run(['pct', 'exec', str(vmid), '--', 'timeout', '-s', 'KILL', '2',
+                                 'sh', '-c', FOOTPRINT_SCRIPT,
+                                 'discovery', record['path'], record['root']], deadline)
+            item = candidate(vmid, probe_record, footprint)
+            items.append(item)
+            if item['metadata_issue']:
+                issues.append(item['metadata_issue'])
+        except ProbeError as exc:
+            issues.append(str(exc))
+            if str(exc) == 'deadline':
+                break
+    return {'vmid': vmid, 'status': 'partial' if issues else 'complete',
+            'candidates': items, 'issues': sorted(set(issues)), 'stale': False,
+            'scanned_at': time.time()}
+
+
+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}
+    if time.time() - value['scanned_at'] >= CACHE_SECONDS:
+        value['stale'] = True
+    return value
+
+
+def mark_stale(vmid):
+    """Lifecycle events retire freshness, never collect new guest metadata."""
+    with _mutex:
+        _generations[int(vmid)] = _generations.get(int(vmid), 0) + 1
+        if int(vmid) in _cache:
+            _cache[int(vmid)]['stale'] = 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 ProbeError as exc:
+            result = {**cached(vmid), 'status': 'error', 'issues': [str(exc)], 'stale': True}
+            result.setdefault('scanned_at', time.time())
+        # Comparing the generation and publishing must be atomic with lifecycle
+        # invalidation, including the first scan when no cache exists yet.
+        with _mutex:
+            previous = copy.deepcopy(_cache.get(vmid))
+            if _generations.get(vmid, 0) != generation:
+                result = {**(previous or {'vmid': vmid, 'candidates': [],
+                                         'scanned_at': time.time()}),
+                          'status': 'error', 'issues': ['guest_changed'], 'stale': True}
+            elif result['status'] != 'complete' and previous and previous['candidates']:
+                # Partial output cannot prove that previously seen apps disappeared.
+                ids = {item['discovery_id'] for item in result['candidates']}
+                result['candidates'].extend(item for item in previous['candidates']
+                                            if item['discovery_id'] not in ids)
+                result['candidates'] = result['candidates'][:MAX_CANDIDATES]
+                result['stale'] = True
+            _cache[vmid] = copy.deepcopy(result)
+        return result
+    finally:
+        lock.release()
+
+
+def filter_candidates(result, suggestions, apps):
+    """Pure composition: never run catalog probes, load or write registries."""
+    known = [*apps, *(suggestions.get('extras') or []),
+             *(suggestions.get('docker_workloads') or [])]
+    if suggestions.get('helper_slug'):
+        known.append({'name': suggestions.get('name_suggestion'),
+                      'helper_slug': suggestions['helper_slug'],
+                      'tracking_suggestion': suggestions.get('tracking_suggestion')})
+    identities = {a.get('discovery_id') for a in known if a.get('discovery_id')}
+    names, paths = set(), set()
+    for app in known:
+        for field in ('name', 'slug', 'helper_slug', 'package'):
+            if app.get(field):
+                names.add(str(app[field]).casefold())
+        detector = app.get('tracking_suggestion') or app
+        for field in ('binary_path', 'file_path'):
+            if detector.get(field):
+                paths.add(detector[field])
+    items = [item for item in result['candidates']
+             if item['discovery_id'] not in identities
+             and item['name'].casefold() not in names
+             and item['installation_path'] not in paths]
+    return {**result, 'candidates': items}
diff --git a/AppImage/scripts/tests/test_native_app_discovery.py b/AppImage/scripts/tests/test_native_app_discovery.py
new file mode 100644
index 00000000..dc24424a
--- /dev/null
+++ b/AppImage/scripts/tests/test_native_app_discovery.py
@@ -0,0 +1,785 @@
+"""Host-independent discovery fixtures; no real pct, guest, auth or registry."""
+import ast
+import copy
+import fcntl
+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 Mock, patch
+
+SCRIPTS = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(SCRIPTS))
+import native_app_discovery as native
+import lxc_apps
+
+
+def service(name='handmade', exe='/opt/handmade/server', directory='/opt/handmade', manager='systemd', script='', source=None):
+    path = source or ('/etc/systemd/system/' + name + '.service' if manager == 'systemd'
+                      else '/etc/init.d/' + name)
+    return '\t'.join(['S', path, manager, exe, script, directory]) + '\nEND\n'
+
+
+class DiscoveryFixtures(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.config = self.root / 'lxc'
+        self.config.mkdir()
+        (self.config / '101.conf').write_text('ostype: debian\n')
+        self.oci = self.root / 'oci'
+        self.oci.mkdir()
+        for context in (patch.object(native, 'LOCAL_CONFIG', self.config),
+                        patch.object(native, 'OCI_ROOT', self.oci),
+                        patch.object(native, 'UPDATE_LOCK_ROOT', self.root)):
+            context.start()
+            self.addCleanup(context.stop)
+        native._cache.clear()
+        native._locks.clear()
+        native._generations.clear()
+        self.calls = []
+
+    def runner(self, services=None, processes='END\n', footprint='FILE\n', manifest=None,
+               status='status: running', ports='END\n', hestia_panels='END\n'):
+        def run(argv, deadline):
+            self.calls.append(argv)
+            self.assertLessEqual(deadline - time.monotonic(), native.TOTAL_SECONDS)
+            if argv[1] == 'status':
+                return status
+            self.assertEqual(argv[:6], ['pct', 'exec', '101', '--', 'timeout', '-s'])
+            self.assertEqual(argv[6:9], ['KILL', '2', 'sh'])
+            script = argv[10]
+            if script == native.SERVICE_SCRIPT:
+                return services if services is not None else service()
+            if script == native.PROCESS_SCRIPT:
+                return processes
+            if script == native.PORT_SCRIPT:
+                return ports
+            if script == native.HESTIA_PANEL_SCRIPT:
+                return hestia_panels
+            if script == native.MANIFEST_FOOTPRINT_SCRIPT:
+                if manifest is not None:
+                    return manifest
+                return 'MANIFEST\tpackage.json\n' + footprint.removeprefix('FILE\n')
+            if script == native.FOOTPRINT_SCRIPT:
+                return footprint
+            self.fail('Unexpected probe')
+        return run
+
+    def test_unknown_version_is_present_with_evidence_and_no_capabilities(self):
+        result = native.scan(101, self.runner())
+        self.assertEqual(result['status'], 'complete')
+        item, = result['candidates']
+        self.assertIsNone(item['installed_version'])
+        self.assertIsNone(item['tracking_suggestion'])
+        self.assertEqual(item['default_ports'], [])
+        self.assertEqual({e['kind'] for e in item['evidence']}, {'installation_file', 'systemd'})
+        for field in ('helper_slug', 'repo', 'update_command', 'binary_path', 'upstream_type'):
+            self.assertNotIn(field, item)
+
+    def test_openrc_node_script_known_version_is_manual_and_secret_free(self):
+        result = native.scan(101, self.runner(
+            service(exe='/usr/bin/node', script='/opt/handmade/server.js', manager='openrc'),
+            footprint='FILE\n' + json.dumps({'version': '1.2.3', 'scripts': {'start': 'SECRET'}, 'env': 'PASSWORD'})))
+        item, = result['candidates']
+        self.assertEqual(item['installed_version'], '1.2.3')
+        self.assertEqual(item['tracking_suggestion']['installed_via'], 'manual')
+        self.assertNotIn('SECRET', json.dumps(result))
+        self.assertNotIn('PASSWORD', json.dumps(result))
+        self.assertFalse(any('systemctl' in arg for call in self.calls for arg in call))
+
+    def test_inactive_service_and_packaged_application_are_not_hidden(self):
+        result = native.scan(101, self.runner(service(name='legit-app', exe='/usr/bin/legit-app', directory='')))
+        self.assertEqual(result['candidates'][0]['name'], 'legit-app')
+
+    def test_system_services_and_ambiguous_runtimes_have_zero_footprint_probes(self):
+        for name, exe in [('ssh', '/usr/sbin/sshd'), ('cron', '/usr/sbin/cron'),
+                          ('postfix', '/usr/lib/postfix/master'), ('rsyslog', '/usr/sbin/rsyslogd'),
+                          ('systemd-networkd', '/usr/lib/systemd/systemd-networkd'),
+                          ('ambiguous', '/usr/bin/python3')]:
+            self.calls.clear()
+            result = native.scan(101, self.runner(service(name=name, exe=exe)))
+            self.assertEqual(result['candidates'], [])
+            self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_common_system_daemons_and_vendor_units_have_zero_footprint_probes(self):
+        cases = [
+            ('nginx', '/usr/sbin/nginx', 'openrc', None),
+            ('apache2', '/usr/sbin/apache2', 'systemd', None),
+            ('redis-server', '/usr/bin/redis-server', 'systemd', None),
+            ('mariadb', '/usr/sbin/mariadbd', 'systemd', None),
+            ('postgresql', '/usr/lib/postgresql/16/bin/postgres', 'systemd', None),
+            ('php8.3-fpm', '/usr/sbin/php-fpm8.3', 'systemd', None),
+            ('vendor-daemon', '/usr/bin/vendor-daemon', 'systemd',
+             '/usr/lib/systemd/system/vendor-daemon.service'),
+        ]
+        for name, exe, manager, source in cases:
+            with self.subTest(name=name):
+                self.calls.clear()
+                result = native.scan(101, self.runner(
+                    service(name=name, exe=exe, manager=manager, source=source)))
+                self.assertEqual(result['candidates'], [])
+                self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_alpine_openrc_infrastructure_has_zero_footprint_probes(self):
+        # Actual Alpine 3.24.1 CT217 init metadata: klogd, user and runsvdir.
+        # The runsvdir definition can exist without /usr/bin/runsvdir installed;
+        # rejecting it must precede probing, not turn the scan into partial.
+        for name, executable in (
+            ('klogd', '/sbin/klogd'),
+            ('user', '/usr/libexec/rc/bin/openrc-user'),
+            ('runsvdir', '/usr/bin/runsvdir'),
+            ('renamed-system-service', '/usr/libexec/rc/bin/openrc-user'),
+        ):
+            with self.subTest(name=name):
+                self.calls.clear()
+                result = native.scan(101, self.runner(
+                    service(name=name, exe=executable, directory='', manager='openrc')))
+                self.assertEqual(result['status'], 'complete')
+                self.assertEqual(result['candidates'], [])
+                self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_helper_services_and_templates_have_zero_footprint_probes(self):
+        cases = [
+            ('netpulse-stable-update', '/bin/bash', '/opt/netpulse', '/opt/netpulse/stable-apply.sh'),
+            ('rent-tool-backup', '/bin/bash', '/home/app/rent-tool', '/home/app/rent-tool/scripts/backup-db.sh'),
+            ('keynest-update', '/bin/bash', '/opt/keynest', '/opt/keynest/keynest-update.sh'),
+            ('cortendesk-bootstrap-worker', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/bootstrap-worker.sh'),
+            ('cortendesk-source-worker', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/source-worker.sh'),
+            ('cortendesk-managed-password@', '/bin/bash', '/opt/cortendesk', '/opt/cortendesk/password-bridge.sh'),
+            ('miminails-watch', '/usr/local/bin/miminails-watch', '/usr/local/bin', ''),
+        ]
+        for name, exe, directory, script in cases:
+            with self.subTest(name=name):
+                self.calls.clear()
+                result = native.scan(101, self.runner(service(
+                    name=name, exe=exe, directory=directory, script=script)))
+                self.assertEqual(result['candidates'], [])
+                self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_hestia_panel_is_recognized_only_from_its_exact_runtime_and_port(self):
+        hestia = '/usr/local/hestia/nginx/sbin/hestia-nginx'
+        panel = f'H\t391\t{hestia}\t/\t8083\nEND\n'
+        result = native.scan(101, self.runner(
+            services='END\n', processes='LIMIT\n', ports='END\n', hestia_panels=panel))
+        item, = result['candidates']
+        self.assertEqual(item['name'], 'Hestia')
+        self.assertEqual(item['installation_path'], hestia)
+        self.assertEqual(item['listening_ports'], [8083])
+        self.assertEqual(item['default_ports'], [8083])
+        self.assertEqual(item['scheme'], 'https')
+        self.assertIn({'kind': 'working_directory', 'value': '/'}, item['evidence'])
+        self.assertIn('metadata_limit', result['issues'])
+
+        unrelated = native.candidate(101, {
+            'name': 'other-panel', 'path': '/opt/other-panel/server', 'root': '/opt/other-panel',
+            'process': '/opt/other-panel/server', 'listening_ports': [8083],
+        }, 'FILE\n')
+        self.assertEqual(unrelated['default_ports'], [])
+        self.assertNotIn('scheme', unrelated)
+
+        self.calls.clear()
+        result = native.scan(101, self.runner(
+            services='END\n', processes='LIMIT\n', ports='END\n'))
+        self.assertEqual(result['candidates'], [])
+        self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+        with self.assertRaises(native.ProbeError):
+            native.parse_records(f'H\t391\t{hestia}\t/\t8443\nEND\n', 'H')
+
+    def test_node_and_npm_wrappers_use_only_bounded_manifest_roots(self):
+        cases = [
+            ('keynest', '/usr/bin/node', '/opt/keynest'),
+            ('rent-tool', '/usr/local/bin/npm', '/home/app/rent-tool'),
+        ]
+        for name, exe, directory in cases:
+            with self.subTest(name=name):
+                self.calls.clear()
+                result = native.scan(101, self.runner(
+                    service(name=name, exe=exe, directory=directory),
+                    footprint='FILE\n{"version":"1.2.3"}'))
+                item, = result['candidates']
+                self.assertEqual(item['name'], name)
+                self.assertEqual(item['installation_path'], directory + '/package.json')
+                self.assertEqual(item['installed_version'], '1.2.3')
+                footprint_calls = [call for call in self.calls if native.MANIFEST_FOOTPRINT_SCRIPT in call]
+                self.assertEqual(len(footprint_calls), 1)
+                if directory.startswith('/home/'):
+                    parent = '/home/app'
+                else:
+                    parent = '/opt'
+                self.assertEqual(footprint_calls[0][-3:], [directory, directory, parent])
+
+        self.calls.clear()
+        result = native.scan(101, self.runner(service(
+            name='ambiguous-home-runtime', exe='/usr/bin/node', directory='/home/app')))
+        self.assertEqual(result['candidates'], [])
+        self.assertFalse(any(native.MANIFEST_FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_node_release_layout_uses_only_server_manifest_under_shared_root(self):
+        self.calls.clear()
+        result = native.scan(101, self.runner(
+            service(name='keynest', exe='/opt/keynest/node/bin/node', directory='/opt/keynest/current'),
+            manifest='MANIFEST\tserver/package.json\n{"version":"1.5.63"}'))
+        item, = result['candidates']
+        self.assertEqual(item['name'], 'keynest')
+        self.assertEqual(item['installation_path'], '/opt/keynest/current/server/package.json')
+        self.assertEqual(item['installed_version'], '1.5.63')
+        manifest_calls = [call for call in self.calls if native.MANIFEST_FOOTPRINT_SCRIPT in call]
+        self.assertEqual(len(manifest_calls), 1)
+        self.assertEqual(manifest_calls[0][-3:],
+                         ['/opt/keynest/current', '/opt/keynest', '/opt'])
+
+    def test_node_release_layout_rejects_manifest_outside_operator_root(self):
+        self.calls.clear()
+        result = native.scan(101, self.runner(
+            service(name='outside-root', exe='/opt/outside/node/bin/node', directory='/opt/outside/current'),
+            manifest='OUTSIDE\n'))
+        self.assertEqual(result['candidates'], [])
+        self.assertIn('unsafe_manifest_root', result['issues'])
+        self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_actual_manifest_probe_accepts_release_and_rejects_symlink_escape(self):
+        parent = self.root / 'opt'
+        anchor = parent / 'keynest'
+        release = anchor / 'releases' / 'v1.5.63'
+        manifest = release / 'server' / 'package.json'
+        manifest.parent.mkdir(parents=True)
+        manifest.write_text('{"version":"1.5.63"}')
+        current = anchor / 'current'
+        current.symlink_to(release)
+        output = native.bounded_run(
+            ['sh', '-c', native.MANIFEST_FOOTPRINT_SCRIPT, 'fixture',
+             str(current), str(anchor), str(parent)], time.monotonic() + 3)
+        self.assertEqual(output, 'MANIFEST\tserver/package.json\n{"version":"1.5.63"}')
+
+        outside = self.root / 'outside' / 'package.json'
+        outside.parent.mkdir()
+        outside.write_text('{"version":"not-readable"}')
+        manifest.unlink()
+        manifest.symlink_to(outside)
+        output = native.bounded_run(
+            ['sh', '-c', native.MANIFEST_FOOTPRINT_SCRIPT, 'fixture',
+             str(current), str(anchor), str(parent)], time.monotonic() + 3)
+        self.assertEqual(output, 'OUTSIDE\n')
+        self.assertNotIn('not-readable', output)
+
+    def test_node_wrapper_with_no_fixed_manifest_is_not_an_error_or_candidate(self):
+        self.calls.clear()
+        result = native.scan(101, self.runner(
+            service(name='missing-manifest', exe='/opt/missing/node/bin/node', directory='/opt/missing/current'),
+            manifest='MISSING\n'))
+        self.assertEqual(result['status'], 'complete')
+        self.assertEqual(result['candidates'], [])
+        self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_helper_process_fallback_has_zero_footprint_probes(self):
+        self.calls.clear()
+        result = native.scan(101, self.runner(
+            services='END\n', processes='P\t123\t/opt/cortendesk/run-worker\t/opt/cortendesk\nEND\n'))
+        self.assertEqual(result['candidates'], [])
+        self.assertFalse(any(native.FOOTPRINT_SCRIPT in call for call in self.calls))
+
+    def test_missing_remote_qemu_stopped_busy_oci_guards_precede_exec(self):
+        scenarios = ['missing', 'stopped', 'busy', 'oci', 'invalid-oci', 'unmanaged']
+        for scenario in scenarios:
+            with self.subTest(scenario=scenario):
+                (self.config / '101.conf').write_text('ostype: debian\n')
+                if scenario == 'missing':
+                    (self.config / '101.conf').unlink()
+                if scenario == 'busy':
+                    (self.config / '101.conf').write_text('lock: backup\n')
+                if scenario == 'unmanaged':
+                    (self.config / '101.conf').write_text('ostype: unmanaged\n')
+                if 'oci' in scenario:
+                    (self.oci / '101').mkdir(exist_ok=True)
+                    (self.oci / '101' / 'oci-compose.json').write_text('broken')
+                self.calls.clear()
+                with self.assertRaises(native.ProbeError):
+                    native.scan(101, self.runner(status='status: stopped' if scenario == 'stopped' else 'status: running'))
+                self.assertFalse(any(call[1] == 'exec' for call in self.calls))
+                if (self.oci / '101').exists():
+                    (self.oci / '101' / 'oci-compose.json').unlink()
+                    (self.oci / '101').rmdir()
+
+    def test_failure_and_partial_output_are_not_successful_empty_results(self):
+        for text in ('', 'malformed\nEND\n', 'S\tpartial', 'LIMIT\n'):
+            result = native.scan(101, self.runner(services=text))
+            self.assertEqual(result['status'], 'partial')
+        run = self.runner()
+        def failed(argv, deadline):
+            if argv[1] == 'exec':
+                raise native.ProbeError('read_failed')
+            return run(argv, deadline)
+        self.assertEqual(native.scan(101, failed)['status'], 'partial')
+
+    def test_optional_invalid_version_metadata_keeps_proven_candidate(self):
+        result = native.scan(101, self.runner(footprint='FILE\n{broken'))
+        self.assertEqual(result['status'], 'partial')
+        item, = result['candidates']
+        self.assertIsNone(item['installed_version'])
+        self.assertIsNone(item['tracking_suggestion'])
+
+    def test_working_directory_manifest_does_not_version_an_unrelated_executable(self):
+        result = native.scan(101, self.runner(service(exe='/opt/other/server'),
+                                             footprint='FILE\n{"version":"1.2.3"}'))
+        self.assertIsNone(result['candidates'][0]['installed_version'])
+
+    def test_path_injection_and_port_only_data_do_not_become_candidates(self):
+        for text in (service(exe='/opt/app/$(id)'), service(exe='/opt/../server'),
+                     service(name='app;id'), service(exe='/bin/sh')):
+            self.assertEqual(native.scan(101, self.runner(text))['candidates'], [])
+        with self.assertRaises(native.ProbeError):
+            native.parse_records('PORT\t8080\nEND\n', 'P')
+
+    def test_same_process_deduplicates_but_two_apps_using_node_do_not_merge(self):
+        services = service(exe='/usr/bin/node', script='/opt/handmade/server.js').removesuffix('END\n')
+        services += service(name='second', exe='/usr/bin/node', script='/opt/second/server.js', directory='/opt/second')
+        result = native.scan(101, self.runner(services))
+        self.assertEqual(len(result['candidates']), 2)
+        result = native.scan(101, self.runner(processes='P\t123\t/opt/handmade/server\t/opt/handmade\nEND\n'))
+        self.assertEqual(len(result['candidates']), 1)
+        other = native.candidate(102, native.parse_records(service(), 'S')[0][0], 'FILE\n')
+        self.assertNotEqual(other['discovery_id'], result['candidates'][0]['discovery_id'])
+
+    def test_record_and_candidate_caps(self):
+        too_many = 'P\t123\t/opt/a/app\t/opt/a\n' * 257 + 'END\n'
+        with self.assertRaises(native.ProbeError):
+            native.parse_records(too_many, 'P')
+        services = ''.join(service(name=f'app{i}', exe=f'/opt/app{i}/server', directory=f'/opt/app{i}').removesuffix('END\n') for i in range(51)) + 'END\n'
+        result = native.scan(101, self.runner(services))
+        self.assertEqual(len(result['candidates']), 50)
+        self.assertIn('candidate_limit', result['issues'])
+
+    def test_cached_ttl_never_runs_scan_and_error_preserves_last_known(self):
+        complete = native.scan(101, self.runner())
+        with patch.object(native, 'scan', return_value=complete) as scan:
+            native.discover(101)
+            for age, stale in ((299.9, False), (300, True), (300.1, True)):
+                with patch.object(native.time, 'time', return_value=complete['scanned_at'] + age):
+                    self.assertEqual(native.cached(101)['stale'], stale)
+            self.assertEqual(scan.call_count, 1)
+        with patch.object(native, 'scan', side_effect=native.ProbeError('timeout')):
+            result = native.discover(101)
+        self.assertEqual(result['candidates'], complete['candidates'])
+        self.assertEqual(native.cached(101)['status'], 'error')
+        self.assertTrue(native.cached(101)['stale'])
+        partial = {**complete, 'status': 'partial', 'candidates': [], 'issues': ['read_failed']}
+        with patch.object(native, 'scan', return_value=partial):
+            self.assertEqual(native.discover(101)['candidates'], complete['candidates'])
+        with patch.object(native, 'scan', return_value={**complete, 'candidates': []}):
+            self.assertEqual(native.discover(101)['candidates'], [])
+
+    def test_lifecycle_during_first_or_cached_scan_cannot_publish_fresh_result(self):
+        complete = native.scan(101, self.runner())
+        for prior in (None, complete):
+            with self.subTest(cached=prior is not None):
+                native._cache.clear()
+                if prior:
+                    native._cache[101] = copy.deepcopy(prior)
+                entered, release = threading.Event(), threading.Event()
+                results = []
+                def delayed(vmid):
+                    entered.set()
+                    if not release.wait(3):
+                        raise AssertionError('scan release timed out')
+                    return copy.deepcopy(complete)
+                with patch.object(native, 'scan', side_effect=delayed):
+                    worker = threading.Thread(target=lambda: results.append(native.discover(101)))
+                    worker.start()
+                    try:
+                        self.assertTrue(entered.wait(3))
+                        native.mark_stale(101)
+                    finally:
+                        release.set()
+                        worker.join(3)
+                self.assertFalse(worker.is_alive())
+                self.assertEqual(len(results), 1)
+                self.assertTrue(results[0]['stale'])
+                self.assertIn('guest_changed', results[0]['issues'])
+                self.assertEqual(results[0]['candidates'], prior['candidates'] if prior else [])
+                self.assertTrue(native.cached(101)['stale'])
+                with patch.object(native, 'scan', return_value=copy.deepcopy(complete)):
+                    self.assertFalse(native.discover(101)['stale'])
+        with patch.object(native, 'scan', side_effect=lambda vmid: (
+                native.mark_stale(102), copy.deepcopy(complete))[1]):
+            self.assertFalse(native.discover(101)['stale'])
+
+    def test_scan_excludes_later_updater_and_releases_lock_after_success_or_failure(self):
+        path = self.root / 'proxmenux-lxc-update-101.lock'
+        for fail in (False, True):
+            with self.subTest(fail=fail):
+                runner = self.runner()
+                probes = []
+                def checked(argv, deadline):
+                    if argv[1] == 'exec':
+                        attempt = subprocess.run(['flock', '-n', str(path), 'true'])
+                        self.assertEqual(attempt.returncode, 1)
+                        probes.append(argv)
+                        if fail:
+                            raise native.ProbeError('timeout')
+                    return runner(argv, deadline)
+                native.scan(101, checked)
+                self.assertTrue(probes)
+                self.assertEqual(subprocess.run(['flock', '-n', str(path), 'true']).returncode, 0)
+                self.assertEqual(path.read_bytes(), b'')
+
+    def test_pending_control_reservation_blocks_scan_without_touching_it(self):
+        path = self.root / 'proxmenux-lxc-update-101.lock'
+        original = b'{"task":"pending-control"}'
+        path.write_bytes(original)
+        with self.assertRaisesRegex(native.ProbeError, 'guest_busy'):
+            native.scan(101, self.runner())
+        self.assertFalse(any(call[1] == 'exec' for call in self.calls))
+        self.assertEqual(path.read_bytes(), original)
+
+    def test_guest_stop_during_scan_never_returns_complete(self):
+        runner = self.runner()
+        statuses = iter(['status: running', 'status: stopped'])
+        def stopped(argv, deadline):
+            return next(statuses) if argv[1] == 'status' else runner(argv, deadline)
+        with self.assertRaisesRegex(native.ProbeError, 'guest_stopped'):
+            native.scan(101, stopped)
+        self.assertTrue(all(call[1] == 'exec' for call in self.calls))
+
+    def test_scan_reservation_closes_race_after_guard_and_releases_after_exception(self):
+        import lxc_check_job
+        path = self.root / 'proxmenux-lxc-update-101.lock'
+        runner = self.runner()
+        handles = []
+        def starts_update(argv, deadline):
+            if argv[1] == 'status':
+                handles.append(lxc_check_job.acquire(101))
+            return runner(argv, deadline)
+        with patch.object(lxc_check_job, 'LOCK_DIR', self.root):
+            try:
+                with self.assertRaisesRegex(native.ProbeError, 'guest_busy'):
+                    native.scan(101, starts_update)
+                self.assertFalse(any(call[1] == 'exec' for call in self.calls))
+            finally:
+                for handle in handles:
+                    handle.close()
+            def crash(argv, deadline):
+                if argv[1] == 'exec':
+                    with self.assertRaises(BlockingIOError):
+                        lxc_check_job.acquire(101)
+                    # Another CT must remain available.
+                    lxc_check_job.acquire(102).close()
+                    raise RuntimeError('unexpected probe failure')
+                return runner(argv, deadline)
+            with self.assertRaisesRegex(RuntimeError, 'unexpected'):
+                native.scan(101, crash)
+            lxc_check_job.acquire(101).close()
+        self.assertEqual(path.read_bytes(), b'')
+
+    def test_failed_scan_releases_discovery_lock_and_keeps_prior_evidence(self):
+        complete = native.scan(101, self.runner())
+        native._cache[101] = copy.deepcopy(complete)
+        with patch.object(native, 'scan', side_effect=RuntimeError('unexpected')):
+            with self.assertRaises(RuntimeError):
+                native.discover(101)
+        self.assertFalse(native._locks[101].locked())
+        self.assertEqual(native.cached(101)['candidates'], complete['candidates'])
+
+    def test_invalid_lock_inode_and_unreadable_config_fail_before_any_probe(self):
+        import os
+        path = self.root / 'proxmenux-lxc-update-101.lock'
+        target = self.root / 'unrelated'
+        target.write_bytes(b'keep')
+        for kind in ('symlink', 'fifo', 'directory'):
+            with self.subTest(kind=kind):
+                if kind == 'symlink':
+                    path.symlink_to(target)
+                elif kind == 'fifo':
+                    os.mkfifo(path)
+                else:
+                    path.mkdir()
+                try:
+                    with self.assertRaisesRegex(native.ProbeError, 'lock_unavailable'):
+                        native.scan(101, self.runner())
+                    self.assertEqual(self.calls, [])
+                    self.assertEqual(target.read_bytes(), b'keep')
+                finally:
+                    path.rmdir() if kind == 'directory' else path.unlink()
+        with patch.object(Path, 'open', side_effect=PermissionError):
+            with self.assertRaisesRegex(native.ProbeError, 'config_unavailable'):
+                native.scan(101, self.runner())
+        self.assertEqual(self.calls, [])
+
+    def test_same_ct_busy_and_different_ct_cache_isolation(self):
+        lock = threading.Lock()
+        lock.acquire()
+        native._locks[101] = lock
+        with patch.object(native, 'scan') as scanner:
+            self.assertEqual(native.discover(101)['status'], 'busy')
+            scanner.assert_not_called()
+        self.assertEqual(native.cached(102)['status'], 'not_scanned')
+
+    def test_filter_registered_curated_and_docker_children_without_mutation(self):
+        result = native.scan(101, self.runner())
+        original = copy.deepcopy(result)
+        for suggestions, apps in (({}, [{'discovery_id': result['candidates'][0]['discovery_id']}]),
+                                   ({'extras': [{'name': 'Handmade'}]}, []),
+                                   ({'docker_workloads': [{'name': 'Handmade'}]}, []),
+                                   ({'helper_slug': 'handmade', 'name_suggestion': 'Handmade'}, [])):
+            self.assertEqual(native.filter_candidates(result, suggestions, apps)['candidates'], [])
+        self.assertEqual(result, original)
+
+    def test_actual_save_first_check_and_recheck_never_probe_or_configure_updater(self):
+        with patch.object(lxc_apps, '_APPS_DIR', str(self.root / 'apps')), \
+             patch.object(lxc_apps, '_pct_exec', side_effect=AssertionError('forbidden guest execution')) as execute, \
+             patch.object(lxc_apps, 'fetch_latest_upstream_details', side_effect=AssertionError('forbidden upstream')) as upstream:
+            for version in (None, '1.2.3'):
+                item = native.candidate(101, native.parse_records(service(), 'S')[0][0],
+                                        'FILE\n' + (json.dumps({'version': version}) if version else ''))
+                payload = {'name': item['name'], 'discovery_id': item['discovery_id'],
+                           'update_method': 'none', **(item['tracking_suggestion'] or {})}
+                ok, sidecar = lxc_apps.add_app(101, payload)
+                self.assertTrue(ok, sidecar)
+                app = sidecar['apps'][-1]
+                self.assertEqual(app['state']['installed_version'], version)
+                self.assertIsNone(app['state']['update_available'])
+                self.assertEqual(app['update_method'], 'none')
+                for field in ('helper_slug', 'update_command', 'repo', 'upstream_type'):
+                    self.assertNotIn(field, app)
+                lxc_apps.check_app(101, app['id'], force=True)
+            self.assertIsNone(lxc_apps.get_schedule(101))
+            self.assertIsNone(lxc_apps.get_bulk_update(101))
+            execute.assert_not_called()
+            upstream.assert_not_called()
+
+    def test_failed_save_has_no_first_check(self):
+        with patch.object(lxc_apps, '_write_sidecar', return_value=False), \
+             patch.object(lxc_apps, '_read_sidecar', return_value=None), \
+             patch.object(lxc_apps, 'check_app') as check:
+            self.assertFalse(lxc_apps.add_app(101, {'name': 'handmade'})[0])
+            check.assert_not_called()
+
+    def test_active_update_lock_blocks_all_guest_probes_without_writing(self):
+        path = self.root / 'proxmenux-lxc-update-101.lock'
+        path.write_bytes(b'unchanged')
+        with path.open('rb') as handle:
+            fcntl.flock(handle, fcntl.LOCK_EX)
+            with self.assertRaisesRegex(native.ProbeError, 'guest_busy'):
+                native.scan(101, self.runner())
+        self.assertEqual(path.read_bytes(), b'unchanged')
+        self.assertEqual(self.calls, [])
+
+    def test_actual_service_parser_reads_data_never_executes_or_sources_it(self):
+        for manager in ('systemd', 'openrc'):
+            directory = self.root / manager
+            directory.mkdir()
+            marker = self.root / ('forbidden-' + manager)
+            unit = directory / ('handmade.service' if manager == 'systemd' else 'handmade')
+            content = ('[Service]\nExecStart=/usr/bin/node /opt/handmade/server.js --password SECRET\nWorkingDirectory=/opt/handmade\n'
+                       if manager == 'systemd' else 'command="/usr/bin/node"\ncommand_args="/opt/handmade/server.js --password SECRET"\ndirectory="/opt/handmade"\n')
+            unit.write_text(content + '\ntouch ' + str(marker) + '\n')
+            script = native.SERVICE_SCRIPT.replace('/etc/systemd/system /usr/lib/systemd/system /lib/systemd/system /etc/init.d', str(directory))
+            # Substitute only fixture paths, not the collector/parser behavior.
+            if manager == 'openrc':
+                script = script.replace('/etc/init.d:*)', str(directory) + ':*)')
+            output = native.bounded_run(['sh', '-c', script], time.monotonic() + 3)
+            record, = native.parse_records(output, 'S')[0]
+            self.assertEqual(record['path'], '/opt/handmade/server.js')
+            self.assertNotIn('SECRET', output)
+            self.assertFalse(marker.exists())
+
+    def test_merged_usr_units_are_counted_once(self):
+        usr = self.root / 'usr-units'
+        usr.mkdir()
+        (usr / 'handmade.service').write_text('[Service]\nExecStart=/opt/handmade/server\n')
+        lib = self.root / 'lib-units'
+        lib.symlink_to(usr)
+        check = '[ "$dir" = /lib/systemd/system ] && [ /lib/systemd/system -ef /usr/lib/systemd/system ]'
+        self.assertIn(check, native.SERVICE_SCRIPT)
+        script = (native.SERVICE_SCRIPT
+                  .replace('/etc/systemd/system /usr/lib/systemd/system /lib/systemd/system /etc/init.d', f'{usr} {lib}')
+                  .replace(check, f'[ "$dir" = {lib} ] && [ {lib} -ef {usr} ]'))
+        output = native.bounded_run(['sh', '-c', script], time.monotonic() + 3)
+        self.assertEqual(output.count('handmade.service'), 1)
+
+    def test_a_guest_without_ss_or_netstat_has_no_listener_evidence(self):
+        bin_dir = self.root / 'no-tools'
+        bin_dir.mkdir()
+        (bin_dir / 'awk').symlink_to('/usr/bin/awk')
+        output = native.bounded_run(['sh', '-c', 'PATH=' + str(bin_dir) + '\n' + native.PORT_SCRIPT, 'fixture', '256'],
+                                    time.monotonic() + 3)
+        self.assertEqual(native.parse_records(output, 'L'), ([], False, 0))
+
+    def test_actual_process_parser_skips_docker_and_never_reads_environ_cmdline(self):
+        proc = self.root / 'proc'
+        proc.mkdir()
+        for pid, group in ((1, '0::/\n'), (2, '0::/docker/child\n')):
+            directory = proc / str(pid)
+            directory.mkdir()
+            (directory / 'cgroup').write_text(group)
+            (directory / 'exe').symlink_to('/opt/handmade/server')
+            (directory / 'cwd').symlink_to('/opt/handmade')
+            (directory / 'cmdline').write_text('SECRET')
+            (directory / 'environ').write_text('PASSWORD')
+        script = native.PROCESS_SCRIPT.replace('/proc/[0-9]*', str(proc) + '/[0-9]*')
+        output = native.bounded_run(['sh', '-c', script, 'fixture', '256'], time.monotonic() + 3)
+        self.assertEqual(output, 'P\t1\t/opt/handmade/server\t/opt/handmade\nEND\n')
+        self.assertNotIn('"$dir/cmdline"', native.PROCESS_SCRIPT)
+        self.assertNotIn('"$dir/environ"', native.PROCESS_SCRIPT)
+        self.assertNotIn('SECRET', output)
+
+    def test_ports_require_exact_process_ownership_and_do_not_infer_http(self):
+        process = 'P\t123\t/opt/handmade/server\t/opt/handmade\nEND\n'
+        result = native.scan(101, self.runner(processes=process, ports='L\t123\t8080\nL\t999\t9000\nEND\n'))
+        item, = result['candidates']
+        self.assertEqual(item['listening_ports'], [8080])
+        self.assertEqual(item['default_ports'], [], 'TCP listener does not prove HTTP')
+        result = native.scan(101, self.runner(ports='L\t123\t8080\nEND\n'))
+        self.assertEqual(result['candidates'][0]['listening_ports'], [])
+
+    def test_shared_runtime_ports_are_assigned_only_by_working_directory(self):
+        services = service(exe='/usr/bin/node', script='/opt/handmade/server.js').removesuffix('END\n')
+        services += service(name='second', exe='/usr/bin/node', script='/opt/second/server.js', directory='/opt/second')
+        process = 'P\t123\t/usr/bin/node\t/opt/second\nEND\n'
+        result = native.scan(101, self.runner(services, processes=process, ports='L\t123\t8080\nEND\n'))
+        self.assertEqual(result['candidates'][0]['listening_ports'], [])
+        self.assertEqual(result['candidates'][1]['listening_ports'], [8080])
+
+    def test_ss_and_alpine_netstat_parser_are_bounded_and_secret_free(self):
+        for tool, line in (('ss', 'LISTEN 0 128 0.0.0.0:8080 0.0.0.0:* users:(("SECRET",pid=123,fd=7))'),
+                           ('netstat', 'tcp 0 0 0.0.0.0:8080 0.0.0.0:* LISTEN 123/SECRET')):
+            bin_dir = self.root / tool
+            bin_dir.mkdir()
+            executable = bin_dir / tool
+            executable.write_text('#!/bin/sh\nprintf \'%s\\n\' \'' + line + '\'\n')
+            executable.chmod(0o755)
+            (bin_dir / 'awk').symlink_to('/usr/bin/awk')
+            (bin_dir / 'timeout').symlink_to('/usr/bin/timeout')
+            script = 'PATH=' + str(bin_dir) + '\n' + native.PORT_SCRIPT
+            output = native.bounded_run(['sh', '-c', script, 'fixture', '256'], time.monotonic() + 3)
+            self.assertEqual(output, 'L\t123\t8080\nEND\n')
+            self.assertNotIn('SECRET', output)
+            executable.write_text('#!/bin/sh\nexit 1\n')
+            output = native.bounded_run(['sh', '-c', script, 'fixture', '256'], time.monotonic() + 3)
+            with self.assertRaises(native.ProbeError):
+                native.parse_records(output, 'L')
+
+
+class BoundedRunner(unittest.TestCase):
+    def test_output_limit_is_enforced_during_read(self):
+        with self.assertRaisesRegex(native.ProbeError, 'output_limit'):
+            native.bounded_run([sys.executable, '-c', 'import os; os.write(2,b"x"*1000000)'], time.monotonic() + 3)
+
+    def test_pipe_holding_child_does_not_bypass_timeout(self):
+        start = time.monotonic()
+        with self.assertRaisesRegex(native.ProbeError, 'timeout'):
+            native.bounded_run([sys.executable, '-c', 'import os,time; os.fork(); time.sleep(10)'], start + .15)
+        self.assertLess(time.monotonic() - start, .8)
+
+    def test_failures_never_reveal_stderr_and_deadline_blocks_spawn(self):
+        with self.assertRaisesRegex(native.ProbeError, '^read_failed$'):
+            native.bounded_run([sys.executable, '-c', 'import sys; sys.stderr.write("SECRET"); sys.exit(1)'], time.monotonic() + 3)
+        with patch.object(native.subprocess, 'Popen') as launch:
+            with self.assertRaisesRegex(native.ProbeError, 'deadline'):
+                native.bounded_run(['forbidden'], time.monotonic() - 1)
+            launch.assert_not_called()
+
+
+class Routes(unittest.TestCase):
+    """Execute actual Flask route bodies; fixture auth, no server startup jobs."""
+    def setUp(self):
+        from flask import Flask, request, jsonify, g
+        self.app = Flask('native-fixture')
+        self.authorized = True
+        def require_auth(fn):
+            @wraps(fn)
+            def wrapped(*args, **kwargs):
+                if not self.authorized:
+                    return jsonify({'error': 'auth_required'}), 401
+                return fn(*args, **kwargs)
+            return wrapped
+        self.globals = {'app': self.app, 'request': request, 'jsonify': jsonify,
+                        'require_auth': require_auth, '_vm_app_suggestions_cache': {},
+                        '_VM_APP_SUGGESTIONS_TTL': 300,
+                        '_vm_cache_get': lambda cache, vmid, ttl: cache.get(vmid),
+                        '_vm_cache_put': lambda cache, vmid, value: cache.update({vmid: value}),
+                        '_pvesh_cache': {}, '_vm_disk_cache': {}, '_vm_schedule_cache': {}, '_vm_cache_invalidate': Mock(),
+                        '_invalidate_lxc_ip': Mock(), '_schedule_started_guest_refresh': Mock(),
+                        '_publish_guest_modal_cache_revision': Mock()}
+        tree = ast.parse((SCRIPTS / 'flask_server.py').read_text())
+        routes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in (
+            'api_vm_apps_suggestions', 'api_vm_apps_native_discovery', 'api_vm_apps_add',
+            '_handle_guest_lifecycle')]
+        self.assertEqual(len(routes), 4)
+        exec(compile(ast.Module(body=routes, type_ignores=[]), str(SCRIPTS / 'flask_server.py'), 'exec'), self.globals)
+        self.client = self.app.test_client()
+
+    def test_explicit_post_only_get_and_existing_post_never_native_scan(self):
+        result = {'vmid': 101, 'status': 'complete', 'candidates': [], 'stale': False}
+        with patch.object(native, 'discover', return_value=result) as scan, \
+             patch.object(native, 'cached', return_value=result), \
+             patch.object(lxc_apps, 'get_suggestions', return_value={'extras': []}) as suggestions, \
+             patch.object(lxc_apps, 'load_sidecar', return_value={'apps': []}), \
+             patch.object(lxc_apps, '_write_sidecar') as write:
+            self.assertEqual(self.client.get('/api/vms/101/apps/native-discovery').status_code, 200)
+            self.client.get('/api/vms/101/apps/suggestions')
+            # OCI refresh and Find's existing catalog path retain their purpose.
+            self.client.post('/api/vms/101/apps/suggestions')
+            with self.app.test_request_context('/api/vms/101/apps/suggestions'):
+                from flask import g
+                g._internal_call = True
+                self.globals['_vm_app_suggestions_cache'].clear()
+                self.globals['api_vm_apps_suggestions'](101)
+            scan.assert_not_called()
+            self.client.post('/api/vms/101/apps/native-discovery')
+            scan.assert_called_once_with(101)
+            self.assertEqual(suggestions.call_count, 2)
+            write.assert_not_called()
+
+    def test_unauthorized_and_wrong_method_have_zero_scans(self):
+        with patch.object(native, 'discover') as scan, patch.object(lxc_apps, 'load_sidecar') as read:
+            self.authorized = False
+            self.assertEqual(self.client.post('/api/vms/101/apps/native-discovery').status_code, 401)
+            self.assertEqual(self.client.put('/api/vms/101/apps/native-discovery').status_code, 405)
+            scan.assert_not_called()
+            read.assert_not_called()
+
+    def test_actual_lifecycle_marks_last_known_stale_without_scanning(self):
+        native._cache[101] = {'vmid': 101, 'status': 'complete', 'stale': False,
+                              'candidates': [], 'scanned_at': time.time()}
+        with patch.object(native, 'discover') as 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'])
+            scan.assert_not_called()
+            self.assertEqual(clear.call_count, 2)
+
+    def test_actual_repeated_prewarmer_has_zero_native_scans(self):
+        tree = ast.parse((SCRIPTS / 'flask_server.py').read_text())
+        warmer = next(node for node in tree.body if isinstance(node, ast.FunctionDef)
+                      and node.name == '_vm_modal_prewarmer_pass')
+        environment = dict(self.globals)
+        for name in ('_vm_details_cache', '_vm_backups_cache', '_vm_schedule_cache', '_vm_mounts_cache'):
+            environment[name] = {}
+        for name in ('_VM_DETAILS_TTL', '_VM_BACKUPS_TTL', '_VM_SCHEDULE_TTL', '_VM_MOUNTS_TTL'):
+            environment[name] = 300
+        for name in ('get_vm_config', 'api_vm_backups', 'api_vm_apps_schedule', 'api_lxc_mount_points'):
+            environment[name] = lambda vmid: None
+        environment.update({'get_cached_pvesh_cluster_resources_vm': lambda: [
+            {'vmid': 101, 'type': 'lxc', 'status': 'running'}], 'time': Mock(), 'sys': sys})
+        exec(compile(ast.Module(body=[warmer], type_ignores=[]), 'flask_server.py', 'exec'), environment)
+        with patch.object(native, 'discover') as scan, \
+             patch.object(lxc_apps, 'ensure_oci_registration'), \
+             patch.object(lxc_apps, 'load_sidecar', return_value={'apps': []}), \
+             patch.object(lxc_apps, 'get_suggestions', return_value={'extras': []}):
+            for _ in range(2):
+                environment['_vm_app_suggestions_cache'].clear()
+                self.assertEqual(environment['_vm_modal_prewarmer_pass'](), 1)
+            scan.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 5e76a028..6e645e4d 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -876,8 +876,8 @@ and traceback before changing catalogs or translation-provider settings.
 #### Qualified offline `tests/` CI lanes
 
 `.github/workflows/test-offline-qualified.yml` runs **only** the explicitly listed
-nine Python files (including the inert active-notification payload fixture and
-the focused health recurrence tests) and eleven
+ten Python files (including the native discovery fixtures, the inert active-notification payload fixture and
+the focused health recurrence tests) and twelve
 Node files in `.github/scripts/run_offline_qualified.py`.
 The Python runner gives each test file a private temporary directory through
 `TMPDIR`. It removes that directory after the test file finishes.
@@ -900,7 +900,7 @@ dangling symlink) rather than overwrite or reuse it. In Bash:
 )
 ```
 
-Requires Python 3.11, Node 22.14, npm and Bash/coreutils for the bounded shell
+Requires Python 3.11, Flask 3.x for the native route fixtures, Node 22.14, npm and Bash/coreutils for the bounded shell
 fixtures. The separate lockfile installs only TypeScript, React, React DOM,
 Lucide icons and the locked scheduler dependency; it does not build the Monitor
 or run npm lifecycle scripts. `--legacy-peer-deps` admits the existing Lucide
diff --git a/tests/lxc_updates/test_native_app_discovery_ui.cjs b/tests/lxc_updates/test_native_app_discovery_ui.cjs
new file mode 100644
index 00000000..2ffa802e
--- /dev/null
+++ b/tests/lxc_updates/test_native_app_discovery_ui.cjs
@@ -0,0 +1,199 @@
+// Execute the real panel, handlers and editor with inert hook/UI/API fixtures.
+// This is component interaction evidence, not browser or Proxmox acceptance.
+const assert = require('node:assert/strict')
+const fs = require('node:fs')
+const path = require('node:path')
+const vm = require('node:vm')
+const root = path.resolve(__dirname, '../../AppImage')
+const ts = require(path.join(root, 'node_modules/typescript'))
+const source = fs.readFileSync(path.join(root, 'components/lxc-app-panel.tsx'), 'utf8')
+const compiled = ts.transpileModule(source, {compilerOptions: {
+  module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020, jsx: ts.JsxEmit.ReactJSX,
+}, reportDiagnostics: true})
+assert.equal(compiled.diagnostics.length, 0)
+
+const flatten = node => !node || typeof node !== 'object' ? [] : [node, ...(
+  [node.props?.children].flat(Infinity).flatMap(flatten)
+)]
+const text = node => node == null || node === false ? '' : typeof node !== 'object'
+  ? String(node) : [node.props?.children].flat(Infinity).map(text).join(' ')
+const candidate = {
+  slug: 'native-' + 'a'.repeat(24), discovery_id: 'native-' + 'a'.repeat(24),
+  name: 'handmade', source: 'native_metadata', installed_version: null,
+  evidence: [{kind: 'installation_file', value: '/opt/handmade/server'}], default_ports: [],
+}
+
+function fixture({nativeCandidate = candidate} = {}) {
+  let cursor = 0, tree, props, effects = [], states = [], refs = [], effectStates = []
+  const calls = [], cachedWrites = []
+  let native = {vmid: 101, status: 'not_scanned', candidates: [], stale: false}
+  let holdScan, scanGate, holdSave, saveGate, scanStatus = 'complete'
+  let cacheGate, resolveCache, cachePending = false
+  const seed = {sidecar: {vmid: 101, apps: []}, suggestions: {extras: [], port_suggestions: []}}
+  const react = {
+    useState(initial) {
+      const index = cursor++
+      if (!(index in states)) states[index] = typeof initial === 'function' ? initial() : initial
+      return [states[index], value => {states[index] = typeof value === 'function' ? value(states[index]) : value}]
+    },
+    useRef(value) {const index = cursor++; return refs[index] ||= {current: value}},
+    useMemo(fn) {cursor++; return fn()}, useCallback(fn) {cursor++; return fn},
+    useEffect(fn, deps) {
+      const index = cursor++, prior = effectStates[index]
+      if (!prior || !deps || deps.some((value, i) => value !== prior.deps?.[i])) {
+        effects.push(() => {prior?.cleanup?.(); effectStates[index] = {deps, cleanup: fn()}})
+      }
+    },
+  }
+  const api = async (url, options = {}) => {
+    calls.push({url, ...options})
+    if (url.endsWith('/native-discovery')) {
+      if (options.method !== 'POST' && cachePending) {
+        cachePending = false
+        const oldSnapshot = structuredClone(native)
+        await cacheGate
+        return oldSnapshot
+      }
+      if (options.method === 'POST') {
+        if (scanGate) await scanGate
+        native = {...native, status: scanStatus, candidates: [nativeCandidate]}
+      }
+      return structuredClone(native)
+    }
+    if (url.endsWith('/apps/suggestions')) return seed.suggestions
+    if (url.endsWith('/apps') && options.method === 'POST') {
+      if (saveGate) await saveGate
+      return {vmid: Number(url.split('/')[3]), apps: [{id: 'saved', ...JSON.parse(options.body), state: {}}]}
+    }
+    if (url.endsWith('/adguard-setup')) return {available: false}
+    return []
+  }
+  const cache = {
+    getLxcAppsCached: () => null, fetchLxcApps: async () => seed,
+    subscribeLxcApps: () => () => {}, setLxcAppsCached: (...args) => cachedWrites.push(args),
+  }
+  const exports = {}
+  vm.runInNewContext(compiled.outputText, {exports, console, setTimeout, clearTimeout,
+    require: name => {
+      if (name === 'react') return react
+      if (name === 'react/jsx-runtime') return {jsx: (type, props) => ({type, props}), jsxs: (type, props) => ({type, props})}
+      if (name.includes('api-config')) return {fetchApi: api}
+      if (name.includes('lxc-apps-cache')) return cache
+      if (name.includes('i18n/provider')) return {useT: () => (key, values) => key + (values ? JSON.stringify(values) : '')}
+      if (name.includes('category-color')) return {useIsLightTheme: () => false, categoryChipStyle: () => ({})}
+      if (name.includes('category-label')) return {getCategoryLabel: value => value}
+      return new Proxy({}, {get: (_, key) => key})
+    },
+  })
+  function render(next = props || {vmid: 101, initialData: seed}) {
+    props = next; cursor = 0; effects = []
+    tree = exports.LxcAppPanel(props)
+    for (const effect of effects) effect()
+    return tree
+  }
+  const settle = async () => {for (let i = 0; i < 6; i++) await Promise.resolve(); render()}
+  const button = key => {
+    const match = flatten(tree).find(n => n.type === 'Button' && text(n).includes('vmLxc.appEditor.' + key))
+    assert.ok(match, 'Missing button ' + key)
+    return match
+  }
+  return {render, settle, calls, cachedWrites, button, tree: () => tree,
+    setNative: value => {native = value},
+    holdInitialCache: () => {cachePending = true; cacheGate = new Promise(resolve => {resolveCache = resolve})},
+    resolveInitialCache: () => resolveCache(),
+    setScanStatus: value => {scanStatus = value},
+    holdScan: () => {scanGate = new Promise(resolve => {holdScan = resolve})}, resolveScan: () => holdScan(),
+    holdSave: () => {saveGate = new Promise(resolve => {holdSave = resolve})}, resolveSave: () => holdSave(),
+  }
+}
+
+async function main() {
+  const f = fixture()
+  f.render(); await f.settle()
+  assert.equal(f.calls.filter(c => c.method === 'POST').length, 0, 'mount and cache reads never scan')
+  await f.button('searchApplications').props.onClick(); f.render()
+  assert.equal(f.calls.filter(c => c.url.endsWith('/native-discovery') && c.method === 'POST').length, 1)
+  assert.ok(text(f.tree()).includes('nativeVersionUnknown'), 'unknown version remains visible')
+  assert.ok(text(f.tree()).includes('/opt/handmade/server'))
+  await f.button('registerButton').props.onClick(); f.render()
+  assert.ok(text(f.tree()).includes('saveButton'), 'candidate opens actual editor')
+  f.button('cancelButton').props.onClick(); f.render()
+  assert.equal(f.calls.filter(c => /\/apps$/.test(c.url) && c.method === 'POST').length, 0, 'cancel never registers')
+  await f.button('registerButton').props.onClick(); f.render()
+  await f.button('saveButton').props.onClick(); f.render()
+  const saved = f.calls.find(c => /\/apps$/.test(c.url) && c.method === 'POST')
+  const payload = JSON.parse(saved.body)
+  assert.equal(saved.url, '/api/vms/101/apps')
+  assert.equal(payload.discovery_id, candidate.discovery_id)
+  assert.ok(!payload.helper_slug, 'native identity never becomes a helper')
+  assert.equal(payload.installed_via, '')
+  assert.equal(payload.update_method, 'none')
+  assert.equal(payload.update_command, '')
+  assert.equal(payload.repo, '')
+  assert.equal(payload.upstream_type, '')
+
+  const hestia = fixture({nativeCandidate: {
+    ...candidate, name: 'Hestia', default_ports: [8083], scheme: 'https',
+  }})
+  hestia.render(); await hestia.settle()
+  await hestia.button('searchApplications').props.onClick(); hestia.render()
+  await hestia.button('registerButton').props.onClick(); hestia.render()
+  await hestia.button('saveButton').props.onClick(); hestia.render()
+  const hestiaPayload = JSON.parse(hestia.calls.find(c => /\/apps$/.test(c.url) && c.method === 'POST').body)
+  assert.deepEqual(hestiaPayload.ports, [{port: 8083, scheme: 'https', web_path: ''}],
+    'a verified Hestia panel pre-fills only its fixed HTTPS endpoint')
+  const delayedCache = fixture()
+  delayedCache.holdInitialCache()
+  delayedCache.render(); await delayedCache.settle()
+  await delayedCache.button('searchApplications').props.onClick(); delayedCache.render()
+  assert.ok(text(delayedCache.tree()).includes('/opt/handmade/server'))
+  delayedCache.resolveInitialCache(); await delayedCache.settle()
+  assert.ok(text(delayedCache.tree()).includes('/opt/handmade/server'),
+    'a delayed initial cache read must not erase a completed explicit search')
+
+  const partial = fixture()
+  partial.render(); await partial.settle(); partial.setScanStatus('partial')
+  await partial.button('searchApplications').props.onClick(); partial.render()
+  assert.ok(text(partial.tree()).includes('nativeScanPartial'))
+  assert.ok(text(partial.tree()).includes('nativeEvidenceStale'))
+  const failed = fixture()
+  failed.render(); await failed.settle(); failed.setScanStatus('error')
+  await failed.button('searchApplications').props.onClick(); failed.render()
+  assert.ok(text(failed.tree()).includes('nativeScanFailed'))
+  assert.equal(failed.calls.filter(c => c.url.endsWith('/apps/suggestions') && c.method === 'POST').length, 0)
+
+  const editorSwitch = fixture()
+  editorSwitch.render(); await editorSwitch.settle()
+  await editorSwitch.button('searchApplications').props.onClick(); editorSwitch.render()
+  await editorSwitch.button('registerButton').props.onClick(); editorSwitch.render()
+  editorSwitch.render({vmid: 102})
+  assert.ok(!text(editorSwitch.tree()).includes('saveButton'), 'old editor must disappear in the very first render of another CT')
+
+  const late = fixture()
+  late.render(); await late.settle(); late.holdScan()
+  const scan = late.button('searchApplications').props.onClick()
+  late.render({vmid: 102}); await late.settle()
+  late.resolveScan(); await scan; late.render()
+  assert.ok(!text(late.tree()).includes('/opt/handmade/server'), 'late A scan cannot appear in B')
+  assert.equal(late.cachedWrites.length, 0)
+  assert.equal(late.calls.filter(c => c.url.endsWith('/apps/suggestions') && c.method === 'POST').length, 0)
+
+  const saveLate = fixture()
+  saveLate.render(); await saveLate.settle()
+  await saveLate.button('searchApplications').props.onClick(); saveLate.render()
+  await saveLate.button('registerButton').props.onClick(); saveLate.render()
+  saveLate.holdSave(); const saving = saveLate.button('saveButton').props.onClick()
+  const priorWrites = saveLate.cachedWrites.length
+  saveLate.render({vmid: 102}); await saveLate.settle()
+  saveLate.resolveSave(); await saving; saveLate.render()
+  assert.equal(saveLate.cachedWrites.length, priorWrites, 'late save response does not update new target')
+  assert.equal(saveLate.calls.filter(c => /\/vms\/102\/apps$/.test(c.url) && c.method === 'POST').length, 0)
+
+  const oci = fixture()
+  oci.render({vmid: 101, oci: {instance: true}, initialData: {sidecar: {apps: []}, suggestions: {oci_instance: {template_id: 'image-app'}, extras: []}}})
+  await oci.settle()
+  await oci.button('refreshData').props.onClick()
+  assert.equal(oci.calls.filter(c => c.url.endsWith('/native-discovery') && c.method === 'POST').length, 0, 'OCI refresh never native-scans')
+  console.log('PASS native discovery: real panel Find → editor → Cancel/Save, no automatic scan, safe payload, CT switching and OCI refresh (fixtures)')
+}
+main().catch(error => {console.error(error); process.exitCode = 1})