mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(gateway): detect and report unclean shutdowns via lifecycle ledger (NS-608)
Hosted agents that die uncleanly (kernel OOM kill, SIGKILL, whole-VM death) leave no trace: shutdown_forensics only covers graceful signals, gateway-exit-diag.log only covers exit paths that actually run, and the VM reboot wipes dmesg before anyone can capture it. NS-608 (BlueAtlas hourly crash cycle, July 12-15) took days of manual log correlation to classify because nothing recorded 'the previous life ended violently'. Add gateway/lifecycle_ledger.py — a sentinel state machine persisted to <HERMES_HOME>/state/gateway.lifecycle.json: - start_gateway() claims the sentinel (phase=running) right after the PID-file/runtime-lock claim, and reports any prior life that never reached an exit path as gateway.previous_unclean_exit in gateway-exit-diag.log + a WARNING log line. - Every exit funnel marks the sentinel exited with a reason: _exit_after_graceful_shutdown (graceful_shutdown), the shutdown watchdog (shutdown_watchdog), and the loop-liveness watchdog (loop_liveness_watchdog). - Ownership-guarded for --replace takeovers: a live matching owner is never reported dead, and the old life cannot clobber the replacement's freshly claimed sentinel on its way out. The 30s loop heartbeat now embeds a cheap /proc memory sample (own RSS, MemAvailable, swap used) so every unclean-death report carries a 'memory N seconds before death' snapshot; the detector flags suspected_oom when the last sample shows <64MiB or <5% available. container-boot.log lines gain prior_exit=clean|unclean|unknown per profile, stamping unclean container deaths into the volume-persisted boot log where support can grep for them. Tests: tests/gateway/test_lifecycle_ledger.py (16 cases) + 4 new container-boot annotation cases. Existing watchdog/forensics/boot suites all green; ruff clean.
This commit is contained in:
Shannon Sands
Teknium
parent
805c1c340c
commit
9c76c133b7
6 changed files with 733 additions and 1 deletions
314
gateway/lifecycle_ledger.py
Normal file
314
gateway/lifecycle_ledger.py
Normal file
|
|
@ -0,0 +1,314 @@
|
|||
"""Gateway lifecycle ledger — durable termination-reason evidence (NS-608).
|
||||
|
||||
The gateway already has *graceful* shutdown forensics
|
||||
(:mod:`gateway.shutdown_forensics` — who sent the SIGTERM) and an exit-path
|
||||
diagnostic log (``gateway-exit-diag.log`` — every way ``asyncio.run`` can
|
||||
return). What it does NOT have is any record of an **unclean death**: a
|
||||
SIGKILL, a kernel OOM kill, or the whole VM dying takes the process out
|
||||
before any handler runs, so the next boot has no idea the previous life
|
||||
ended violently — support tickets like NS-608 then require manually
|
||||
cross-correlating four log files and two external APIs to answer "what
|
||||
killed the gateway?".
|
||||
|
||||
This module closes that gap with a tiny state machine persisted to
|
||||
``<HERMES_HOME>/state/gateway.lifecycle.json``:
|
||||
|
||||
* On startup, :func:`record_startup` reads the sentinel left by the
|
||||
previous life. ``phase == "running"`` means that life never reached any
|
||||
exit path → it died uncleanly. The finding — including the last
|
||||
heartbeat's memory sample, which is the closest thing to a pre-death
|
||||
telemetry snapshot — is appended to ``gateway-exit-diag.log`` as a
|
||||
``gateway.previous_unclean_exit`` record and logged at WARNING. The
|
||||
sentinel is then rewritten as ``phase=running`` for the new life.
|
||||
* On every clean exit path, :func:`mark_exited` rewrites the sentinel as
|
||||
``phase=exited`` with the exit code and a reason string. Wired into
|
||||
``_exit_after_graceful_shutdown`` (the single funnel for all graceful
|
||||
exits, #53107) and the two watchdog ``os._exit`` sites in
|
||||
:mod:`gateway.shutdown_watchdog`.
|
||||
|
||||
:func:`sample_memory` provides the cheap (<1ms, pure /proc reads) memory
|
||||
snapshot that :func:`gateway.shutdown_watchdog.write_loop_heartbeat`
|
||||
embeds in the 30s heartbeat — giving every unclean-death report a
|
||||
"memory available N seconds before death" data point so OOM crash cycles
|
||||
are classifiable from the volume alone (no Prometheus retention races).
|
||||
|
||||
Everything here is best-effort: a forensics failure must never affect the
|
||||
gateway lifecycle it is observing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_LIFECYCLE_RELATIVE = ("state", "gateway.lifecycle.json")
|
||||
_EXIT_DIAG_RELATIVE = ("logs", "gateway-exit-diag.log")
|
||||
|
||||
# Heuristic OOM-suspicion thresholds applied to the last heartbeat's memory
|
||||
# sample. Deliberately conservative: this only annotates the report with a
|
||||
# hint; classification stays with the human reading the evidence.
|
||||
_LOW_MEM_AVAILABLE_KIB = 64 * 1024 # < 64 MiB available
|
||||
_LOW_MEM_AVAILABLE_FRACTION = 0.05 # < 5% of MemTotal available
|
||||
|
||||
|
||||
def _process_hermes_home() -> Path:
|
||||
"""HERMES_HOME for process-level identity files (ignore task overrides)."""
|
||||
val = os.environ.get("HERMES_HOME", "").strip()
|
||||
if val:
|
||||
return Path(val)
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
return get_hermes_home()
|
||||
|
||||
|
||||
def get_lifecycle_sentinel_path(home: Optional[Path] = None) -> Path:
|
||||
"""Return ``<HERMES_HOME>/state/gateway.lifecycle.json``."""
|
||||
base = home if home is not None else _process_hermes_home()
|
||||
return base.joinpath(*_LIFECYCLE_RELATIVE)
|
||||
|
||||
|
||||
def sample_memory() -> Dict[str, Any]:
|
||||
"""Cheap memory snapshot: own RSS + system availability + swap.
|
||||
|
||||
Pure ``/proc`` reads, Linux-only (returns ``{}`` elsewhere), never
|
||||
raises. Values in KiB to match the kernel's units.
|
||||
"""
|
||||
sample: Dict[str, Any] = {}
|
||||
try:
|
||||
with open("/proc/self/status", encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
if line.startswith("VmRSS:"):
|
||||
sample["rss_kib"] = int(line.split()[1])
|
||||
break
|
||||
except (OSError, ValueError, IndexError):
|
||||
pass
|
||||
try:
|
||||
meminfo: Dict[str, int] = {}
|
||||
wanted = {"MemTotal", "MemAvailable", "SwapTotal", "SwapFree"}
|
||||
with open("/proc/meminfo", encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
key = line.split(":", 1)[0]
|
||||
if key in wanted:
|
||||
meminfo[key] = int(line.split()[1])
|
||||
if len(meminfo) == len(wanted):
|
||||
break
|
||||
if "MemTotal" in meminfo:
|
||||
sample["mem_total_kib"] = meminfo["MemTotal"]
|
||||
if "MemAvailable" in meminfo:
|
||||
sample["mem_available_kib"] = meminfo["MemAvailable"]
|
||||
if "SwapTotal" in meminfo and "SwapFree" in meminfo:
|
||||
sample["swap_used_kib"] = meminfo["SwapTotal"] - meminfo["SwapFree"]
|
||||
except (OSError, ValueError, IndexError):
|
||||
pass
|
||||
return sample
|
||||
|
||||
|
||||
def _read_json(path: Path) -> Optional[Dict[str, Any]]:
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return data if isinstance(data, dict) else None
|
||||
|
||||
|
||||
def _write_sentinel(payload: Dict[str, Any], home: Optional[Path]) -> None:
|
||||
path = get_lifecycle_sentinel_path(home)
|
||||
try:
|
||||
from utils import atomic_json_write
|
||||
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
atomic_json_write(path, payload, indent=None)
|
||||
except Exception:
|
||||
logger.debug("Failed to write lifecycle sentinel", exc_info=True)
|
||||
|
||||
|
||||
def _append_exit_diag(record: Dict[str, Any], home: Optional[Path]) -> None:
|
||||
"""Append a JSON line to gateway-exit-diag.log (same format as the CLI's
|
||||
``_exit_diag`` records so existing tooling greps both)."""
|
||||
base = home if home is not None else _process_hermes_home()
|
||||
path = base.joinpath(*_EXIT_DIAG_RELATIVE)
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(record, default=str) + "\n")
|
||||
except OSError:
|
||||
logger.debug("Failed to append unclean-exit record", exc_info=True)
|
||||
|
||||
|
||||
def _pid_alive_with_start_time(pid: Any, start_time: Any) -> bool:
|
||||
"""True when ``pid`` is a live process matching ``start_time`` (±2s).
|
||||
|
||||
Guards the takeover race: during ``--replace`` the old gateway can still
|
||||
be mid-teardown when the new one boots — a live matching owner is a
|
||||
planned handover, not an unclean death.
|
||||
"""
|
||||
try:
|
||||
pid_int = int(pid)
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
if pid_int <= 0:
|
||||
return False
|
||||
try:
|
||||
os.kill(pid_int, 0)
|
||||
except (ProcessLookupError, PermissionError, OSError):
|
||||
return False
|
||||
if start_time is None:
|
||||
return True # alive; can't disambiguate PID reuse — err on "alive"
|
||||
try:
|
||||
from gateway.status import get_process_start_time
|
||||
|
||||
actual = get_process_start_time(pid_int)
|
||||
if actual is None:
|
||||
return True
|
||||
return abs(float(actual) - float(start_time)) <= 2.0
|
||||
except Exception:
|
||||
return True
|
||||
|
||||
|
||||
def detect_unclean_exit(home: Optional[Path] = None) -> Optional[Dict[str, Any]]:
|
||||
"""Inspect the previous life's sentinel; return an evidence dict when it
|
||||
died uncleanly, else ``None``. Read-only — does not rewrite the sentinel.
|
||||
"""
|
||||
sentinel = _read_json(get_lifecycle_sentinel_path(home))
|
||||
if not sentinel or sentinel.get("phase") != "running":
|
||||
return None
|
||||
if _pid_alive_with_start_time(sentinel.get("pid"), sentinel.get("start_time")):
|
||||
return None # live owner — planned takeover in flight, not a death
|
||||
|
||||
evidence: Dict[str, Any] = {
|
||||
"prior_pid": sentinel.get("pid"),
|
||||
"prior_started_at": sentinel.get("started_at"),
|
||||
"prior_start_time": sentinel.get("start_time"),
|
||||
}
|
||||
|
||||
# Enrich with the last heartbeat: when did the loop last prove liveness,
|
||||
# and what did memory look like at that moment?
|
||||
try:
|
||||
from gateway.shutdown_watchdog import get_loop_heartbeat_path
|
||||
|
||||
hb = _read_json(get_loop_heartbeat_path(home))
|
||||
except Exception:
|
||||
hb = None
|
||||
if hb:
|
||||
evidence["last_heartbeat_at"] = hb.get("updated_at")
|
||||
mem = hb.get("mem")
|
||||
if isinstance(mem, dict):
|
||||
evidence["last_heartbeat_mem"] = mem
|
||||
total = mem.get("mem_total_kib")
|
||||
avail = mem.get("mem_available_kib")
|
||||
if isinstance(avail, int) and (
|
||||
avail < _LOW_MEM_AVAILABLE_KIB
|
||||
or (
|
||||
isinstance(total, int)
|
||||
and total > 0
|
||||
and avail / total < _LOW_MEM_AVAILABLE_FRACTION
|
||||
)
|
||||
):
|
||||
evidence["suspected_oom"] = True
|
||||
return evidence
|
||||
|
||||
|
||||
def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]:
|
||||
"""Boot-time entry point: report any unclean previous exit, then claim
|
||||
the sentinel for the current life.
|
||||
|
||||
Returns the unclean-exit evidence dict (also persisted to
|
||||
``gateway-exit-diag.log`` and logged at WARNING) or ``None``. Never
|
||||
raises.
|
||||
"""
|
||||
evidence: Optional[Dict[str, Any]] = None
|
||||
try:
|
||||
evidence = detect_unclean_exit(home)
|
||||
if evidence is not None:
|
||||
record = {
|
||||
"ts": datetime.now(timezone.utc).isoformat(),
|
||||
"tag": "gateway.previous_unclean_exit",
|
||||
"pid": os.getpid(),
|
||||
**evidence,
|
||||
}
|
||||
_append_exit_diag(record, home)
|
||||
logger.warning(
|
||||
"Previous gateway life (pid=%s, started_at=%s) exited UNCLEANLY "
|
||||
"(no exit path ran — SIGKILL / OOM / VM death). "
|
||||
"last_heartbeat_at=%s last_mem=%s suspected_oom=%s",
|
||||
evidence.get("prior_pid"),
|
||||
evidence.get("prior_started_at"),
|
||||
evidence.get("last_heartbeat_at"),
|
||||
evidence.get("last_heartbeat_mem"),
|
||||
evidence.get("suspected_oom", False),
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Unclean-exit detection failed", exc_info=True)
|
||||
|
||||
try:
|
||||
_write_sentinel(
|
||||
{
|
||||
"phase": "running",
|
||||
"pid": os.getpid(),
|
||||
"start_time": time.time(),
|
||||
"started_at": datetime.now(timezone.utc).isoformat(),
|
||||
},
|
||||
home,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Failed to claim lifecycle sentinel", exc_info=True)
|
||||
return evidence
|
||||
|
||||
|
||||
def mark_exited(
|
||||
exit_code: Optional[int] = None,
|
||||
reason: str = "graceful_shutdown",
|
||||
home: Optional[Path] = None,
|
||||
) -> None:
|
||||
"""Mark the current life as cleanly exited. Idempotent, never raises.
|
||||
|
||||
Only rewrites the sentinel when it is still owned by this process —
|
||||
during a ``--replace`` takeover the replacement claims the sentinel
|
||||
before the old process finishes teardown, and the old life must not
|
||||
clobber the new owner's ``running`` phase on its way out.
|
||||
"""
|
||||
try:
|
||||
sentinel = _read_json(get_lifecycle_sentinel_path(home))
|
||||
if sentinel and sentinel.get("pid") not in (None, os.getpid()):
|
||||
return
|
||||
_write_sentinel(
|
||||
{
|
||||
"phase": "exited",
|
||||
"pid": os.getpid(),
|
||||
"exit_code": exit_code,
|
||||
"exit_reason": reason,
|
||||
"exited_at": datetime.now(timezone.utc).isoformat(),
|
||||
},
|
||||
home,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Failed to mark lifecycle sentinel exited", exc_info=True)
|
||||
|
||||
|
||||
def read_prior_exit_label(profile_home: Path) -> str:
|
||||
"""Container-boot helper: one-word summary of how the profile's last
|
||||
gateway life ended. ``clean`` / ``unclean`` / ``unknown`` (no sentinel
|
||||
or never ran). Read-only and exception-free — used by
|
||||
``hermes_cli.container_boot`` to annotate ``container-boot.log``.
|
||||
"""
|
||||
try:
|
||||
sentinel = _read_json(get_lifecycle_sentinel_path(profile_home))
|
||||
if not sentinel:
|
||||
return "unknown"
|
||||
phase = sentinel.get("phase")
|
||||
if phase == "exited":
|
||||
return "clean"
|
||||
if phase == "running":
|
||||
# At container boot the old PID namespace is gone — any
|
||||
# "running" sentinel is from a life that never exited cleanly.
|
||||
return "unclean"
|
||||
except Exception:
|
||||
pass
|
||||
return "unknown"
|
||||
|
|
@ -24948,6 +24948,17 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
|
|||
atexit.register(remove_pid_file)
|
||||
atexit.register(release_gateway_runtime_lock)
|
||||
|
||||
# Lifecycle ledger (NS-608): report if the previous gateway life died
|
||||
# uncleanly (SIGKILL / OOM / VM death — no exit path ran), then claim
|
||||
# the sentinel for this life. Placed after the PID-file/lock claim so
|
||||
# only the authoritative gateway for this HERMES_HOME touches the
|
||||
# sentinel — a --replace loser exiting above must not clobber it.
|
||||
try:
|
||||
from gateway.lifecycle_ledger import record_startup as _lifecycle_record_startup
|
||||
_lifecycle_record_startup()
|
||||
except Exception as _lc_exc:
|
||||
logger.debug("Lifecycle ledger startup record failed: %s", _lc_exc)
|
||||
|
||||
try:
|
||||
from hermes_cli.nous_auth_keepalive import start_nous_auth_keepalive
|
||||
|
||||
|
|
@ -25265,6 +25276,16 @@ def _exit_after_graceful_shutdown(exit_code: int) -> None:
|
|||
release_gateway_runtime_lock()
|
||||
except Exception:
|
||||
pass
|
||||
# Mark this life cleanly exited in the lifecycle sentinel (NS-608). This
|
||||
# is the single funnel every graceful exit passes through, so the next
|
||||
# boot's unclean-death detector only fires for genuine SIGKILL/OOM/VM
|
||||
# deaths. Ownership-guarded internally: a --replace old life won't
|
||||
# clobber the replacement's freshly claimed "running" sentinel.
|
||||
try:
|
||||
from gateway.lifecycle_ledger import mark_exited
|
||||
mark_exited(exit_code, reason="graceful_shutdown")
|
||||
except Exception:
|
||||
pass
|
||||
# Drain the async log queue: os._exit bypasses atexit, so the listener's
|
||||
# atexit drain won't fire. Use drain_log_queue() (bounded, no restart), NOT
|
||||
# flush_log_queue(): if the listener is wedged on the rotation lock — the
|
||||
|
|
|
|||
|
|
@ -185,6 +185,14 @@ def start_loop_liveness_watchdog(
|
|||
logger.debug("Loop liveness faulthandler dump failed", exc_info=True)
|
||||
if stop_event.is_set():
|
||||
return
|
||||
# Record the watchdog exit in the lifecycle sentinel so the next
|
||||
# boot reports "watchdog hard-exit" instead of misclassifying
|
||||
# this as an unclean SIGKILL/OOM death (NS-608).
|
||||
try:
|
||||
from gateway.lifecycle_ledger import mark_exited
|
||||
mark_exited(exit_code, reason="loop_liveness_watchdog")
|
||||
except Exception:
|
||||
pass
|
||||
os._exit(exit_code)
|
||||
return
|
||||
|
||||
|
|
@ -241,6 +249,19 @@ def write_loop_heartbeat(
|
|||
}
|
||||
if start_time is not None:
|
||||
payload["start_time"] = float(start_time)
|
||||
# Embed a cheap memory sample (own RSS + MemAvailable + swap) so the
|
||||
# heartbeat doubles as a rolling pre-death telemetry snapshot: after an
|
||||
# unclean death (SIGKILL/OOM/VM loss) the last heartbeat is the closest
|
||||
# surviving record of memory pressure — see gateway.lifecycle_ledger
|
||||
# (NS-608). Best-effort; <1ms of /proc reads on Linux, {} elsewhere.
|
||||
try:
|
||||
from gateway.lifecycle_ledger import sample_memory
|
||||
|
||||
mem = sample_memory()
|
||||
if mem:
|
||||
payload["mem"] = mem
|
||||
except Exception:
|
||||
pass
|
||||
if extra:
|
||||
payload.update(extra)
|
||||
try:
|
||||
|
|
@ -391,6 +412,13 @@ def arm_shutdown_watchdog(
|
|||
drain_log_queue(timeout=1.0)
|
||||
except Exception:
|
||||
pass
|
||||
# Record the watchdog exit so the next boot's unclean-death detector
|
||||
# reports "shutdown watchdog fired" instead of SIGKILL/OOM (NS-608).
|
||||
try:
|
||||
from gateway.lifecycle_ledger import mark_exited
|
||||
mark_exited(exit_code, reason="shutdown_watchdog")
|
||||
except Exception:
|
||||
pass
|
||||
os._exit(exit_code)
|
||||
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -83,6 +83,13 @@ class ReconcileAction:
|
|||
profile: str
|
||||
prior_state: str | None
|
||||
action: ReconcileActionLabel
|
||||
# How the profile's previous gateway life ended: "clean" (exit path ran),
|
||||
# "unclean" (sentinel still says running — SIGKILL/OOM/VM death), or
|
||||
# "unknown" (no sentinel / never ran). See gateway.lifecycle_ledger
|
||||
# (NS-608): at container boot this is the one place that can stamp
|
||||
# "the previous container life ended violently" into a durable,
|
||||
# volume-persisted log line.
|
||||
prior_exit: str = "unknown"
|
||||
|
||||
|
||||
def reconcile_profile_gateways(
|
||||
|
|
@ -156,6 +163,7 @@ def reconcile_profile_gateways(
|
|||
profile="default",
|
||||
prior_state=default_prior_state,
|
||||
action="started" if default_should_start else "registered",
|
||||
prior_exit=_read_prior_exit_label(hermes_home),
|
||||
))
|
||||
|
||||
profiles_root = hermes_home / "profiles"
|
||||
|
|
@ -195,6 +203,7 @@ def reconcile_profile_gateways(
|
|||
profile=entry.name,
|
||||
prior_state=prior_state,
|
||||
action="started" if should_start else "registered",
|
||||
prior_exit=_read_prior_exit_label(entry),
|
||||
))
|
||||
|
||||
if not dry_run:
|
||||
|
|
@ -405,6 +414,20 @@ def _cleanup_stale_runtime_files(profile_dir: Path) -> None:
|
|||
(profile_dir / name).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def _read_prior_exit_label(profile_dir: Path) -> str:
|
||||
"""How the profile's previous gateway life ended (clean/unclean/unknown).
|
||||
|
||||
Thin, exception-free wrapper over
|
||||
:func:`gateway.lifecycle_ledger.read_prior_exit_label` — cont-init runs
|
||||
in a minimal environment and forensics must never block reconciliation
|
||||
(NS-608)."""
|
||||
try:
|
||||
from gateway.lifecycle_ledger import read_prior_exit_label
|
||||
return read_prior_exit_label(profile_dir)
|
||||
except Exception:
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
|
||||
"""Recreate the s6 service slot for one profile.
|
||||
|
||||
|
|
@ -541,7 +564,7 @@ def _write_reconcile_log(
|
|||
for a in actions:
|
||||
f.write(
|
||||
f"{ts} profile={a.profile} prior_state={a.prior_state} "
|
||||
f"action={a.action}\n"
|
||||
f"action={a.action} prior_exit={a.prior_exit}\n"
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
266
tests/gateway/test_lifecycle_ledger.py
Normal file
266
tests/gateway/test_lifecycle_ledger.py
Normal file
|
|
@ -0,0 +1,266 @@
|
|||
"""Tests for gateway.lifecycle_ledger — unclean-shutdown detection (NS-608).
|
||||
|
||||
The ledger is a tiny sentinel state machine:
|
||||
``record_startup`` claims ``state/gateway.lifecycle.json`` as
|
||||
``phase=running``; every exit path calls ``mark_exited``; the next boot's
|
||||
``record_startup``/``detect_unclean_exit`` reports a still-``running``
|
||||
sentinel from a dead process as an unclean death (SIGKILL / OOM / VM loss)
|
||||
and enriches the report with the last heartbeat's memory sample.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.lifecycle_ledger import (
|
||||
detect_unclean_exit,
|
||||
get_lifecycle_sentinel_path,
|
||||
mark_exited,
|
||||
read_prior_exit_label,
|
||||
record_startup,
|
||||
sample_memory,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DEAD_PID = 2 ** 22 + 12345 # beyond default pid_max on Linux; never alive
|
||||
|
||||
|
||||
def _write_sentinel(home: Path, payload: dict) -> Path:
|
||||
path = get_lifecycle_sentinel_path(home)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _read_sentinel(home: Path) -> dict:
|
||||
return json.loads(get_lifecycle_sentinel_path(home).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _write_heartbeat(home: Path, payload: dict) -> Path:
|
||||
path = home / "state" / "gateway.heartbeat"
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _exit_diag_records(home: Path) -> list[dict]:
|
||||
path = home / "logs" / "gateway-exit-diag.log"
|
||||
if not path.exists():
|
||||
return []
|
||||
return [
|
||||
json.loads(line)
|
||||
for line in path.read_text(encoding="utf-8").splitlines()
|
||||
if line.strip()
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# sample_memory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sample_memory_never_raises() -> None:
|
||||
sample = sample_memory()
|
||||
assert isinstance(sample, dict)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "linux", reason="/proc is Linux-only")
|
||||
def test_sample_memory_has_expected_keys_on_linux() -> None:
|
||||
sample = sample_memory()
|
||||
assert sample.get("rss_kib", 0) > 0
|
||||
assert sample.get("mem_total_kib", 0) > 0
|
||||
assert "mem_available_kib" in sample
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# First boot / clean lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_first_boot_reports_nothing_and_claims_sentinel(tmp_path: Path) -> None:
|
||||
assert record_startup(home=tmp_path) is None
|
||||
sentinel = _read_sentinel(tmp_path)
|
||||
assert sentinel["phase"] == "running"
|
||||
assert sentinel["pid"] == os.getpid()
|
||||
assert "start_time" in sentinel
|
||||
|
||||
|
||||
def test_clean_exit_then_boot_reports_nothing(tmp_path: Path) -> None:
|
||||
record_startup(home=tmp_path)
|
||||
mark_exited(0, reason="graceful_shutdown", home=tmp_path)
|
||||
|
||||
sentinel = _read_sentinel(tmp_path)
|
||||
assert sentinel["phase"] == "exited"
|
||||
assert sentinel["exit_code"] == 0
|
||||
assert sentinel["exit_reason"] == "graceful_shutdown"
|
||||
|
||||
assert record_startup(home=tmp_path) is None
|
||||
assert _exit_diag_records(tmp_path) == []
|
||||
|
||||
|
||||
def test_mark_exited_records_watchdog_reason(tmp_path: Path) -> None:
|
||||
record_startup(home=tmp_path)
|
||||
mark_exited(70, reason="loop_liveness_watchdog", home=tmp_path)
|
||||
sentinel = _read_sentinel(tmp_path)
|
||||
assert sentinel["exit_reason"] == "loop_liveness_watchdog"
|
||||
assert sentinel["exit_code"] == 70
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unclean-death detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_running_sentinel_from_dead_pid_is_unclean(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running",
|
||||
"pid": _DEAD_PID,
|
||||
"start_time": 1000.0,
|
||||
"started_at": "2026-07-11T04:30:00+00:00",
|
||||
})
|
||||
|
||||
evidence = detect_unclean_exit(home=tmp_path)
|
||||
assert evidence is not None
|
||||
assert evidence["prior_pid"] == _DEAD_PID
|
||||
assert evidence["prior_started_at"] == "2026-07-11T04:30:00+00:00"
|
||||
|
||||
|
||||
def test_record_startup_persists_unclean_report_and_reclaims(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running",
|
||||
"pid": _DEAD_PID,
|
||||
"start_time": 1000.0,
|
||||
"started_at": "2026-07-11T04:30:00+00:00",
|
||||
})
|
||||
|
||||
evidence = record_startup(home=tmp_path)
|
||||
assert evidence is not None
|
||||
|
||||
records = _exit_diag_records(tmp_path)
|
||||
assert len(records) == 1
|
||||
assert records[0]["tag"] == "gateway.previous_unclean_exit"
|
||||
assert records[0]["prior_pid"] == _DEAD_PID
|
||||
assert records[0]["pid"] == os.getpid()
|
||||
|
||||
# Sentinel reclaimed for the new life.
|
||||
sentinel = _read_sentinel(tmp_path)
|
||||
assert sentinel["phase"] == "running"
|
||||
assert sentinel["pid"] == os.getpid()
|
||||
|
||||
|
||||
def test_unclean_report_includes_last_heartbeat_memory(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running", "pid": _DEAD_PID, "start_time": 1000.0,
|
||||
})
|
||||
_write_heartbeat(tmp_path, {
|
||||
"pid": _DEAD_PID,
|
||||
"updated_at": "2026-07-12T19:33:00+00:00",
|
||||
"mem": {
|
||||
"rss_kib": 900_000,
|
||||
"mem_total_kib": 2_015_136,
|
||||
"mem_available_kib": 40_000, # ~2% available → OOM suspicion
|
||||
"swap_used_kib": 900_000,
|
||||
},
|
||||
})
|
||||
|
||||
evidence = detect_unclean_exit(home=tmp_path)
|
||||
assert evidence is not None
|
||||
assert evidence["last_heartbeat_at"] == "2026-07-12T19:33:00+00:00"
|
||||
assert evidence["last_heartbeat_mem"]["mem_available_kib"] == 40_000
|
||||
assert evidence["suspected_oom"] is True
|
||||
|
||||
|
||||
def test_healthy_memory_heartbeat_does_not_suspect_oom(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running", "pid": _DEAD_PID, "start_time": 1000.0,
|
||||
})
|
||||
_write_heartbeat(tmp_path, {
|
||||
"pid": _DEAD_PID,
|
||||
"updated_at": "2026-07-12T19:33:00+00:00",
|
||||
"mem": {
|
||||
"mem_total_kib": 2_015_136,
|
||||
"mem_available_kib": 1_000_000,
|
||||
},
|
||||
})
|
||||
|
||||
evidence = detect_unclean_exit(home=tmp_path)
|
||||
assert evidence is not None
|
||||
assert "suspected_oom" not in evidence
|
||||
|
||||
|
||||
def test_live_owner_is_not_reported_as_unclean(tmp_path: Path) -> None:
|
||||
"""A live matching PID means a --replace takeover is in flight, not a
|
||||
death — the detector must stay quiet (start_time omitted → assume alive)."""
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running",
|
||||
"pid": os.getpid(),
|
||||
})
|
||||
assert detect_unclean_exit(home=tmp_path) is None
|
||||
|
||||
|
||||
def test_corrupt_sentinel_is_ignored(tmp_path: Path) -> None:
|
||||
path = get_lifecycle_sentinel_path(tmp_path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("{not json", encoding="utf-8")
|
||||
assert detect_unclean_exit(home=tmp_path) is None
|
||||
assert record_startup(home=tmp_path) is None
|
||||
# And the boot still claims a fresh sentinel.
|
||||
assert _read_sentinel(tmp_path)["phase"] == "running"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Takeover ownership guard on mark_exited
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_old_life_cannot_clobber_new_owner_sentinel(tmp_path: Path) -> None:
|
||||
"""--replace: the replacement claims the sentinel while the old process
|
||||
is mid-teardown; the old life's mark_exited must be a no-op."""
|
||||
_write_sentinel(tmp_path, {
|
||||
"phase": "running",
|
||||
"pid": os.getpid() + 1, # someone else owns it
|
||||
"start_time": 2000.0,
|
||||
})
|
||||
mark_exited(0, reason="graceful_shutdown", home=tmp_path)
|
||||
sentinel = _read_sentinel(tmp_path)
|
||||
assert sentinel["phase"] == "running"
|
||||
assert sentinel["pid"] == os.getpid() + 1
|
||||
|
||||
|
||||
def test_mark_exited_without_prior_sentinel_writes_exited(tmp_path: Path) -> None:
|
||||
mark_exited(1, reason="graceful_shutdown", home=tmp_path)
|
||||
assert _read_sentinel(tmp_path)["phase"] == "exited"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# read_prior_exit_label (container-boot annotation)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_prior_exit_label_unknown_when_no_sentinel(tmp_path: Path) -> None:
|
||||
assert read_prior_exit_label(tmp_path) == "unknown"
|
||||
|
||||
|
||||
def test_prior_exit_label_clean_after_exit(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {"phase": "exited", "pid": 123, "exit_code": 0})
|
||||
assert read_prior_exit_label(tmp_path) == "clean"
|
||||
|
||||
|
||||
def test_prior_exit_label_unclean_when_still_running(tmp_path: Path) -> None:
|
||||
_write_sentinel(tmp_path, {"phase": "running", "pid": _DEAD_PID})
|
||||
assert read_prior_exit_label(tmp_path) == "unclean"
|
||||
|
||||
|
||||
def test_prior_exit_label_survives_corrupt_sentinel(tmp_path: Path) -> None:
|
||||
path = get_lifecycle_sentinel_path(tmp_path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("garbage", encoding="utf-8")
|
||||
assert read_prior_exit_label(tmp_path) == "unknown"
|
||||
|
|
@ -1094,3 +1094,83 @@ def test_main_ignores_removed_skip_reconcile_env_var(
|
|||
assert rc == 0
|
||||
# Reconcile still ran despite the stale env var.
|
||||
assert (scandir / "gateway-worker").exists()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# prior_exit annotation (NS-608 — unclean-shutdown forensics)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _write_lifecycle_sentinel(profile_dir: Path, payload: dict) -> None:
|
||||
state_dir = profile_dir / "state"
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
(state_dir / "gateway.lifecycle.json").write_text(json.dumps(payload))
|
||||
|
||||
|
||||
def test_reconcile_log_annotates_unclean_prior_exit(tmp_path: Path) -> None:
|
||||
"""A profile whose lifecycle sentinel still says phase=running at
|
||||
container boot died uncleanly (OOM / SIGKILL / VM death) — the
|
||||
container-boot.log line must say so (NS-608)."""
|
||||
scandir = tmp_path / "run-service"; scandir.mkdir()
|
||||
crashed = _make_profile(tmp_path, "crashy", state="running")
|
||||
_write_lifecycle_sentinel(crashed, {
|
||||
"phase": "running", "pid": 2**22 + 999, "start_time": 1000.0,
|
||||
})
|
||||
|
||||
actions = reconcile_profile_gateways(
|
||||
hermes_home=tmp_path, scandir=scandir, dry_run=False,
|
||||
)
|
||||
|
||||
(crashy,) = [a for a in actions if a.profile == "crashy"]
|
||||
assert crashy.prior_exit == "unclean"
|
||||
log = (tmp_path / "logs" / "container-boot.log").read_text()
|
||||
assert "profile=crashy" in log
|
||||
assert "prior_exit=unclean" in log
|
||||
|
||||
|
||||
def test_reconcile_log_annotates_clean_prior_exit(tmp_path: Path) -> None:
|
||||
scandir = tmp_path / "run-service"; scandir.mkdir()
|
||||
stopped = _make_profile(tmp_path, "tidy", state="stopped")
|
||||
_write_lifecycle_sentinel(stopped, {
|
||||
"phase": "exited", "pid": 12345, "exit_code": 0,
|
||||
"exit_reason": "graceful_shutdown",
|
||||
})
|
||||
|
||||
actions = reconcile_profile_gateways(
|
||||
hermes_home=tmp_path, scandir=scandir, dry_run=False,
|
||||
)
|
||||
|
||||
(tidy,) = [a for a in actions if a.profile == "tidy"]
|
||||
assert tidy.prior_exit == "clean"
|
||||
log = (tmp_path / "logs" / "container-boot.log").read_text()
|
||||
assert "prior_exit=clean" in log
|
||||
|
||||
|
||||
def test_reconcile_log_prior_exit_unknown_without_sentinel(tmp_path: Path) -> None:
|
||||
"""No lifecycle sentinel (fresh profile, pre-upgrade volume) →
|
||||
prior_exit=unknown, and reconciliation is unaffected."""
|
||||
scandir = tmp_path / "run-service"; scandir.mkdir()
|
||||
_make_profile(tmp_path, "fresh", state="running")
|
||||
|
||||
actions = reconcile_profile_gateways(
|
||||
hermes_home=tmp_path, scandir=scandir, dry_run=False,
|
||||
)
|
||||
|
||||
(fresh,) = [a for a in actions if a.profile == "fresh"]
|
||||
assert fresh.prior_exit == "unknown"
|
||||
assert fresh.action == "started"
|
||||
|
||||
|
||||
def test_default_root_prior_exit_annotated(tmp_path: Path) -> None:
|
||||
scandir = tmp_path / "run-service"; scandir.mkdir()
|
||||
_seed_default_root(tmp_path, state="running")
|
||||
_write_lifecycle_sentinel(tmp_path, {
|
||||
"phase": "running", "pid": 2**22 + 999, "start_time": 1000.0,
|
||||
})
|
||||
|
||||
actions = reconcile_profile_gateways(
|
||||
hermes_home=tmp_path, scandir=scandir, dry_run=False,
|
||||
)
|
||||
|
||||
(default,) = [a for a in actions if a.profile == "default"]
|
||||
assert default.prior_exit == "unclean"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue