Merge pull request #449 from Vaso73/fix/vm-control-task-completion

fix: wait for VM power tasks before completing controls

.github/scripts/tests/test_command_descriptions.py

@@ -220,9 +220,14 @@ class CommandDescriptionsTests(unittest.TestCase):
                                 "installedVersionUnverified", "versionStatusUnknown"):
                         messages["vmLxc"]["updates"].setdefault(
                             key, english["vmLxc"]["updates"][key])
+                    for key in ("startPending", "shutdownPending", "rebootPending", "stopPending"):
+                        messages["vmLxc"]["actions"].setdefault(
+                            key, english["vmLxc"]["actions"][key])
                     for key in ("osDockerScope", "dockerImageMethod", "unverifiedTargetsHelp"):
                         messages["vmLxc"]["bulkUpdate"].setdefault(
                             key, english["vmLxc"]["bulkUpdate"][key])
+                    messages["vmLxc"]["errors"].setdefault(
+                        "controlNotConfirmed", english["vmLxc"]["errors"]["controlNotConfirmed"])
                     path.write_text(json.dumps(messages, ensure_ascii=False))
             before = {p: p.read_bytes() for p in root.glob("*/common.json")}
             argv = [str(SCRIPT), "--source", str(root / "en/common.json"),

AppImage/components/virtual-machines.tsx

@@ -28,6 +28,7 @@ import { AppUpdaterEditor, type AppUpdateMethod } from "./app-updater-editor"
 import { formatStorage } from "../lib/utils"
 import { formatNetworkTraffic, getNetworkUnit } from "../lib/format-network"
 import { fetchApi } from "../lib/api-config"
+import { waitForVmControlTask } from "../lib/vm-control"
 import { matchesVmSearch, normalizeVmSearchValue } from "../lib/vm-search"
 import { refreshCheckLxcUpdates, getLxcOsUpdateState, isLxcUpdateStale } from "../lib/lxc-update-observation"
 import DOMPurify from "dompurify"
@@ -872,6 +873,8 @@ export function VirtualMachines() {
   const lifecycleCacheRevisionsRef = useRef(new Map<number, number>())
   const [dockerInventoryRefreshingVmid, setDockerInventoryRefreshingVmid] = useState<number | null>(null)
   const [controlLoading, setControlLoading] = useState(false)
+  const controlRequestRef = useRef(false)
+  const [controlAction, setControlAction] = useState<{ vmid: number; action: string } | null>(null)
   // Destructive control confirmation. `Force Stop` and `Reboot` skip the OS
   // shutdown sequence and can corrupt running guests; gate them behind a
   // typed-VMID match prompt to prevent misclicks. See audit Tier 2 #17.
@@ -1797,12 +1800,16 @@ export function VirtualMachines() {
   }
 
   const handleVMControl = async (vmid: number, action: string) => {
+    if (controlRequestRef.current) return
+    controlRequestRef.current = true
+    setControlAction({ vmid, action })
     setControlLoading(true)
     try {
-      await fetchApi(`/api/vms/${vmid}/control`, {
+      const accepted = await fetchApi<{ task: string }>(`/api/vms/${vmid}/control`, {
         method: "POST",
         body: JSON.stringify({ action }),
       })
+      const completed = await waitForVmControlTask(vmid, action, accepted.task)
 
       // Any control action can change what the guest reports: a stop
       // hides runtime-only fields, a start may bring a new config that
@@ -1818,16 +1825,37 @@ export function VirtualMachines() {
       cache.firewall.delete(vmid)
       invalidateLxcApps(vmid)
 
-      mutate()
+      // Publish the confirmed result as an SWR mutation before revalidating.
-      setSelectedVM(null)
+      // SWR then discards list requests started before this completion.
-      setVMDetails(null)
+      try {
+        await mutate(current => current?.map(vm => vm.vmid === vmid
+          ? { ...vm, status: completed.guest_status as VMData["status"] } : vm), { revalidate: false })
+        await mutate()
+      } catch { /* Polling will retry; control already succeeded. */ }
+      // Keep the user's tab/scroll and last complete detail while revalidating.
+      // Never resurrect a modal the user closed, or overwrite another guest.
+      if (selectedVMRef.current?.vmid === vmid) {
+        setSelectedVM(current => current?.vmid === vmid
+          ? { ...current, status: completed.guest_status as VMData["status"] } : current)
+        try {
+          const details = await fetchApi<VMDetails>(`/api/vms/${vmid}`)
+          cache.details.set(vmid, details)
+          if (selectedVMRef.current?.vmid === vmid) setVMDetails(details)
+        } catch {
+          // The control succeeded; detail polling will retry independently.
+        }
+      }
     } catch (error) {
       console.error(`Failed to ${action} VM ${vmid}:`, error)
       // Same UX issue as handleCreateBackup: a silent console.error left the
       // user looking at a "Stop"/"Start" button that just never reacted.
-      const msg = error instanceof Error ? error.message : t("vmLxc.errors.unknown")
+      const msg = error instanceof Error
+        ? (error.message === "controlNotConfirmed" ? t("vmLxc.errors.controlNotConfirmed") : error.message)
+        : t("vmLxc.errors.unknown")
       alert(t("vmLxc.errors.controlFailed", { action, vmid, message: msg }))
     } finally {
+      controlRequestRef.current = false
+      setControlAction(null)
       setControlLoading(false)
     }
   }
@@ -7623,8 +7651,11 @@ const handleDownloadLogs = async (vmid: number, vmName: string) => {
                     tab we added. */}
                 {(() => {
                   const hasTerminal = selectedVM?.type === "lxc" && selectedVM?.status === "running"
+                  const pendingAction = controlAction?.vmid === selectedVM?.vmid ? controlAction?.action : null
+                  const controlLabel = (action: string, key: string) => pendingAction === action
+                    ? t(`vmLxc.actions.${action}Pending`) : t(key)
                   return (
-                    <div className={`grid gap-2 sm:gap-3 ${hasTerminal ? "grid-cols-4 lg:grid-cols-5" : "grid-cols-4"}`}>
+                    <div aria-busy={controlLoading} className={`grid gap-2 sm:gap-3 ${hasTerminal ? "grid-cols-4 lg:grid-cols-5" : "grid-cols-4"}`}>
                       {hasTerminal && (
                         <Button
                           className="w-full bg-zinc-600/20 border border-zinc-600/50 text-zinc-300 hover:bg-zinc-600/30 col-span-4 lg:col-span-1"
@@ -7638,19 +7669,19 @@ const handleDownloadLogs = async (vmid: number, vmName: string) => {
                         className="w-full bg-green-600/20 border border-green-600/50 text-green-400 hover:bg-green-600/30"
                         disabled={selectedVM?.status === "running" || controlLoading}
                         onClick={() => selectedVM && handleVMControl(selectedVM.vmid, "start")}
-                        title={t("vmLxc.actions.start")}
+                        title={controlLabel("start", "vmLxc.actions.start")}
                       >
-                        <Play className="h-4 w-4 sm:mr-2" />
+                        {pendingAction === "start" ? <Loader2 className="h-4 w-4 sm:mr-2 animate-spin" /> : <Play className="h-4 w-4 sm:mr-2" />}
-                        <span className="hidden sm:inline">{t("vmLxc.actions.start")}</span>
+                        <span className="hidden sm:inline">{controlLabel("start", "vmLxc.actions.start")}</span>
                       </Button>
                       <Button
                         className="w-full bg-blue-600/20 border border-blue-600/50 text-blue-400 hover:bg-blue-600/30"
                         disabled={selectedVM?.status !== "running" || controlLoading}
                         onClick={() => selectedVM && handleVMControl(selectedVM.vmid, "shutdown")}
-                        title={t("vmLxc.actions.shutdown")}
+                        title={controlLabel("shutdown", "vmLxc.actions.shutdown")}
                       >
-                        <Power className="h-4 w-4 sm:mr-2" />
+                        {pendingAction === "shutdown" ? <Loader2 className="h-4 w-4 sm:mr-2 animate-spin" /> : <Power className="h-4 w-4 sm:mr-2" />}
-                        <span className="hidden sm:inline">{t("vmLxc.actions.shutdown")}</span>
+                        <span className="hidden sm:inline">{controlLabel("shutdown", "vmLxc.actions.shutdown")}</span>
                       </Button>
                       <Button
                         className="w-full bg-blue-600/20 border border-blue-600/50 text-blue-400 hover:bg-blue-600/30"
@@ -7660,10 +7691,10 @@ const handleDownloadLogs = async (vmid: number, vmName: string) => {
                           vmid: selectedVM.vmid,
                           vmName: selectedVM.name,
                         })}
-                        title={t("vmLxc.actions.reboot")}
+                        title={controlLabel("reboot", "vmLxc.actions.reboot")}
                       >
-                        <RotateCcw className="h-4 w-4 sm:mr-2" />
+                        {pendingAction === "reboot" ? <Loader2 className="h-4 w-4 sm:mr-2 animate-spin" /> : <RotateCcw className="h-4 w-4 sm:mr-2" />}
-                        <span className="hidden sm:inline">{t("vmLxc.actions.reboot")}</span>
+                        <span className="hidden sm:inline">{controlLabel("reboot", "vmLxc.actions.reboot")}</span>
                       </Button>
                       <Button
                         className="w-full bg-red-600/20 border border-red-600/50 text-red-400 hover:bg-red-600/30"
@@ -7673,10 +7704,10 @@ const handleDownloadLogs = async (vmid: number, vmName: string) => {
                           vmid: selectedVM.vmid,
                           vmName: selectedVM.name,
                         })}
-                        title={t("vmLxc.actions.forceStop")}
+                        title={controlLabel("stop", "vmLxc.actions.forceStop")}
                       >
-                        <StopCircle className="h-4 w-4 sm:mr-2" />
+                        {pendingAction === "stop" ? <Loader2 className="h-4 w-4 sm:mr-2 animate-spin" /> : <StopCircle className="h-4 w-4 sm:mr-2" />}
-                        <span className="hidden sm:inline">{t("vmLxc.actions.forceStop")}</span>
+                        <span className="hidden sm:inline">{controlLabel("stop", "vmLxc.actions.forceStop")}</span>
                       </Button>
                     </div>
                   )

AppImage/lib/vm-control.ts

@@ -0,0 +1,32 @@
+import { fetchApi } from './api-config'
+
+// A successful control POST accepts an asynchronous Proxmox task. In
+// particular, an unchanged "running" status cannot confirm a reboot.
+export async function waitForVmControlTask(vmid: number, action: string, task: string) {
+  if (!task) throw new Error('controlNotConfirmed')
+  const expected = action === 'shutdown' || action === 'stop' ? 'stopped' : 'running'
+  const deadline = Date.now() + 180_000
+  let readFailures = 0
+  while (Date.now() < deadline) {
+    let result: { status?: string; exitstatus?: string; guest_status?: string }
+    try {
+      result = await fetchApi(`/api/vms/${vmid}/control-task/${encodeURIComponent(task)}`, {
+        signal: AbortSignal.timeout(10_000),
+      })
+      readFailures = 0
+    } catch (error) {
+      if (++readFailures >= 3) throw new Error('controlNotConfirmed')
+      await new Promise(resolve => setTimeout(resolve, 1000))
+      continue
+    }
+    if (result?.status === 'stopped') {
+      if (!result.exitstatus) throw new Error('controlNotConfirmed')
+      if (result.exitstatus !== 'OK') throw new Error(result.exitstatus)
+      if (result.guest_status === expected) return result
+    } else if (result?.status !== 'running') {
+      throw new Error('controlNotConfirmed')
+    }
+    await new Promise(resolve => setTimeout(resolve, 1000))
+  }
+  throw new Error('controlNotConfirmed')
+}

AppImage/messages/en/common.json

@@ -1100,7 +1100,11 @@
       "shutdown": "Shutdown",
       "reboot": "Reboot",
       "forceStop": "Force Stop",
-      "working": "Working..."
+      "working": "Working...",
+      "startPending": "Starting...",
+      "shutdownPending": "Shutting down...",
+      "rebootPending": "Restarting...",
+      "stopPending": "Stopping..."
     },
     "confirm": {
       "forceStopTitle": "Force Stop",
@@ -1144,6 +1148,7 @@
       "unknown": "Unknown error",
       "backupStartFailed": "Failed to start backup: {message}",
       "controlFailed": "Failed to {action} VM {vmid}: {message}",
+      "controlNotConfirmed": "The operation has not been confirmed complete. Check the guest status before trying again.",
       "saveNotesFailed": "Error saving notes. Please try again.",
       "appNotFound": "This application is no longer available. Refresh the page and try again.",
       "saveCustomCommandFailed": "Could not save the custom update command: {message}",

AppImage/messages/sk/common.json

@@ -1100,7 +1100,11 @@
       "shutdown": "Vypnúť",
       "reboot": "Reštartovať",
       "forceStop": "Vynútiť vypnutie",
-      "working": "Pracujem..."
+      "working": "Pracujem...",
+      "startPending": "Spúšťa sa...",
+      "shutdownPending": "Vypína sa...",
+      "rebootPending": "Reštartuje sa...",
+      "stopPending": "Vypína sa..."
     },
     "confirm": {
       "forceStopTitle": "Vynútiť vypnutie",
@@ -1144,6 +1148,7 @@
       "unknown": "Neznáma chyba",
       "backupStartFailed": "Zálohu sa nepodarilo spustiť: {message}",
       "controlFailed": "Akcia {action} pre VM {vmid} zlyhala: {message}",
+      "controlNotConfirmed": "Dokončenie operácie sa zatiaľ nepodarilo potvrdiť. Pred ďalším pokusom skontrolujte stav VM alebo LXC.",
       "saveNotesFailed": "Poznámky sa nepodarilo uložiť. Skúste to znova.",
       "appNotFound": "Táto aplikácia už nie je dostupná. Obnovte stránku a skúste to znova.",
       "saveCustomCommandFailed": "Vlastný príkaz na aktualizáciu sa nepodarilo uložiť: {message}",

AppImage/scripts/flask_server.py

@@ -6847,10 +6847,15 @@ def get_proxmox_vms():
                         continue
 
                     vm_type = 'lxc' if resource.get('type') == 'lxc' else 'qemu'
+                    # The PVE summary can lag behind a completed LXC task.
+                    # Use the same direct local cgroup check as the detail
+                    # route; never resurrect a stopped CT from that summary.
+                    vm_status = (_fast_guest_status(resource.get('vmid'), vm_type)
+                                 if vm_type == 'lxc' else resource.get('status', 'unknown'))
                     vm_data = {
                         'vmid': resource.get('vmid'),
                         'name': resource.get('name', f"VM-{resource.get('vmid')}"),
-                        'status': resource.get('status', 'unknown'),
+                        'status': vm_status,
                         'type': vm_type,
                         'cpu': resource.get('cpu', 0),
                         'mem': resource.get('mem', 0),
@@ -6905,7 +6910,7 @@ def get_proxmox_vms():
                         # detail-aware clients.  The cache is still warmed
                         # at startup and invalidated by LXC lifecycle events.
                         vm_data['ips'] = []
-                        if resource.get('status') == 'running':
+                        if vm_data['status'] == 'running':
                             lxc_ip_info = _get_lxc_ip_info_cached(resource.get('vmid'))
                             lxc_ips = _normalise_lxc_ips(lxc_ip_info)
                             vm_data['ips'] = lxc_ips
@@ -15870,6 +15875,8 @@ def api_vm_config_set(vmid):
 @require_auth
 def api_vm_control(vmid):
     """Control VM/LXC (start, stop, shutdown, reboot)"""
+    operation_lock = None
+    task_identity_confirmed = False
     try:
         data = request.get_json()
         action = data.get('action')  # start, stop, shutdown, reboot
@@ -15892,6 +15899,13 @@ def api_vm_control(vmid):
             
             vm_type = 'lxc' if vm_info.get('type') == 'lxc' else 'qemu'
             node = vm_info.get('node', 'pve')
+            if vm_type == 'lxc':
+                import lxc_check_job
+                try:
+                    operation_lock = lxc_check_job.acquire(vmid)
+                except BlockingIOError:
+                    return jsonify({'error': 'Another container operation is running'}), 409
+                lxc_check_job.begin_control(operation_lock, node)
             
             # Execute action
             control_result = subprocess.run(
@@ -15899,6 +15913,37 @@ def api_vm_control(vmid):
                 capture_output=True, text=True, timeout=30)
 
             if control_result.returncode == 0:
+                import re
+                # pvesh can print progress lines before the task identity.
+                # The final non-empty line is the only task associated with
+                # this request; accepting a different log line would make the
+                # frontend poll an unrelated guest operation.
+                lines = [line.strip().strip('"') for line in
+                         (control_result.stdout or '').splitlines() if line.strip()]
+                task = lines[-1] if lines else ''
+                task_types = [f"{'vz' if vm_type == 'lxc' else 'qm'}{action}"]
+                if action == 'start':
+                    task_types.append('hastart')
+                elif action in ('stop', 'shutdown'):
+                    # Proxmox 9 routes HA shutdown through hastop for both
+                    # LXC and QEMU guests.
+                    task_types.append('hastop')
+                task_pattern = (r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:'
+                                r'[0-9A-Fa-f]+:(?:' + '|'.join(task_types) +
+                                rf'):{vmid}:[\w.@!-]+:')
+                if not re.fullmatch(task_pattern, task):
+                    # A successful process exit without a verifiable task
+                    # cannot be polled.  Do not leave a task-less control
+                    # marker that would block this CT forever.
+                    if operation_lock is not None:
+                        lxc_check_job.cancel_control(operation_lock)
+                    return jsonify({
+                        'success': False,
+                        'vmid': vmid,
+                        'action': action,
+                        'error': 'Control task was not confirmed by Proxmox',
+                    }), 502
+                task_identity_confirmed = True
                 # Invalidate VM resources cache so the next /api/vms call
                 # returns fresh status instead of the pre-action snapshot.
                 _pvesh_cache['cluster_resources_vm_time'] = 0
@@ -15906,13 +15951,21 @@ def api_vm_control(vmid):
                 # (lxc_ip_info, os_info) — a start/stop flips those, so
                 # drop the modal caches for this vmid.
                 _vm_cache_invalidate(vmid)
+                # The frontend waits for this exact task rather than assuming
+                # that accepting the request means the action completed.
+                if operation_lock is not None:
+                    lxc_check_job.retain_control(vmid, task, operation_lock)
+                    operation_lock = None
                 return jsonify({
                     'success': True,
                     'vmid': vmid,
                     'action': action,
+                    'task': task,
                     'message': f'Successfully executed {action} on {vm_info.get("name")}'
                 })
             else:
+                if operation_lock is not None:
+                    lxc_check_job.cancel_control(operation_lock)
                 # `pvesh` failed → fire the matching vm_fail / ct_fail
                 # notification so the user gets paged on their channels
                 # too, not just an in-dashboard alert. Previously this
@@ -15964,8 +16017,70 @@ def api_vm_control(vmid):
         else:
             return jsonify({'error': 'Failed to get VM details'}), 500
     except Exception as e:
+        # A transport error happens before we have an identity that can be
+        # recovered later.  Leaving the pre-submission marker would otherwise
+        # make the CT permanently busy after one timed-out pvesh invocation.
+        if operation_lock is not None and not task_identity_confirmed:
+            try:
+                lxc_check_job.cancel_control(operation_lock)
+            except Exception:
+                pass
         return jsonify({'error': str(e)}), 500
 
+    finally:
+        if operation_lock is not None:
+            operation_lock.close()
+
+@app.route('/api/vms/<int:vmid>/control-task/<path:task>', methods=['GET'])
+@require_auth
+def api_vm_control_task(vmid, task):
+    """Read only the power-control task bound to this guest and node."""
+    import re
+    import json
+    if not re.fullmatch(r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:'
+                        r'(?:(?:vz|qm)(?:start|stop|shutdown|reboot)|ha(?:start|stop)):'
+                        r'[0-9]+:[\w.@!-]+:', task):
+        return jsonify({'error': 'Invalid control task'}), 400
+    parts = task.split(':')
+    resource = next((item for item in get_cached_pvesh_cluster_resources_vm() or []
+                     if item.get('vmid') == vmid), None)
+    if resource is None:
+        return jsonify({'error': 'VM/LXC not found'}), 404
+    vm_type = 'lxc' if resource.get('type') == 'lxc' else 'qemu'
+    node = resource.get('node', 'pve')
+    prefix = 'vz' if vm_type == 'lxc' else 'qm'
+    allowed_task_types = {f'{prefix}{action}' for action in
+                          ('start', 'stop', 'shutdown', 'reboot')}
+    allowed_task_types.update(('hastart', 'hastop'))
+    if (parts[1] != node or parts[6] != str(vmid)
+            or parts[5] not in allowed_task_types):
+        return jsonify({'error': 'Control task does not belong to this guest'}), 400
+    try:
+        result = subprocess.run(['pvesh', 'get', f'/nodes/{node}/tasks/{task}/status',
+                                 '--output-format', 'json'],
+                                capture_output=True, text=True, timeout=5)
+        if result.returncode != 0:
+            return jsonify({'error': 'Control task status unavailable'}), 502
+        status = json.loads(result.stdout)
+        if not isinstance(status, dict) or status.get('status') not in ('running', 'stopped'):
+            return jsonify({'error': 'Invalid control task status'}), 502
+        payload = {'status': status['status'], 'exitstatus': status.get('exitstatus')}
+        if status['status'] == 'stopped' and status.get('exitstatus') == 'OK':
+            current = subprocess.run(['pvesh', 'get', f'/nodes/{node}/{vm_type}/{vmid}/status/current',
+                                      '--output-format', 'json'],
+                                     capture_output=True, text=True, timeout=5)
+            if current.returncode != 0:
+                return jsonify({'error': 'Guest status unavailable'}), 502
+            guest = json.loads(current.stdout)
+            if not isinstance(guest, dict) or guest.get('status') not in ('running', 'stopped'):
+                return jsonify({'error': 'Invalid guest status'}), 502
+            payload['guest_status'] = guest['status']
+            _pvesh_cache['cluster_resources_vm_time'] = 0
+            _vm_cache_invalidate(vmid)
+        return jsonify(payload)
+    except (ValueError, subprocess.TimeoutExpired):
+        return jsonify({'error': 'Control task status unavailable'}), 502
+
 @app.route('/api/vms/<int:vmid>/description', methods=['PUT'])
 @app.route('/api/vms/<int:vmid>/config', methods=['PUT'])  # legacy alias
 @require_auth

AppImage/scripts/lxc_check_job.py

@@ -25,6 +25,7 @@ _jobs = {}
 _attempts = {}
 COOLDOWN = 60
 JOB_TTL = 3600
+CONTROL_TASK_WAIT_SECONDS = 180
 
 
 def acquire(vmid):
@@ -32,6 +33,7 @@ def acquire(vmid):
     handle = os.fdopen(os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600), 'r+')
     try:
         fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        check_control_task(vmid, handle)
         return handle
     except BaseException:
         handle.close()
@@ -85,6 +87,92 @@ def clear_invalid_marker(vmid, *, check_started_ns=None):
         handle.close()
 
 
+def _write_control(handle, record):
+    handle.seek(0)
+    if record is not None:
+        handle.write(json.dumps(record))
+    handle.flush()
+    # Do not erase the old reservation before the replacement was written.
+    # A partial/failed write stays non-empty and therefore fails closed.
+    handle.truncate(handle.tell())
+
+
+def begin_control(handle, node):
+    """Persist uncertainty before submission, including across Monitor restart."""
+    if not isinstance(node, str) or not re.fullmatch(r'[\w.-]+', node):
+        raise ValueError('Invalid control node')
+    _write_control(handle, {'node': node, 'task': None})
+
+
+def cancel_control(handle):
+    """Clear a request for which no recoverable task identity was recorded."""
+    _write_control(handle, None)
+
+
+def check_control_task(vmid, handle=None):
+    """Called under the shared flock, also by the shell update runner.
+
+    No record means no probe. A restart releases OS file descriptors, but the
+    record still forbids a new operation until PVE confirms task termination.
+    Unknown submission/status cannot safely be treated as completion.
+    """
+    if handle is None:
+        with open(LOCK_DIR / f'proxmenux-lxc-update-{int(vmid)}.lock', 'r+') as stream:
+            return check_control_task(vmid, stream)
+    handle.seek(0)
+    raw = handle.read(1025)
+    if not raw:
+        return
+    try:
+        record = json.loads(raw)
+        node, task = record['node'], record['task']
+        if not isinstance(node, str) or not re.fullmatch(r'[\w.-]+', node):
+            raise ValueError('Invalid node')
+        if (not isinstance(task, str)
+                or not re.fullmatch(r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:'
+                                    r'[0-9A-Fa-f]+:(?:vz(?:start|stop|shutdown|reboot)'
+                                    r'|ha(?:start|stop)):'
+                                    r'[0-9]+:[\w.@!-]+:', task)
+                or task.split(':')[1] != node or task.split(':')[6] != str(vmid)):
+            raise ValueError('Unconfirmed control task')
+        result = subprocess.run(['pvesh', 'get', f'/nodes/{node}/tasks/{task}/status',
+                                '--output-format', 'json'],
+                               capture_output=True, text=True, timeout=5)
+        status = json.loads(result.stdout) if result.returncode == 0 else None
+        if (not isinstance(status, dict) or status.get('status') != 'stopped'
+                or not isinstance(status.get('exitstatus'), str) or not status['exitstatus']):
+            raise ValueError('Control task completion not confirmed')
+    except (ValueError, KeyError, TypeError, OSError, subprocess.TimeoutExpired) as exc:
+        raise BlockingIOError('Container control completion is not confirmed') from exc
+    _write_control(handle, None)
+
+
+def _wait_control(vmid, reservation):
+    deadline = time.monotonic() + CONTROL_TASK_WAIT_SECONDS
+    try:
+        while time.monotonic() < deadline:
+            try:
+                check_control_task(vmid, reservation)
+                return
+            except BlockingIOError:
+                # Keep the known task record for recovery after this bounded
+                # waiter exits; a later caller asks Proxmox to verify it.
+                remaining = deadline - time.monotonic()
+                if remaining > 0:
+                    time.sleep(min(2, remaining))
+    finally:
+        reservation.close()
+
+
+def retain_control(vmid, task, reservation):
+    reservation.seek(0)
+    record = json.loads(reservation.read(1025))
+    record['task'] = task
+    _write_control(reservation, record)
+    # Reservation ownership transfers only after Thread.start succeeds.
+    threading.Thread(target=_wait_control, args=(vmid, reservation), daemon=True).start()
+
+
 def guard(vmid):
     if not managed_installs._lxc_updates_detection_enabled():
         raise RuntimeError('LXC update detection is disabled')

AppImage/scripts/tests/test_lxc_vm_ip_inventory.py

@@ -36,6 +36,7 @@ class LxcVmIpInventoryTests(unittest.TestCase):
             "_get_lxc_app_watch_map": lambda: {},
             "_get_lxc_docker_inventory_map": lambda: {},
             "get_proxmox_node_name": lambda: "pve",
+            "_fast_guest_status": lambda _vmid, _vm_type: "running",
             "_get_cached_vm_list_search_notes": lambda _resources, _node: {},
             "get_cached_vm_disk": lambda _vmid: None,
             "_guest_lifecycle_lock": threading.RLock(),

scripts/lxc/apply_updates.sh

@@ -124,12 +124,19 @@ extract_helper_slug_from_wrapper() {
 # One update per CT at a time, regardless of whether it came from the
 # UI or the scheduler. The descriptor remains open for this process.
 LOCK_DIR="${PROXMENUX_LOCK_DIR:-/run/lock}"
-exec 9>"${LOCK_DIR}/proxmenux-lxc-update-${VMID}.lock"
+exec 9>>"${LOCK_DIR}/proxmenux-lxc-update-${VMID}.lock"
 if ! flock -n 9; then
   echo "ERROR: another ProxMenux update is already running for CT $VMID." >&2
   exit 7
 fi
 
+if [[ -s "${LOCK_DIR}/proxmenux-lxc-update-${VMID}.lock" ]]; then
+  if ! python3 -c 'import sys; sys.path.insert(0, "/usr/local/share/proxmenux/monitor-app/usr/bin"); import lxc_check_job; lxc_check_job.check_control_task(int(sys.argv[1]))' "$VMID"; then
+    echo "ERROR: container control completion is not confirmed; nothing was changed." >&2
+    exit 7
+  fi
+fi
+
 if [[ "${REQUIRE_CHECKED_OS:-0}" == "1" ]]; then
   if ! python3 -c 'import sys; sys.path.insert(0, "/usr/local/share/proxmenux/monitor-app/usr/bin"); import lxc_check_job; lxc_check_job.checked_os(int(sys.argv[1]))' "$VMID"; then
     echo "ERROR: a fresh successful positive OS check is required; nothing was installed." >&2

tests/lxc_updates/test_check_only_job.py

@@ -288,6 +288,45 @@ class CheckOnlyTests(unittest.TestCase):
                     self.assertTrue(stored['error'])
                     self.assertFalse(any('apt list' in ' '.join(command) for command in commands))
 
+    def test_control_route_uses_same_lock_and_preserves_qemu(self):
+        app = Flask(__name__)
+        def control_result(argv, **_):
+            path = argv[2]
+            vm_type, vmid, action = path.split('/')[3], path.split('/')[4], path.rsplit('/', 1)[1]
+            task_type = ('vz' if vm_type == 'lxc' else 'qm') + action
+            return subprocess.CompletedProcess(
+                [], 0, f'"UPID:fixture:000A:000B:000C:{task_type}:{vmid}:root@pam:"\n', '')
+        command = Mock(side_effect=control_result)
+        scope = {'app': app, 'jsonify': jsonify, 'request': request,
+                 'require_auth': lambda fn: fn,
+                 'get_cached_pvesh_cluster_resources_vm': lambda: [
+                     {'vmid': 100, 'type': 'lxc', 'node': 'fixture'},
+                     {'vmid': 200, 'type': 'qemu', 'node': 'fixture'}],
+                 'subprocess': SimpleNamespace(run=command), '_pvesh_cache': {},
+                 '_vm_cache_invalidate': Mock()}
+        node = next(n for n in ast.parse((ROOT / 'AppImage/scripts/flask_server.py').read_text()).body
+                    if isinstance(n, ast.FunctionDef) and n.name == 'api_vm_control')
+        exec(compile(ast.Module(body=[node], type_ignores=[]), 'route', 'exec'), scope)
+        client = app.test_client()
+        reservation = job.acquire(100)
+        try:
+            self.assertEqual(client.post('/api/vms/100/control', json={'action': 'stop'}).status_code, 409)
+            command.assert_not_called()
+            self.assertEqual(client.post('/api/vms/200/control', json={'action': 'reboot'}).status_code, 200)
+        finally:
+            reservation.close()
+        # The production waiter owns a long-lived reservation. Isolate its
+        # launch here so it cannot probe through another test's transport.
+        with patch.object(job.threading, 'Thread') as worker:
+            response = client.post('/api/vms/100/control', json={'action': 'start'})
+            worker.return_value.start.assert_called_once_with()
+            retained = worker.call_args.kwargs['args'][1]
+            self.addCleanup(retained.close)
+            job.cancel_control(retained)
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(response.json['success'])
+        self.assertEqual(response.json['task'], 'UPID:fixture:000A:000B:000C:vzstart:100:root@pam:')
+        self.assertIn('/nodes/fixture/lxc/100/status/start', command.call_args.args[0])
 
     def test_actual_routes_submit_poll_and_control_lock(self):
         app = Flask(__name__)

tests/lxc_updates/test_control_task_reservation.py

@@ -0,0 +1,218 @@
+"""Actual Flask submission, real flock, isolated Proxmox reads and shell caller."""
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import test_vm_control_task as control_fixture
+
+TASK, ROOT = control_fixture.TASK, control_fixture.ROOT
+
+sys.path.insert(0, str(ROOT / 'AppImage/scripts'))
+import lxc_check_job as job
+
+
+class ControlReservationTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.folder = Path(self.temp.name)
+        self.lock_path = self.folder / 'proxmenux-lxc-update-100.lock'
+        self.lock_patch = patch.object(job, 'LOCK_DIR', self.folder)
+        self.lock_patch.start()
+        self.addCleanup(self.lock_patch.stop)
+        job._jobs.clear()
+        job._attempts.clear()
+        self.route = control_fixture.ControlTaskTests()
+        self.route.setUp()
+        self.thread = Mock()
+        self.factory = patch.object(job.threading, 'Thread', return_value=self.thread)
+        self.thread_factory = self.factory.start()
+        self.addCleanup(self.factory.stop)
+        self.reservations = []
+        self.addCleanup(lambda: [handle.close() for handle in self.reservations if not handle.closed])
+
+    def accepted(self, action='reboot'):
+        task = TASK.replace('vzreboot', 'vz' + action)
+        self.route.run.return_value = self.route.result(task)
+        response = self.route.client.post('/api/vms/100/control', json={'action': action})
+        self.assertEqual(response.status_code, 200, response.json)
+        handle = self.thread_factory.call_args.kwargs['args'][1]
+        self.reservations.append(handle)
+        self.assertEqual(response.json['task'], task)
+        return handle
+
+    @staticmethod
+    def result(payload, code=0):
+        return subprocess.CompletedProcess([], code, json.dumps(payload), '')
+
+    def test_every_power_action_keeps_reservation_after_post_and_blocks_second_caller(self):
+        for action in ('start', 'shutdown', 'reboot', 'stop'):
+            with self.subTest(action=action):
+                handle = self.accepted(action)
+                self.route.run.reset_mock()
+                response = self.route.client.post('/api/vms/100/control', json={'action': 'start'})
+                self.assertEqual(response.status_code, 409)
+                self.route.run.assert_not_called()
+                with self.assertRaises(BlockingIOError):
+                    job.start(100)
+                self.assertFalse(handle.closed)
+                job.cancel_control(handle)
+                handle.close()
+
+    def test_waiter_survives_read_failure_and_running_until_confirmed_terminal_status(self):
+        handle = self.accepted()
+        probes = [subprocess.TimeoutExpired('pvesh', 5),
+                  self.result({'status': 'running'}), self.result({'status': 'stopped'}),
+                  self.result({'status': 'stopped', 'exitstatus': 'OK'})]
+        def still_reserved(_):
+            with self.assertRaises(BlockingIOError):
+                job.acquire(100)
+            self.assertFalse(handle.closed)
+        with patch.object(job.subprocess, 'run', side_effect=probes) as reads, \
+             patch.object(job.time, 'sleep', side_effect=still_reserved) as sleeps:
+            job._wait_control(100, handle)
+        self.assertEqual(sleeps.call_count, 3)
+        self.assertTrue(handle.closed)
+        self.assertEqual(self.lock_path.read_text(), '')
+        self.assertTrue(all(call.args[0][:2] == ['pvesh', 'get'] for call in reads.call_args_list))
+        job.acquire(100).close()
+
+    def test_restart_releases_fd_but_marker_blocks_until_pve_verifies_completion(self):
+        handle = self.accepted()
+        handle.close()  # simulate Monitor shutdown, not guest task termination
+        with patch.object(job.subprocess, 'run', return_value=self.result({'status': 'running'})):
+            with self.assertRaises(BlockingIOError):
+                job.acquire(100)
+        with patch.object(job.subprocess, 'run', return_value=self.result(
+                {'status': 'stopped', 'exitstatus': 'start failed'})):
+            job.acquire(100).close()
+        self.assertEqual(self.lock_path.read_text(), '')
+
+    def test_unknown_or_malformed_records_make_no_pve_probe_and_keep_blocked(self):
+        for content in ['not-json', '{}', json.dumps({'node': 'fixture', 'task': None}),
+                        json.dumps({'node': 'other', 'task': TASK}),
+                        json.dumps({'node': 'fixture', 'task': TASK.replace(':100:', ':200:')}),
+                        'x' * 1025]:
+            self.lock_path.write_text(content)
+            with patch.object(job.subprocess, 'run') as probe:
+                with self.assertRaises(BlockingIOError):
+                    job.acquire(100)
+                probe.assert_not_called()
+            self.assertEqual(self.lock_path.read_text(), content)
+
+    def test_plain_lock_acquire_has_no_new_probe(self):
+        with patch.object(job.subprocess, 'run') as probe:
+            job.acquire(100).close()
+            probe.assert_not_called()
+
+    def test_definite_failure_or_unidentified_submission_never_leaves_a_permanent_reservation(self):
+        self.route.run.return_value = self.result({}, code=1)
+        self.assertEqual(self.route.client.post('/api/vms/100/control', json={'action': 'start'}).status_code, 500)
+        self.assertEqual(self.lock_path.read_text(), '')
+        for response in [subprocess.TimeoutExpired('pvesh', 30), self.result('')]:
+            self.route.run.side_effect = response if isinstance(response, Exception) else None
+            self.route.run.return_value = response
+            response = self.route.client.post('/api/vms/100/control', json={'action': 'start'})
+            self.assertIn(response.status_code, (500, 502), response.json)
+            self.assertEqual(self.lock_path.read_text(), '')
+            job.acquire(100).close()
+
+    def test_ha_task_recovers_like_a_native_lxc_task(self):
+        task = TASK.replace('vzreboot', 'hastop')
+        self.lock_path.write_text(json.dumps({'node': 'fixture', 'task': task}))
+        with patch.object(job.subprocess, 'run', return_value=self.result(
+                {'status': 'stopped', 'exitstatus': 'OK'})) as probe:
+            job.acquire(100).close()
+        self.assertEqual(probe.call_count, 1)
+        self.assertEqual(self.lock_path.read_text(), '')
+
+    def test_wait_deadline_releases_fd_but_keeps_known_task_for_recovery(self):
+        handle = self.accepted()
+        with patch.object(job, 'CONTROL_TASK_WAIT_SECONDS', 0):
+            job._wait_control(100, handle)
+        self.assertTrue(handle.closed)
+        record = json.loads(self.lock_path.read_text())
+        self.assertEqual(record['task'], TASK)
+        with patch.object(job.subprocess, 'run', return_value=self.result(
+                {'status': 'stopped', 'exitstatus': 'OK'})):
+            job.acquire(100).close()
+        self.assertEqual(self.lock_path.read_text(), '')
+
+    def test_thread_launch_failure_retains_marker_for_recovery(self):
+        self.thread.start.side_effect = RuntimeError('thread failed')
+        self.route.run.return_value = self.route.result(TASK)
+        self.assertEqual(self.route.client.post('/api/vms/100/control', json={'action': 'reboot'}).status_code, 500)
+        self.assertEqual(json.loads(self.lock_path.read_text())['task'], TASK)
+        with patch.object(job.subprocess, 'run', return_value=self.result({'status': 'running'})):
+            with self.assertRaises(BlockingIOError):
+                job.acquire(100)
+
+    def test_failed_task_record_write_keeps_uncertainty_before_any_next_operation(self):
+        original = job._write_control
+        calls = 0
+        def fail_second(handle, record):
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                raise OSError('persistence failed')
+            original(handle, record)
+        self.route.run.return_value = self.route.result(TASK)
+        with patch.object(job, '_write_control', side_effect=fail_second):
+            response = self.route.client.post('/api/vms/100/control', json={'action': 'reboot'})
+        self.assertEqual(response.status_code, 500)
+        self.assertIsNone(json.loads(self.lock_path.read_text())['task'])
+        with patch.object(job.subprocess, 'run') as probe:
+            with self.assertRaises(BlockingIOError):
+                job.acquire(100)
+            probe.assert_not_called()
+
+    def test_shell_blocked_by_live_reservation_runs_no_probe_or_guest_action(self):
+        self.accepted()
+        import os
+        env = {**os.environ, 'PROXMENUX_LOCK_DIR': str(self.folder),
+               'VMID': '100', 'TARGET': 'os', 'BACKUP': '1'}
+        result = subprocess.run(['/bin/bash', str(ROOT / 'scripts/lxc/apply_updates.sh')],
+                                env=env, capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 7, result.stdout + result.stderr)
+        self.assertIn('another ProxMenux update', result.stderr)
+        self.assertEqual(result.stdout, '')
+
+    def test_retired_combined_mode_fails_before_lock_or_any_guest_operation(self):
+        import os
+        env = {**os.environ, 'PROXMENUX_LOCK_DIR': str(self.folder),
+               'VMID': '100', 'TARGET': 'os', 'CHECK_APPLY_OS': '1'}
+        result = subprocess.run(['/bin/bash', str(ROOT / 'scripts/lxc/apply_updates.sh')],
+                                env=env, capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 4, result.stdout + result.stderr)
+        self.assertFalse(self.lock_path.exists())
+        self.assertEqual(result.stdout, '')
+
+    def test_shell_caller_preserves_pending_record_and_performs_no_guest_operation(self):
+        # Covers process-restart recovery: no Python thread holds this flock.
+        self.lock_path.write_text(json.dumps({'node': 'fixture', 'task': TASK}))
+        binary = self.folder / 'bin'
+        binary.mkdir()
+        call_log = self.folder / 'calls'
+        for name, body in [('pvesh', 'echo \'{"status":"running"}\''),
+                           ('pct', f'echo pct >> "{call_log}"; exit 99')]:
+            script = binary / name
+            script.write_text('#!/bin/sh\n' + body + '\n')
+            script.chmod(0o755)
+        import os
+        before = self.lock_path.read_text()
+        env = {**os.environ, 'PATH': str(binary) + ':' + os.environ['PATH'],
+               'PYTHONPATH': str(ROOT / 'AppImage/scripts'),
+               'PROXMENUX_LOCK_DIR': str(self.folder), 'VMID': '100', 'TARGET': 'os', 'BACKUP': '1'}
+        result = subprocess.run(['/bin/bash', str(ROOT / 'scripts/lxc/apply_updates.sh')],
+                                env=env, capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 7, result.stdout + result.stderr)
+        self.assertFalse(call_log.exists())
+        self.assertEqual(self.lock_path.read_text(), before)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/lxc_updates/test_observed_state.py

@@ -145,7 +145,25 @@ class ObservedStateRoutesTests(unittest.TestCase):
         self.assertEqual(response.status_code, 200, response.json)
         return response.json[0]
 
+    def test_completed_shutdown_cannot_revert_to_stale_cluster_running(self):
+        self.running = False
+        self.scope['get_cached_pvesh_cluster_resources_vm'] = lambda: [
+            dict(type='lxc', node='fixture-pve', vmid=100, name='dev-trixie', status='running')]
+        for _ in range(3):
+            row = self.row()
+            self.assertEqual(row['status'], 'stopped', 'cached summary cannot override live local state')
+            self.assertEqual(row['ips'], [])
+        self.ip_probe.assert_not_called()
+        self.assertEqual(self.commands, [], 'status polling never enters the guest')
 
+    def test_completed_start_cannot_revert_to_stale_cluster_stopped(self):
+        self.running = True
+        self.scope['get_cached_pvesh_cluster_resources_vm'] = lambda: [
+            dict(type='lxc', node='fixture-pve', vmid=100, name='dev-trixie', status='stopped')]
+        row = self.row()
+        self.assertEqual(row['status'], 'running')
+        self.assertEqual(row['ip'], self.ip)
+        self.assertEqual(self.commands, [])
 
     def test_dhcp_change_after_ttl_without_lifecycle_and_api_compatibility(self):
         first = self.row()

tests/lxc_updates/test_observed_state_ui.cjs

@@ -473,5 +473,39 @@ context.selectedVMRef.current = { vmid: 100 }
       assert.ok(messages[key]?.trim())
     assert.match(messages.checkCooldown, /\{seconds\}/)
   }
+  const oldFetch = context.fetchApi
+  const oldMutate = context.mutate
+  context.controlRequestRef = { current: false }
+  context.setControlAction = () => {}
+  context.waitForVmControlTask = async () => ({ guest_status: 'running' })
+  context.setSelectedVM = update => {
+    const previous = context.selectedVM
+    const next = typeof update === 'function' ? update(previous) : update
+    assert.equal(next?.vmid, previous?.vmid, 'control preserves the selected guest')
+    context.selectedVM = next
+  }
+  context.fetchApi = async () => ({ config: { onboot: 1 } })
+  context.mutate = async () => {}
+  for (const action of ["start", "shutdown", "reboot", "stop"]) {
+    context.selectedVMRef.current = { vmid: 100 }
+    context.selectionChanged = undefined
+    await ui.control(100, action)
+    assert.equal(context.selectionChanged, undefined, "control must not close or reset the selected modal")
+    assert.equal(context.controlLoading, false)
+  }
+  for (const selection of [null, { vmid: 200 }]) {
+    context.selectedVMRef.current = { vmid: 100 }
+    context.mutate = async () => { context.selectedVMRef.current = selection }
+    details = null
+    await ui.control(100, "start")
+    assert.equal(details, null, "closing or switching guests during control must not reopen/overwrite a modal")
+  }
+  context.selectedVMRef.current = { vmid: 100 }
+  context.mutate = async () => { throw new Error("poll failed") }
+  context.alerted = null
+  await ui.control(100, "shutdown")
+  assert.equal(context.alerted, null, "failed revalidation is not failed control")
+  context.fetchApi = oldFetch
+  context.mutate = oldMutate
   console.log("PASS observed-state UI: freshness, safe actions, per-guest preferences/bulk races, EN/SK")
 })().catch(error => { console.error(error); process.exitCode = 1 })

tests/lxc_updates/test_vm_control_task.py

@@ -0,0 +1,183 @@
+"""Actual Flask routes: additive task metadata and confined read-only status."""
+import ast
+import json
+import subprocess
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+from flask import Flask, jsonify, request
+
+ROOT = Path(__file__).resolve().parents[2]
+TASK = 'UPID:fixture:000A:000B:000C:vzreboot:100:root@pam:'
+
+class ControlTaskTests(unittest.TestCase):
+    def setUp(self):
+        self.app = Flask(__name__)
+        self.run = Mock()
+        self.cache = {}
+        self.invalidate = Mock()
+        scope = dict(app=self.app, jsonify=jsonify, request=request, require_auth=lambda fn: fn,
+                     get_cached_pvesh_cluster_resources_vm=lambda: [
+                         dict(vmid=100, type='lxc', node='fixture'),
+                         dict(vmid=200, type='qemu', node='fixture')],
+                     subprocess=SimpleNamespace(run=self.run, TimeoutExpired=subprocess.TimeoutExpired),
+                     _pvesh_cache=self.cache, _vm_cache_invalidate=self.invalidate)
+        nodes = [n for n in ast.parse((ROOT / 'AppImage/scripts/flask_server.py').read_text()).body
+                 if isinstance(n, ast.FunctionDef) and n.name in ('api_vm_control_task', 'api_vm_control')]
+        exec(compile(ast.Module(body=nodes, type_ignores=[]), 'routes', 'exec'), scope)
+        self.client = self.app.test_client()
+
+    def get(self, task=TASK, vmid=100):
+        return self.client.get(f'/api/vms/{vmid}/control-task/{task}')
+
+    def result(self, payload, code=0):
+        return subprocess.CompletedProcess([], code, json.dumps(payload), '')
+
+    def test_actual_control_post_preserves_each_action_and_adds_its_task(self):
+        lock = Mock()
+        retained = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=retained, cancel_control=Mock())
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            for family, vmid in [('vz', 100), ('qm', 200)]:
+                for action in ['start', 'shutdown', 'reboot', 'stop']:
+                    self.run.reset_mock()
+                    task = TASK.replace('vzreboot:100', f'{family}{action}:{vmid}')
+                    self.run.return_value = self.result(task)
+                    response = self.client.post(f'/api/vms/{vmid}/control', json={'action': action})
+                    self.assertEqual(response.status_code, 200)
+                    self.assertTrue(response.json['success'])
+                    self.assertEqual(response.json['action'], action)
+                    self.assertEqual(response.json['vmid'], vmid)
+                    self.assertEqual(response.json['task'], task)
+                    self.assertEqual(self.run.call_count, 1, 'POST accepts once without polling or extra action')
+                    self.assertEqual(self.run.call_args.args[0],
+                        ['pvesh', 'create', f'/nodes/fixture/{"lxc" if family == "vz" else "qemu"}/{vmid}/status/{action}'])
+            self.assertEqual(retained.call_count, 4)
+            lock.close.assert_not_called()
+
+    def test_control_post_reads_final_upid_after_pvesh_log_output(self):
+        lock = Mock()
+        retained = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=retained, cancel_control=Mock())
+        task = TASK.replace('vzreboot', 'vzstart')
+        self.run.return_value = subprocess.CompletedProcess(
+            [], 0, f'waiting for server lock\n"{task}"\n', '')
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            response = self.client.post('/api/vms/100/control', json={'action': 'start'})
+        self.assertEqual(response.status_code, 200, response.json)
+        self.assertEqual(response.json['task'], task)
+        retained.assert_called_once_with(100, task, lock)
+
+    def test_control_post_accepts_ha_shutdown_task_for_lxc_and_qemu(self):
+        lock = Mock()
+        retained = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=retained, cancel_control=Mock())
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            for vmid in (100, 200):
+                with self.subTest(vmid=vmid):
+                    self.run.reset_mock()
+                    task = TASK.replace('vzreboot:100', f'hastop:{vmid}')
+                    self.run.return_value = self.result(task)
+                    response = self.client.post(
+                        f'/api/vms/{vmid}/control', json={'action': 'shutdown'})
+                    self.assertEqual(response.status_code, 200, response.json)
+                    self.assertTrue(response.json['success'])
+                    self.assertEqual(response.json['task'], task)
+                    self.assertEqual(self.run.call_count, 1)
+                    self.assertEqual(self.run.call_args.args[0], [
+                        'pvesh', 'create',
+                        f'/nodes/fixture/{"lxc" if vmid == 100 else "qemu"}/{vmid}/status/shutdown',
+                    ])
+        retained.assert_called_once_with(100, TASK.replace('vzreboot:100', 'hastop:100'), lock)
+
+    def test_control_post_rejects_success_without_a_confirmable_task(self):
+        lock = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=Mock(), cancel_control=Mock())
+        self.run.return_value = subprocess.CompletedProcess([], 0, 'waiting for server lock\n', '')
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            response = self.client.post('/api/vms/100/control', json={'action': 'start'})
+        self.assertEqual(response.status_code, 502, response.json)
+        self.assertFalse(response.json['success'])
+        self.assertIn('task', response.json['error'].lower())
+        module.retain_control.assert_not_called()
+        module.cancel_control.assert_called_once_with(lock)
+        lock.close.assert_called_once()
+        self.run.reset_mock()
+        self.assertEqual(self.client.post('/api/vms/100/control', json={'action': 'destroy'}).status_code, 400)
+        self.run.assert_not_called()
+
+    def test_running_task_never_probes_guest_or_mutates(self):
+        self.run.return_value = self.result(dict(status='running'))
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json, dict(status='running', exitstatus=None))
+        self.assertEqual(self.run.call_count, 1)
+        self.assertEqual(self.run.call_args.args[0][:2], ['pvesh', 'get'])
+        self.invalidate.assert_not_called()
+
+    def test_complete_task_checks_actual_guest_for_all_power_actions_and_qemu(self):
+        for family, vmid in [('vz', 100), ('qm', 200)]:
+            for action in ['start', 'shutdown', 'reboot', 'stop']:
+                with self.subTest(family=family, action=action):
+                    self.run.reset_mock()
+                    expected = 'stopped' if action in ('shutdown', 'stop') else 'running'
+                    self.run.side_effect = [self.result(dict(status='stopped', exitstatus='OK')),
+                                            self.result(dict(status=expected))]
+                    task = TASK.replace('vzreboot:100', f'{family}{action}:{vmid}')
+                    response = self.get(task, vmid)
+                    self.assertEqual(response.status_code, 200)
+                    self.assertEqual(response.json['guest_status'], expected)
+                    self.assertEqual(self.run.call_count, 2)
+                    self.assertTrue(all(call.args[0][:2] == ['pvesh', 'get'] for call in self.run.call_args_list))
+
+    def test_ha_start_and_stop_tasks_belong_to_the_guest(self):
+        for action, expected in [('hastart', 'running'), ('hastop', 'stopped')]:
+            with self.subTest(action=action):
+                self.run.reset_mock()
+                self.run.side_effect = [self.result(dict(status='stopped', exitstatus='OK')),
+                                        self.result(dict(status=expected))]
+                task = TASK.replace('vzreboot', action)
+                response = self.get(task)
+                self.assertEqual(response.status_code, 200, response.json)
+                self.assertEqual(response.json['guest_status'], expected)
+                self.assertEqual(self.run.call_count, 2)
+
+    def test_wrong_guest_node_type_and_path_make_zero_probes(self):
+        for task in [TASK.replace(':100:', ':200:'), TASK.replace('fixture', 'other'),
+                     TASK.replace('vzreboot', 'qmreboot'), TASK.replace('vzreboot', 'vzdump'),
+                     TASK + '/extra', '../../etc/passwd', 'not-a-task']:
+            with self.subTest(task=task):
+                self.run.reset_mock()
+                self.assertIn(self.get(task).status_code, [400, 404])
+                self.run.assert_not_called()
+        self.assertEqual(self.get(vmid=999).status_code, 404)
+        self.run.assert_not_called()
+
+    def test_task_failure_never_probes_guest_or_claims_success(self):
+        self.run.return_value = self.result(dict(status='stopped', exitstatus='start failed'))
+        self.assertEqual(self.get().json['exitstatus'], 'start failed')
+        self.assertEqual(self.run.call_count, 1)
+        self.invalidate.assert_not_called()
+
+    def test_unknown_and_invalid_reads_are_errors_not_success(self):
+        for value in [[], {}, dict(status='unknown'), dict(status='running')]:
+            self.run.return_value = self.result(value, code=1 if value == dict(status='running') else 0)
+            self.assertEqual(self.get().status_code, 502)
+        self.run.return_value = subprocess.CompletedProcess([], 0, '', '')
+        self.assertEqual(self.get().status_code, 502)
+        self.run.side_effect = subprocess.TimeoutExpired('pvesh', 5)
+        self.assertEqual(self.get().status_code, 502)
+
+    def test_invalid_final_guest_never_clears_cache(self):
+        for value in [{}, [], dict(status='unknown')]:
+            self.run.side_effect = [self.result(dict(status='stopped', exitstatus='OK')), self.result(value)]
+            self.assertEqual(self.get().status_code, 502)
+            self.invalidate.assert_not_called()
+
+if __name__ == '__main__':
+    unittest.main()

tests/lxc_updates/test_vm_control_ui.cjs

@@ -0,0 +1,135 @@
+// Execute the actual handler and footer: acceptance is not task completion.
+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/virtual-machines.tsx'), 'utf8')
+const tree = ts.createSourceFile('vm.tsx', source, ts.ScriptTarget.Latest, true, ts.ScriptKind.TSX)
+let handler, footer
+function walk(node) {
+  if (ts.isVariableDeclaration(node) && node.name.getText(tree) === 'handleVMControl') handler = node.getText(tree)
+  if (ts.isJsxElement(node) && node.openingElement.tagName.getText(tree) === 'div'
+    && node.openingElement.attributes.getText(tree).includes('border-t border-border bg-background px-6 py-4 mt-auto'))
+    footer = node.getText(tree)
+  ts.forEachChild(node, walk)
+}
+walk(tree)
+function compile(code, context) {
+  vm.runInNewContext(ts.transpileModule(code, { compilerOptions: {
+    module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020, jsx: ts.JsxEmit.ReactJSX,
+  }}).outputText, context)
+  return context.exports
+}
+function deferred() { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no }); return { promise, resolve, reject } }
+function buttons(node) {
+  if (!node || typeof node !== 'object') return []
+  if (Array.isArray(node)) return node.flatMap(buttons)
+  return [...(node.type === 'Button' ? [node] : []), ...buttons(node.props?.children)]
+}
+;(async () => {
+  for (const action of ['start', 'shutdown', 'reboot', 'stop']) {
+    const completion = deferred(), accepted = deferred()
+    const context = {
+      exports: {}, console: { error() {} }, selectedVM: { vmid: 100, status: action === 'start' ? 'stopped' : 'running', type: 'lxc' },
+      selectedVMRef: { current: { vmid: 100 } }, controlRequestRef: { current: false },
+      controlLoading: false, controlAction: null, posts: 0, alerts: [],
+      setControlLoading(value) { context.controlLoading = value },
+      setControlAction(value) { context.controlAction = value },
+      setSelectedVM(value) { context.selectedVM = typeof value === 'function' ? value(context.selectedVM) : value },
+      rows: [{ vmid: 100, status: 'running' }, { vmid: 200, status: 'stopped' }], publications: 0,
+      setVMDetails() {}, mutate: async (update, options) => {
+        if (!update) return
+        assert.equal(options.revalidate, false, 'publish the confirmation before another GET')
+        context.rows = update(context.rows)
+        assert.equal(context.rows[1].status, 'stopped', 'only the operated guest changes')
+        context.publications++
+      }, invalidateLxcApps() {},
+      vmModalCacheRef: { current: Object.fromEntries(['details','backups','mountPoints','schedule','firewall'].map(n => [n, new Map()])) },
+      fetchApi: async (url, options) => {
+        if (options?.method === 'POST') { context.posts++; return accepted.promise }
+        return { config: {} }
+      },
+      waitForVmControlTask: () => completion.promise,
+      t: key => key, alert: text => context.alerts.push(text),
+      require: () => ({ jsx: (type, props) => ({ type, props }), jsxs: (type, props) => ({ type, props }) }),
+    }
+    for (const name of ['Button','Terminal','Play','Power','RotateCcw','StopCircle','Loader2']) context[name] = name
+    const ui = compile(`const ${handler}; exports.control = handleVMControl; exports.footer = () => (${footer});`, context)
+    const run = ui.control(100, action)
+    assert.equal(context.controlLoading, true, 'busy starts before the POST')
+    accepted.resolve({ success: true, task: 'task-id' })
+    await new Promise(resolve => setImmediate(resolve))
+    assert.equal(context.controlLoading, true, 'accepted control must stay busy until the Proxmox task completes')
+    assert.equal(context.publications, 0, 'acceptance never publishes an assumed final status')
+    assert(buttons(ui.footer()).filter(b => b.props.title).every(b => b.props.disabled), 'all power buttons stay disabled')
+    await ui.control(100, action)
+    assert.equal(context.posts, 1, 'a double click never submits a second operation')
+    completion.resolve({ guest_status: action === 'shutdown' || action === 'stop' ? 'stopped' : 'running' })
+    await run
+    assert.equal(context.controlLoading, false)
+    assert.equal(context.controlRequestRef.current, false)
+    assert.equal(context.selectedVM.status, action === 'shutdown' || action === 'stop' ? 'stopped' : 'running')
+    assert.equal(context.alerts.length, 0)
+    assert.equal(context.controlAction, null)
+    assert.equal(context.publications, 1)
+    assert.equal(context.rows[0].status, action === 'shutdown' || action === 'stop' ? 'stopped' : 'running')
+    const failed = deferred()
+    context.fetchApi = async () => ({ task: 'task-id' })
+    context.waitForVmControlTask = () => failed.promise
+    const failureRun = ui.control(100, action)
+    failed.reject(new Error('task failed'))
+    await failureRun
+    assert.equal(context.controlLoading, false)
+    assert.equal(context.alerts.length, 1, 'a task error cannot look like completion')
+    const switched = deferred()
+    context.waitForVmControlTask = () => switched.promise
+    const switchRun = ui.control(100, action)
+    context.selectedVMRef.current = { vmid: 200 }
+    context.selectedVM = { vmid: 200, status: 'stopped' }
+    switched.resolve({ guest_status: 'running' })
+    await switchRun
+    assert.equal(context.selectedVM.vmid, 200)
+    assert.equal(context.selectedVM.status, 'stopped', 'completion cannot overwrite another guest')
+    assert.equal(buttons(ui.footer()).filter(b => b.props.title).some(b => /Pending/.test(b.props.title)), false)
+  }
+  const polls = [], sequence = []
+  let clock = 0, readError = null
+  const helper = compile(fs.readFileSync(path.join(root, 'lib/vm-control.ts'), 'utf8'), {
+    exports: {}, Date: { now: () => clock }, AbortSignal: { timeout: () => ({}) },
+    setTimeout: callback => { clock += 1000; callback() },
+    require: () => ({ fetchApi: async (url, options) => {
+      polls.push([url, options]); if (readError) throw readError
+      return sequence.shift() || { status: 'running' }
+    }}),
+  })
+  // A reboot with "running" status is still pending until its task reports OK.
+  for (const action of ['start','shutdown','reboot','stop']) {
+    clock = 0; polls.length = 0
+    const expected = ['shutdown','stop'].includes(action) ? 'stopped' : 'running'
+    sequence.push({ status: 'running', guest_status: expected },
+      { status: 'stopped', exitstatus: 'OK', guest_status: expected === 'running' ? 'stopped' : 'running' },
+      { status: 'stopped', exitstatus: 'OK', guest_status: expected })
+    assert.equal((await helper.waitForVmControlTask(100, action, 'UPID:test')).guest_status, expected)
+    assert.equal(polls.length, 3, 'wait for both the task and actual target state')
+    assert(polls.every(([url, options]) => url === '/api/vms/100/control-task/UPID%3Atest' && !options.method),
+      'polling only reads the accepted task; no second POST or package check')
+  }
+  for (const response of [null, {}, { status:'stopped' }, { status:'stopped', exitstatus:'failed' }]) {
+    sequence.push(response)
+    // null must reach the helper, rather than the fake's default running state.
+    if (response === null) sequence[sequence.length - 1] = { status: null }
+    await assert.rejects(helper.waitForVmControlTask(100, 'reboot', 'task'), /controlNotConfirmed|failed/)
+  }
+  clock = 0; polls.length = 0
+  await assert.rejects(helper.waitForVmControlTask(100, 'reboot', 'task'), /controlNotConfirmed/)
+  assert.equal(clock, 180000, 'bounded wait includes the exact deadline')
+  polls.length = 0
+  await assert.rejects(helper.waitForVmControlTask(100, 'start', ''), /controlNotConfirmed/)
+  assert.equal(polls.length, 0, 'a missing task never starts speculative probes')
+  clock = 0; readError = new Error('unavailable')
+  await assert.rejects(helper.waitForVmControlTask(100, 'start', 'task'), /controlNotConfirmed/)
+  assert.equal(polls.length, 3, 'repeated read failures do not report success')
+  console.log('PASS VM control UI: accepted/pending/completed, all actions, duplicate guard and task failure')
+})().catch(error => { console.error(error); process.exitCode = 1 })