From 790e1eb6bd576261e6b9e70fcfe4fbdf17f866bb Mon Sep 17 00:00:00 2001
From: Casey <caseyrussell1976@gmail.com>
Date: Tue, 25 Aug 2026 22:41:47 -0500
Subject: [PATCH] fix(update): pause SCM-supervised Windows gateway services
 before venv mutation
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

On Windows installs where the gateway runs as an SCM service (WinSW,
NSSM, sc.exe create), the existing pause machinery kills the gateway
process directly — and the service wrapper's failure ladder resurrects
it within seconds, re-taking the venv file locks mid-update. The update
then dies partway through dependency sync with access-denied errors.

This extends _pause_windows_gateways_for_update() to detect when a
gateway's process tree is owned by a running SCM service, and to stop
the SERVICE through sc.exe instead of killing the child:

- gateway/status.py: expose service-ownership discovery for gateway
  runtimes (find_windows_gateway_services maps validated gateway PIDs
  through process ancestry to running SCM service PIDs, with
  create-time identity checks against PID reuse).
- hermes_cli/update_cmd.py: stop verified services via sc.exe before
  venv mutation and restart them afterward. Stops wait for a stable
  SCM 'stopped' state AND for the original descendant processes to
  exit (service 'Stopped' is not proof the child released its
  handles). Failure to prove ownership, stop a service, or restart it
  fails closed; rollback restores attempted services, and rollback
  failures are surfaced rather than swallowed.
- Fail-closed throughout: unreadable identities, ambiguous ancestry,
  or a service that will not reach a stable state abort the update
  before any file mutation.

Complements #37039 (gateway-only concurrent instances no longer abort):
that fix lets the update proceed past the gate; this one makes the
pause actually stick when the gateway is service-supervised.

Note: tests/gateway/test_status.py::TestReadProcessCmdlinePsFallback::
test_ps_fallback_when_proc_unavailable fails on Windows on current main
before this change as well (POSIX ps fallback asserted on a platform
without it); all other touched suites pass (155 passed, 5 skipped).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../emails/caseyrussell1976@gmail.com         |   1 +
 gateway/status.py                             |  87 ++++
 hermes_cli/gateway.py                         | 171 ++++++-
 hermes_cli/update_cmd.py                      | 454 ++++++++++++++++--
 hermes_cli/update_inventory.py                |  43 +-
 tests/gateway/test_status.py                  |  50 ++
 tests/hermes_cli/test_gateway.py              | 148 ++++++
 .../test_restart_plan_reconciliation.py       |  45 ++
 ...test_update_cold_start_gateway_liveness.py |  10 +-
 .../test_update_concurrent_quarantine.py      | 313 ++++++++++++
 .../test_update_fleet_check_fail_closed.py    |  11 +
 ...ws_gateway_cold_start_desktop_lifecycle.py |   6 +
 ...t_windows_update_restart_reconciliation.py |   8 +-
 13 files changed, 1294 insertions(+), 53 deletions(-)
 create mode 100644 contributors/emails/caseyrussell1976@gmail.com

diff --git a/contributors/emails/caseyrussell1976@gmail.com b/contributors/emails/caseyrussell1976@gmail.com
new file mode 100644
index 0000000000..34cdfe1ed0
--- /dev/null
+++ b/contributors/emails/caseyrussell1976@gmail.com
@@ -0,0 +1 @@
+SmelterLabs
diff --git a/gateway/status.py b/gateway/status.py
index 161cd30e4a..9f2d984f29 100644
--- a/gateway/status.py
+++ b/gateway/status.py
@@ -1034,6 +1034,38 @@ def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool:
             pass
 
 
+def _strict_path_exists(path: Path, label: str) -> bool:
+    try:
+        path.stat()
+        return True
+    except FileNotFoundError:
+        return False
+    except OSError as exc:
+        raise RuntimeError(f"{label} metadata is not inspectable: {exc}") from exc
+
+
+def _is_gateway_runtime_lock_active_strict(lock_path: Path) -> bool:
+    """Probe ownership without treating access failures as absence."""
+    try:
+        handle = open(lock_path, "r+", encoding="utf-8")
+    except FileNotFoundError:
+        return False
+    except OSError as exc:
+        raise RuntimeError(f"gateway runtime lock is not inspectable: {exc}") from exc
+    try:
+        if _try_acquire_file_lock(handle):
+            _release_file_lock(handle)
+            return False
+        return True
+    except OSError as exc:
+        raise RuntimeError(f"gateway runtime lock probe failed: {exc}") from exc
+    finally:
+        try:
+            handle.close()
+        except OSError:
+            pass
+
+
 def write_pid_file() -> None:
     """Write the current process PID and metadata to the gateway PID file.
 
@@ -2356,6 +2388,61 @@ def get_running_pid(
     return None
 
 
+def get_running_pid_identity_strict(pid_path: Path) -> Optional[tuple[int, float]]:
+    """Return a verified process identity or fail on ambiguous runtime state."""
+    resolved_pid_path = Path(pid_path)
+    resolved_lock_path = _get_gateway_lock_path(resolved_pid_path)
+    pid_exists = _strict_path_exists(resolved_pid_path, "gateway PID")
+    lock_exists = _strict_path_exists(resolved_lock_path, "gateway lock")
+    if not pid_exists and not lock_exists:
+        return None
+    if not lock_exists:
+        # No runtime lock can be owned. A stale PID file is not a live gateway.
+        return None
+    if not _is_gateway_runtime_lock_active_strict(resolved_lock_path):
+        # The lock probe is authoritative for absence. Stale or malformed files
+        # may remain after a crash, but no process currently owns this runtime.
+        return None
+    if not pid_exists:
+        raise RuntimeError("active gateway lock has no PID metadata")
+    pid_record = _read_pid_record(resolved_pid_path)
+    lock_record = _read_gateway_lock_record(resolved_lock_path)
+    if not pid_record or not lock_record:
+        raise RuntimeError("gateway PID or lock metadata is malformed")
+    pid = _pid_from_record(pid_record)
+    if pid is None or pid <= 0 or _pid_from_record(lock_record) != pid:
+        raise RuntimeError("gateway PID and lock identities disagree")
+    if not _pid_exists(pid):
+        raise RuntimeError("gateway identity is not live")
+    current_start = _get_process_start_time(pid)
+    starts = (pid_record.get("start_time"), lock_record.get("start_time"))
+    if current_start is None or any(start is None for start in starts):
+        raise RuntimeError("gateway creation time is unavailable")
+    try:
+        current = float(current_start)
+        recorded = tuple(float(start) for start in starts)
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError("gateway creation time is malformed") from exc
+    if current <= 0 or any(start <= 0 or abs(start - current) > 0.001 for start in recorded):
+        raise RuntimeError("gateway process identity changed")
+    if not all(_record_matches_live_gateway_pid(record, pid) for record in (pid_record, lock_record)):
+        raise RuntimeError("runtime metadata does not identify a live gateway")
+    # Windows persists a centisecond fingerprint; SCM ownership checks need the
+    # exact psutil epoch timestamp. Re-read it only after the persisted identity
+    # has been validated, and prove it still rounds to that same fingerprint.
+    if _IS_WINDOWS:
+        try:
+            import psutil  # type: ignore
+
+            exact_create_time = float(psutil.Process(pid).create_time())
+        except Exception as exc:
+            raise RuntimeError("exact gateway creation time is unavailable") from exc
+        if int(round(exact_create_time * 100)) != int(current):
+            raise RuntimeError("gateway process identity changed")
+        return pid, exact_create_time
+    return pid, current
+
+
 def get_running_pid_cached(
     pid_path: Optional[Path] = None,
     *,
diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py
index addd30f4bc..260822b0e8 100644
--- a/hermes_cli/gateway.py
+++ b/hermes_cli/gateway.py
@@ -96,6 +96,21 @@ class ProfileGatewayProcess:
     profile: str
     path: Path
     pid: int
+    create_time: float = 0.0
+
+
+@dataclass(frozen=True)
+class WindowsGatewayService:
+    """A real Windows service supervising a profile gateway process tree."""
+
+    name: str
+    profile: str
+    service_pid: int
+    gateway_pid: int
+    descendant_pids: frozenset[int]
+    descendant_identities: tuple[tuple[int, float], ...]
+    service_create_time: float = 0.0
+    gateway_create_time: float = 0.0
 
 
 def _get_service_pids(all_profiles: bool = False) -> set:
@@ -784,31 +799,177 @@ def find_gateway_pids(
 
 def find_profile_gateway_processes(
     exclude_pids: set | None = None,
+    *,
+    strict: bool = False,
 ) -> list[ProfileGatewayProcess]:
     """Return running gateway PIDs mapped to Hermes profiles via PID files."""
     _exclude = set(exclude_pids or set())
     processes: list[ProfileGatewayProcess] = []
     try:
-        from gateway.status import get_running_pid
+        from gateway.status import get_running_pid, get_running_pid_identity_strict
         from hermes_cli.profiles import list_profiles
     except Exception:
+        if strict:
+            raise
         return processes
 
     seen: set[int] = set()
-    for profile in list_profiles():
+    try:
+        profiles = list_profiles()
+    except Exception:
+        if strict:
+            raise
+        return processes
+    for profile in profiles:
         try:
-            pid = get_running_pid(profile.path / "gateway.pid", cleanup_stale=False)
-        except Exception:
+            if strict:
+                identity = get_running_pid_identity_strict(profile.path / "gateway.pid")
+                pid = identity[0] if identity else None
+                create_time = identity[1] if identity else 0.0
+            else:
+                pid = get_running_pid(profile.path / "gateway.pid", cleanup_stale=False)
+                create_time = 0.0
+        except Exception as exc:
+            if strict:
+                raise RuntimeError(
+                    f"Could not inspect gateway PID for profile {profile.name}"
+                ) from exc
             continue
         if pid is None or pid <= 0 or pid in _exclude or pid in seen:
             continue
         seen.add(pid)
         processes.append(
-            ProfileGatewayProcess(profile=profile.name, path=profile.path, pid=pid)
+            ProfileGatewayProcess(
+                profile=profile.name,
+                path=profile.path,
+                pid=pid,
+                create_time=create_time,
+            )
         )
     return processes
 
 
+def find_windows_gateway_services(
+    *,
+    psutil_module=None,
+    profile_processes: list[ProfileGatewayProcess] | None = None,
+) -> list[WindowsGatewayService]:
+    """Find profile gateways supervised by real Windows services.
+
+    Service-logon processes can deny the interactive Desktop access to their
+    command lines. The updater can still identify them without guessing: a
+    validated profile gateway PID comes from Hermes's own PID file, and its
+    parent chain terminates at a running SCM service PID. The complete service
+    subtree is returned so the Desktop preflight exempts only processes the CLI
+    updater will stop through the Service Control Manager.
+    """
+    if sys.platform != "win32":
+        return []
+    try:
+        if psutil_module is None:
+            import psutil as psutil_module  # type: ignore[no-redef]  # noqa: PLC0415
+        if profile_processes is None:
+            profile_processes = find_profile_gateway_processes(strict=True)
+        service_names_by_pid: dict[int, set[str]] = {}
+        for service in psutil_module.win_service_iter():
+            try:
+                if all(
+                    callable(getattr(service, field, None))
+                    for field in ("name", "status", "pid")
+                ):
+                    service_name = str(service.name() or "")
+                    service_status = service.status()
+                    service_pid = int(service.pid() or 0)
+                else:
+                    data = service.as_dict()
+                    service_name = str(data.get("name") or "")
+                    service_status = data.get("status")
+                    service_pid = int(data.get("pid") or 0)
+            except FileNotFoundError:
+                # The service was deleted between enumeration and inspection;
+                # it cannot still supervise a live gateway tree.
+                continue
+            except Exception as exc:
+                raise RuntimeError("SCM service inspection failed") from exc
+            if not service_name:
+                raise RuntimeError("SCM service has an empty name")
+            if service_status == "stopped":
+                continue
+            if service_status != "running":
+                raise RuntimeError(
+                    f"SCM service {service_name} has indeterminate status: {service_status}"
+                )
+            if service_pid <= 0:
+                raise RuntimeError(
+                    f"Running SCM service {service_name} has no valid process ID"
+                )
+            service_names_by_pid.setdefault(service_pid, set()).add(service_name)
+    except Exception as exc:
+        raise RuntimeError("SCM service enumeration failed") from exc
+
+    found: dict[str, WindowsGatewayService] = {}
+    for profile_process in profile_processes:
+        try:
+            gateway_process = psutil_module.Process(int(profile_process.pid))
+            gateway_create_time = float(gateway_process.create_time())
+            if profile_process.create_time <= 0 or abs(
+                gateway_create_time - profile_process.create_time
+            ) > 0.001:
+                raise RuntimeError("Gateway process identity changed during SCM discovery")
+            ancestor_pids = [int(parent.pid) for parent in gateway_process.parents()]
+            shared_service_pids = [
+                pid
+                for pid in ancestor_pids
+                if len(service_names_by_pid.get(pid, set())) > 1
+            ]
+            if shared_service_pids:
+                raise RuntimeError(
+                    "Gateway ownership is ambiguous under shared SCM host PID(s): "
+                    + ", ".join(str(pid) for pid in shared_service_pids)
+                )
+            service_pid = next(
+                (
+                    pid
+                    for pid in ancestor_pids
+                    if len(service_names_by_pid.get(pid, set())) == 1
+                ),
+                None,
+            )
+            if service_pid is None:
+                continue
+            service_name = next(iter(service_names_by_pid[service_pid]))
+            service_process = psutil_module.Process(service_pid)
+            service_create_time = float(service_process.create_time())
+            descendant_processes = service_process.children(recursive=True)
+            descendants = frozenset(int(child.pid) for child in descendant_processes)
+            if int(profile_process.pid) not in descendants:
+                continue
+            descendant_identities = tuple(
+                sorted(
+                    (int(child.pid), float(child.create_time()))
+                    for child in descendant_processes
+                )
+            )
+            found[service_name] = WindowsGatewayService(
+                name=service_name,
+                profile=str(profile_process.profile),
+                service_pid=service_pid,
+                gateway_pid=int(profile_process.pid),
+                descendant_pids=descendants,
+                descendant_identities=descendant_identities,
+                service_create_time=service_create_time,
+                gateway_create_time=gateway_create_time,
+            )
+        except RuntimeError:
+            raise
+        except Exception as exc:
+            raise RuntimeError(
+                "Could not determine SCM ownership for gateway profile "
+                f"{profile_process.profile}"
+            ) from exc
+    return [found[name] for name in sorted(found)]
+
+
 def _gateway_run_args_for_profile(profile: str) -> list[str]:
     args = [get_python_path(), "-m", "hermes_cli.main"]
     if profile != "default":
diff --git a/hermes_cli/update_cmd.py b/hermes_cli/update_cmd.py
index 7b5b70b404..5fd27ae055 100644
--- a/hermes_cli/update_cmd.py
+++ b/hermes_cli/update_cmd.py
@@ -5033,6 +5033,173 @@ def _desktop_owns_gateway_lifecycle() -> bool:
     return False
 
 
+def _stop_windows_gateway_service(
+    name: str,
+    *,
+    expected_processes: tuple[tuple[int, float], ...] = (),
+    expected_service_identity: tuple[int, float] | None = None,
+    expected_gateway_identity: tuple[int, float] | None = None,
+    timeout: float = 30.0,
+) -> None:
+    """Stop one verified Windows service and wait until SCM reports it down."""
+    import psutil  # noqa: PLC0415
+
+    service = psutil.win_service_get(name)
+    if expected_service_identity is not None:
+        try:
+            current_status = str(service.status())
+            current_service_pid = int(service.pid() or 0)
+        except Exception as exc:
+            raise RuntimeError(
+                f"Windows service {name} SCM identity is unavailable before stop"
+            ) from exc
+        if current_status != "running":
+            raise RuntimeError(
+                f"Windows service {name} is not stably running before stop: {current_status}"
+            )
+        if current_service_pid != int(expected_service_identity[0]):
+            raise RuntimeError(
+                f"Windows service {name} SCM process identity changed before stop"
+            )
+    for label, identity in (
+        ("service", expected_service_identity),
+        ("gateway", expected_gateway_identity),
+    ):
+        if identity is None:
+            continue
+        pid, create_time = identity
+        try:
+            current = float(psutil.Process(int(pid)).create_time())
+        except Exception as exc:
+            raise RuntimeError(
+                f"Windows {label} process identity is unavailable before stop"
+            ) from exc
+        if abs(current - float(create_time)) > 0.001:
+            raise RuntimeError(
+                f"Windows {label} process identity changed before stop"
+            )
+    if expected_service_identity is not None and expected_gateway_identity is not None:
+        service_pid = int(expected_service_identity[0])
+        gateway_pid = int(expected_gateway_identity[0])
+        try:
+            ancestor_pids = {
+                int(parent.pid) for parent in psutil.Process(gateway_pid).parents()
+            }
+        except Exception as exc:
+            raise RuntimeError(
+                "Windows gateway ancestry is unavailable before service stop"
+            ) from exc
+        if service_pid not in ancestor_pids:
+            raise RuntimeError(
+                f"Windows gateway is no longer owned by service {name}"
+            )
+    result = subprocess.run(
+        ["sc.exe", "stop", name],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        timeout=10,
+        check=False,
+    )
+    if result.returncode != 0 and service.status() != "stopped":
+        detail = (result.stderr or result.stdout).strip()
+        raise RuntimeError(detail or f"sc.exe stop failed with {result.returncode}")
+
+    def _original_process_is_alive(pid: int, create_time: float) -> bool:
+        try:
+            current = float(psutil.Process(pid).create_time())
+        except (psutil.NoSuchProcess, psutil.ZombieProcess):
+            # A vanished process is clear.
+            return False
+        except Exception:
+            # AccessDenied or any unknown probe failure stays fail-closed
+            # because the venv may still be locked.
+            return True
+        return abs(current - create_time) <= 0.001
+
+    alive = [
+        pid
+        for pid, create_time in expected_processes
+        if _original_process_is_alive(pid, create_time)
+    ]
+    deadline = _time.monotonic() + timeout
+    while _time.monotonic() < deadline:
+        service_stopped = service.status() == "stopped"
+        alive = [
+            pid
+            for pid, create_time in expected_processes
+            if _original_process_is_alive(pid, create_time)
+        ]
+        if service_stopped and not alive:
+            return
+        _time.sleep(0.2)
+    if service.status() == "stopped":
+        # We only return if the original processes have also exited their identity.
+        # A lingering process with a matching creation time means the venv mutation
+        # must not proceed — fail closed.
+        alive_after_stop = [
+            pid
+            for pid, create_time in expected_processes
+            if _original_process_is_alive(pid, create_time)
+        ]
+        if alive_after_stop:
+            raise RuntimeError(
+                f"Windows service {name} stopped but its process tree is still alive: "
+                f"{alive_after_stop}"
+            )
+        return
+    # If we reach here, the timeout elapsed without the service reaching a stable stopped state
+    # while its original descendants are still alive. Fail closed — venv mutation is unsafe.
+    raise RuntimeError(
+        f"Windows service {name} did not stop within {timeout:.0f}s; venv mutation unsafe."
+    )
+
+
+def _start_windows_gateway_service(name: str, *, timeout: float = 30.0) -> None:
+    """Start one previously paused Windows service and verify it is running."""
+    import psutil  # noqa: PLC0415
+
+    service = psutil.win_service_get(name)
+    result = subprocess.run(
+        ["sc.exe", "start", name],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        timeout=10,
+        check=False,
+    )
+    if result.returncode != 0 and service.status() != "running":
+        detail = (result.stderr or result.stdout).strip()
+        raise RuntimeError(detail or f"sc.exe start failed with {result.returncode}")
+    deadline = _time.monotonic() + timeout
+    while _time.monotonic() < deadline:
+        if service.status() == "running":
+            return
+        _time.sleep(0.2)
+    raise RuntimeError(f"Windows service {name} did not start within {timeout:.0f}s")
+
+
+def _restore_windows_gateway_service(name: str, *, timeout: float = 60.0) -> None:
+    """Restore a service after an uncertain stop, including STOP_PENDING."""
+    import psutil  # noqa: PLC0415
+
+    service = psutil.win_service_get(name)
+    deadline = _time.monotonic() + timeout
+    while _time.monotonic() < deadline:
+        status = service.status()
+        if status == "running":
+            return
+        if status == "stopped":
+            _start_windows_gateway_service(name)
+            return
+        _time.sleep(0.2)
+    raise RuntimeError(
+        f"Windows service {name} did not reach a restorable state within {timeout:.0f}s"
+    )
+
+
 def _pause_windows_gateways_for_update() -> dict | None:
     """Stop running Windows gateways before mutating the checkout or venv.
 
@@ -5051,16 +5218,45 @@ def _pause_windows_gateways_for_update() -> dict | None:
             _get_restart_drain_timeout,
             find_gateway_pids,
             find_profile_gateway_processes,
+            find_windows_gateway_services,
         )
     except Exception as exc:
-        logger.debug("Could not prepare Windows gateway pause for update: %s", exc)
-        return None
+        raise RuntimeError(
+            f"Could not prepare Windows gateway pause for update: {exc}"
+        ) from exc
 
     try:
-        running_pids = list(dict.fromkeys(find_gateway_pids(all_profiles=True)))
+        profile_process_list = find_profile_gateway_processes(strict=True)
+        profile_processes = {proc.pid: proc for proc in profile_process_list}
     except Exception as exc:
-        logger.debug("Could not discover Windows gateway PIDs before update: %s", exc)
-        return None
+        raise RuntimeError(
+            f"Could not map Windows gateway PIDs to profiles: {exc}"
+        ) from exc
+
+    try:
+        service_gateways = find_windows_gateway_services(
+            profile_processes=profile_process_list
+        )
+    except Exception as exc:
+        raise RuntimeError(
+            f"Could not determine Windows gateway service ownership: {exc}"
+        ) from exc
+
+    service_gateway_pids = {int(service.gateway_pid) for service in service_gateways}
+    try:
+        running_pids = list(
+            dict.fromkeys(
+                [
+                    *find_gateway_pids(all_profiles=True),
+                    *sorted(profile_processes),
+                    *sorted(service_gateway_pids),
+                ]
+            )
+        )
+    except Exception as exc:
+        raise RuntimeError(
+            f"Could not discover Windows gateway PIDs before update: {exc}"
+        ) from exc
     if not running_pids:
         # No gateway is running right now, but the user may have installed an
         # autostart entry (Scheduled Task or Startup-folder login item) — that
@@ -5107,18 +5303,12 @@ def _pause_windows_gateways_for_update() -> dict | None:
             )
         return None
 
-    profile_processes = {}
-    try:
-        profile_processes = {
-            proc.pid: proc for proc in find_profile_gateway_processes()
-        }
-    except Exception as exc:
-        logger.debug("Could not map Windows gateway PIDs to profiles: %s", exc)
-
     profiles: dict[str, int] = {}
     mapped_pids = []
     socket_acks: list[dict] = []
     for pid in running_pids:
+        if pid in service_gateway_pids:
+            continue
         proc = profile_processes.get(pid)
         if proc is None:
             continue
@@ -5183,7 +5373,11 @@ def _pause_windows_gateways_for_update() -> dict | None:
         mapped_pids,
         timeout=drain_timeout,
     )
-    unmapped_pids = [pid for pid in running_pids if pid not in profile_processes]
+    unmapped_pids = [
+        pid
+        for pid in running_pids
+        if pid not in profile_processes and pid not in service_gateway_pids
+    ]
 
     # Snapshot each unmapped gateway's command line *before* we force-kill it,
     # so ``_resume_windows_gateways_after_update`` can respawn it by replaying
@@ -5228,14 +5422,76 @@ def _pause_windows_gateways_for_update() -> dict | None:
             # denied, already gone): those still need a manual restart.
             print("    Restart manually after update: hermes gateway run")
 
-    return {
+    token = {
         "resume_needed": True,
         "profiles": profiles,
         "unmapped_pids": unmapped_pids,
         "unmapped": unmapped,
     }
 
-def _cold_start_windows_gateway_after_update() -> None:
+    # Stop SCM-supervised gateways only after every fallible preparation step
+    # for ordinary gateways is complete. From this point to return, any error
+    # restores both the attempted services and the already-paused ordinary
+    # gateways before aborting the update.
+    paused_services = []
+    current_service_name = None
+    try:
+        for service in service_gateways:
+            current_service_name = str(service.name)
+            _stop_windows_gateway_service(
+                current_service_name,
+                expected_processes=tuple(
+                    getattr(service, "descendant_identities", ())
+                ),
+                expected_service_identity=(
+                    int(service.service_pid),
+                    float(service.service_create_time),
+                ),
+                expected_gateway_identity=(
+                    int(service.gateway_pid),
+                    float(service.gateway_create_time),
+                ),
+            )
+            paused_services.append(current_service_name)
+            current_service_name = None
+        if paused_services:
+            token["services"] = paused_services
+            token["expected_services"] = list(paused_services)
+            token["restarted_services"] = []
+            token["service_profiles"] = {
+                str(service.name): str(service.profile)
+                for service in service_gateways
+                if str(service.name) in paused_services
+            }
+            print(
+                "  ✓ Paused Windows gateway service(s): "
+                + ", ".join(paused_services)
+            )
+        return token
+    except Exception as exc:
+        restore_names = []
+        if current_service_name:
+            restore_names.append(current_service_name)
+        restore_names.extend(reversed(paused_services))
+        rollback_failures = []
+        for service_name in dict.fromkeys(restore_names):
+            try:
+                _restore_windows_gateway_service(service_name)
+            except Exception as restore_exc:
+                rollback_failures.append(f"{service_name}: {restore_exc}")
+        if profiles or unmapped:
+            try:
+                _resume_windows_gateways_after_update(token)
+            except Exception as restore_exc:
+                rollback_failures.append(f"ordinary gateways: {restore_exc}")
+        failed_service = current_service_name or "unknown"
+        detail = f"Could not stop Windows gateway service {failed_service}: {exc}"
+        if rollback_failures:
+            detail += "; rollback failures: " + "; ".join(rollback_failures)
+        raise RuntimeError(detail) from exc
+
+
+def _cold_start_windows_gateway_after_update() -> bool:
     """Start a fresh detached gateway after update when one is installed but down.
 
     Invoked from ``_resume_windows_gateways_after_update`` for the
@@ -5258,45 +5514,57 @@ def _cold_start_windows_gateway_after_update() -> None:
     unconditionally from the returned PID.
     """
     if not _m()._is_windows():
-        return
+        return True
     try:
         from hermes_cli import gateway_windows
         from hermes_cli.gateway import find_gateway_pids
     except Exception as exc:
-        logger.debug("Could not load Windows gateway cold-start helpers: %s", exc)
-        return
+        raise RuntimeError(
+            f"Could not load Windows gateway cold-start helpers: {exc}"
+        ) from exc
 
     # Re-check liveness right before spawning — between pause and resume the
     # autostart entry may have already brought a gateway up, or a leftover
     # process may have re-registered. Don't double-start.
     try:
         if list(find_gateway_pids(all_profiles=True)):
-            return
+            return True
     except Exception as exc:
-        logger.debug("Could not re-check gateway liveness before cold-start: %s", exc)
-        return
+        raise RuntimeError(
+            f"Could not re-check gateway liveness before cold-start: {exc}"
+        ) from exc
 
     try:
         if _desktop_owns_gateway_lifecycle():
             logger.debug(
                 "Skipping Windows gateway cold-start: Desktop owns gateway lifecycle"
             )
-            return
+            return True
     except Exception as exc:
-        logger.debug(
-            "Could not re-check Desktop gateway-lifecycle ownership before cold-start: %s",
-            exc,
-        )
+        raise RuntimeError(
+            "Could not re-check Desktop gateway-lifecycle ownership before cold-start: "
+            f"{exc}"
+        ) from exc
 
     try:
         pid = gateway_windows._spawn_detached()
     except Exception as exc:
-        logger.debug("Could not cold-start Windows gateway after update: %s", exc)
-        return
+        raise RuntimeError(f"Could not cold-start Windows gateway after update: {exc}") from exc
+
+    if not pid:
+        raise RuntimeError("Windows gateway cold-start did not return a process ID")
+    ready_pids = gateway_windows._wait_for_gateway_ready()
+    if not ready_pids:
+        raise RuntimeError(
+            f"Windows gateway cold-start PID {pid} did not become ready"
+        )
+    print()
+    print(
+        "✓ Gateway started via cold-start after update "
+        f"(PID: {', '.join(map(str, ready_pids))})"
+    )
+    return True
 
-    if pid:
-        print()
-        gateway_windows._report_gateway_start(f"cold-start after update (PID {pid})")
 
 def _for_each_systemd_gateway_unit(
     list_units_stdout: str,
@@ -5979,21 +6247,60 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
     """Restart Windows profile gateways previously paused for update."""
     if not token or not token.get("resume_needed"):
         return
-    token["resume_needed"] = False
     if not _m()._is_windows():
+        token["resume_needed"] = False
         return
 
     # Regenerate the persisted launcher scripts before respawning anything,
     # so a legacy pythonw-era Scheduled Task / Startup entry comes back on
-    # the current hidden-console design at the next login too.
+    # current hidden-console design at the next login too.
     _m()._refresh_windows_gateway_launchers()
 
+    services = list(token.get("services") or [])
+    token.setdefault("expected_services", list(services))
+    verified_restarts = list(token.get("restarted_services") or [])
+    restarted_services = []
+    failed_services = []
+    for service_name in services:
+        try:
+            _start_windows_gateway_service(str(service_name))
+            restarted_services.append(str(service_name))
+            if str(service_name) not in verified_restarts:
+                verified_restarts.append(str(service_name))
+        except Exception as exc:
+            logger.warning(
+                "Could not restart Windows gateway service %s after update: %s",
+                service_name,
+                exc,
+            )
+            print(f"  ⚠ Could not restart Windows gateway service: {service_name}")
+            failed_services.append(str(service_name))
+
+    if failed_services:
+        token["services"] = failed_services
+        token["restarted_services"] = verified_restarts
+        raise RuntimeError(
+            "Could not restart Windows gateway service(s): "
+            + ", ".join(failed_services)
+        )
+    token["services"] = []
+    token["restarted_services"] = verified_restarts
+    if restarted_services:
+        print()
+        print(
+            "  ✓ Restarted Windows gateway service(s): "
+            + ", ".join(restarted_services)
+        )
+
     profiles = token.get("profiles") or {}
     unmapped = token.get("unmapped") or []
     cold_start = bool(token.get("cold_start_if_installed"))
     if not profiles and not any(u.get("argv") for u in unmapped):
         if cold_start:
-            _m()._cold_start_windows_gateway_after_update()
+            if not _m()._cold_start_windows_gateway_after_update():
+                raise RuntimeError("Windows gateway cold-start was not verified")
+            token["cold_start_if_installed"] = False
+        token["resume_needed"] = False
         return
 
     try:
@@ -6002,20 +6309,25 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
             launch_detached_profile_gateway_restart,
         )
     except Exception as exc:
-        logger.debug("Could not load Windows gateway restart helper: %s", exc)
-        return
+        raise RuntimeError(
+            f"Could not load Windows gateway restart helper: {exc}"
+        ) from exc
 
     relaunched = []
+    failed_profiles = {}
     for profile, old_pid in sorted(profiles.items()):
         try:
             if launch_detached_profile_gateway_restart(str(profile), int(old_pid)):
                 relaunched.append(str(profile))
+            else:
+                failed_profiles[str(profile)] = int(old_pid)
         except Exception as exc:
             logger.debug(
                 "Could not restart Windows gateway profile %s after update: %s",
                 profile,
                 exc,
             )
+            failed_profiles[str(profile)] = int(old_pid)
 
     # Surface the outcome on the token (#91277 Phase 2 plan-vs-execution
     # reconciliation): the git-based update path's fleet reconciliation
@@ -6034,20 +6346,31 @@ def _resume_windows_gateways_after_update(token: dict | None) -> None:
     # Respawn unmapped gateways (no profile→PID-file mapping, e.g. a Scheduled
     # Task) by replaying the argv we snapshotted before force-killing them.
     unmapped_relaunched = 0
+    failed_unmapped = []
     for entry in unmapped:
         argv = entry.get("argv")
         old_pid = entry.get("pid")
         if not argv or not old_pid:
+            failed_unmapped.append(entry)
             continue
         try:
             if launch_detached_gateway_restart_by_cmdline(int(old_pid), list(argv)):
                 unmapped_relaunched += 1
+            else:
+                failed_unmapped.append(entry)
         except Exception as exc:
             logger.debug(
                 "Could not restart unmapped Windows gateway (pid %s) after update: %s",
                 old_pid,
                 exc,
             )
+            failed_unmapped.append(entry)
+
+    token["profiles"] = failed_profiles
+    token["unmapped"] = failed_unmapped
+    if failed_profiles or failed_unmapped:
+        raise RuntimeError("Could not restart every paused Windows gateway")
+    token["resume_needed"] = False
 
     if relaunched:
         print()
@@ -8087,6 +8410,7 @@ def _cmd_update_impl(args, gateway_mode: bool):
             _write_gateway_update_exit_code(desktop_build_ok)
 
         gateway_fleet_restart_incomplete = False
+        gateway_restart_phase_errors: list[str] = []
         # Snapshot of gateways running before we touch anything. Stays empty
         # until we successfully import the probe and are about to stop/drain —
         # so an exception raised before we touch any gateway keeps this empty
@@ -8865,6 +9189,7 @@ def _cmd_update_impl(args, gateway_mode: bool):
 
         except Exception as e:
             logger.debug("Gateway restart during update failed: %s", e)
+            gateway_restart_phase_errors.append(str(e))
             # An exception escaping the whole phase means the drain/restart
             # output the user relies on never printed. Don't let that pass for
             # a clean update: surface it and treat the fleet as stale unless we
@@ -8953,8 +9278,23 @@ def _cmd_update_impl(args, gateway_mode: bool):
             except Exception:
                 pass
 
-        _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
-        if _windows_gateway_resume:
+        try:
+            _m()._resume_windows_gateways_after_update(_windows_gateway_resume)
+        except Exception as _windows_resume_exc:
+            gateway_fleet_restart_incomplete = True
+            gateway_restart_phase_errors.append(str(_windows_resume_exc))
+            print(
+                "  ⚠ Windows gateway service restart incomplete: "
+                f"{_windows_resume_exc}"
+            )
+            if gateway_mode:
+                _exit_code_path = get_hermes_home() / ".update_exit_code"
+                try:
+                    _exit_code_path.write_text("1", encoding="utf-8")
+                except OSError:
+                    pass
+
+        if isinstance(_windows_gateway_resume, dict):
             # Feed Windows's own pause/resume outcome into the same
             # relaunched_profiles bookkeeping the systemd/launchd restart
             # phase populates, so the #91277 Phase 2 reconciliation below
@@ -8976,6 +9316,40 @@ def _cmd_update_impl(args, gateway_mode: bool):
                     "reconciliation bookkeeping: %s",
                     _win_reconcile_exc,
                 )
+            windows_restarted = list(
+                _windows_gateway_resume.get("restarted_services") or []
+            )
+            for service_name in windows_restarted:
+                if service_name not in restarted_services:
+                    restarted_services.append(service_name)
+            service_profiles = _windows_gateway_resume.get("service_profiles") or {}
+            for service_name in windows_restarted:
+                profile_name = service_profiles.get(service_name)
+                if profile_name and profile_name not in relaunched_profiles:
+                    relaunched_profiles.append(profile_name)
+            pending_services = list(_windows_gateway_resume.get("services") or [])
+            for service_name in pending_services:
+                label = str(service_profiles.get(service_name) or service_name)
+                if label not in failed_or_stale_units:
+                    failed_or_stale_units.append(label)
+
+            try:
+                from hermes_cli.update_receipt import record_gateway_restart
+
+                record_gateway_restart(
+                    restarted_services=restarted_services,
+                    relaunched_profiles=relaunched_profiles,
+                    externally_supervised_profiles=externally_supervised_profiles,
+                    killed_pids=sorted(killed_pids),
+                    failed_units=failed_or_stale_units,
+                    incomplete=(
+                        gateway_fleet_restart_incomplete
+                        or bool(failed_or_stale_units)
+                    ),
+                    phase_error="; ".join(gateway_restart_phase_errors) or None,
+                )
+            except Exception:
+                pass
 
         # Warn if legacy Hermes gateway unit files are still installed.
         # When both hermes.service (from a pre-rename install) and the
@@ -9271,6 +9645,8 @@ def _fleet_probe_expected_runtimes(
     if isinstance(windows_resume_token, dict) and (
         windows_resume_token.get("profiles")
         or windows_resume_token.get("unmapped")
+        or windows_resume_token.get("services")
+        or windows_resume_token.get("expected_services")
     ):
         return True
     return False
diff --git a/hermes_cli/update_inventory.py b/hermes_cli/update_inventory.py
index eb0d17abcd..9be2dc8108 100644
--- a/hermes_cli/update_inventory.py
+++ b/hermes_cli/update_inventory.py
@@ -78,8 +78,16 @@ class UpdatePlan:
         return payload
 
 
-def _detect_supervisor_for_pid(pid: int, service_pids: set) -> str:
+def _detect_supervisor_for_pid(
+    pid: int, service_pids: set, windows_service_pids: set | None = None
+) -> str:
     """Classify how a live gateway PID is supervised."""
+    if windows_service_pids and pid in windows_service_pids:
+        # SCM-supervised Windows gateway (WinSW/NSSM/sc.exe create): the
+        # update pause machinery stops the SERVICE via sc.exe instead of
+        # killing the child, so #91277 Phase 2 reconciliation must plan it
+        # under its own mechanism id, not "manual".
+        return "windows-service"
     if pid in service_pids:
         try:
             from hermes_cli.gateway import is_macos, supports_systemd_services
@@ -109,6 +117,8 @@ def _restart_mechanism(supervisor: str, profile: str) -> str:
         return "launchd"
     if supervisor == "desktop":
         return "desktop"
+    if supervisor == "windows-service":
+        return "windows-service"
     if supervisor == "manual-serve":
         return "respawn-argv"
     return "manual"
@@ -122,6 +132,8 @@ def describe_restart_mechanism(mechanism: str, profile: str) -> str:
         return "launchctl kickstart -k (drain-first, per-label domain)"
     if mechanism == "desktop":
         return "Desktop app respawns its serve backend"
+    if mechanism == "windows-service":
+        return "sc.exe stop before venv mutation, sc.exe start after update"
     if mechanism == "respawn-argv":
         return "stop before code swap, relaunch with recorded launch args"
     if profile != "default":
@@ -215,6 +227,23 @@ def collect_runtime_inventory() -> UpdatePlan:
     except Exception as exc:
         logger.debug("Service-PID probe failed: %s", exc)
 
+    # --- SCM-supervised gateway PIDs (Windows) ------------------------------
+    # find_windows_gateway_services() maps validated gateway PIDs through
+    # process ancestry to running SCM service PIDs (no-op off Windows). The
+    # update's pause phase stops these via `sc.exe stop` / restarts via
+    # `sc.exe start`, so the plan must carry the matching mechanism id for
+    # the #91277 Phase 2 reconciliation and the fleet check.
+    windows_service_pids: set = set()
+    try:
+        from hermes_cli.gateway import find_windows_gateway_services
+
+        windows_service_pids = {
+            int(service.gateway_pid)
+            for service in find_windows_gateway_services()
+        }
+    except Exception as exc:
+        logger.debug("Windows SCM service-ownership probe failed: %s", exc)
+
     # --- per-profile gateways (PID files + runtime status stamps) ----------
     seen_pids: set[int] = set()
     try:
@@ -247,7 +276,9 @@ def collect_runtime_inventory() -> UpdatePlan:
                     supervisor = (
                         str(declared)
                         if declared
-                        else _detect_supervisor_for_pid(sock_pid, service_pids)
+                        else _detect_supervisor_for_pid(
+                            sock_pid, service_pids, windows_service_pids
+                        )
                     )
                     sock_sha = identity.get("code_sha")
                     plan.runtimes.append(
@@ -275,7 +306,9 @@ def collect_runtime_inventory() -> UpdatePlan:
             if pid is None or not _pid_exists(pid):
                 continue
             seen_pids.add(pid)
-            supervisor = _detect_supervisor_for_pid(pid, service_pids)
+            supervisor = _detect_supervisor_for_pid(
+                pid, service_pids, windows_service_pids
+            )
             plan.runtimes.append(
                 RuntimeRecord(
                     kind="gateway",
@@ -298,7 +331,9 @@ def collect_runtime_inventory() -> UpdatePlan:
             if proc.pid in seen_pids:
                 continue
             seen_pids.add(proc.pid)
-            supervisor = _detect_supervisor_for_pid(proc.pid, service_pids)
+            supervisor = _detect_supervisor_for_pid(
+                proc.pid, service_pids, windows_service_pids
+            )
             plan.runtimes.append(
                 RuntimeRecord(
                     kind="gateway",
diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py
index b2e926ad02..8306e305e7 100644
--- a/tests/gateway/test_status.py
+++ b/tests/gateway/test_status.py
@@ -1387,3 +1387,53 @@ class TestResolveGatewayLiveness:
         # profile's live gateway from being reported as this profile's.
         assert seen["expected_home"] == profile_dir
 
+
+def test_strict_gateway_identity_returns_none_for_confirmed_absence(tmp_path):
+    assert status.get_running_pid_identity_strict(tmp_path / "gateway.pid") is None
+
+
+def test_strict_gateway_identity_raises_when_metadata_stat_is_denied(
+    tmp_path, monkeypatch
+):
+    pid_path = tmp_path / "gateway.pid"
+    original_stat = Path.stat
+
+    def denied_stat(self, *args, **kwargs):
+        if self == pid_path:
+            raise PermissionError("denied")
+        return original_stat(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "stat", denied_stat)
+
+    with pytest.raises(RuntimeError, match="not inspectable"):
+        status.get_running_pid_identity_strict(pid_path)
+
+
+def test_strict_gateway_identity_raises_on_malformed_active_metadata(
+    tmp_path, monkeypatch
+):
+    pid_path = tmp_path / "gateway.pid"
+    lock_path = tmp_path / "gateway.lock"
+    pid_path.write_text("bad", encoding="utf-8")
+    lock_path.write_text("bad", encoding="utf-8")
+    monkeypatch.setattr(status, "_get_gateway_lock_path", lambda _path=None: lock_path)
+    monkeypatch.setattr(status, "_is_gateway_runtime_lock_active_strict", lambda _path=None: True)
+
+    with pytest.raises(RuntimeError, match="malformed"):
+        status.get_running_pid_identity_strict(pid_path)
+
+
+def test_strict_gateway_identity_rejects_reused_pid(tmp_path, monkeypatch):
+    pid_path = tmp_path / "gateway.pid"
+    lock_path = tmp_path / "gateway.lock"
+    record = {"pid": 123, "start_time": 10.0, "kind": "hermes-gateway"}
+    pid_path.write_text(json.dumps(record), encoding="utf-8")
+    lock_path.write_text(json.dumps(record), encoding="utf-8")
+    monkeypatch.setattr(status, "_get_gateway_lock_path", lambda _path=None: lock_path)
+    monkeypatch.setattr(status, "_is_gateway_runtime_lock_active_strict", lambda _path=None: True)
+    monkeypatch.setattr(status, "_pid_exists", lambda _pid: True)
+    monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 20.0)
+
+    with pytest.raises(RuntimeError, match="identity changed"):
+        status.get_running_pid_identity_strict(pid_path)
+
diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py
index 73fce456d5..02ff059ccc 100644
--- a/tests/hermes_cli/test_gateway.py
+++ b/tests/hermes_cli/test_gateway.py
@@ -1007,3 +1007,151 @@ class TestWindowsScheduledTaskSupervisorGuard:
             monkeypatch.setattr(gateway, "_windows_scheduled_task_state", lambda name, s=state: s)
             assert gateway._windows_scheduled_task_supervises("Hermes_Gateway") is expected, state
             assert gateway._windows_scheduled_task_running("Hermes_Gateway") is (state == "Running")
+
+
+def test_find_windows_gateway_services_maps_verified_pid_tree(monkeypatch):
+    """Only an SCM service whose subtree contains a validated gateway PID is returned."""
+    monkeypatch.setattr(gateway.sys, "platform", "win32")
+    profile = SimpleNamespace(profile="default", pid=300, create_time=300.0)
+
+    class FakeService:
+        def __init__(self, name, pid):
+            self.name = name
+            self.pid = pid
+
+        def as_dict(self):
+            return {
+                "name": self.name,
+                "pid": self.pid,
+                "status": "running",
+            }
+
+    class FakeProcess:
+        def __init__(self, pid):
+            self.pid = pid
+
+        def parents(self):
+            return [FakeProcess(200), FakeProcess(100)]
+
+        def children(self, recursive=False):
+            assert self.pid == 100
+            assert recursive is True
+            return [FakeProcess(200), FakeProcess(300)]
+
+        def create_time(self):
+            return float(self.pid)
+
+    fake_psutil = SimpleNamespace(
+        win_service_iter=lambda: [
+            FakeService("HermesGateway", 100),
+            FakeService("UnrelatedService", 900),
+        ],
+        Process=FakeProcess,
+    )
+
+    result = gateway.find_windows_gateway_services(
+        psutil_module=fake_psutil,
+        profile_processes=[profile],
+    )
+
+    assert result == [
+        gateway.WindowsGatewayService(
+            name="HermesGateway",
+            profile="default",
+            service_pid=100,
+            gateway_pid=300,
+            descendant_pids=frozenset({200, 300}),
+            descendant_identities=((200, 200.0), (300, 300.0)),
+            service_create_time=100.0,
+            gateway_create_time=300.0,
+        )
+    ]
+
+
+def test_find_windows_gateway_services_rejects_shared_service_host_pid(monkeypatch):
+    """A shared host PID cannot prove which service owns the gateway subtree."""
+    monkeypatch.setattr(gateway.sys, "platform", "win32")
+    profile = SimpleNamespace(profile="default", pid=300, create_time=300.0)
+
+    class FakeService:
+        def __init__(self, name):
+            self.name = name
+
+        def as_dict(self):
+            return {"name": self.name, "pid": 100, "status": "running"}
+
+    class FakeProcess:
+        def __init__(self, pid):
+            self.pid = pid
+
+        def parents(self):
+            return [FakeProcess(100)]
+
+        def children(self, recursive=False):
+            return [FakeProcess(300)]
+
+        def create_time(self):
+            return float(self.pid)
+
+    fake_psutil = SimpleNamespace(
+        win_service_iter=lambda: [FakeService("ServiceA"), FakeService("ServiceB")],
+        Process=FakeProcess,
+    )
+
+    with pytest.raises(RuntimeError, match="shared SCM host"):
+        gateway.find_windows_gateway_services(
+            psutil_module=fake_psutil,
+            profile_processes=[profile],
+        )
+
+
+def test_find_windows_gateway_services_fails_closed_on_service_access_error(
+    monkeypatch,
+):
+    monkeypatch.setattr(gateway.sys, "platform", "win32")
+    profile = SimpleNamespace(profile="default", pid=300, create_time=300.0)
+
+    class InaccessibleService:
+        def as_dict(self):
+            raise PermissionError("access denied")
+
+    fake_psutil = SimpleNamespace(
+        win_service_iter=lambda: [InaccessibleService()],
+    )
+
+    with pytest.raises(RuntimeError, match="SCM"):
+        gateway.find_windows_gateway_services(
+            psutil_module=fake_psutil,
+            profile_processes=[profile],
+        )
+
+
+def test_find_windows_gateway_services_fails_closed_when_scm_scan_is_indeterminate(
+    monkeypatch,
+):
+    monkeypatch.setattr(gateway.sys, "platform", "win32")
+    profile = SimpleNamespace(profile="default", pid=300, create_time=300.0)
+    fake_psutil = SimpleNamespace(
+        win_service_iter=lambda: (_ for _ in ()).throw(OSError("SCM unavailable")),
+    )
+
+    with pytest.raises(RuntimeError, match="SCM"):
+        gateway.find_windows_gateway_services(
+            psutil_module=fake_psutil,
+            profile_processes=[profile],
+        )
+
+
+def test_find_profile_gateway_processes_strict_propagates_profile_listing_failure(
+    monkeypatch,
+):
+    import hermes_cli.profiles as profiles_mod
+
+    monkeypatch.setattr(
+        profiles_mod,
+        "list_profiles",
+        lambda: (_ for _ in ()).throw(RuntimeError("profile listing failed")),
+    )
+
+    with pytest.raises(RuntimeError, match="profile listing failed"):
+        gateway.find_profile_gateway_processes(strict=True)
diff --git a/tests/hermes_cli/test_restart_plan_reconciliation.py b/tests/hermes_cli/test_restart_plan_reconciliation.py
index 7c0ecfb1fd..eebe427728 100644
--- a/tests/hermes_cli/test_restart_plan_reconciliation.py
+++ b/tests/hermes_cli/test_restart_plan_reconciliation.py
@@ -40,11 +40,56 @@ def test_mechanism_ids_are_machine_readable_and_described():
     assert _restart_mechanism("launchd", "work") == "launchd"
     assert _restart_mechanism("desktop", "default") == "desktop"
     assert _restart_mechanism("manual", "work") == "manual"
+    assert _restart_mechanism("windows-service", "default") == "windows-service"
     # display derives FROM the id
     assert "systemctl" in describe_restart_mechanism("systemd", "default")
     assert "kickstart" in describe_restart_mechanism("launchd", "work")
     assert "-p work" in describe_restart_mechanism("manual", "work")
     assert describe_restart_mechanism("manual", "default") == "hermes gateway restart"
+    assert "sc.exe" in describe_restart_mechanism("windows-service", "default")
+
+
+def test_windows_service_supervisor_classification():
+    from hermes_cli.update_inventory import _detect_supervisor_for_pid
+
+    # An SCM-owned gateway PID classifies as windows-service even when the
+    # generic service-PID probe also knows the pid.
+    assert (
+        _detect_supervisor_for_pid(41, set(), {41}) == "windows-service"
+    )
+    assert (
+        _detect_supervisor_for_pid(41, {41}, {41}) == "windows-service"
+    )
+    # Without SCM ownership the existing classification is untouched.
+    assert _detect_supervisor_for_pid(42, set(), set()) == "manual"
+    assert _detect_supervisor_for_pid(42, set(), None) == "manual"
+
+
+def test_windows_service_runtime_reconciles_via_service_profiles():
+    # The update path merges the pause token's service_profiles into
+    # relaunched_profiles after sc.exe start — a restarted SCM gateway
+    # must not trip the unaccounted tripwire.
+    outcomes = match_runtime_outcomes(
+        _plan(_rt("default", 500, supervisor="windows-service")),
+        restarted_services=["hermes-gateway"], relaunched_profiles=["default"],
+        externally_supervised_profiles=[], killed_pids=set(), failed_units=[],
+    )
+    assert outcomes == [
+        {"kind": "gateway", "profile": "default", "pid": 500,
+         "mechanism": "windows-service", "outcome": "restarted"}
+    ]
+    assert report_unaccounted_runtimes(outcomes) is False
+
+
+def test_windows_service_runtime_unaccounted_when_restart_fails():
+    outcomes = match_runtime_outcomes(
+        _plan(_rt("work", 501, supervisor="windows-service")),
+        restarted_services=[], relaunched_profiles=[],
+        externally_supervised_profiles=[], killed_pids=set(), failed_units=[],
+    )
+    assert outcomes[0]["mechanism"] == "windows-service"
+    assert outcomes[0]["outcome"] == "unaccounted"
+    assert report_unaccounted_runtimes(outcomes) is True
 
 
 def test_relaunched_profile_is_restarted():
diff --git a/tests/hermes_cli/test_update_cold_start_gateway_liveness.py b/tests/hermes_cli/test_update_cold_start_gateway_liveness.py
index de5a82cd21..5b779291a9 100644
--- a/tests/hermes_cli/test_update_cold_start_gateway_liveness.py
+++ b/tests/hermes_cli/test_update_cold_start_gateway_liveness.py
@@ -10,6 +10,8 @@ same as every other ``_spawn_detached`` caller.
 
 from __future__ import annotations
 
+import pytest
+
 from hermes_cli import gateway as hermes_gateway
 from hermes_cli import gateway_windows
 from hermes_cli import main as cli_main
@@ -37,11 +39,11 @@ def _run_cold_start(monkeypatch, capsys, *, surviving_pids):
     return capsys.readouterr().out
 
 
-def test_cold_start_reports_failure_when_process_does_not_survive(monkeypatch, capsys):
-    out = _run_cold_start(monkeypatch, capsys, surviving_pids=[])
+def test_cold_start_raises_when_process_does_not_survive(monkeypatch, capsys):
+    with pytest.raises(RuntimeError, match="did not become ready"):
+        _run_cold_start(monkeypatch, capsys, surviving_pids=[])
 
-    assert "✓ Starting Windows gateway after update" not in out
-    assert "no process detected" in out
+    assert "✓ Starting Windows gateway after update" not in capsys.readouterr().out
 
 
 def test_cold_start_reports_success_when_process_survives(monkeypatch, capsys):
diff --git a/tests/hermes_cli/test_update_concurrent_quarantine.py b/tests/hermes_cli/test_update_concurrent_quarantine.py
index 143a9eaf98..f23d9d858b 100644
--- a/tests/hermes_cli/test_update_concurrent_quarantine.py
+++ b/tests/hermes_cli/test_update_concurrent_quarantine.py
@@ -204,6 +204,9 @@ def test_pause_windows_gateways_for_update_stops_profile_and_unmapped_pids(
     profile_proc = SimpleNamespace(profile="work", path=profile_home, pid=101)
 
     monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_k: [101, 202])
+    monkeypatch.setattr(
+        gateway_mod, "find_windows_gateway_services", lambda **_k: []
+    )
     monkeypatch.setattr(
         gateway_mod,
         "find_profile_gateway_processes",
@@ -262,6 +265,292 @@ def test_pause_windows_gateways_for_update_stops_profile_and_unmapped_pids(
     assert "Restart manually after update" not in captured
 
 
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_pause_and_resume_windows_gateway_service(
+    _winp,
+    monkeypatch,
+    tmp_path,
+):
+    """A real Windows service is stopped before venv mutation and restarted
+    afterward instead of spawning a competing detached gateway."""
+    import hermes_cli.gateway as gateway_mod
+    import hermes_cli.update_cmd as update_cmd
+
+    profile_home = tmp_path / "profiles" / "default"
+    profile_home.mkdir(parents=True)
+    profile_proc = SimpleNamespace(profile="default", path=profile_home, pid=101)
+    service = SimpleNamespace(
+        name="HermesGateway",
+        profile="default",
+        service_pid=11,
+        service_create_time=11.0,
+        gateway_pid=101,
+        gateway_create_time=101.0,
+        descendant_pids=frozenset({11, 22, 101}),
+        descendant_identities=((22, 22.0), (101, 101.0)),
+    )
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(
+        gateway_mod, "find_profile_gateway_processes", lambda **_k: [profile_proc]
+    )
+    monkeypatch.setattr(
+        gateway_mod,
+        "find_windows_gateway_services",
+        lambda **_k: [service],
+        raising=False,
+    )
+    monkeypatch.setattr(gateway_mod, "_get_restart_drain_timeout", lambda: 0.1)
+
+    stopped = []
+    started = []
+    monkeypatch.setattr(
+        update_cmd,
+        "_stop_windows_gateway_service",
+        lambda name, **_kwargs: stopped.append(name),
+        raising=False,
+    )
+    monkeypatch.setattr(
+        update_cmd,
+        "_start_windows_gateway_service",
+        lambda name: started.append(name),
+        raising=False,
+    )
+    monkeypatch.setattr(cli_main, "_refresh_windows_gateway_launchers", lambda: None)
+    monkeypatch.setattr(
+        cli_main,
+        "_cold_start_windows_gateway_after_update",
+        lambda: (_ for _ in ()).throw(AssertionError("service resume must not cold-start")),
+    )
+
+    token = cli_main._pause_windows_gateways_for_update()
+    assert token == {
+        "resume_needed": True,
+        "profiles": {},
+        "unmapped_pids": [],
+        "unmapped": [],
+        "services": ["HermesGateway"],
+        "expected_services": ["HermesGateway"],
+        "restarted_services": [],
+        "service_profiles": {"HermesGateway": "default"},
+    }
+    assert stopped == ["HermesGateway"]
+
+    cli_main._resume_windows_gateways_after_update(token)
+    assert started == ["HermesGateway"]
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_pause_windows_gateway_service_failure_restores_every_attempted_service(
+    _winp,
+    monkeypatch,
+):
+    """A service that times out after accepting stop is restarted too."""
+    import hermes_cli.gateway as gateway_mod
+    import hermes_cli.update_cmd as update_cmd
+
+    services = [
+        SimpleNamespace(name="HermesGateway", service_pid=11, service_create_time=11.0, gateway_pid=101, gateway_create_time=101.0, descendant_identities=()),
+        SimpleNamespace(name="HermesGatewayPicasso", service_pid=22, service_create_time=22.0, gateway_pid=202, gateway_create_time=202.0, descendant_identities=()),
+    ]
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(
+        gateway_mod, "find_windows_gateway_services", lambda **_k: services
+    )
+
+    def fake_stop(name, **_kwargs):
+        if name == "HermesGatewayPicasso":
+            raise RuntimeError("simulated stop timeout")
+
+    restarted = []
+    monkeypatch.setattr(update_cmd, "_stop_windows_gateway_service", fake_stop)
+    monkeypatch.setattr(
+        update_cmd,
+        "_restore_windows_gateway_service",
+        lambda name: restarted.append(name),
+        raising=False,
+    )
+
+    with pytest.raises(RuntimeError, match="HermesGatewayPicasso"):
+        cli_main._pause_windows_gateways_for_update()
+
+    assert restarted == ["HermesGatewayPicasso", "HermesGateway"]
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_pause_windows_gateway_service_surfaces_rollback_start_failure(
+    _winp,
+    monkeypatch,
+):
+    import hermes_cli.gateway as gateway_mod
+    import hermes_cli.update_cmd as update_cmd
+
+    services = [
+        SimpleNamespace(name="HermesGateway", service_pid=11, service_create_time=11.0, gateway_pid=101, gateway_create_time=101.0, descendant_identities=()),
+        SimpleNamespace(name="HermesGatewayPicasso", service_pid=22, service_create_time=22.0, gateway_pid=202, gateway_create_time=202.0, descendant_identities=()),
+    ]
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(
+        gateway_mod, "find_windows_gateway_services", lambda **_k: services
+    )
+
+    def fake_stop(name, **_kwargs):
+        if name == "HermesGatewayPicasso":
+            raise RuntimeError("simulated stop timeout")
+
+    def fake_start(name):
+        if name == "HermesGateway":
+            raise RuntimeError("simulated rollback start failure")
+
+    monkeypatch.setattr(update_cmd, "_stop_windows_gateway_service", fake_stop)
+    monkeypatch.setattr(
+        update_cmd,
+        "_restore_windows_gateway_service",
+        fake_start,
+        raising=False,
+    )
+
+    with pytest.raises(RuntimeError, match="rollback failures: HermesGateway"):
+        cli_main._pause_windows_gateways_for_update()
+
+
+def test_restore_windows_gateway_service_waits_out_stop_pending(monkeypatch):
+    import hermes_cli.update_cmd as update_cmd
+
+    statuses = iter(["stop_pending", "stopped"])
+    service = SimpleNamespace(status=lambda: next(statuses))
+    fake_psutil = SimpleNamespace(win_service_get=lambda _name: service)
+    restarted = []
+    monkeypatch.setitem(sys.modules, "psutil", fake_psutil)
+    monkeypatch.setattr(update_cmd._time, "sleep", lambda _seconds: None)
+    monkeypatch.setattr(
+        update_cmd,
+        "_start_windows_gateway_service",
+        lambda name: restarted.append(name),
+    )
+
+    update_cmd._restore_windows_gateway_service("HermesGateway")
+
+    assert restarted == ["HermesGateway"]
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_pause_windows_gateways_aborts_when_service_discovery_is_indeterminate(
+    _winp,
+    monkeypatch,
+):
+    import hermes_cli.gateway as gateway_mod
+
+    monkeypatch.setattr(
+        gateway_mod,
+        "find_windows_gateway_services",
+        lambda **_k: (_ for _ in ()).throw(RuntimeError("SCM scan indeterminate")),
+    )
+    monkeypatch.setattr(
+        gateway_mod,
+        "find_gateway_pids",
+        lambda **_k: (_ for _ in ()).throw(
+            AssertionError("ordinary gateway teardown must not begin")
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="SCM scan indeterminate"):
+        cli_main._pause_windows_gateways_for_update()
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_pause_windows_gateways_aborts_when_gateway_pid_discovery_is_indeterminate(
+    _winp,
+    monkeypatch,
+):
+    import hermes_cli.gateway as gateway_mod
+
+    monkeypatch.setattr(gateway_mod, "find_windows_gateway_services", lambda **_k: [])
+    monkeypatch.setattr(
+        gateway_mod,
+        "find_gateway_pids",
+        lambda **_k: (_ for _ in ()).throw(RuntimeError("PID discovery failed")),
+    )
+
+    with pytest.raises(RuntimeError, match="PID discovery failed"):
+        cli_main._pause_windows_gateways_for_update()
+
+
+def test_stop_windows_gateway_service_waits_for_original_descendants(
+    monkeypatch,
+):
+    """SCM STOPPED is insufficient while the original process identity lives."""
+    import hermes_cli.update_cmd as update_cmd
+
+    service = SimpleNamespace(status=lambda: "stopped")
+    fake_psutil = SimpleNamespace(
+        win_service_get=lambda _name: service,
+        Process=lambda pid: SimpleNamespace(create_time=lambda: 12.5),
+    )
+    monkeypatch.setitem(sys.modules, "psutil", fake_psutil)
+    monkeypatch.setattr(
+        update_cmd.subprocess,
+        "run",
+        lambda *_a, **_k: SimpleNamespace(returncode=0, stdout="", stderr=""),
+    )
+    with pytest.raises(RuntimeError, match="process tree"):
+        update_cmd._stop_windows_gateway_service(
+            "HermesGateway",
+            expected_processes=((123, 12.5),),
+            timeout=0,
+        )
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_resume_windows_gateway_service_failure_stays_retryable(
+    _winp,
+    monkeypatch,
+):
+    import hermes_cli.update_cmd as update_cmd
+
+    token = {
+        "resume_needed": True,
+        "profiles": {},
+        "unmapped": [],
+        "services": ["HermesGateway"],
+    }
+    monkeypatch.setattr(cli_main, "_refresh_windows_gateway_launchers", lambda: None)
+    monkeypatch.setattr(
+        update_cmd,
+        "_start_windows_gateway_service",
+        lambda _name: (_ for _ in ()).throw(RuntimeError("simulated start failure")),
+    )
+
+    with pytest.raises(RuntimeError, match="HermesGateway"):
+        cli_main._resume_windows_gateways_after_update(token)
+
+    assert token["resume_needed"] is True
+    assert token["services"] == ["HermesGateway"]
+
+
+@patch.object(cli_main, "_is_windows", return_value=True)
+def test_resume_windows_gateway_launcher_refresh_failure_stays_retryable(
+    _winp,
+    monkeypatch,
+):
+    token = {
+        "resume_needed": True,
+        "profiles": {},
+        "unmapped": [],
+        "services": ["HermesGateway"],
+    }
+    monkeypatch.setattr(
+        cli_main,
+        "_refresh_windows_gateway_launchers",
+        lambda: (_ for _ in ()).throw(RuntimeError("refresh failed")),
+    )
+
+    with pytest.raises(RuntimeError, match="refresh failed"):
+        cli_main._resume_windows_gateways_after_update(token)
+
+    assert token["resume_needed"] is True
+    assert token["services"] == ["HermesGateway"]
+
+
 # ---------------------------------------------------------------------------
 # venv-side launcher ancestors (the uv launcher/worker split)
 #
@@ -370,6 +659,9 @@ def test_pause_kill_set_covers_venv_guard_abort_set(
     )
 
     monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda **_k: [worker_pid])
+    monkeypatch.setattr(
+        gateway_mod, "find_windows_gateway_services", lambda **_k: []
+    )
     monkeypatch.setattr(
         gateway_mod, "find_profile_gateway_processes", lambda **_k: [profile_proc]
     )
@@ -720,5 +1012,26 @@ def test_update_gate_still_aborts_on_non_gateway_concurrent(
     assert "--force" in captured
 
 
+def test_stop_service_refuses_pid_reuse_before_sc_stop(monkeypatch):
+    import hermes_cli.update_cmd as update_cmd
+
+    fake_psutil = SimpleNamespace(
+        win_service_get=lambda _name: SimpleNamespace(
+            status=lambda: "running", pid=lambda: 11
+        ),
+        Process=lambda _pid: SimpleNamespace(create_time=lambda: 99.0),
+    )
+    calls = []
+    monkeypatch.setitem(sys.modules, "psutil", fake_psutil)
+    monkeypatch.setattr(update_cmd.subprocess, "run", lambda *_a, **_k: calls.append(True))
+
+    with pytest.raises(RuntimeError, match="identity changed"):
+        update_cmd._stop_windows_gateway_service(
+            "HermesGateway", expected_service_identity=(11, 11.0)
+        )
+
+    assert calls == []
+
+
 
 
diff --git a/tests/hermes_cli/test_update_fleet_check_fail_closed.py b/tests/hermes_cli/test_update_fleet_check_fail_closed.py
index 8ec1781d6e..ab71d5e896 100644
--- a/tests/hermes_cli/test_update_fleet_check_fail_closed.py
+++ b/tests/hermes_cli/test_update_fleet_check_fail_closed.py
@@ -64,6 +64,17 @@ class TestEmptySnapshotFailClosed:
             _fleet_probe_expected_runtimes(None, [], token, [], set()) is True
         )
 
+    def test_incomplete_when_windows_resume_token_has_services(self):
+        token = {
+            "resume_needed": False,
+            "profiles": {},
+            "unmapped": [],
+            "services": ["HermesGateway"],
+        }
+        assert (
+            _fleet_probe_expected_runtimes(None, [], token, [], set()) is True
+        )
+
     def test_incomplete_when_restart_phase_touched_gateways(self):
         # The original #93410 signal still counts.
         assert (
diff --git a/tests/hermes_cli/test_windows_gateway_cold_start_desktop_lifecycle.py b/tests/hermes_cli/test_windows_gateway_cold_start_desktop_lifecycle.py
index 196c54c13a..bc359d5c19 100644
--- a/tests/hermes_cli/test_windows_gateway_cold_start_desktop_lifecycle.py
+++ b/tests/hermes_cli/test_windows_gateway_cold_start_desktop_lifecycle.py
@@ -87,6 +87,9 @@ def test_orphaned_control_plane_does_not_own_lifecycle(monkeypatch):
 def test_pause_skips_cold_start_plan_when_desktop_owns_lifecycle(monkeypatch):
     monkeypatch.setattr(cli_main, "_is_windows", lambda: True)
     monkeypatch.setattr(hermes_gateway, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(
+        hermes_gateway, "find_windows_gateway_services", lambda **_k: []
+    )
     monkeypatch.setattr(gateway_windows, "is_installed", lambda: True)
     monkeypatch.setattr(update_cmd, "_desktop_owns_gateway_lifecycle", lambda: True)
 
@@ -96,6 +99,9 @@ def test_pause_skips_cold_start_plan_when_desktop_owns_lifecycle(monkeypatch):
 def test_pause_still_cold_starts_when_autostart_and_no_desktop_owner(monkeypatch):
     monkeypatch.setattr(cli_main, "_is_windows", lambda: True)
     monkeypatch.setattr(hermes_gateway, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(
+        hermes_gateway, "find_windows_gateway_services", lambda **_k: []
+    )
     monkeypatch.setattr(gateway_windows, "is_installed", lambda: True)
     monkeypatch.setattr(update_cmd, "_desktop_owns_gateway_lifecycle", lambda: False)
 
diff --git a/tests/hermes_cli/test_windows_update_restart_reconciliation.py b/tests/hermes_cli/test_windows_update_restart_reconciliation.py
index 89737bcf6b..0de2d6bbd9 100644
--- a/tests/hermes_cli/test_windows_update_restart_reconciliation.py
+++ b/tests/hermes_cli/test_windows_update_restart_reconciliation.py
@@ -21,6 +21,8 @@ reconciliation runs (mirrored here directly, since driving the full
 
 from unittest.mock import patch
 
+import pytest
+
 import hermes_cli.gateway as gateway
 import hermes_cli.main as hm
 from hermes_cli.update_cmd import _resume_windows_gateways_after_update
@@ -77,7 +79,11 @@ def test_resume_omits_profiles_whose_relaunch_failed(monkeypatch):
 
     token = _token({"default": 1111, "work": 2222})
     with patch("builtins.print"):
-        _resume_windows_gateways_after_update(token)
+        # Fail-closed contract: a profile whose relaunch failed
+        # raises so the update is marked incomplete (the caller catches,
+        # records the phase error, and exits 1 in gateway mode).
+        with pytest.raises(RuntimeError, match="Could not restart every paused"):
+            _resume_windows_gateways_after_update(token)
 
     assert token["relaunched_profiles"] == ["default"]