fix(update): verify updater-owning gateway restart out of cgroup

gateway/run.py

@@ -1702,6 +1702,31 @@ def _clear_planned_restart_notification() -> None:
 from utils import atomic_json_write, is_truthy_value
 _hermes_home = get_hermes_home()
 
+
+def _current_systemd_gateway_service() -> str | None:
+    """Return the exact gateway service component containing this process."""
+    from hermes_cli.update_owner_restart import current_systemd_gateway_service
+
+    return current_systemd_gateway_service()
+
+
+def _read_update_terminal_exit_code(home: Path) -> int | None:
+    """Read a generic exit marker or the atomic owner-restart result."""
+    exit_code_path = Path(home) / ".update_exit_code"
+    if exit_code_path.exists():
+        try:
+            return int(exit_code_path.read_text(encoding="utf-8").strip() or "1")
+        except (OSError, TypeError, ValueError):
+            return 1
+    try:
+        from hermes_cli.update_owner_restart import (
+            read_owner_restart_result_exit_code,
+        )
+
+        return read_owner_restart_result_exit_code(Path(home))
+    except Exception:
+        return None
+
 # Load environment variables from ~/.hermes/.env first.
 # User-managed env files should override stale shell exports on restart.
 from dotenv import load_dotenv  # noqa: F401  # backward-compat for tests that monkeypatch this symbol
@@ -11271,6 +11296,25 @@ async def start(self) -> bool:
             logger.info("Channel directory built: %d target(s)", ch_count)
         except Exception as e:
             logger.warning("Channel directory build failed: %s", e)
+
+        # A transient verifier in a separate systemd cgroup owns the terminal
+        # update result.  Acknowledge only after this new owner has completed
+        # adapter/config startup and published its running runtime state.  The
+        # verifier independently checks ActiveState, MainPID, and start
+        # generation before it exposes .update_exit_code.
+        try:
+            from hermes_cli.update_owner_restart import (
+                acknowledge_owner_restart_ready,
+            )
+
+            if acknowledge_owner_restart_ready(
+                _hermes_home,
+                current_service=_current_systemd_gateway_service(),
+                current_pid=os.getpid(),
+            ):
+                logger.info("Acknowledged updater-owner restart readiness")
+        except Exception as exc:
+            logger.warning("Could not acknowledge updater-owner readiness: %s", exc)
 
         # Check if we're restarting after a /update command. If the update is
         # still running, keep watching so we notify once it actually finishes.
@@ -20565,6 +20609,7 @@ async def _watch_update_progress(
         claimed_path = _hermes_home / ".update_pending.claimed.json"
         output_path = _hermes_home / ".update_output.txt"
         exit_code_path = _hermes_home / ".update_exit_code"
+        owner_result_path = _hermes_home / ".update_owner_restart_result.json"
         prompt_path = _hermes_home / ".update_prompt.json"
 
         loop = asyncio.get_running_loop()
@@ -20613,10 +20658,15 @@ async def _watch_update_progress(
             # after the first completion check — otherwise a platform that
             # reconnects a few seconds after completion never gets notified.
             while (pending_path.exists() or claimed_path.exists()) and loop.time() < deadline:
-                if exit_code_path.exists() and await self._send_update_notification():
+                if (
+                    _read_update_terminal_exit_code(_hermes_home) is not None
+                    and await self._send_update_notification()
+                ):
                     return
                 await asyncio.sleep(poll_interval)
-            if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists():
+            if (
+                pending_path.exists() or claimed_path.exists()
+            ) and _read_update_terminal_exit_code(_hermes_home) is None:
                 exit_code_path.write_text("124", encoding="utf-8")
                 await self._send_update_notification()
             return
@@ -20655,8 +20705,10 @@ async def _flush_buffer() -> None:
                     logger.debug("Update stream send failed: %s", e)
 
         while loop.time() < deadline:
-            # Check for completion
-            if exit_code_path.exists():
+            # Check for completion.  Owner-managed updates commit their atomic
+            # result JSON before best-effort projection to .update_exit_code.
+            terminal_exit_code = _read_update_terminal_exit_code(_hermes_home)
+            if terminal_exit_code is not None:
                 # Read any remaining output
                 if output_path.exists():
                     try:
@@ -20670,8 +20722,7 @@ async def _flush_buffer() -> None:
 
                 # Send final status
                 try:
-                    exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1"
-                    exit_code = int(exit_code_raw)
+                    exit_code = terminal_exit_code
                     if exit_code == 0:
                         await adapter.send(
                             chat_id,
@@ -20689,8 +20740,14 @@ async def _flush_buffer() -> None:
                     logger.warning("Update final notification failed: %s", e)
 
                 # Cleanup
-                for p in (pending_path, claimed_path, output_path,
-                          exit_code_path, prompt_path):
+                for p in (
+                    pending_path,
+                    claimed_path,
+                    output_path,
+                    exit_code_path,
+                    owner_result_path,
+                    prompt_path,
+                ):
                     p.unlink(missing_ok=True)
                 (_hermes_home / ".update_response").unlink(missing_ok=True)
                 _up_done = self._peek_session_state(session_key)
@@ -20773,7 +20830,7 @@ async def _flush_buffer() -> None:
             await asyncio.sleep(poll_interval)
 
         # Timeout
-        if not exit_code_path.exists():
+        if _read_update_terminal_exit_code(_hermes_home) is None:
             logger.warning("Update watcher timed out after %.0fs", timeout)
             exit_code_path.write_text("124", encoding="utf-8")
             await _flush_buffer()
@@ -20785,8 +20842,14 @@ async def _flush_buffer() -> None:
                 )
             except Exception:
                 pass
-            for p in (pending_path, claimed_path, output_path,
-                      exit_code_path, prompt_path):
+            for p in (
+                pending_path,
+                claimed_path,
+                output_path,
+                exit_code_path,
+                owner_result_path,
+                prompt_path,
+            ):
                 p.unlink(missing_ok=True)
             (_hermes_home / ".update_response").unlink(missing_ok=True)
             _up_timeout_state = self._peek_session_state(session_key)
@@ -20807,6 +20870,7 @@ async def _send_update_notification(self) -> bool:
         claimed_path = _hermes_home / ".update_pending.claimed.json"
         output_path = _hermes_home / ".update_output.txt"
         exit_code_path = _hermes_home / ".update_exit_code"
+        owner_result_path = _hermes_home / ".update_owner_restart_result.json"
 
         if not pending_path.exists() and not claimed_path.exists():
             return False
@@ -20830,16 +20894,14 @@ async def _send_update_notification(self) -> bool:
             thread_id = pending.get("thread_id")
             message_id = pending.get("message_id")
 
-            if not exit_code_path.exists():
+            exit_code = _read_update_terminal_exit_code(_hermes_home)
+            if exit_code is None:
                 logger.info("Update notification deferred: update still running")
                 cleanup = False
                 active_pending_path = pending_path
                 claimed_path.replace(pending_path)
                 return False
 
-            exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1"
-            exit_code = int(exit_code_raw)
-
             # Read the captured update output
             output = ""
             if output_path.exists():
@@ -20909,6 +20971,7 @@ async def _send_update_notification(self) -> bool:
                 claimed_path.unlink(missing_ok=True)
                 output_path.unlink(missing_ok=True)
                 exit_code_path.unlink(missing_ok=True)
+                owner_result_path.unlink(missing_ok=True)
 
         return True
 

gateway/slash_commands.py

@@ -5427,6 +5427,12 @@ async def _handle_update_command(self, event: MessageEvent) -> str:
         pending_path = _hermes_home / ".update_pending.json"
         output_path = _hermes_home / ".update_output.txt"
         exit_code_path = _hermes_home / ".update_exit_code"
+        owner_restart_pending_path = (
+            _hermes_home / ".update_owner_restart_pending.json"
+        )
+        owner_restart_result_path = (
+            _hermes_home / ".update_owner_restart_result.json"
+        )
         session_key = self._session_key_for_source(event.source)
         pending = {
             "platform": event.source.platform.value,
@@ -5444,6 +5450,8 @@ async def _handle_update_command(self, event: MessageEvent) -> str:
         _tmp_pending.write_text(json.dumps(pending), encoding="utf-8")
         _tmp_pending.replace(pending_path)
         exit_code_path.unlink(missing_ok=True)
+        owner_restart_pending_path.unlink(missing_ok=True)
+        owner_restart_result_path.unlink(missing_ok=True)
 
         # Spawn `hermes update --gateway` detached so it survives gateway restart.
         # --gateway enables file-based IPC for interactive prompts (stash
@@ -5510,7 +5518,13 @@ async def _handle_update_command(self, event: MessageEvent) -> str:
                     # in zsh, and this command string is copied/reused in macOS/zsh
                     # operator wrappers. Keep the template zsh-safe even though this
                     # specific subprocess currently runs under bash.
-                    f"rc=$?; printf '%s' \"$rc\" > {shlex.quote(str(exit_code_path))}"
+                    #
+                    # A same-cgroup updater delegates the terminal restart and
+                    # result marker to an external verifier.  Never publish the
+                    # updater shell's return code while that proof is pending.
+                    f"rc=$?; if [ ! -e {shlex.quote(str(owner_restart_pending_path))} ] "
+                    f"&& [ ! -e {shlex.quote(str(owner_restart_result_path))} ]; "
+                    f"then printf '%s' \"$rc\" > {shlex.quote(str(exit_code_path))}; fi"
                 )
                 setsid_bin = shutil.which("setsid")
                 if setsid_bin:

hermes_cli/dashboard_procs.py

@@ -209,13 +209,34 @@ def _kill_stale_dashboard_processes(
     # stay a stop, not a restart.
     pid_cgroup: dict[int, str | None] = {}
     pid_service: dict[int, str | None] = {}
+    pid_service_snapshot: dict[int, dict] = {}
     pid_cmdline: dict[int, list[str]] = {}
     if restart_managed and sys.platform != "win32":
         for pid in pids:
             cg_path = _m()._get_pid_cgroup_path(pid)
             pid_cgroup[pid] = cg_path
             pid_service[pid] = _m()._get_systemd_service_for_pid(pid)
-            if not pid_service[pid]:
+            if pid_service[pid]:
+                scope_name = _m()._extract_scope_from_cgroup(cg_path)
+                if scope_name in {"user", "system"}:
+                    scope = ("--user",) if scope_name == "user" else ()
+                    state = _m()._read_managed_dashboard_service_state(
+                        scope, pid_service[pid]
+                    )
+                    # The PID-to-cgroup proof is stronger than a transient
+                    # is-active read failure.  Preserve the killed PID as the
+                    # old generation so post-restart verification cannot pass
+                    # on an unchanged service process.
+                    state["active"] = True
+                    if int(state.get("main_pid") or 0) <= 0:
+                        state["main_pid"] = pid
+                    state.update(
+                        scope=scope,
+                        unit=pid_service[pid],
+                        runtime=None,
+                    )
+                    pid_service_snapshot[pid] = state
+            else:
                 # Manually-started process: preserve its exact argv so we
                 # can respawn it after the update (#40449, #68934).
                 cmdline = _m()._dashboard_cmdline_for_pid(pid)
@@ -295,6 +316,8 @@ def _kill_stale_dashboard_processes(
     #  - manually-started PIDs: respawn the argv captured before the kill
     #    (#40449) — detached, headless, logged to logs/dashboard-restart.log.
     restarted_services: list[str] = []
+    restarted_service_snapshots: list[dict] = []
+    unverified_services: list[str] = []
     unrecovered: list[int] = []
     if killed and restart_managed:
         failed_restarts: list[tuple[str, str]] = []
@@ -308,6 +331,11 @@ def _kill_stale_dashboard_processes(
                 seen_services.add(svc_name)
                 if _m()._try_restart_systemd_service(svc_name, pid_cgroup.get(pid)):
                     restarted_services.append(svc_name)
+                    snapshot = pid_service_snapshot.get(pid)
+                    if snapshot is not None:
+                        restarted_service_snapshots.append(snapshot)
+                    else:
+                        unverified_services.append(svc_name)
                 else:
                     failed_restarts.append((svc_name, "systemctl restart returned non-zero"))
                     unrecovered.append(pid)
@@ -339,6 +367,8 @@ def _kill_stale_dashboard_processes(
         "killed": list(killed),
         "failed": list(failed),
         "unrecovered": list(unrecovered),
+        "restarted_service_snapshots": list(restarted_service_snapshots),
+        "unverified_services": list(unverified_services),
     }
 
 def _detect_concurrent_hermes_instances(