fix(service): supervise canonical Mission Core lifecycle

2026-08-25 21:58:22 +03:00
parent fb5bf943c9
commit f50e0077c5
10 changed files with 1095 additions and 6 deletions
@@ -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.
@@ -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 <current-sha256> \
  --expected-desired-sha256 <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.
@@ -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())
@@ -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.")
        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:
@@ -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()
@@ -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,
    }
@@ -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
@@ -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
@@ -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")
@@ -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