fix(update): a killed Windows updater no longer strands paused gateways, and never resumes them onto a torn tree

gateway/status.py

@@ -21,6 +21,11 @@
 from pathlib import Path
 from typing import Any, Callable, NamedTuple, Optional
 
+from gateway.status_inline_source import (
+    command_line_runs_inline_source,
+    inline_bootstrap_argv,
+    inline_source_flag_index,
+)
 from hermes_constants import _get_platform_default_hermes_home, get_hermes_home, get_process_hermes_home
 from hermes_cli._subprocess_compat import pid_exists_stdlib
 from utils import atomic_json_write
@@ -529,129 +534,6 @@ def _read_process_cmdline(pid: int) -> Optional[str]:
     return None
 
 
-
-def inline_source_flag_index(tokens: list[str]) -> int | None:
-    """Index of the ``-c`` token when *tokens* is an interpreter running INLINE SOURCE, else None.
-
-    Everything after ``-c`` is data the inline program receives, not this process's own identity.
-    The detached gateway restart watcher (``gateway._spawn_gateway_restart_watcher``) is spawned as
-    ``python -c <watcher source> <old_pid> <python> -m hermes_cli.main gateway run``: its trailing
-    argv is the command the watcher will LATER spawn, so every argv matcher used to read it as a
-    live gateway. See #107002 and the "never infer process identity from argv substrings" rule.
-
-    Only interpreter options may precede ``-c``; the first non-option token ends the option block
-    (``python -m hermes_cli.main …`` therefore never matches).
-
-    The walk is VALUE-AWARE: ``-X``/``-W``/``-Q`` and ``--check-hash-based-pycs``/``--jit`` take a
-    SEPARATE operand, so a naive "first non-option token ends the block" walk mistakes that operand
-    for the end of the block and never reaches the ``-c`` behind it (``python -X utf8 -c <src> …``
-    was still read as a live gateway). The operand sets are the canonical ones in
-    ``hermes_state_holders``, not a second hand-rolled copy.
-
-    *tokens* must be CASE-PRESERVING: the operand-taking ``-Q``/``-W``/``-X`` differ from the
-    operand-less ``-q``/``-b``, so a lowercased argv would skip the token after a plain ``-q``.
-    """
-    from hermes_state_holders import (
-        _PYTHON_LONG_OPTIONS_WITH_OPERANDS,
-        _PYTHON_SHORT_OPTIONS_WITH_OPERANDS,
-    )
-
-    index = 1
-    while index < len(tokens):
-        token = tokens[index]
-        if token == "--":
-            return None
-        if token in _PYTHON_LONG_OPTIONS_WITH_OPERANDS:
-            index += 2  # the next token is this option's operand, not the end of the option block
-            continue
-        if token.startswith("--"):
-            index += 1  # ``--opt=value`` and operand-less long options
-            continue
-        if not token.startswith("-") or token == "-":
-            return None
-        # Clustered short options (``-uc``, ``-IsB``). An operand-taking letter consumes the rest of
-        # the cluster as its attached value, or the following token when the cluster ends there --
-        # so ``-Xc`` is ``-X c``, NOT an inline-source ``-c``.
-        cluster = token[1:]
-        for position, letter in enumerate(cluster):
-            if letter == "c":
-                return index
-            if letter in _PYTHON_SHORT_OPTIONS_WITH_OPERANDS:
-                index += 1 if cluster[position + 1 :] else 2
-                break
-        else:
-            index += 1
-    return None
-
-
-def command_line_runs_inline_source(tokens: list[str]) -> bool:
-    """True when *tokens* is an interpreter running INLINE SOURCE (``python -c <src> [args]``)."""
-    return inline_source_flag_index(tokens) is not None
-
-
-# Hermes' own inline bootstraps hand control to a Hermes entry point IN this process, so the argv
-# they run with is this process's own identity; every other ``-c`` program keeps its trailing argv
-# as data (#107002). Each pattern is one emitted source shape, anchored at both ends so a program
-# merely CARRYING a bootstrap command line (the restart watcher's respawn argv) never matches.
-_Q = r"""['"]?"""
-_MAIN = rf"{_Q}__main__{_Q}"
-_RUN_MODULE = rf"runpy\.run_module\(\s*{_Q}(?P<target>[\w.]+){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*,\s*alter_sys\s*=\s*True\s*\)"
-_BOOTSTRAPS = (
-    # hermes_cli._launchers.runtime_command (store launcher, the Windows updater's relaunch)
-    ("module", re.compile(rf"import os, sys, runpy;.*\b{_RUN_MODULE}", re.S)),
-    # hermes_cli.venv_sync.relaunch_command: argv is assigned inside the source
-    ("module", re.compile(rf"import sys, runpy; sys\.path\.insert\(.*\b{_RUN_MODULE}", re.S)),
-    ("path", re.compile(
-        rf"import sys, runpy; sys\.path\.insert\(.*\brunpy\.run_path\(\s*{_Q}(?P<target>[^'\"]+?){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*\)",
-        re.S)),
-    # hermes_cli._launchers._launcher_script (the published POSIX shell / Windows .cmd launcher)
-    ("entry", re.compile(r"import os, re, sys\s.*\bfrom\s+(?P<target>[\w.]+)\s+import\s+(?P<func>\w+)\b.*\bsys\.exit\(\s*(?P=func)\(\)\s*\)", re.S)),
-    # hermes_cli._launchers._write_cmd_launcher: the launcher script, base64-encoded
-    ("base64", re.compile(rf"import base64; exec\(base64\.b64decode\({_Q}(?P<target>[A-Za-z0-9+/=]+){_Q}\)\)")),
-)
-_ASSIGNED_ARGV = re.compile(r"\bsys\.argv\s*=\s*\[(.*?)\]\s*;")
-
-
-def _bootstrap_entry(source: str, argv: list[str]) -> list[str] | None:
-    """``[-m, <module>, *argv]`` (or ``[<path>, *argv]``) the inline *source* runs in-process, else None."""
-    source = source.strip()
-    kind, match = next(((k, m) for k, p in _BOOTSTRAPS if (m := p.fullmatch(source))), (None, None))
-    if match is None:
-        return None
-    target = match["target"]
-    if kind == "base64":
-        import base64
-        import binascii
-        try:
-            return _bootstrap_entry(base64.b64decode(target, validate=True).decode("utf-8"), argv)
-        except (binascii.Error, UnicodeDecodeError):
-            return None
-    if kind == "entry":  # the launcher script's own ``--run-module <module>`` switch
-        return ["-m", argv[1], *argv[2:]] if argv[:1] == ["--run-module"] and len(argv) > 1 else ["-m", target, *argv]
-    if assigned := _ASSIGNED_ARGV.search(source):
-        argv = [item.strip().strip("'\"") for item in assigned.group(1).split(",")][1:]
-    return [target, *argv] if kind == "path" else ["-m", target, *argv]
-
-
-def inline_bootstrap_argv(tokens: list[str]) -> list[str] | None:
-    """*tokens* as the equivalent ``python -m <module> <argv…>`` when this interpreter's ``-c`` source
-    is a Hermes bootstrap running an entry point in-process; None for any other inline source.
-
-    Command lines usually arrive space-joined (``/proc``, psutil, ``ps``), which splits the source
-    across tokens; the shortest token run that ends in a recognised tail is the source, whatever
-    joined it, and the tokens after it are the entry point's argv.
-    """
-    index = inline_source_flag_index(tokens)
-    if index is None:
-        return None
-    for end in range(index + 1, len(tokens)):
-        if tokens[end].rstrip().endswith(")"):
-            entry = _bootstrap_entry(" ".join(tokens[index + 1 : end + 1]), tokens[end + 1 :])
-            if entry is not None:
-                return [tokens[0], *entry]
-    return None
-
-
 def _gateway_command_subcommand(command: str | None) -> str | None:
     """Hermes gateway lifecycle subcommand from a command line, or None. No loose substring matches
     (``"gateway" in cmdline`` also matched ``gateway status`` / ``python -m tui_gateway``): needs a
@@ -1791,13 +1673,17 @@ def _marker_is_stale(written_at: str, ttl_s: int) -> bool:
 
 def _read_live_pid_marker(path: Path, ttl_s: int) -> Optional[tuple[dict[str, Any], int, Any]]:
     """``(record, target_pid, target_start_time)`` for a usable marker, else None. Malformed/expired
-    markers can never match anyone, so they are unlinked here (must not wedge a new instance)."""
+    markers can never match anyone, so they are unlinked here (must not wedge a new instance) --
+    except a request stamped ``accepted`` (``update_pause_record.mark_stop_accepted``): expired, it
+    matches nobody, but it can be the only trace that a still-draining gateway accepted an update's
+    stop, so it stays until that pause is settled."""
     record = _read_json_file(path)
     if not record:
         return None
     target_pid = _pid_from_record(record, "target_pid")
     if target_pid is None or _marker_is_stale(record.get("written_at") or "", ttl_s):
-        _unlink_quietly(path)
+        if record.get("accepted") is not True or not _update_pause_on_disk():
+            _unlink_quietly(path)
         return None
     return record, target_pid, record.get("target_start_time")
 
@@ -1814,7 +1700,7 @@ def _pid_marker_names_self(target_pid: int, target_start_time: Any) -> bool:
     return None in (target_start_time, our_start_time) or target_start_time == our_start_time
 
 
-def _consume_pid_marker_for_self(path: Path, *, ttl_s: int) -> bool:
+def _consume_pid_marker_for_self(path: Path, *, ttl_s: int, on_consume=None, keep: bool = False) -> bool:
     parsed = _read_live_pid_marker(path, ttl_s)
     if parsed is None:
         return False
@@ -1833,7 +1719,10 @@ def _consume_pid_marker_for_self(path: Path, *, ttl_s: int) -> bool:
         if replacer_home is not None and not _same_hermes_home(replacer_home, our_home):
             return False
     matches = _pid_marker_names_self(target_pid, target_start_time)
-    _unlink_quietly(path)
+    if matches and on_consume is not None:
+        on_consume(path, record)
+    if not keep:
+        _unlink_quietly(path)
     return matches
 
 
@@ -2069,11 +1958,55 @@ def write_planned_stop_marker(target_pid: int) -> bool:
     })
 
 
+_PLANNED_STOP_MUTEX_WAIT_S = 2.0
+
+
+def _update_pause_on_disk() -> bool:
+    """A Windows update pause of this checkout is on disk (its record or a recovery's claim), or the
+    record directory cannot be listed. Only then does a consume need the pause lock and a checkpoint,
+    and only then can accepted-stop evidence beside a request still be owed: with none, every pause
+    was settled by recovery or the updater (or never made: every non-Windows host)."""
+    from hermes_cli import update_pause_record
+    record = update_pause_record.record_path()
+    try:
+        with os.scandir(record.parent) as entries:
+            return any(e.name == record.name or (e.name.startswith(record.name + ".") and e.name.endswith(".claim"))
+                       for e in entries)
+    except FileNotFoundError:
+        return False
+    except OSError:
+        return True
+
+
+def _checkpoint_planned_stop(path: Path, record: dict[str, Any]) -> None:
+    """``on_consume``, decided after the request was read: an update writes its record before its
+    request, so a pause that appeared since the consumer first looked is checkpointed too."""
+    from hermes_cli import update_pause_record
+    if _update_pause_on_disk():
+        with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S):
+            update_pause_record.mark_stop_consumed(path, record)
+
+
 def consume_planned_stop_marker_for_self() -> bool:
     """Return True when the current process is being intentionally stopped."""
-    return _consume_pid_marker_for_self(
-        _get_planned_stop_marker_path(), ttl_s=_PLANNED_STOP_MARKER_TTL_S
-    )
+    from hermes_cli import update_pause_record
+    path = _get_planned_stop_marker_path()
+    try:
+        paused = _update_pause_on_disk()
+        if not paused:  # an earlier drain's receipt: its debt is settled
+            _unlink_quietly(update_pause_record._accepted_path(path))
+        # With a pause on disk, recovery must see either the request or its checkpoint, never the
+        # gap between validating/consuming the marker and scheduling asynchronous stop.
+        with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S) if paused else contextlib.nullcontext():
+            return _consume_pid_marker_for_self(path, ttl_s=_PLANNED_STOP_MARKER_TTL_S,
+                                                on_consume=_checkpoint_planned_stop)
+    except OSError as exc:
+        # The pause bookkeeping (a busy mutex, a checkpoint that cannot be written) never decides
+        # whether this stop was planned: classify without consuming, and leave a receipt beside the
+        # request so recovery keeps this drain owed after the request's TTL (mark_stop_accepted).
+        logger.warning("Planned-stop checkpoint skipped (%s); the stop request stays on disk", exc)
+        return _consume_pid_marker_for_self(_get_planned_stop_marker_path(), ttl_s=_PLANNED_STOP_MARKER_TTL_S,
+                                            keep=True, on_consume=update_pause_record.mark_stop_accepted)
 
 
 def planned_stop_marker_targets_self() -> bool:

gateway/status_inline_source.py

@@ -0,0 +1,132 @@
+"""Inline-source (``python -c <src> …``) command-line parsing for gateway process identity.
+
+Everything after ``-c`` is data for the inline program, not the process's own identity -- except
+Hermes' own bootstraps, which run an entry point in-process and are mapped back to the equivalent
+``python -m <module> <argv…>`` (see #107002).
+"""
+
+from __future__ import annotations
+
+import re
+
+
+def inline_source_flag_index(tokens: list[str]) -> int | None:
+    """Index of the ``-c`` token when *tokens* is an interpreter running INLINE SOURCE, else None.
+
+    Everything after ``-c`` is data the inline program receives, not this process's own identity.
+    The detached gateway restart watcher (``gateway._spawn_gateway_restart_watcher``) is spawned as
+    ``python -c <watcher source> <old_pid> <python> -m hermes_cli.main gateway run``: its trailing
+    argv is the command the watcher will LATER spawn, so every argv matcher used to read it as a
+    live gateway. See #107002 and the "never infer process identity from argv substrings" rule.
+
+    Only interpreter options may precede ``-c``; the first non-option token ends the option block
+    (``python -m hermes_cli.main …`` therefore never matches).
+
+    The walk is VALUE-AWARE: ``-X``/``-W``/``-Q`` and ``--check-hash-based-pycs``/``--jit`` take a
+    SEPARATE operand, so a naive "first non-option token ends the block" walk mistakes that operand
+    for the end of the block and never reaches the ``-c`` behind it (``python -X utf8 -c <src> …``
+    was still read as a live gateway). The operand sets are the canonical ones in
+    ``hermes_state_holders``, not a second hand-rolled copy.
+
+    *tokens* must be CASE-PRESERVING: the operand-taking ``-Q``/``-W``/``-X`` differ from the
+    operand-less ``-q``/``-b``, so a lowercased argv would skip the token after a plain ``-q``.
+    """
+    from hermes_state_holders import (
+        _PYTHON_LONG_OPTIONS_WITH_OPERANDS,
+        _PYTHON_SHORT_OPTIONS_WITH_OPERANDS,
+    )
+
+    index = 1
+    while index < len(tokens):
+        token = tokens[index]
+        if token == "--":
+            return None
+        if token in _PYTHON_LONG_OPTIONS_WITH_OPERANDS:
+            index += 2  # the next token is this option's operand, not the end of the option block
+            continue
+        if token.startswith("--"):
+            index += 1  # ``--opt=value`` and operand-less long options
+            continue
+        if not token.startswith("-") or token == "-":
+            return None
+        # Clustered short options (``-uc``, ``-IsB``). An operand-taking letter consumes the rest of
+        # the cluster as its attached value, or the following token when the cluster ends there --
+        # so ``-Xc`` is ``-X c``, NOT an inline-source ``-c``.
+        cluster = token[1:]
+        for position, letter in enumerate(cluster):
+            if letter == "c":
+                return index
+            if letter in _PYTHON_SHORT_OPTIONS_WITH_OPERANDS:
+                index += 1 if cluster[position + 1 :] else 2
+                break
+        else:
+            index += 1
+    return None
+
+
+def command_line_runs_inline_source(tokens: list[str]) -> bool:
+    """True when *tokens* is an interpreter running INLINE SOURCE (``python -c <src> [args]``)."""
+    return inline_source_flag_index(tokens) is not None
+
+
+# Hermes' own inline bootstraps hand control to a Hermes entry point IN this process, so the argv
+# they run with is this process's own identity; every other ``-c`` program keeps its trailing argv
+# as data (#107002). Each pattern is one emitted source shape, anchored at both ends so a program
+# merely CARRYING a bootstrap command line (the restart watcher's respawn argv) never matches.
+_Q = r"""['"]?"""
+_MAIN = rf"{_Q}__main__{_Q}"
+_RUN_MODULE = rf"runpy\.run_module\(\s*{_Q}(?P<target>[\w.]+){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*,\s*alter_sys\s*=\s*True\s*\)"
+_BOOTSTRAPS = (
+    # hermes_cli._launchers.runtime_command (store launcher, the Windows updater's relaunch)
+    ("module", re.compile(rf"import os, sys, runpy;.*\b{_RUN_MODULE}", re.S)),
+    # hermes_cli.venv_sync.relaunch_command: argv is assigned inside the source
+    ("module", re.compile(rf"import sys, runpy; sys\.path\.insert\(.*\b{_RUN_MODULE}", re.S)),
+    ("path", re.compile(
+        rf"import sys, runpy; sys\.path\.insert\(.*\brunpy\.run_path\(\s*{_Q}(?P<target>[^'\"]+?){_Q}\s*,\s*run_name\s*=\s*{_MAIN}\s*\)",
+        re.S)),
+    # hermes_cli._launchers._launcher_script (the published POSIX shell / Windows .cmd launcher)
+    ("entry", re.compile(r"import os, re, sys\s.*\bfrom\s+(?P<target>[\w.]+)\s+import\s+(?P<func>\w+)\b.*\bsys\.exit\(\s*(?P=func)\(\)\s*\)", re.S)),
+    # hermes_cli._launchers._write_cmd_launcher: the launcher script, base64-encoded
+    ("base64", re.compile(rf"import base64; exec\(base64\.b64decode\({_Q}(?P<target>[A-Za-z0-9+/=]+){_Q}\)\)")),
+)
+_ASSIGNED_ARGV = re.compile(r"\bsys\.argv\s*=\s*\[(.*?)\]\s*;")
+
+
+def _bootstrap_entry(source: str, argv: list[str]) -> list[str] | None:
+    """``[-m, <module>, *argv]`` (or ``[<path>, *argv]``) the inline *source* runs in-process, else None."""
+    source = source.strip()
+    kind, match = next(((k, m) for k, p in _BOOTSTRAPS if (m := p.fullmatch(source))), (None, None))
+    if match is None:
+        return None
+    target = match["target"]
+    if kind == "base64":
+        import base64
+        import binascii
+        try:
+            return _bootstrap_entry(base64.b64decode(target, validate=True).decode("utf-8"), argv)
+        except (binascii.Error, UnicodeDecodeError):
+            return None
+    if kind == "entry":  # the launcher script's own ``--run-module <module>`` switch
+        return ["-m", argv[1], *argv[2:]] if argv[:1] == ["--run-module"] and len(argv) > 1 else ["-m", target, *argv]
+    if assigned := _ASSIGNED_ARGV.search(source):
+        argv = [item.strip().strip("'\"") for item in assigned.group(1).split(",")][1:]
+    return [target, *argv] if kind == "path" else ["-m", target, *argv]
+
+
+def inline_bootstrap_argv(tokens: list[str]) -> list[str] | None:
+    """*tokens* as the equivalent ``python -m <module> <argv…>`` when this interpreter's ``-c`` source
+    is a Hermes bootstrap running an entry point in-process; None for any other inline source.
+
+    Command lines usually arrive space-joined (``/proc``, psutil, ``ps``), which splits the source
+    across tokens; the shortest token run that ends in a recognised tail is the source, whatever
+    joined it, and the tokens after it are the entry point's argv.
+    """
+    index = inline_source_flag_index(tokens)
+    if index is None:
+        return None
+    for end in range(index + 1, len(tokens)):
+        if tokens[end].rstrip().endswith(")"):
+            entry = _bootstrap_entry(" ".join(tokens[index + 1 : end + 1]), tokens[end + 1 :])
+            if entry is not None:
+                return [tokens[0], *entry]
+    return None