fix(update): never disarm the host restart obligation, and never discharge unknown terms

Review findings on the host-scoped update→restart obligation.

- update_cmd_fleet: an unwritable host state dir (read-only HERMES_GATEWAY_LOCK_DIR,
  container UID that does not own $HOME) made write_host_obligation return False and the
  caller ignored it, so an interrupted update left ZERO obligation — stale code, no
  warning, no catch-up restart. The return is now propagated: the legacy per-home marker
  (still read by every reader) carries it, and a host that can write neither says so.
- update_cmd_fleet::_obligation_fields: a PRESENT but unparseable/foreign-versioned host
  record no longer falls through to the legacy marker; terms nobody can read cannot be
  discharged by another record's terms.
- update_cmd_fleet::_restart_identity_sha: zip/pip/Docker installs resolve no checkout
  SHA, so the restart-once stamp was "" and could never match — every profile's update
  re-killed the one shared multiplexer. Falls back to the record's expected_sha, then the
  receipt's post-update identity.
- update_host_obligation: any main_pid probe error is unproven identity (keep its own
  restart), never an aborted restart pass.
- run_notifications: the online notice dedupes per home CHAT, so two served profiles
  sharing one chat get one message (accounting stays per profile); transport resolution is
  isolated per profile, so one broken adapter no longer starves the rest of the fan-out.
- run_adapters / run_profile_reconcile: _profile_configs is pruned with the served set, so
  a failed or removed profile no longer owes a notice nothing can deliver and
  .restart_pending.json is unlinked.

Tests cover the new format's own hazards: unwritable record dir, foreign-version record
with a legacy marker present, non-git install, shared home chat, broken adapter, pruned
config, plus a parity test for the duplicated host-state-dir resolver.
This commit is contained in:
teknium1
2026-09-21 04:51:36 -07:00
committed by Teknium
parent 061195fac1
commit 953b6f6f08
7 changed files with 317 additions and 15 deletions

View File

@@ -898,6 +898,13 @@ class GatewayAdapterLifecycleMixin:
# would park a transiently-failed profile before the first watcher tick can retry it.
for profile_name in transient_failed:
self._served_profile_signatures.pop(profile_name, None)
# Cached configs follow the served set: a profile that failed to start (or stopped being
# served) keeps no home channel in the host-wide notice fan-out, where it would be owed a
# notice no transport can deliver and ``.restart_pending.json`` would never be unlinked.
configs = getattr(self, "_profile_configs", None)
if configs is not None:
for profile_name in [p for p in configs if p not in self._served_profile_signatures]:
configs.pop(profile_name, None)
self._restore_secondary_completion_ledgers(profile_homes)
return connected

View File

@@ -44,6 +44,31 @@ def _served_notice_target_key(profile: Optional[str], platform_value: str, chat_
platform_value if profile is None else f"{profile}:{platform_value}", chat_id, thread_id)
def _delivery_target_key(platform_value: str, chat_id, thread_id) -> tuple:
"""Dedupe key for one DELIVERED chat, profile-independent.
Two served profiles can share a single home chat (one Telegram group for the whole host);
keyed per profile they would each post their own "Gateway online" notice into it.
"""
return _notice_target_key(platform_value, chat_id, thread_id)
def _safe_delivery_transport(platform, config, adapters, *, profile: Optional[str] = None):
"""``resolve_delivery_transport`` isolated to one target: ``None`` (logged) on failure.
The fan-out spans every served profile, so one profile's broken adapter must not abort the
pass and starve every profile after it in dict order.
"""
from gateway.delivery import resolve_delivery_transport
try:
return resolve_delivery_transport(platform, config, adapters)
except Exception as exc:
logger.debug(
"Home-channel transport unavailable for %s%s: %s",
f"{profile}:" if profile else "", getattr(platform, "value", platform), exc)
return None
def _update_output_tail(output: str, limit: int) -> str:
"""Last ``limit`` chars of an update log, prefixed with an ellipsis when cut."""
return output if len(output) <= limit else "…" + output[-limit:]
@@ -786,12 +811,11 @@ class GatewayNotificationsMixin:
def _home_channel_transports(self):
"""Yield ``(platform, platform_cfg, home, transport)`` for every home channel with a live transport."""
from gateway.delivery import resolve_delivery_transport
for platform, platform_cfg in self.config.platforms.items():
home = platform_cfg.home_channel
if not home or not home.chat_id:
continue
transport = resolve_delivery_transport(platform, self.config, self.adapters)
transport = _safe_delivery_transport(platform, self.config, self.adapters)
if transport is None:
continue
yield platform, platform_cfg, home, transport
@@ -813,7 +837,6 @@ class GatewayNotificationsMixin:
def _served_home_channel_transports(self):
"""``(profile, platform, platform_cfg, home, transport)`` for every served profile's home
channel with a live transport — the launch profile's (``profile`` ``None``) first."""
from gateway.delivery import resolve_delivery_transport
for platform, platform_cfg, home, transport in self._home_channel_transports():
yield None, platform, platform_cfg, home, transport
for profile, profile_cfg in (getattr(self, "_profile_configs", None) or {}).items():
@@ -822,7 +845,7 @@ class GatewayNotificationsMixin:
home = platform_cfg.home_channel
if not home or not home.chat_id:
continue
transport = resolve_delivery_transport(platform, profile_cfg, adapters)
transport = _safe_delivery_transport(platform, profile_cfg, adapters, profile=profile)
if transport is None:
continue
yield profile, platform, platform_cfg, home, transport
@@ -917,8 +940,9 @@ class GatewayNotificationsMixin:
) -> set[tuple[str, str, Optional[str]]]:
"""Notify EVERY served profile's configured home channels that the gateway is back online.
Best-effort, once per (profile, platform) home channel — one host process serves them all,
so a notice restricted to the launch profile leaves every other profile's channel silent.
Best-effort, once per home CHAT — several served profiles can share one chat (a single
Telegram group for the whole host), and one host process restarting once owes that chat
one notice. Accounting stays per profile so the marker's owed set still discharges.
``skip_targets`` lets startup avoid duplicate messages when a more specific restart
notification is queued for the same chat.
"""
@@ -928,7 +952,14 @@ class GatewayNotificationsMixin:
free_tier_line = self._free_tier_startup_line()
if free_tier_line:
message = f"{message}\n{free_tier_line}"
for profile, platform, platform_cfg, home, transport in self._served_home_channel_transports():
targets = list(self._served_home_channel_transports())
# A chat already notified for ANOTHER profile is not notified again.
notified_chats = {
_delivery_target_key(platform.value, home.chat_id, home.thread_id)
for profile, platform, _cfg, home, _transport in targets
if _served_notice_target_key(profile, platform.value, home.chat_id, home.thread_id) in skipped
}
for profile, platform, platform_cfg, home, transport in targets:
if not platform_cfg.gateway_restart_notification:
logger.info(
"Home-channel startup notification suppressed: %s has gateway_restart_notification=false",
@@ -938,9 +969,14 @@ class GatewayNotificationsMixin:
target = _served_notice_target_key(profile, platform.value, home.chat_id, home.thread_id)
if target in skipped or target in delivered:
continue
chat = _delivery_target_key(platform.value, home.chat_id, home.thread_id)
if chat in notified_chats:
delivered.add(target)
continue
if await self._send_home_channel_message(
platform, home, transport, message, "Home-channel startup notification failed for %s:%s: %s",
):
notified_chats.add(chat)
delivered.add(target)
logger.info("Sent home-channel startup notification to %s:%s", platform.value, home.chat_id)
return delivered

View File

@@ -156,6 +156,11 @@ class GatewayProfileReconcileMixin:
for name in transient_failed:
if isinstance(self._served_profile_signatures, dict):
self._served_profile_signatures.pop(name, None)
# A cached config with no live adapters is owed a home-channel notice nothing can
# deliver, and the planned-restart marker then never clears.
configs = getattr(self, "_profile_configs", None)
if isinstance(configs, dict):
configs.pop(name, None)
if added:
await self._after_profiles_added([(n, current[n]) for n in added])
result["served_profiles"] = self.served_profile_names()
@@ -215,7 +220,7 @@ class GatewayProfileReconcileMixin:
# Its ``<name>:<platform>`` runtime entries describe a profile that no longer exists.
_write_runtime_status_quiet(drop_profile_platforms=name)
for attr in ("pairing_stores", "_busy_text_modes_by_profile", "_busy_input_modes_by_profile",
"_busy_text_timing_by_profile", "_human_delay_by_profile"):
"_busy_text_timing_by_profile", "_human_delay_by_profile", "_profile_configs"):
store = getattr(self, attr, None)
if isinstance(store, dict):
store.pop(name, None)

View File

@@ -56,20 +56,63 @@ def _fleet_restart_pending_marker_path() -> Path:
return get_hermes_home() / _FLEET_RESTART_PENDING_NAME
def _write_legacy_fleet_restart_pending_marker(
*, expected_sha: str = "", runtimes: list[dict] | None = None
) -> bool:
"""Arm the LEGACY per-``HERMES_HOME`` marker. True when written. Never raises.
Fallback only: ``$HERMES_HOME`` is writable by construction (the updater already writes its
receipts there), so it still carries the obligation when the host state dir cannot.
"""
path = _fleet_restart_pending_marker_path()
try:
lines = [f"started={_time.time()}", f"pid={os.getpid()}"]
if expected_sha:
lines.append(f"expected_sha={expected_sha}")
if runtimes is not None:
lines.append("inventory=" + json.dumps({"version": 1, "runtimes": runtimes}))
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return True
except OSError as exc:
logger.debug("Could not write legacy fleet-restart-pending marker: %s", exc)
return False
def _write_fleet_restart_pending_marker(*, expected_sha: str = "", runtimes: list[dict] | None = None) -> None:
"""Arm the HOST pull→restart obligation. Never raises."""
"""Arm the HOST pull→restart obligation. Never raises.
An unwritable host state dir (``HERMES_GATEWAY_LOCK_DIR`` on a read-only mount, a container
UID that does not own ``$HOME``) must never disarm the obligation: an update interrupted
after this point would then leave stale code running with no warning and no catch-up restart
(#117275). The legacy per-home marker — which every reader here still honours — carries it
instead, and a host that can write neither says so out loud.
"""
if runtimes == []:
# An explicit empty inventory owes no restart (e.g. Desktop-hosted `serve` with no
# gateway services). Arming the marker here leaves a breadcrumb nothing can discharge:
# a no-gateway host would then fail every later ``hermes update`` (#115311).
return
from hermes_cli.update_cmd import _m
from hermes_cli.update_host_obligation import write_host_obligation
from hermes_cli.update_host_obligation import host_obligation_path, write_host_obligation
if _m()._pytest_owns_live_checkout(_fleet_restart_pending_marker_path().parent):
logger.debug("Skipping fleet-restart-pending obligation under pytest (live checkout)")
return
write_host_obligation(
expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name())
if write_host_obligation(
expected_sha=expected_sha, runtimes=runtimes, profile=_current_profile_name()):
return
if _write_legacy_fleet_restart_pending_marker(expected_sha=expected_sha, runtimes=runtimes):
logger.warning(
"Host update-restart obligation (%s) is unwritable; armed the per-home marker %s instead.",
host_obligation_path(), _fleet_restart_pending_marker_path())
return
logger.error(
"Could not arm the update-restart obligation in %s or %s; an interrupted update will not warn.",
host_obligation_path(), _fleet_restart_pending_marker_path())
print(
" ⚠ Could not record the pending gateway-restart obligation (state dir not writable) — "
"restart gateways with `hermes gateway restart` if this update is interrupted.",
file=sys.stderr,
)
def _current_profile_name() -> str:
@@ -104,10 +147,14 @@ def _obligation_fields() -> dict[str, str] | None:
``None`` means nothing armed OR a malformed record; both must leave the obligation standing.
"""
from hermes_cli.update_host_obligation import obligation_fields
from hermes_cli.update_host_obligation import host_obligation_present, obligation_fields
fields = obligation_fields()
if fields is not None:
return fields
if host_obligation_present():
# The record exists but its terms are unknown (corrupt, or a NEWER CLI's version). An
# unrelated legacy marker's inventory cannot discharge terms nobody can read: fail closed.
return None
try:
text = _fleet_restart_pending_marker_path().read_text(encoding="utf-8")
except (OSError, UnicodeError):
@@ -566,6 +613,29 @@ def _live_fleet_current_rows() -> list[dict] | None:
return None
def _restart_identity_sha() -> str:
"""The SHA a completed host restart is stamped with; ``""`` when nothing names the code.
``_current_checkout_sha()`` is ``None`` on every non-git install (zip, pip, Docker), and an
empty stamp can never match, so the per-host restart-once guard would be inert exactly on the
installs it exists for: each profile's ``hermes update`` would re-kill the one shared
multiplexer. The obligation's own ``expected_sha`` — else the receipt's post-update identity —
names the same pulled code.
"""
sha = _current_checkout_sha()
if sha:
return str(sha)
sha = ((_obligation_fields() or {}).get("expected_sha") or "").strip()
if sha:
return sha
with suppress(Exception):
from hermes_cli.update_receipt import read_latest_receipt
post_update = (read_latest_receipt() or {}).get("post_update")
if isinstance(post_update, dict):
return str(post_update.get("sha") or "")
return ""
def _run_pending_fleet_restart() -> bool:
"""Catch-up restart for gateways left on pre-update code. Never raises.
@@ -579,7 +649,7 @@ def _run_pending_fleet_restart() -> bool:
"""
from hermes_cli.update_cmd import _m
from hermes_cli.update_host_obligation import host_restart_already_completed, mark_host_restart_completed
checkout_sha = _current_checkout_sha()
checkout_sha = _restart_identity_sha()
if host_restart_already_completed(checkout_sha):
print(" ✓ This host's gateway was already restarted for this update — not restarting it again.")
return True

View File

@@ -205,7 +205,9 @@ def collapse_units_to_host_processes(
for unit in units:
try:
pid = int(main_pid(unit) or 0)
except (TypeError, ValueError):
except Exception:
# Identity that cannot be proved keeps its own restart; a probe failure of any kind
# must never abort the whole pass.
pid = 0
if pid <= 0:
restart.append(unit)

View File

@@ -12,6 +12,7 @@ from unittest.mock import AsyncMock, Mock
import pytest
import gateway.delivery as gateway_delivery
import gateway.run as gateway_run
from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig
from gateway.platforms.base import SendResult
@@ -92,3 +93,84 @@ async def test_marker_survives_until_a_served_profile_is_reachable(multiplex_run
coder.send.assert_awaited_once()
assert launch.send.await_count == 1, "a reached home is never notified twice"
assert not marker.exists()
@pytest.mark.asyncio
async def test_profiles_sharing_one_home_chat_get_one_notice(tmp_path, monkeypatch):
"""One host process restarting once owes a shared chat ONE notice, not one per profile.
A single Telegram group as the home channel of both the launch profile and a served profile
is a common setup; keyed per profile it received two "Gateway online" messages.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
runner = object.__new__(gateway_run.GatewayRunner)
runner.config = _home_config(Platform.TELEGRAM, "-100999")
runner.config.sessions_dir = tmp_path / "sessions"
launch, coder = _adapter(), _adapter()
runner.adapters = {Platform.TELEGRAM: launch}
runner._profile_configs = {"coder": _home_config(Platform.TELEGRAM, "-100999")}
runner._profile_adapters = {"coder": {Platform.TELEGRAM: coder}}
runner._free_tier_startup_line = Mock(return_value=None)
runner._planned_restart_notice_lock = None
marker = tmp_path / ".restart_pending.json"
marker.write_text("{}", encoding="utf-8")
await runner._replay_pending_planned_restart_notification()
assert launch.send.await_count + coder.send.await_count == 1, "one chat, one restart, one notice"
assert not marker.exists(), "the shared chat was reached, so every owed profile is discharged"
@pytest.mark.asyncio
async def test_one_broken_profile_does_not_starve_the_rest(multiplex_runner, monkeypatch):
"""A profile whose transport resolution raises is skipped; the fan-out continues."""
runner, marker = multiplex_runner
runner.adapters[Platform.DISCORD] = _adapter()
ok = _adapter()
runner._profile_configs = {
"b": _home_config(Platform.TELEGRAM, "b-home"),
"c": _home_config(Platform.SLACK, "c-home"),
}
runner._profile_adapters = {"b": {Platform.TELEGRAM: _adapter()}, "c": {Platform.SLACK: ok}}
real = gateway_delivery.resolve_delivery_transport
def resolve(platform, config, adapters):
if platform is Platform.TELEGRAM:
raise RuntimeError("broken adapter")
return real(platform, config, adapters)
monkeypatch.setattr(gateway_delivery, "resolve_delivery_transport", resolve)
await runner._send_home_channel_startup_notifications()
ok.send.assert_awaited_once(), "a profile after the broken one is still notified"
@pytest.mark.asyncio
async def test_unserved_profile_config_is_pruned_from_the_fan_out(tmp_path, monkeypatch):
"""A profile whose adapters failed keeps no cached config, or it is owed a notice forever.
``owed`` is built from ``_profile_configs`` while delivery needs a live transport, so a stale
entry makes ``owed <= delivered`` permanently false and ``.restart_pending.json`` immortal.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
runner = object.__new__(gateway_run.GatewayRunner)
runner.config = _home_config(Platform.DISCORD, "launch-home")
runner._profile_configs = {"ghost": _home_config(Platform.TELEGRAM, "-200")}
runner._profile_adapters = {}
runner._multiplex_on = Mock(return_value=True)
runner._primary_resource_claims = Mock(return_value={})
runner._record_served_profiles = Mock()
runner._restore_secondary_completion_ledgers = Mock()
runner._start_one_profile_adapters = AsyncMock(side_effect=RuntimeError("adapters failed"))
monkeypatch.setattr(gateway_run, "_multiplex_profile_homes", lambda cfg: [("ghost", tmp_path / "ghost")])
monkeypatch.setattr("hermes_cli.profiles.get_active_profile_name", lambda: "default")
monkeypatch.setattr(
"gateway.run_profile_reconcile.profile_serve_signature", lambda home: ("sig",))
await runner._start_secondary_profile_adapters()
assert "ghost" not in runner._profile_configs
assert list(runner._served_home_channel_configs()) == [
(None, Platform.DISCORD, runner.config.platforms[Platform.DISCORD])]

View File

@@ -202,3 +202,103 @@ def test_host_obligation_lives_beside_the_host_rendezvous_record(two_profiles, m
assert path == tmp_path / "gateway-locks" / "host-update-restart.json"
assert path.is_file()
assert not (two_profiles["coder"] / "fleet_restart_pending").exists()
@pytest.mark.skipif(getattr(os, "geteuid", lambda: 1)() == 0, reason="root ignores directory permissions")
def test_unwritable_host_state_dir_still_arms_the_obligation(two_profiles, no_live_fleet, monkeypatch, tmp_path):
"""An unwritable host state dir must never silently disarm the update→restart obligation.
The host record moved out of ``$HERMES_HOME`` (writable by construction) into the host state
dir, which a read-only mount or a container UID mismatch can make unwritable. Losing the
obligation there is the #117275 outage shape: an interrupted update leaves stale code running
with no warning and no catch-up restart.
"""
_enter(monkeypatch, two_profiles["coder"])
lock_dir = tmp_path / "gateway-locks"
lock_dir.mkdir(parents=True, exist_ok=True)
lock_dir.chmod(0o500)
try:
_arm("coder")
assert not host_obligation.host_obligation_present(), "precondition: the record could not be written"
assert fleet._fleet_restart_obligation_armed() is True
assert fleet._pending_fleet_restart_needed() is True
finally:
lock_dir.chmod(0o700)
def test_unreadable_host_record_is_never_discharged_by_the_legacy_marker(two_profiles, no_live_fleet, monkeypatch, tmp_path):
"""A record whose terms are UNKNOWN cannot be settled by another record's terms.
A foreign version (a NEWER CLI wrote it) or a corrupt record is fail-closed by contract; the
legacy per-home marker describes a different obligation and must not discharge it.
"""
_enter(monkeypatch, two_profiles["coder"])
lock_dir = tmp_path / "gateway-locks"
lock_dir.mkdir(parents=True, exist_ok=True)
(lock_dir / host_obligation.HOST_OBLIGATION_NAME).write_text(
json.dumps({"version": 99, "expected_sha": SHA}), encoding="utf-8")
fleet._fleet_restart_pending_marker_path().write_text(
f"started=1.0\npid=1\nexpected_sha={SHA}\n"
+ "inventory=" + json.dumps({"version": 1, "runtimes": []}) + "\n",
encoding="utf-8",
)
assert fleet._obligation_fields() is None
assert fleet._pending_fleet_restart_needed() is True
def test_restart_runs_once_per_host_on_a_non_git_install(two_profiles, monkeypatch, capsys):
"""zip/pip/Docker installs resolve no checkout SHA; the restart-once guard must still hold.
``mark_host_restart_completed("")`` can never match, so every profile's ``hermes update``
re-killed the one shared multiplexer on exactly the installs this record exists for.
"""
monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: None)
monkeypatch.setattr("hermes_cli.update_receipt.collect_fleet_versions", lambda: [])
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda **k: [4242])
monkeypatch.setattr("hermes_cli.gateway.supports_systemd_services", lambda: False)
monkeypatch.setattr("hermes_cli.gateway.is_macos", lambda: False)
monkeypatch.setattr("hermes_cli.gateway.is_windows", lambda: False)
monkeypatch.setattr("hermes_cli.gateway._wait_for_gateway_exit", lambda **k: True)
kills: list = []
monkeypatch.setattr("hermes_cli.gateway.kill_gateway_processes", lambda **k: kills.append(k))
_enter(monkeypatch, two_profiles["coder"])
_arm("coder")
assert update_cmd._run_pending_fleet_restart() is True
_enter(monkeypatch, two_profiles["writer"])
assert update_cmd._run_pending_fleet_restart() is True
assert len(kills) == 1, "the host gateway must be stopped once per update, not once per profile"
assert "already restarted for this update" in capsys.readouterr().out
def test_a_failing_main_pid_probe_keeps_its_own_restart():
"""Any probe error is unproven identity (its own restart), never an aborted restart pass."""
def boom(unit):
raise RuntimeError("systemctl exploded")
restart, covered = host_obligation.collapse_units_to_host_processes(["a.service", "b.service"], boom)
assert restart == ["a.service", "b.service"]
assert covered == {}
@pytest.mark.parametrize("env", [
{"HERMES_GATEWAY_LOCK_DIR": "/srv/override/locks"},
{"XDG_STATE_HOME": "/srv/xdg-state"},
{"XDG_STATE_HOME": "relative/state"},
{},
])
def test_recovery_host_state_dir_matches_the_gateway_resolver(monkeypatch, env):
"""``update_restart_recovery`` re-implements the lock-dir rule (it may import no Hermes code
at runtime); the duplicate must not drift from ``gateway.status._get_lock_dir``."""
from gateway.status import _get_lock_dir
for name in ("HERMES_GATEWAY_LOCK_DIR", "XDG_STATE_HOME"):
monkeypatch.delenv(name, raising=False)
for name, value in env.items():
monkeypatch.setenv(name, value)
assert recovery._host_state_dir() == str(_get_lock_dir())