From f50e0077c554a93b288cae152f156b228b5ebc7b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 21:58:22 +0300 Subject: [PATCH] fix(service): supervise canonical Mission Core lifecycle --- ...25_MISSION_CORE_BACKEND_LIFECYCLE_AUDIT.md | 116 ++++++++ .../MISSION_CORE_LOCAL_SERVICE_RECOVERY.md | 110 +++++++ scripts/manage_mission_core_launch_agent.py | 221 ++++++++++++++ src/k1link/device_plugins/xgrids_k1/cli.py | 24 +- src/k1link/local_service_launchd.py | 132 +++++++++ src/k1link/service_watchdog.py | 272 ++++++++++++++++++ tests/test_cli.py | 1 + tests/test_local_service_launchd.py | 46 +++ .../test_manage_mission_core_launch_agent.py | 63 ++++ tests/test_service_watchdog.py | 116 ++++++++ 10 files changed, 1095 insertions(+), 6 deletions(-) create mode 100644 docs/2026-08-25_MISSION_CORE_BACKEND_LIFECYCLE_AUDIT.md create mode 100644 docs/runbooks/MISSION_CORE_LOCAL_SERVICE_RECOVERY.md create mode 100644 scripts/manage_mission_core_launch_agent.py create mode 100644 src/k1link/local_service_launchd.py create mode 100644 src/k1link/service_watchdog.py create mode 100644 tests/test_local_service_launchd.py create mode 100644 tests/test_manage_mission_core_launch_agent.py create mode 100644 tests/test_service_watchdog.py diff --git a/docs/2026-08-25_MISSION_CORE_BACKEND_LIFECYCLE_AUDIT.md b/docs/2026-08-25_MISSION_CORE_BACKEND_LIFECYCLE_AUDIT.md new file mode 100644 index 0000000..ecb1b85 --- /dev/null +++ b/docs/2026-08-25_MISSION_CORE_BACKEND_LIFECYCLE_AUDIT.md @@ -0,0 +1,116 @@ +# Mission Core backend lifecycle audit — 2026-08-25 + +## Outcome + +The observed outage was operator-tool induced, not an unexplained Python crash, +GPU out-of-memory event or host memory leak. A Codex session repeatedly used +forced LaunchAgent restarts while the LAB camera/timeline endpoints still had +active work. The final forced termination left the old Uvicorn child draining +while a replacement tried to acquire the same singleton service lease. + +The failure class is a **process-lifecycle termination leak**: the old process +remained alive beyond the restart command's assumption. It is not evidence of +unbounded heap growth. + +## Evidence and causal chain + +The private Codex rollout journal +`~/.codex/sessions/2026/08/25/rollout-2026-08-25T01-08-27-01a035d1-587d-7112-9506-7cc801c2863c.jsonl` +for the active 2026-08-25 task records +`launchctl kickstart -k` against `com.nodedc.mission-core.local` at 14:40, +15:38, 15:44, 16:22, 16:32, 17:14, 17:29, 17:31 and 17:42 Moscow time. At +17:15 it additionally records `SIGTERM`, a ten-second wait, then `SIGKILL` +against the exact old process before another kickstart. + +The user-visible 17:09 Moscow-time outage occurred after the 16:32 forced +restart and before the later 17:15 TERM/KILL recovery attempt. This ordering +rules out the later kill as the start of that outage while still attributes the +failure window to the same repeated forced-restart sequence. + +The LaunchAgent evidence showed 139 historical runs and last exit code 143 +(`SIGTERM`), with no jetsam/OOM record. The application log showed Uvicorn +entering graceful shutdown and waiting for connections/background tasks while +M4.8S LAB camera/timeline requests were open. Replacement processes reported +that Mission Core was already starting or stopping because the old child still +held `.runtime/mission-core/.serve.lock`. + +The complete causal chain was: + +```text +Codex forced launchctl restart + -> SIGTERM reached the uv/Uvicorn generation + -> Uvicorn waited without a configured graceful-shutdown deadline + -> old child retained the singleton flock during active LAB work + -> launchd observed its wrapper transition and attempted a replacement + -> replacement failed closed on the singleton lease + -> port 8000 remained unavailable until the old generation was killed +``` + +The Codex agent caused the outage. The backend did not spontaneously fall over. + +## Code and runtime audit + +The audit covered Python service startup/shutdown, ASGI lifespan cleanup, +thread joins, subprocess calls, the LaunchAgent declaration and operator +restart paths under `src/k1link` and `scripts`. + +| Finding | Severity | State | Resolution | +| --- | --- | --- | --- | +| MC-LIFE-001: Uvicorn graceful drain had no deadline | P0 | fixed | `timeout_graceful_shutdown=10` | +| MC-LIFE-002: PID-only launchd supervision could not detect a live unhealthy service | P0 | fixed | exact-health self-watchdog, three-failure gate, TERM then KILL | +| MC-LIFE-003: forced restart did not prove old label/process release | P0 | fixed | SHA-bound plan/apply, full `bootout` disappearance wait, health acceptance and rollback | +| MC-LIFE-004: dependency resolution could mutate or delay a recovery launch | P1 | fixed | canonical launcher uses `uv run --no-sync` | +| MC-LIFE-004A: launcher parent exit could leave its Python child generation | P1 | fixed | `AbandonProcessGroup=false` makes launchd own the complete group | +| MC-LIFE-005: ASGI/plugin close functions can individually block | P1 | bounded externally | Uvicorn 10 s, watchdog 12 s escalation and launchd 20 s deadline bound the whole generation | +| MC-LIFE-006: self-health state previously had no separate durable evidence | P1 | fixed | private rotating JSONL watchdog journal | +| MC-LIFE-007: three unbounded joins exist in the offline E33 qualification runner | P2 | isolated | daemon-only offline worker path; not imported or executed by the backend lifecycle | +| MC-LIFE-008: artifact-build/guardrail scripts contain subprocess calls without local deadlines | P3 | isolated | developer/CI paths only; not service-reachable and cannot hold port 8000 | + +All subprocess calls reachable through the backend probes and compute-network +control paths already carry explicit deadlines. Service-owned joins found in +the active backend, camera, viewer, preparation, LiDAR-shadow, simulation and +protocol lifecycles are bounded. The remaining no-timeout calls identified by +the syntax scan are offline artifact/qualification tooling, not request or +lifespan paths. + +The direct `.venv/bin/k1link` LaunchAgent entrypoint was also tested and +rejected by macOS with `EPERM` while reading `.venv/pyvenv.cfg` below the +`Downloads` privacy boundary. The apply tool restored the previous plist and +health. The accepted declaration therefore retains the already-authorized +Homebrew `uv` boundary and disables syncing; it does not weaken macOS privacy +controls. + +## Recovery and resume semantics + +`KeepAlive` now restores a dead process. The self-watchdog converts a +live-but-unhealthy event-loop/application generation into a bounded process +exit so `KeepAlive` can act. Startup reconstructs read-only plugin runtimes, +catalogs and background reconciliation from durable artifacts. + +Auto-resume is intentionally selective: + +- idempotent read/catalog/preparation reconciliation restarts automatically; +- browser WebSockets reconnect to the new generation; +- interrupted physical acquisition is reconciled or marked interrupted from + durable ledgers; +- physical commands, acquisition continuation, navigation and actuation are + never silently resumed. + +This distinction prevents availability recovery from becoming an authority +escalation. + +## Qualification evidence + +Focused lifecycle/perception tests passed before deployment. The installed +LaunchAgent accepted exact health with watchdog enabled and a 20-second exit +deadline. A controlled termination of the complete Mission Core process group +produced a new LaunchAgent PID and exact health in 26.588 seconds without +manual intervention. A separate `SIGKILL` crash injection changed PID/PGID +`42895` to PID `42942`, restored exact health in 22.996 seconds and left no old +group residue. + +The local qualification does not make the current Mac LaunchAgent an onboard +deployment artifact. The eventual onboard init declaration must independently +prove boot start, crash restart, health-hang restart, power-loss recovery, +bounded shutdown, single-generation fencing and fail-closed physical-state +reconciliation. diff --git a/docs/runbooks/MISSION_CORE_LOCAL_SERVICE_RECOVERY.md b/docs/runbooks/MISSION_CORE_LOCAL_SERVICE_RECOVERY.md new file mode 100644 index 0000000..b14d22d --- /dev/null +++ b/docs/runbooks/MISSION_CORE_LOCAL_SERVICE_RECOVERY.md @@ -0,0 +1,110 @@ +# Mission Core local service recovery + +## Scope + +This runbook owns the single local Mission Core backend at +`http://127.0.0.1:8000`. It covers startup, health supervision, bounded +shutdown and recovery on the current macOS operator station. It does not grant +physical acquisition, actuation, navigation or safety authority. + +The canonical LaunchAgent label is: + +```text +com.nodedc.mission-core.local +``` + +Do not start a second backend on another port. Do not use +`launchctl kickstart -k` for this service: it combines termination and restart +without proving that the old process group has released the singleton lease. + +## Recovery contract + +The installed LaunchAgent and application form one bounded recovery ladder: + +1. `launchd` starts one `uv run --no-sync k1link serve` process group with + `RunAtLoad=true`, `KeepAlive=true`, `AbandonProcessGroup=false` and a + five-second throttle. +2. Mission Core starts a private self-health thread after acquiring the + singleton backend lease. +3. After the 45-second cold-start grace, three consecutive failed exact + `/api/health` probes request `SIGTERM` for the complete process group. +4. Uvicorn stops accepting work and has ten seconds to drain active requests + and ASGI lifespan work. +5. If the service remains alive, the watchdog escalates to `SIGKILL` after + twelve seconds. `launchd` also owns a 20-second exit deadline. +6. `launchd` starts a fresh process group. Startup reconstructs plugin + runtimes and the recording-preparation reconciler from durable state. + +The watchdog journal is private, bounded and rotated: + +```text +.runtime/mission-core/service-watchdog.jsonl +.runtime/mission-core/service-watchdog.jsonl.1 +``` + +Physical operations are deliberately not resumed from an assumed state. +Interrupted preparation and catalog work is reconciled from durable evidence; +physical acquisition, commands and actuation remain fail-closed and require a +new confirmed authority transition. + +## Plan and apply + +Always plan from the repository root before changing the installed agent: + +```bash +uv run python scripts/manage_mission_core_launch_agent.py plan \ + --repository-root "$PWD" +``` + +Copy the exact `current_sha256` and `desired_sha256` from that output into the +apply command: + +```bash +uv run python scripts/manage_mission_core_launch_agent.py apply \ + --repository-root "$PWD" \ + --expected-current-sha256 \ + --expected-desired-sha256 +``` + +Apply writes a mode-0600 backup below +`.runtime/mission-core/launch-agent-backups`, atomically replaces the plist, +waits until `bootout` has fully removed the old label, bootstraps the new +declaration and accepts only the exact Mission Core health document. Any +failure restores the previous plist and repeats the same health acceptance. + +Read-only status: + +```bash +uv run python scripts/manage_mission_core_launch_agent.py status \ + --repository-root "$PWD" +``` + +## Acceptance after recovery + +The service is recovered only when all of the following are true: + +- `launchctl print gui/$(id -u)/com.nodedc.mission-core.local` reports + `state = running`; +- the arguments include `uv run --no-sync k1link serve`; +- the environment contains `MISSIONCORE_SERVICE_WATCHDOG => 1`; +- the launchd exit timeout is 20 seconds; +- `GET http://127.0.0.1:8000/api/health` returns HTTP 200 with + `ok=true`, `status=ok` and + `service=mission-core-control-plane`; +- the watchdog journal contains `watchdog-started` for the current child PID; +- there is only one `uv` parent and one Mission Core Python child in their + exact process group. + +## 2026-08-25 recovery qualification + +The reviewed declaration SHA-256 was +`80fca5ec6bdab21f11a544d8dbee65a35f73a6c34752c6a48e9a1181e5da256a`. +A controlled `SIGTERM` of the exact Mission Core process group changed the +LaunchAgent PID from `40866` to `40957`; exact health recovered automatically +in `26.588` seconds without a manual start. After the explicit +`AbandonProcessGroup=false` fence was installed, a controlled `SIGKILL` of PID +and PGID `42895` produced a new LaunchAgent PID `42942` and exact health in +`22.996` seconds, with no old process-group residue. This is local +operator-station evidence only. An onboard Linux deployment must express the +same contract in its init system and pass its own power-loss, crash, hang and +durable-state reconciliation qualification. diff --git a/scripts/manage_mission_core_launch_agent.py b/scripts/manage_mission_core_launch_agent.py new file mode 100644 index 0000000..08dc0ed --- /dev/null +++ b/scripts/manage_mission_core_launch_agent.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +"""Plan/apply the user-owned Mission Core LaunchAgent with rollback acceptance.""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import os +import subprocess +import tempfile +import time +from contextlib import suppress +from pathlib import Path + +from k1link.local_service_launchd import ( + MISSION_CORE_LAUNCH_AGENT_LABEL, + MissionCoreLaunchAgentError, + plan_mission_core_launch_agent, +) + +_LAUNCHCTL_TRANSITION_TIMEOUT_SECONDS = 30.0 +_LAUNCHCTL_TRANSITION_POLL_SECONDS = 0.1 + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("action", choices=("plan", "apply", "status")) + parser.add_argument("--repository-root", type=Path, required=True) + parser.add_argument( + "--agent-path", + type=Path, + default=Path.home() + / "Library/LaunchAgents/com.nodedc.mission-core.local.plist", + ) + parser.add_argument("--expected-current-sha256") + parser.add_argument("--expected-desired-sha256") + arguments = parser.parse_args() + if arguments.action == "status": + print(json.dumps(_status(), sort_keys=True)) + return 0 + plan = plan_mission_core_launch_agent( + repository_root=arguments.repository_root, + agent_path=arguments.agent_path, + ) + if arguments.action == "plan": + print(json.dumps(plan.to_dict(), indent=2, sort_keys=True)) + return 0 + if ( + arguments.expected_current_sha256 != plan.current_sha256 + or arguments.expected_desired_sha256 != plan.desired_sha256 + ): + raise MissionCoreLaunchAgentError("launch agent plan changed before apply") + previous = plan.agent_path.read_bytes() + backup_root = ( + arguments.repository_root.expanduser().resolve() + / ".runtime/mission-core/launch-agent-backups" + ) + backup_root.mkdir(mode=0o700, parents=True, exist_ok=True) + backup = backup_root / f"{plan.current_sha256}.plist" + if backup.exists() and hashlib.sha256(backup.read_bytes()).hexdigest() != plan.current_sha256: + raise MissionCoreLaunchAgentError("launch agent backup identity collision") + if not backup.exists(): + _write_atomic(backup, previous) + _write_atomic(plan.agent_path, plan.desired_payload) + try: + _reload_launch_agent(plan.agent_path) + accepted = _wait_for_health(60.0) + if not accepted: + raise MissionCoreLaunchAgentError("reloaded Mission Core did not become healthy") + except BaseException: + _write_atomic(plan.agent_path, previous) + _reload_launch_agent(plan.agent_path) + if not _wait_for_health(60.0): + raise MissionCoreLaunchAgentError( + "launch agent apply failed and rollback did not recover health" + ) from None + raise + result = { + **plan.to_dict(), + "applied": True, + "backup_path": str(backup), + "health_accepted": True, + "status": _status(), + } + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + + +def _write_atomic(path: Path, payload: bytes) -> None: + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "wb", closefd=True) as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + except BaseException: + with suppress(OSError): + os.close(descriptor) + with suppress(FileNotFoundError): + os.unlink(temporary) + raise + + +def _reload_launch_agent(path: Path) -> None: + """Reload only after launchd proves the previous job is fully absent. + + ``launchctl bootout`` can return while the job is still represented by an + ``xpcproxy`` transition. An immediate bootstrap is then rejected even + though the plist is valid. Waiting on the exact label avoids that race and + keeps apply/rollback symmetric. + """ + + domain = f"gui/{os.getuid()}" + target = f"{domain}/{MISSION_CORE_LAUNCH_AGENT_LABEL}" + bootout = subprocess.run( + ["launchctl", "bootout", f"{domain}/{MISSION_CORE_LAUNCH_AGENT_LABEL}"], + check=False, + capture_output=True, + text=True, + timeout=30.0, + ) + if bootout.returncode != 0 and _launch_agent_loaded(target): + raise MissionCoreLaunchAgentError( + f"launchctl bootout failed with exit code {bootout.returncode}" + ) + _wait_until_launch_agent_unloaded(target) + completed = subprocess.run( + ["launchctl", "bootstrap", domain, str(path)], + check=False, + capture_output=True, + text=True, + timeout=30.0, + ) + if completed.returncode != 0: + raise MissionCoreLaunchAgentError( + "launchctl bootstrap rejected the Mission Core agent " + f"with exit code {completed.returncode}" + ) + + +def _launch_agent_loaded(target: str) -> bool: + completed = subprocess.run( + ["launchctl", "print", target], + check=False, + capture_output=True, + text=True, + timeout=5.0, + ) + return completed.returncode == 0 + + +def _wait_until_launch_agent_unloaded(target: str) -> None: + deadline = time.monotonic() + _LAUNCHCTL_TRANSITION_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if not _launch_agent_loaded(target): + return + time.sleep(_LAUNCHCTL_TRANSITION_POLL_SECONDS) + raise MissionCoreLaunchAgentError( + "Mission Core launch agent did not finish bootout before the deadline" + ) + + +def _wait_for_health(timeout_seconds: float) -> bool: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if _healthy(): + return True + time.sleep(0.25) + return False + + +def _healthy() -> bool: + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1.0) + try: + connection.request("GET", "/api/health", headers={"Connection": "close"}) + response = connection.getresponse() + payload = response.read(64 * 1024 + 1) + except (OSError, TimeoutError, http.client.HTTPException): + return False + finally: + connection.close() + if response.status != 200 or len(payload) > 64 * 1024: + return False + try: + document = json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError): + return False + return bool( + isinstance(document, dict) + and document.get("ok") is True + and document.get("status") == "ok" + and document.get("service") == "mission-core-control-plane" + ) + + +def _status() -> dict[str, object]: + completed = subprocess.run( + [ + "launchctl", + "print", + f"gui/{os.getuid()}/{MISSION_CORE_LAUNCH_AGENT_LABEL}", + ], + check=False, + capture_output=True, + text=True, + timeout=5.0, + ) + return { + "label": MISSION_CORE_LAUNCH_AGENT_LABEL, + "launchd_loaded": completed.returncode == 0, + "health_ok": _healthy(), + } + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index 9274729..5622ef5 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -81,6 +81,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_authority import ( ) from k1link.device_plugins.xgrids_k1.usb.snapshot import snapshot as usb_snapshot from k1link.macos_credentials import CredentialDialogError, prompt_wifi_credentials +from k1link.service_watchdog import MissionCoreSelfWatchdog, watchdog_enabled from k1link.sessions import SessionIntegrityError app = typer.Typer( @@ -116,6 +117,7 @@ app.add_typer(artifact_app, name="artifact") _CANONICAL_MISSION_CORE_PORT = 8000 _MISSION_CORE_SERVE_LOCK_FILENAME = ".serve.lock" +_MISSION_CORE_GRACEFUL_SHUTDOWN_SECONDS = 10 class _MissionCoreServeLeaseError(RuntimeError): @@ -1166,13 +1168,23 @@ def serve_console( f"NODEDC MISSION CORE: http://127.0.0.1:{_CANONICAL_MISSION_CORE_PORT}" ) console.print("The credential endpoint is bound to this Mac only.") - uvicorn.run( - "k1link.web.app:app", - host="127.0.0.1", - port=_CANONICAL_MISSION_CORE_PORT, - log_level="info", - access_log=True, + watchdog = ( + MissionCoreSelfWatchdog(repository_root) if watchdog_enabled() else None ) + if watchdog is not None: + watchdog.start() + try: + uvicorn.run( + "k1link.web.app:app", + host="127.0.0.1", + port=_CANONICAL_MISSION_CORE_PORT, + log_level="info", + access_log=True, + timeout_graceful_shutdown=_MISSION_CORE_GRACEFUL_SHUTDOWN_SECONDS, + ) + finally: + if watchdog is not None: + watchdog.stop() def _print_existing_mission_core(port: int) -> None: diff --git a/src/k1link/local_service_launchd.py b/src/k1link/local_service_launchd.py new file mode 100644 index 0000000..3f9b4be --- /dev/null +++ b/src/k1link/local_service_launchd.py @@ -0,0 +1,132 @@ +"""Versioned launchd declaration for the canonical local Mission Core service.""" + +from __future__ import annotations + +import hashlib +import plistlib +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +MISSION_CORE_LAUNCH_AGENT_LABEL: Final = "com.nodedc.mission-core.local" +MISSION_CORE_LAUNCH_AGENT_SCHEMA: Final = "missioncore.local-launch-agent-plan/v1" + + +class MissionCoreLaunchAgentError(RuntimeError): + """The local launch agent cannot be planned without weakening its boundary.""" + + +@dataclass(frozen=True, slots=True) +class MissionCoreLaunchAgentPlan: + agent_path: Path + current_sha256: str + desired_sha256: str + current_program_arguments: tuple[str, ...] + desired_program_arguments: tuple[str, ...] + desired_payload: bytes + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": MISSION_CORE_LAUNCH_AGENT_SCHEMA, + "label": MISSION_CORE_LAUNCH_AGENT_LABEL, + "agent_path": str(self.agent_path), + "current_sha256": self.current_sha256, + "desired_sha256": self.desired_sha256, + "current_program_arguments": list(self.current_program_arguments), + "desired_program_arguments": list(self.desired_program_arguments), + "changes": { + "dependency_sync_disabled": "--no-sync" + in self.desired_program_arguments, + "self_health_watchdog": True, + "bounded_launchd_exit_timeout_seconds": 20, + "keep_alive": True, + "process_group_owned": True, + }, + } + + +def plan_mission_core_launch_agent( + *, + repository_root: Path, + agent_path: Path, +) -> MissionCoreLaunchAgentPlan: + repository = repository_root.expanduser().resolve(strict=True) + path = agent_path.expanduser().absolute() + current_payload = _read_private_regular_file(path) + try: + current = plistlib.loads(current_payload) + except plistlib.InvalidFileException as exc: + raise MissionCoreLaunchAgentError("current Mission Core launch agent is invalid") from exc + if not isinstance(current, dict) or current.get("Label") != MISSION_CORE_LAUNCH_AGENT_LABEL: + raise MissionCoreLaunchAgentError("current launch agent identity changed") + current_arguments = _program_arguments(current) + current_working_directory = current.get("WorkingDirectory") + if current_working_directory != str(repository): + raise MissionCoreLaunchAgentError("current launch agent targets another repository") + environment = current.get("EnvironmentVariables") + if not isinstance(environment, dict) or any( + not isinstance(key, str) or not isinstance(value, str) + for key, value in environment.items() + ): + raise MissionCoreLaunchAgentError("current launch agent environment is invalid") + # A LaunchAgent started directly from this repository's venv is denied + # access to ``.venv/pyvenv.cfg`` by macOS privacy controls because the + # checkout is below Downloads. The Homebrew uv launcher is already the + # accepted local execution boundary. ``--no-sync`` keeps launch startup + # deterministic and prevents dependency mutation during recovery. + uv_entrypoint = Path(current_arguments[0]) + if ( + not uv_entrypoint.is_absolute() + or uv_entrypoint.name != "uv" + or not uv_entrypoint.exists() + ): + raise MissionCoreLaunchAgentError("Mission Core uv entrypoint is unavailable") + desired_environment = dict(environment) + desired_environment["MISSIONCORE_SERVICE_WATCHDOG"] = "1" + log_path = repository / ".runtime/mission-core/k1link-serve-launchd.log" + desired: dict[str, object] = { + "Label": MISSION_CORE_LAUNCH_AGENT_LABEL, + "ProgramArguments": [ + str(uv_entrypoint), + "run", + "--no-sync", + "k1link", + "serve", + ], + "WorkingDirectory": str(repository), + "EnvironmentVariables": desired_environment, + "KeepAlive": True, + "RunAtLoad": True, + "AbandonProcessGroup": False, + "ProcessType": "Background", + "ThrottleInterval": 5, + "ExitTimeOut": 20, + "StandardOutPath": str(log_path), + "StandardErrorPath": str(log_path), + } + desired_payload = plistlib.dumps(desired, fmt=plistlib.FMT_XML, sort_keys=True) + return MissionCoreLaunchAgentPlan( + agent_path=path, + current_sha256=_sha256(current_payload), + desired_sha256=_sha256(desired_payload), + current_program_arguments=current_arguments, + desired_program_arguments=tuple(desired["ProgramArguments"]), + desired_payload=desired_payload, + ) + + +def _program_arguments(document: dict[str, object]) -> tuple[str, ...]: + value = document.get("ProgramArguments") + if not isinstance(value, list) or not value or any(not isinstance(item, str) for item in value): + raise MissionCoreLaunchAgentError("launch agent program arguments are invalid") + return tuple(value) + + +def _read_private_regular_file(path: Path) -> bytes: + if path.is_symlink() or not path.is_file(): + raise MissionCoreLaunchAgentError("Mission Core launch agent is unavailable") + return path.read_bytes() + + +def _sha256(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() diff --git a/src/k1link/service_watchdog.py b/src/k1link/service_watchdog.py new file mode 100644 index 0000000..a69f7a1 --- /dev/null +++ b/src/k1link/service_watchdog.py @@ -0,0 +1,272 @@ +"""Fail-closed health watchdog for the canonical Mission Core service process.""" + +from __future__ import annotations + +import http.client +import json +import os +import signal +import stat +import time +from collections.abc import Callable, Mapping +from contextlib import suppress +from dataclasses import dataclass +from pathlib import Path +from threading import Event, Lock, Thread +from typing import Final + +WATCHDOG_SCHEMA: Final = "missioncore.local-service-watchdog/v1" +WATCHDOG_ENV: Final = "MISSIONCORE_SERVICE_WATCHDOG" +WATCHDOG_ENABLED_VALUE: Final = "1" +MISSION_CORE_SERVICE_ID: Final = "mission-core-control-plane" +DEFAULT_JOURNAL_MAX_BYTES: Final = 4 * 1024 * 1024 + + +class MissionCoreWatchdogError(RuntimeError): + """The watchdog cannot establish a trustworthy local safety boundary.""" + + +@dataclass(frozen=True, slots=True) +class MissionCoreWatchdogPolicy: + startup_grace_seconds: float = 45.0 + probe_interval_seconds: float = 2.0 + probe_timeout_seconds: float = 1.0 + consecutive_failure_limit: int = 3 + graceful_shutdown_seconds: float = 12.0 + + def __post_init__(self) -> None: + if ( + self.startup_grace_seconds <= 0 + or self.probe_interval_seconds <= 0 + or self.probe_timeout_seconds <= 0 + or self.consecutive_failure_limit < 1 + or self.graceful_shutdown_seconds <= 0 + ): + raise ValueError("Mission Core watchdog policy must be positive") + + +class ConsecutiveHealthGate: + """Trigger only after a bounded sequence of genuine probe failures.""" + + def __init__(self, failure_limit: int) -> None: + if failure_limit < 1: + raise ValueError("health failure limit must be positive") + self.failure_limit = failure_limit + self.consecutive_failures = 0 + + def observe(self, healthy: bool) -> bool: + if healthy: + self.consecutive_failures = 0 + return False + self.consecutive_failures += 1 + return self.consecutive_failures >= self.failure_limit + + +class MissionCoreWatchdogJournal: + """Append bounded, private lifecycle evidence outside the application log.""" + + def __init__( + self, + path: Path, + *, + max_bytes: int = DEFAULT_JOURNAL_MAX_BYTES, + ) -> None: + if max_bytes < 1: + raise ValueError("watchdog journal limit must be positive") + self.path = path.expanduser().absolute() + self.max_bytes = max_bytes + self._guard = Lock() + self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + metadata = self.path.parent.lstat() + if not stat.S_ISDIR(metadata.st_mode): + raise MissionCoreWatchdogError("watchdog journal parent is not a directory") + + def append(self, event: str, **details: object) -> None: + document = { + "schema_version": WATCHDOG_SCHEMA, + "event": event, + "utc_ns": time.time_ns(), + "monotonic_ns": time.monotonic_ns(), + "pid": os.getpid(), + **details, + } + payload = json.dumps( + document, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + b"\n" + with self._guard: + self._rotate_if_needed(len(payload)) + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(self.path, flags, 0o600) + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or stat.S_IMODE(metadata.st_mode) != 0o600 + or metadata.st_nlink != 1 + ): + raise MissionCoreWatchdogError( + "watchdog journal is not a private regular file" + ) + os.write(descriptor, payload) + os.fsync(descriptor) + finally: + os.close(descriptor) + + def _rotate_if_needed(self, incoming_bytes: int) -> None: + try: + metadata = self.path.lstat() + except FileNotFoundError: + return + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: + raise MissionCoreWatchdogError("watchdog journal identity changed") + if metadata.st_size + incoming_bytes <= self.max_bytes: + return + previous = self.path.with_name(f"{self.path.name}.1") + with suppress(FileNotFoundError): + previous.unlink() + os.replace(self.path, previous) + + +Probe = Callable[[], bool] +SignalAction = Callable[[], None] + + +class MissionCoreSelfWatchdog: + """Terminate a live-but-unhealthy service so its init system can restart it.""" + + def __init__( + self, + repository_root: Path, + *, + policy: MissionCoreWatchdogPolicy | None = None, + probe: Probe | None = None, + request_shutdown: SignalAction | None = None, + force_shutdown: SignalAction | None = None, + journal: MissionCoreWatchdogJournal | None = None, + ) -> None: + self.policy = policy or MissionCoreWatchdogPolicy() + self.probe = probe or _mission_core_health_probe( + self.policy.probe_timeout_seconds + ) + self.request_shutdown = request_shutdown or _process_group_signal(signal.SIGTERM) + self.force_shutdown = force_shutdown or _process_group_signal(signal.SIGKILL) + self.journal = journal or MissionCoreWatchdogJournal( + repository_root.expanduser().absolute() + / ".runtime/mission-core/service-watchdog.jsonl" + ) + self._stop = Event() + self._thread = Thread( + target=self._run, + name="mission-core-self-health-watchdog", + daemon=True, + ) + self._started = False + + def start(self) -> None: + if self._started: + raise MissionCoreWatchdogError("Mission Core watchdog already started") + self._started = True + self.journal.append("watchdog-started", policy=_policy_dict(self.policy)) + self._thread.start() + + def stop(self) -> None: + if not self._started: + return + self._stop.set() + self._thread.join(timeout=self.policy.probe_timeout_seconds + 1.0) + self.journal.append( + "watchdog-stopped", + worker_alive=self._thread.is_alive(), + ) + + def _run(self) -> None: + if self._stop.wait(self.policy.startup_grace_seconds): + return + gate = ConsecutiveHealthGate(self.policy.consecutive_failure_limit) + last_reported_health: bool | None = None + while not self._stop.is_set(): + healthy = False + try: + healthy = self.probe() + except Exception: + healthy = False + triggered = gate.observe(healthy) + if healthy != last_reported_health: + self.journal.append( + "health-state-changed", + healthy=healthy, + consecutive_failures=gate.consecutive_failures, + ) + last_reported_health = healthy + if triggered: + self.journal.append( + "restart-requested", + reason="consecutive-health-probe-failures", + consecutive_failures=gate.consecutive_failures, + ) + self.request_shutdown() + if not self._stop.wait(self.policy.graceful_shutdown_seconds): + self.journal.append( + "restart-escalated", + reason="graceful-shutdown-timeout", + ) + self.force_shutdown() + return + if self._stop.wait(self.policy.probe_interval_seconds): + return + + +def watchdog_enabled(environ: Mapping[str, str] = os.environ) -> bool: + return environ.get(WATCHDOG_ENV) == WATCHDOG_ENABLED_VALUE + + +def _mission_core_health_probe(timeout_seconds: float) -> Probe: + def probe() -> bool: + connection = http.client.HTTPConnection( + "127.0.0.1", + 8000, + timeout=timeout_seconds, + ) + try: + connection.request("GET", "/api/health", headers={"Connection": "close"}) + response = connection.getresponse() + payload = response.read(64 * 1024 + 1) + except (OSError, TimeoutError, http.client.HTTPException): + return False + finally: + connection.close() + if response.status != 200 or len(payload) > 64 * 1024: + return False + try: + document = json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError): + return False + return bool( + isinstance(document, dict) + and document.get("ok") is True + and document.get("status") == "ok" + and document.get("service") == MISSION_CORE_SERVICE_ID + ) + + return probe + + +def _process_group_signal(signal_number: signal.Signals) -> SignalAction: + def send() -> None: + os.killpg(os.getpgrp(), signal_number) + + return send + + +def _policy_dict(policy: MissionCoreWatchdogPolicy) -> dict[str, object]: + return { + "startup_grace_seconds": policy.startup_grace_seconds, + "probe_interval_seconds": policy.probe_interval_seconds, + "probe_timeout_seconds": policy.probe_timeout_seconds, + "consecutive_failure_limit": policy.consecutive_failure_limit, + "graceful_shutdown_seconds": policy.graceful_shutdown_seconds, + } diff --git a/tests/test_cli.py b/tests/test_cli.py index 44944cc..a2ee440 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -83,6 +83,7 @@ def test_serve_resolves_frontend_from_repository_root(monkeypatch: Any) -> None: "port": 8000, "log_level": "info", "access_log": True, + "timeout_graceful_shutdown": 10, } assert lease.active is False diff --git a/tests/test_local_service_launchd.py b/tests/test_local_service_launchd.py new file mode 100644 index 0000000..e40195b --- /dev/null +++ b/tests/test_local_service_launchd.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import plistlib +from pathlib import Path + +from k1link.local_service_launchd import plan_mission_core_launch_agent + + +def test_launch_agent_plan_disables_sync_and_enables_watchdog(tmp_path: Path) -> None: + repository = tmp_path / "repo" + repository.mkdir() + uv_entrypoint = tmp_path / "uv" + uv_entrypoint.write_text("#!/bin/sh\n") + uv_entrypoint.chmod(0o700) + agent = tmp_path / "agent.plist" + agent.write_bytes( + plistlib.dumps( + { + "Label": "com.nodedc.mission-core.local", + "ProgramArguments": [str(uv_entrypoint), "run", "k1link", "serve"], + "WorkingDirectory": str(repository), + "EnvironmentVariables": {"PATH": "/usr/bin:/bin"}, + } + ) + ) + agent.chmod(0o600) + + plan = plan_mission_core_launch_agent( + repository_root=repository, + agent_path=agent, + ) + desired = plistlib.loads(plan.desired_payload) + + assert desired["ProgramArguments"] == [ + str(uv_entrypoint), + "run", + "--no-sync", + "k1link", + "serve", + ] + assert desired["EnvironmentVariables"]["MISSIONCORE_SERVICE_WATCHDOG"] == "1" + assert desired["KeepAlive"] is True + assert desired["RunAtLoad"] is True + assert desired["AbandonProcessGroup"] is False + assert desired["ExitTimeOut"] == 20 + assert plan.current_sha256 != plan.desired_sha256 diff --git a/tests/test_manage_mission_core_launch_agent.py b/tests/test_manage_mission_core_launch_agent.py new file mode 100644 index 0000000..de291de --- /dev/null +++ b/tests/test_manage_mission_core_launch_agent.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import importlib.util +import subprocess +from pathlib import Path +from typing import Any + +_SCRIPT = Path(__file__).parents[1] / "scripts/manage_mission_core_launch_agent.py" +_SPEC = importlib.util.spec_from_file_location( + "manage_mission_core_launch_agent", + _SCRIPT, +) +assert _SPEC is not None and _SPEC.loader is not None +manager = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(manager) + + +def test_reload_waits_for_launchd_transition_before_bootstrap( + monkeypatch: Any, + tmp_path: Path, +) -> None: + commands: list[tuple[str, ...]] = [] + print_results = iter((0, 0, 1)) + + def fake_run(arguments: list[str], **_: object) -> subprocess.CompletedProcess[str]: + command = tuple(arguments) + commands.append(command) + if arguments[1] == "print": + return subprocess.CompletedProcess(arguments, next(print_results), "", "") + return subprocess.CompletedProcess(arguments, 0, "", "") + + monkeypatch.setattr(manager.subprocess, "run", fake_run) + monkeypatch.setattr(manager.time, "sleep", lambda _: None) + + agent = tmp_path / "agent.plist" + manager._reload_launch_agent(agent) + + assert [command[1] for command in commands] == [ + "bootout", + "print", + "print", + "print", + "bootstrap", + ] + + +def test_reload_rejects_failed_bootout_while_job_is_still_loaded( + monkeypatch: Any, + tmp_path: Path, +) -> None: + def fake_run(arguments: list[str], **_: object) -> subprocess.CompletedProcess[str]: + if arguments[1] == "bootout": + return subprocess.CompletedProcess(arguments, 5, "", "") + return subprocess.CompletedProcess(arguments, 0, "", "") + + monkeypatch.setattr(manager.subprocess, "run", fake_run) + + try: + manager._reload_launch_agent(tmp_path / "agent.plist") + except manager.MissionCoreLaunchAgentError as exc: + assert "bootout failed with exit code 5" in str(exc) + else: + raise AssertionError("failed bootout was accepted") diff --git a/tests/test_service_watchdog.py b/tests/test_service_watchdog.py new file mode 100644 index 0000000..1b19e88 --- /dev/null +++ b/tests/test_service_watchdog.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import json +from pathlib import Path +from threading import Event + +from k1link.service_watchdog import ( + ConsecutiveHealthGate, + MissionCoreSelfWatchdog, + MissionCoreWatchdogJournal, + MissionCoreWatchdogPolicy, + watchdog_enabled, +) + + +def test_consecutive_health_gate_resets_after_recovery() -> None: + gate = ConsecutiveHealthGate(3) + + assert gate.observe(False) is False + assert gate.observe(False) is False + assert gate.observe(True) is False + assert gate.consecutive_failures == 0 + assert gate.observe(False) is False + assert gate.observe(False) is False + assert gate.observe(False) is True + + +def test_self_watchdog_escalates_a_persistently_unhealthy_process(tmp_path: Path) -> None: + requested = Event() + forced = Event() + journal_path = tmp_path / "watchdog.jsonl" + watchdog = MissionCoreSelfWatchdog( + tmp_path, + policy=MissionCoreWatchdogPolicy( + startup_grace_seconds=0.01, + probe_interval_seconds=0.01, + probe_timeout_seconds=0.01, + consecutive_failure_limit=2, + graceful_shutdown_seconds=0.02, + ), + probe=lambda: False, + request_shutdown=requested.set, + force_shutdown=forced.set, + journal=MissionCoreWatchdogJournal(journal_path), + ) + + watchdog.start() + + assert requested.wait(0.5) + assert forced.wait(0.5) + watchdog.stop() + events = [json.loads(line)["event"] for line in journal_path.read_text().splitlines()] + assert events == [ + "watchdog-started", + "health-state-changed", + "restart-requested", + "restart-escalated", + "watchdog-stopped", + ] + + +def test_self_watchdog_leaves_a_healthy_process_running(tmp_path: Path) -> None: + probed = Event() + requested = Event() + forced = Event() + journal_path = tmp_path / "watchdog.jsonl" + + def healthy_probe() -> bool: + probed.set() + return True + + watchdog = MissionCoreSelfWatchdog( + tmp_path, + policy=MissionCoreWatchdogPolicy( + startup_grace_seconds=0.01, + probe_interval_seconds=0.01, + probe_timeout_seconds=0.01, + consecutive_failure_limit=2, + graceful_shutdown_seconds=0.02, + ), + probe=healthy_probe, + request_shutdown=requested.set, + force_shutdown=forced.set, + journal=MissionCoreWatchdogJournal(journal_path), + ) + + watchdog.start() + + assert probed.wait(0.5) + watchdog.stop() + assert requested.is_set() is False + assert forced.is_set() is False + events = [json.loads(line)["event"] for line in journal_path.read_text().splitlines()] + assert events == [ + "watchdog-started", + "health-state-changed", + "watchdog-stopped", + ] + + +def test_watchdog_journal_rotates_before_exceeding_bound(tmp_path: Path) -> None: + path = tmp_path / "watchdog.jsonl" + journal = MissionCoreWatchdogJournal(path, max_bytes=300) + + journal.append("first", payload="x" * 180) + journal.append("second", payload="y" * 180) + + assert path.is_file() + assert path.with_name("watchdog.jsonl.1").is_file() + assert json.loads(path.read_text())["event"] == "second" + + +def test_watchdog_requires_exact_enable_marker() -> None: + assert watchdog_enabled({"MISSIONCORE_SERVICE_WATCHDOG": "1"}) is True + assert watchdog_enabled({"MISSIONCORE_SERVICE_WATCHDOG": "true"}) is False + assert watchdog_enabled({}) is False