diff --git a/solstone/think/app_supervised.py b/solstone/think/app_supervised.py new file mode 100644 index 000000000..47d376ba8 --- /dev/null +++ b/solstone/think/app_supervised.py @@ -0,0 +1,33 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import logging +import os +import sys +from typing import Sequence + +FLAG = "--app-supervised" +SELECTOR_ENV = "SOLSTONE_APP_SUPERVISED" +PARENT_FD_ENV = "SOLSTONE_PARENT_FD" + + +def is_app_supervised(argv: Sequence[str] | None = None) -> bool: + args = sys.argv if argv is None else argv + return FLAG in args or os.environ.get(SELECTOR_ENV) == "1" + + +def resolve_parent_fd() -> int: + raw = os.environ.get(PARENT_FD_ENV) + if raw is None: + return 0 + try: + return int(raw) + except ValueError: + logging.getLogger(__name__).warning( + "invalid %s value %r; defaulting parent-death watch fd to stdin", + PARENT_FD_ENV, + raw, + ) + return 0 diff --git a/solstone/think/start.py b/solstone/think/start.py index 08a8c6c94..64a4deffc 100644 --- a/solstone/think/start.py +++ b/solstone/think/start.py @@ -10,6 +10,7 @@ import sys from pathlib import Path import solstone +from solstone.think.app_supervised import is_app_supervised from solstone.think.install_guard import alias_paths, install_wrappers from solstone.think.journal_io import atomic_replace from solstone.think.service import reconcile_installed_unit @@ -44,21 +45,24 @@ def _refresh_skill_links() -> None: raise RuntimeError(f"skill refresh failed with {report.error_count} error(s)") -def _refresh_for_version_marker() -> None: +def _refresh_for_version_marker(skip_reconcile: bool = False) -> None: marker_path = _version_marker_path() if _version_marker_is_current(marker_path): return _install_current_wrappers() - reconcile_installed_unit() + if not skip_reconcile: + reconcile_installed_unit() _refresh_skill_links() atomic_replace(marker_path, f"{solstone.__version__}\n") def main() -> None: + app_supervised = is_app_supervised(sys.argv) try: - reconcile_installed_unit() - _refresh_for_version_marker() + if not app_supervised: + reconcile_installed_unit() + _refresh_for_version_marker(skip_reconcile=app_supervised) except Exception: logger.exception("journal start failed during service reconciliation") sys.exit(1) diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index 6d5958cc2..e0e83783e 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -6,12 +6,14 @@ from __future__ import annotations import argparse import asyncio import concurrent.futures +import fcntl import getpass import json import logging import os import signal import socket +import stat import subprocess import sys import tempfile @@ -20,11 +22,12 @@ import time from collections import OrderedDict, deque from datetime import datetime, timezone from pathlib import Path -from typing import Any, Iterable +from typing import Any, Callable, Iterable, NoReturn import psutil from solstone.think import routines, scheduler +from solstone.think.app_supervised import FLAG, is_app_supervised, resolve_parent_fd from solstone.think.callosum import CallosumConnection, CallosumServer from solstone.think.maint import run_pending_tasks from solstone.think.models import LOCAL_MODEL, is_local_provider_needed @@ -61,6 +64,12 @@ DEFAULT_THRESHOLD = 60 CHECK_INTERVAL = 30 MAX_UPDATED_CATCHUP = 4 TEMPFAIL_DELAY = 15 # seconds to wait before retrying a tempfail exit +HANDLE_SHUTDOWN_REAP_S = 3.0 +APP_SUPERVISED_SHUTDOWN_CEILING_S = 10.0 +APP_SUPERVISED_TASK_DRAIN_S = 2.0 +APP_SUPERVISED_CHILD_STOP_S = 2.0 +APP_SUPERVISED_CALLOSUM_JOIN_S = 2.0 +PARENT_DEATH_POLL_INTERVAL_S = 1.0 STOPPED_TICKS_THRESHOLD = 2 LOCAL_SERVER_PROCESS_NAME = "llama-server" LOCAL_WEDGE_THRESHOLD = 3 @@ -88,6 +97,16 @@ _sync_conflict_shutdown: bool = False # Supervisor identity (set in main() once ref is assigned) _supervisor_ref: str | None = None _supervisor_start: float | None = None +_parent_death_sigterm_sent = threading.Event() + + +def app_supervised_graceful_budget_s() -> float: + """Worst-case non-wedge graceful time after watcher self-SIGTERM.""" + return ( + HANDLE_SHUTDOWN_REAP_S + + APP_SUPERVISED_TASK_DRAIN_S + + APP_SUPERVISED_CALLOSUM_JOIN_S + ) def _sd_notify(state: str) -> None: @@ -103,6 +122,101 @@ def _sd_notify(state: str) -> None: logging.warning("sd_notify failed: %s", exc) +def _parent_fd_is_usable(fd: int) -> bool: + try: + mode = os.fstat(fd).st_mode + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + except OSError: + return False + access_mode = flags & os.O_ACCMODE + return stat.S_ISFIFO(mode) and access_mode in (os.O_RDONLY, os.O_RDWR) + + +def wait_until_parent_gone( + parent_fd: int, *, poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S +) -> str: + if _parent_fd_is_usable(parent_fd): + while True: + try: + data = os.read(parent_fd, 4096) + except OSError: + return "fd-error" + if data == b"": + return "eof" + + while True: + if os.getppid() == 1: + return "orphaned" + time.sleep(poll_interval) + + +def enforce_parent_death_shutdown_deadline( + reason: str, + *, + ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S, + managed_procs: Iterable[RunnerManagedProcess] | None = None, + sent_event: "threading.Event | None" = None, + kill: Callable[[int, int], None] | None = None, + exit_now: Callable[[int], NoReturn] | None = None, + sleep: Callable[[float], None] | None = None, +) -> None: + del reason + kill_fn = kill or os.kill + exit_now_fn = exit_now or os._exit + sleep_fn = sleep or time.sleep + sigterm_sent = sent_event if sent_event is not None else _parent_death_sigterm_sent + + if not sigterm_sent.is_set(): + sigterm_sent.set() + kill_fn(os.getpid(), signal.SIGTERM) + + sleep_fn(ceiling) + + procs = managed_procs if managed_procs is not None else _managed_procs + for managed in procs: + if not managed.is_running(): + continue + try: + managed.process.kill() + except Exception: + logger.exception( + "parent-death backstop: SIGKILL failed for %s", managed.name + ) + + exit_now_fn(1) + + +def _parent_death_watcher_main( + parent_fd: int, + *, + poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S, + ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S, +) -> None: + reason = wait_until_parent_gone(parent_fd, poll_interval=poll_interval) + logger.warning( + "parent-death detected (%s); converging to graceful shutdown", reason + ) + enforce_parent_death_shutdown_deadline(reason, ceiling=ceiling) + + +def start_parent_death_watcher( + parent_fd: int | None = None, + *, + poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S, + ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S, +) -> threading.Thread: + fd = parent_fd if parent_fd is not None else resolve_parent_fd() + thread = threading.Thread( + target=_parent_death_watcher_main, + args=(fd,), + kwargs={"poll_interval": poll_interval, "ceiling": ceiling}, + name="parent-death-watcher", + daemon=True, + ) + thread.start() + return thread + + def _candidate_journal(proc: "psutil.Process") -> Path | None: """Return the resolved SOLSTONE_JOURNAL of ``proc``, or None on any failure. @@ -970,8 +1084,12 @@ def _start_termination_thread( thread.start() -def _stop_process(managed: RunnerManagedProcess) -> None: +def _stop_process( + managed: RunnerManagedProcess, *, timeout_cap: float | None = None +) -> None: timeout = _SERVICE_STATE.get(managed.name, {}).get("shutdown_timeout", 15) + if timeout_cap is not None: + timeout = min(timeout, timeout_cap) _terminate_managed(managed, timeout, reason="shutdown") managed.cleanup() @@ -1557,7 +1675,7 @@ def wait_for_convey_ready( return False -def stop_callosum_in_process() -> None: +def stop_callosum_in_process(join_timeout: float = 5.0) -> None: """Stop the in-process Callosum server.""" global _callosum_server, _callosum_thread @@ -1566,7 +1684,7 @@ def stop_callosum_in_process() -> None: _callosum_server.stop() if _callosum_thread: - _callosum_thread.join(timeout=5) + _callosum_thread.join(timeout=join_timeout) if _callosum_thread.is_alive(): logging.warning("Callosum server thread did not stop cleanly") @@ -2183,6 +2301,14 @@ def parse_args() -> argparse.ArgumentParser: action="store_true", help="Disable periodic task scheduler", ) + parser.add_argument( + FLAG, + action="store_true", + help=( + "App-supervised mode: skip all service-unit work and self-exit when " + "the parent process dies (used by the macOS app)." + ), + ) parser.add_argument( "--remote", type=str, @@ -2206,7 +2332,7 @@ def handle_shutdown(signum, frame): except Exception: logger.exception("shutdown: terminate failed for %s", managed.name) - deadline = time.monotonic() + 3.0 + deadline = time.monotonic() + HANDLE_SHUTDOWN_REAP_S while time.monotonic() < deadline: if all(not managed.is_running() for managed in live): break @@ -2258,6 +2384,7 @@ def main() -> None: journal_info = get_journal_info() args = setup_cli(parser) + app_supervised = is_app_supervised(sys.argv) _ensure_venv_bin_on_path() journal_path = _get_journal_path() @@ -2285,7 +2412,6 @@ def main() -> None: health_dir = journal_path / "health" lock_path = health_dir / "supervisor.lock" pid_path = health_dir / "supervisor.pid" - import fcntl lock_fd = open(lock_path, "w") try: @@ -2499,6 +2625,8 @@ def main() -> None: print(" Supervisor ready", flush=True) _sd_notify("READY=1") signal_ready() + if app_supervised: + start_parent_death_watcher() asyncio.run( supervise( daily=daily_enabled, @@ -2523,11 +2651,13 @@ def main() -> None: print("\nShutting down gracefully (this may take a moment)...", flush=True) if _task_queue: - _task_queue.shutdown(timeout=10) + task_drain_timeout = APP_SUPERVISED_TASK_DRAIN_S if app_supervised else 10 + _task_queue.shutdown(timeout=task_drain_timeout) # Stop services in reverse order + child_stop_timeout = APP_SUPERVISED_CHILD_STOP_S if app_supervised else None for managed in reversed(procs): - _stop_process(managed) + _stop_process(managed, timeout_cap=child_stop_timeout) if schedule_enabled: try: @@ -2541,7 +2671,10 @@ def main() -> None: logging.info("Supervisor disconnected from Callosum") # Stop in-process Callosum server last - stop_callosum_in_process() + callosum_join_timeout = ( + APP_SUPERVISED_CALLOSUM_JOIN_S if app_supervised else 5.0 + ) + stop_callosum_in_process(join_timeout=callosum_join_timeout) logging.info("Supervisor shutdown complete.") print("Shutdown complete.", flush=True) diff --git a/tests/test_app_supervised.py b/tests/test_app_supervised.py new file mode 100644 index 000000000..fa00ed32b --- /dev/null +++ b/tests/test_app_supervised.py @@ -0,0 +1,37 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.think.app_supervised import ( + FLAG, + PARENT_FD_ENV, + SELECTOR_ENV, + is_app_supervised, + resolve_parent_fd, +) + + +def test_is_app_supervised_uses_cli_flag_or_env_with_or_semantics(monkeypatch): + monkeypatch.delenv(SELECTOR_ENV, raising=False) + assert is_app_supervised(["journal", "start"]) is False + + assert is_app_supervised(["journal", "start", FLAG]) is True + + monkeypatch.setenv(SELECTOR_ENV, "1") + assert is_app_supervised(["journal", "start"]) is True + + monkeypatch.setenv(SELECTOR_ENV, "0") + assert is_app_supervised(["journal", "start"]) is False + assert is_app_supervised(["journal", "start", FLAG]) is True + + +def test_resolve_parent_fd_defaults_to_stdin_and_ignores_bad_env(monkeypatch): + monkeypatch.delenv(PARENT_FD_ENV, raising=False) + assert resolve_parent_fd() == 0 + + monkeypatch.setenv(PARENT_FD_ENV, "7") + assert resolve_parent_fd() == 7 + + monkeypatch.setenv(PARENT_FD_ENV, "garbage") + assert resolve_parent_fd() == 0 diff --git a/tests/test_journal_start.py b/tests/test_journal_start.py index 5f54c0e7f..2a540bc79 100644 --- a/tests/test_journal_start.py +++ b/tests/test_journal_start.py @@ -3,12 +3,14 @@ from __future__ import annotations +import sys from pathlib import Path from unittest.mock import MagicMock import pytest from solstone.think import start +from solstone.think.app_supervised import FLAG, SELECTOR_ENV from solstone.think.service import Reconciled @@ -52,6 +54,80 @@ def test_start_version_marker_mismatch_triggers_refresh(monkeypatch, tmp_path): assert marker.read_text(encoding="utf-8") == f"{start.solstone.__version__}\n" +@pytest.mark.parametrize("selector", ["flag", "env"]) +def test_app_supervised_start_skips_reconcile_but_refreshes_version_marker_artifacts( + selector, monkeypatch, tmp_path +): + marker = tmp_path / ".last-start-version" + marker.write_text("old-version\n", encoding="utf-8") + _patch_marker(monkeypatch, marker) + monkeypatch.delenv(SELECTOR_ENV, raising=False) + argv = ["journal", "start"] + if selector == "flag": + argv.append(FLAG) + else: + monkeypatch.setenv(SELECTOR_ENV, "1") + monkeypatch.setattr(sys, "argv", argv) + + calls: list[str] = [] + reconcile = MagicMock(return_value=Reconciled(False, None, None, None)) + supervisor = MagicMock() + monkeypatch.setattr(start, "reconcile_installed_unit", reconcile) + monkeypatch.setattr( + start, "_install_current_wrappers", lambda: calls.append("wrappers") + ) + monkeypatch.setattr(start, "_refresh_skill_links", lambda: calls.append("skills")) + monkeypatch.setattr("solstone.think.supervisor.main", supervisor) + + start.main() + + reconcile.assert_not_called() + assert calls == ["wrappers", "skills"] + assert marker.read_text(encoding="utf-8") == f"{start.solstone.__version__}\n" + supervisor.assert_called_once_with() + + +def test_default_start_reconciles_both_sites_with_stale_marker(monkeypatch, tmp_path): + marker = tmp_path / ".last-start-version" + marker.write_text("old-version\n", encoding="utf-8") + _patch_marker(monkeypatch, marker) + monkeypatch.delenv(SELECTOR_ENV, raising=False) + monkeypatch.setattr(sys, "argv", ["journal", "start"]) + + reconcile = MagicMock(return_value=Reconciled(False, None, None, None)) + supervisor = MagicMock() + monkeypatch.setattr(start, "reconcile_installed_unit", reconcile) + monkeypatch.setattr(start, "_install_current_wrappers", lambda: None) + monkeypatch.setattr(start, "_refresh_skill_links", lambda: None) + monkeypatch.setattr("solstone.think.supervisor.main", supervisor) + + start.main() + + assert reconcile.call_count == 2 + supervisor.assert_called_once_with() + + +@pytest.mark.skipif(sys.platform != "linux", reason="linux reconcile regression") +def test_default_start_reconciles_both_sites_with_stale_marker_on_linux( + monkeypatch, tmp_path +): + marker = tmp_path / ".last-start-version" + marker.write_text("old-version\n", encoding="utf-8") + _patch_marker(monkeypatch, marker) + monkeypatch.delenv(SELECTOR_ENV, raising=False) + monkeypatch.setattr(sys, "argv", ["journal", "start"]) + + reconcile = MagicMock(return_value=Reconciled(False, None, None, None)) + monkeypatch.setattr(start, "reconcile_installed_unit", reconcile) + monkeypatch.setattr(start, "_install_current_wrappers", lambda: None) + monkeypatch.setattr(start, "_refresh_skill_links", lambda: None) + monkeypatch.setattr("solstone.think.supervisor.main", MagicMock()) + + start.main() + + assert reconcile.call_count == 2 + + def test_start_version_marker_match_is_noop(monkeypatch, tmp_path): marker = tmp_path / ".last-start-version" marker.write_text(f"{start.solstone.__version__}\n", encoding="utf-8") diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 97beaefb5..b6a9c9918 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -149,6 +149,20 @@ def test_parse_args_remote_flag_optional(): assert args.remote is None +def test_parse_args_app_supervised_flag(): + mod = importlib.reload(importlib.import_module("solstone.think.supervisor")) + + parser = mod.parse_args() + args = parser.parse_args( + ["5016", "--app-supervised", "--no-daily", "--no-schedule"] + ) + + assert args.port == 5016 + assert args.app_supervised is True + assert args.no_daily is True + assert args.no_schedule is True + + def test_parse_args_lifecycle_verb_hint(monkeypatch, capsys): mod = importlib.reload(importlib.import_module("solstone.think.supervisor")) monkeypatch.setattr(sys, "argv", ["sol", "supervisor", "stop"]) @@ -224,7 +238,7 @@ def test_graceful_shutdown_calls_stop_process_for_each_managed_proc( monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0) monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None) monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None) - monkeypatch.setattr(mod, "stop_callosum_in_process", lambda: None) + monkeypatch.setattr(mod, "stop_callosum_in_process", lambda **_kwargs: None) monkeypatch.setattr(mod, "wait_for_convey_ready", lambda _proc: True) monkeypatch.setattr(mod, "_maybe_submit_startup_digest", lambda *, no_cortex: None) @@ -262,7 +276,7 @@ def test_graceful_shutdown_calls_stop_process_for_each_managed_proc( monkeypatch.setattr( mod, "_stop_process", - lambda managed: stop_order.append(managed.name), + lambda managed, **_kwargs: stop_order.append(managed.name), ) def interrupt_supervise(coro): @@ -279,6 +293,151 @@ def test_graceful_shutdown_calls_stop_process_for_each_managed_proc( assert stop_order == ["spl", "cortex", "sense", "convey"] +def _run_supervisor_main_for_shutdown_knobs(tmp_path, monkeypatch, *, argv): + mod = importlib.reload(importlib.import_module("solstone.think.supervisor")) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + monkeypatch.delenv("SOL_SUPERVISOR_SPAWNED", raising=False) + monkeypatch.delenv("SOLSTONE_APP_SUPERVISED", raising=False) + monkeypatch.setattr(sys, "argv", argv) + monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: (0, 0)) + monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0) + monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None) + monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None) + monkeypatch.setattr(mod, "is_local_provider_needed", lambda: False) + monkeypatch.setattr(mod, "_maybe_submit_startup_digest", lambda *, no_cortex: None) + + class FakeCallosumConnection: + def __init__(self, *args, **kwargs): + pass + + def start(self, *args, **kwargs): + pass + + def emit(self, *args, **kwargs): + pass + + def stop(self): + pass + + monkeypatch.setattr(mod, "CallosumConnection", FakeCallosumConnection) + + managed = _TaskManagedStub(cmd=["journal", "sense"]) + managed.name = "sense" + monkeypatch.setattr(mod, "start_sense", lambda: managed) + + events: list[str] = [] + captures: dict[str, list] = { + "task_shutdown": [], + "stop_process": [], + "callosum_join": [], + } + + class FakeTaskQueue: + def __init__(self, *args, **kwargs): + pass + + def set_ready(self): + pass + + def shutdown(self, *, timeout): + captures["task_shutdown"].append(timeout) + + monkeypatch.setattr(mod, "TaskQueue", FakeTaskQueue) + monkeypatch.setattr(mod, "signal_ready", lambda: events.append("ready")) + + def record_watcher_start(): + events.append("watcher") + assert mod._managed_procs == [managed] + + monkeypatch.setattr(mod, "start_parent_death_watcher", record_watcher_start) + monkeypatch.setattr( + mod, + "_stop_process", + lambda proc, *, timeout_cap=None: captures["stop_process"].append( + (proc.name, timeout_cap) + ), + ) + monkeypatch.setattr( + mod, + "stop_callosum_in_process", + lambda *, join_timeout=5.0: captures["callosum_join"].append(join_timeout), + ) + exit_now = MagicMock() + monkeypatch.setattr(mod.os, "_exit", exit_now) + + def interrupt_supervise(coro): + events.append("run") + coro.close() + raise KeyboardInterrupt + + monkeypatch.setattr(mod.asyncio, "run", interrupt_supervise) + + try: + mod.main() + finally: + os.environ.pop("SOL_SUPERVISOR_SPAWNED", None) + + return mod, captures, events, exit_now + + +def test_app_supervised_main_uses_watcher_and_compressed_shutdown_knobs( + tmp_path, monkeypatch +): + from solstone.think import install_guard, service, skills_cli + + reconcile = MagicMock() + install_wrappers = MagicMock() + install_project = MagicMock() + monkeypatch.setattr(service, "reconcile_installed_unit", reconcile) + monkeypatch.setattr(install_guard, "install_wrappers", install_wrappers) + monkeypatch.setattr(skills_cli, "install_project", install_project) + + mod, captures, events, exit_now = _run_supervisor_main_for_shutdown_knobs( + tmp_path, + monkeypatch, + argv=[ + "supervisor", + "0", + "--app-supervised", + "--no-daily", + "--no-schedule", + "--no-convey", + "--no-cortex", + "--no-spl", + ], + ) + + assert events == ["ready", "watcher", "run"] + assert captures["task_shutdown"] == [mod.APP_SUPERVISED_TASK_DRAIN_S] + assert captures["stop_process"] == [("sense", mod.APP_SUPERVISED_CHILD_STOP_S)] + assert captures["callosum_join"] == [mod.APP_SUPERVISED_CALLOSUM_JOIN_S] + exit_now.assert_not_called() + reconcile.assert_not_called() + install_wrappers.assert_not_called() + install_project.assert_not_called() + + +def test_default_main_uses_default_shutdown_knobs(tmp_path, monkeypatch): + mod, captures, events, _exit_now = _run_supervisor_main_for_shutdown_knobs( + tmp_path, + monkeypatch, + argv=[ + "supervisor", + "0", + "--no-daily", + "--no-schedule", + "--no-convey", + "--no-cortex", + "--no-spl", + ], + ) + + assert events == ["ready", "run"] + assert captures["task_shutdown"] == [10] + assert captures["stop_process"] == [("sense", None)] + assert captures["callosum_join"] == [5.0] + + def test_get_command_name(): """Test command name extraction for queue serialization.""" mod = importlib.import_module("solstone.think.supervisor") diff --git a/tests/test_supervisor_shutdown.py b/tests/test_supervisor_shutdown.py index ef5169aa4..cccbda720 100644 --- a/tests/test_supervisor_shutdown.py +++ b/tests/test_supervisor_shutdown.py @@ -3,6 +3,9 @@ """Tests for handle_shutdown's reap pass.""" +import os +import signal +import threading from unittest.mock import MagicMock import pytest @@ -87,3 +90,125 @@ def test_reap_swallows_oserror_on_kill(monkeypatch, caplog): supervisor.handle_shutdown(15, None) assert "shutdown: kill failed for bad" in caplog.text + + +def test_parent_watcher_waits_for_pipe_eof_without_ppid_poll(monkeypatch): + read_fd, write_fd = os.pipe() + getppid = MagicMock() + monkeypatch.setattr(supervisor.os, "getppid", getppid) + os.close(write_fd) + try: + assert supervisor.wait_until_parent_gone(read_fd) == "eof" + finally: + os.close(read_fd) + getppid.assert_not_called() + + +@pytest.mark.parametrize("fd_kind", ["devnull", "bad"]) +def test_parent_watcher_non_pipe_and_bad_fd_fall_back_to_ppid(fd_kind, monkeypatch): + fd = os.open(os.devnull, os.O_RDONLY) if fd_kind == "devnull" else -1 + pids = iter([999, 1]) + monkeypatch.setattr(supervisor.os, "getppid", lambda: next(pids)) + monkeypatch.setattr(supervisor.time, "sleep", lambda _seconds: None) + monkeypatch.setattr( + supervisor.os, + "read", + lambda *_args, **_kwargs: pytest.fail("non-pipe fd should not be read"), + ) + try: + assert supervisor.wait_until_parent_gone(fd) == "orphaned" + finally: + if fd_kind == "devnull": + os.close(fd) + + +def test_parent_watcher_shutdown_deadline_sends_sigterm_once_kills_children(): + sent_event = threading.Event() + managed = FakeManaged("stuck", exits_after_terminate=False) + kill = MagicMock() + exit_now = MagicMock() + + supervisor.enforce_parent_death_shutdown_deadline( + "eof", + ceiling=0, + managed_procs=[managed], + sent_event=sent_event, + kill=kill, + exit_now=exit_now, + sleep=lambda _seconds: None, + ) + + kill.assert_called_once_with(os.getpid(), signal.SIGTERM) + assert managed.process.kill.called + exit_now.assert_called_once_with(1) + + +def test_parent_watcher_shutdown_deadline_does_not_resend_sigterm(): + sent_event = threading.Event() + sent_event.set() + managed = FakeManaged("stuck", exits_after_terminate=False) + kill = MagicMock() + exit_now = MagicMock() + + supervisor.enforce_parent_death_shutdown_deadline( + "eof", + ceiling=0, + managed_procs=[managed], + sent_event=sent_event, + kill=kill, + exit_now=exit_now, + sleep=lambda _seconds: None, + ) + + kill.assert_not_called() + assert managed.process.kill.called + exit_now.assert_called_once_with(1) + + +def test_app_supervised_graceful_budget_stays_under_hard_ceiling(): + assert ( + supervisor.app_supervised_graceful_budget_s() + < supervisor.APP_SUPERVISED_SHUTDOWN_CEILING_S + ) + + +def test_stop_process_applies_timeout_cap(monkeypatch): + managed = MagicMock() + managed.name = "spl" + supervisor._SERVICE_STATE.clear() + supervisor._SERVICE_STATE["spl"] = { + "restart": True, + "shutdown_timeout": 15, + } + calls: list[tuple[float, str]] = [] + monkeypatch.setattr( + supervisor, + "_terminate_managed", + lambda _managed, timeout, *, reason: calls.append((timeout, reason)), + ) + + supervisor._stop_process(managed, timeout_cap=2.0) + + assert calls == [(2.0, "shutdown")] + managed.cleanup.assert_called_once_with() + + +def test_stop_process_defaults_to_service_timeout(monkeypatch): + managed = MagicMock() + managed.name = "spl" + supervisor._SERVICE_STATE.clear() + supervisor._SERVICE_STATE["spl"] = { + "restart": True, + "shutdown_timeout": 15, + } + calls: list[tuple[float, str]] = [] + monkeypatch.setattr( + supervisor, + "_terminate_managed", + lambda _managed, timeout, *, reason: calls.append((timeout, reason)), + ) + + supervisor._stop_process(managed) + + assert calls == [(15, "shutdown")] + managed.cleanup.assert_called_once_with() diff --git a/tests/test_supervisor_sync_gate.py b/tests/test_supervisor_sync_gate.py index eb81f8634..b3517e1ce 100644 --- a/tests/test_supervisor_sync_gate.py +++ b/tests/test_supervisor_sync_gate.py @@ -159,9 +159,9 @@ def test_finally_clears_self_heartbeat_on_normal_shutdown(tmp_path, monkeypatch) ], ) monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None) - monkeypatch.setattr(mod, "stop_callosum_in_process", lambda: None) + monkeypatch.setattr(mod, "stop_callosum_in_process", lambda **_kwargs: None) monkeypatch.setattr(mod, "start_sense", lambda: SimpleNamespace(name="sense")) - monkeypatch.setattr(mod, "_stop_process", lambda _managed: None) + monkeypatch.setattr(mod, "_stop_process", lambda _managed, **_kwargs: None) class FakeCallosumConnection: def __init__(self, *args, **kwargs): @@ -210,9 +210,9 @@ def test_finally_does_not_clear_self_heartbeat_after_mid_run_conflict( ], ) monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None) - monkeypatch.setattr(mod, "stop_callosum_in_process", lambda: None) + monkeypatch.setattr(mod, "stop_callosum_in_process", lambda **_kwargs: None) monkeypatch.setattr(mod, "start_sense", lambda: SimpleNamespace(name="sense")) - monkeypatch.setattr(mod, "_stop_process", lambda _managed: None) + monkeypatch.setattr(mod, "_stop_process", lambda _managed, **_kwargs: None) class FakeCallosumConnection: def __init__(self, *args, **kwargs): diff --git a/vpe/workspace/l1-app-supervised-design.md b/vpe/workspace/l1-app-supervised-design.md new file mode 100644 index 000000000..3eaa94096 --- /dev/null +++ b/vpe/workspace/l1-app-supervised-design.md @@ -0,0 +1,211 @@ +# L1 Design: `journal start --app-supervised` + +Opt-in app-supervised start mode for the L2 macOS Swift child-runner. This +stage designs the contract only; no production code is implemented here. + +## Contract Pinned For L2 + +- Selector constants live in new `solstone/think/app_supervised.py`: + - `FLAG = "--app-supervised"` + - `SELECTOR_ENV = "SOLSTONE_APP_SUPERVISED"` + - `PARENT_FD_ENV = "SOLSTONE_PARENT_FD"` +- App-supervised selection is OR semantics: + - true if `--app-supervised` is present in argv. + - true if `SOLSTONE_APP_SUPERVISED=1`. + - false otherwise. There is no "off" override. +- Env namespace rationale: new Swift child-runner cross-process vars use + `SOLSTONE_`; `SOL_*` remains the internal supervisor-spawn convention + (`SOL_SUPERVISOR_SPAWNED`, `SOL_SKIP_SUPERVISOR_CHECK`). +- Parent fd handshake: + - Parent creates a pipe, keeps the write end open, and passes the read fd as + `SOLSTONE_PARENT_FD=`. + - EOF on that read fd means parent is dead. + - If `SOLSTONE_PARENT_FD` is absent or not parseable, default watched fd is + stdin, fd `0`. + - Usability gate: only EOF-watch a fd when it is readable and + `stat.S_ISFIFO(os.fstat(fd).st_mode)` is true. Non-pipe, bad, or unreadable + fd falls back to the `getppid()` belt. This prevents `/dev/null` stdin from + causing instant false shutdown. +- Watcher integration: + - In app-supervised mode only, `supervisor.main()` starts one daemon thread + after children are started, after `_managed_procs = procs`, and after + `signal_ready()`, immediately before `asyncio.run(supervise(...))`. + - Primary path for usable pipe fd: block in `os.read(fd, ...)`; EOF returns + parent-gone; `OSError` after the fd was accepted is fail-safe parent-gone. + Nonempty bytes are ignored and the read continues. + - Belt path for no usable pipe fd: poll `os.getppid() == 1` every about + `1.0s`. This is the only polling path. + - Convergence: parent-gone detection sends `os.kill(os.getpid(), + signal.SIGTERM)` at most once and uses the existing `handle_shutdown` + signal path. No second teardown sequence is added. +- Backstop: the same daemon thread waits `10.0s` from sending SIGTERM. If the + supervisor process is still alive, it best-effort SIGKILLs every entry in + `_managed_procs`, then calls `os._exit(1)`. +- Compressed graceful timeout ceiling: + - Pin `HANDLE_SHUTDOWN_REAP_S = 3.0` and use it in `handle_shutdown`; do not + branch the shared signal handler by mode. + - Pin `APP_SUPERVISED_SHUTDOWN_CEILING_S = 10.0`. + - In app-supervised mode only, compress the `finally` block knobs: + `_task_queue.shutdown(timeout=APP_SUPERVISED_TASK_DRAIN_S)`, + `_stop_process(..., timeout_cap=APP_SUPERVISED_CHILD_STOP_S)`, and + `stop_callosum_in_process(join_timeout=APP_SUPERVISED_CALLOSUM_JOIN_S)`. + - Pin the compressed knobs at `2.0s` each: + `APP_SUPERVISED_TASK_DRAIN_S = 2.0`, + `APP_SUPERVISED_CHILD_STOP_S = 2.0`, and + `APP_SUPERVISED_CALLOSUM_JOIN_S = 2.0`. + - Add `app_supervised_graceful_budget_s() -> float`, returning + `HANDLE_SHUTDOWN_REAP_S + APP_SUPERVISED_TASK_DRAIN_S + + APP_SUPERVISED_CALLOSUM_JOIN_S`. + - New AC: the budget must stay strictly under + `APP_SUPERVISED_SHUTDOWN_CEILING_S` (`3.0 + 2.0 + 2.0 = 7.0 < 10.0`) so a + non-wedge app-supervised quit completes without the `os._exit` backstop. + - `os._exit` remains reserved for a genuine wedge: a child that survives + SIGKILL or hung teardown after the compressed normal path. + - `os._exit` bypasses `finally` cleanup (`clear_ready`, callosum stop, lock + release), but the kernel releases flock/fds on process exit, and SIGKILLed + children free their ports. + +## Residual Confirmations + +- A. Skill refresh does no service-unit work. + - `start._refresh_skill_links()` calls only `install_project(...)` + (`solstone/think/start.py:41-42`). + - `install_project()` discovers sources, creates project skill dirs, installs + symlinks, removes stale symlinks, and returns a report + (`solstone/think/skills_cli.py:384-399`). + - Project source install is symlink/unlink only + (`solstone/think/skills_cli.py:323-362`); stale cleanup is unlink only + (`solstone/think/skills_cli.py:365-381`). + - `rg "launchctl|systemctl|plist|systemd|subprocess" solstone/think/skills_cli.py` + returns no matches. Service-unit work is in `solstone/think/service.py`, not + skill refresh. +- B. Supervision and SIGTERM handling run on the main thread. + - `supervisor.main()` is the synchronous CLI entry (`supervisor.py:2254`) and + installs `signal.signal(SIGINT/SIGTERM, handle_shutdown)` before the + supervise loop (`supervisor.py:2344-2346`). + - `asyncio.run(supervise(...))` is called directly in that same `main()` + (`supervisor.py:2502-2508`). + - `rg "add_signal_handler|loop\\.add_signal_handler" solstone/think/supervisor.py` + returns no matches, so asyncio does not replace the SIGTERM handler. + - A daemon thread's `os.kill(os.getpid(), signal.SIGTERM)` therefore routes + through `handle_shutdown()`, sets `shutdown_requested`, and raises + `KeyboardInterrupt` out of the `asyncio.run` call + (`supervisor.py:2194-2235`, `supervisor.py:2509-2511`). +- C. Local model server is covered by `_managed_procs`. + - Darwin MLX path launches a managed process in `_start_mlx_local_server()` + (`supervisor.py:1362-1399`), and `start_local_server()` returns it on Darwin + (`supervisor.py:1424-1427`). + - Non-Darwin local path launches a managed process at + `supervisor.py:1471-1494`. + - Startup appends the returned local process to `procs` + (`supervisor.py:2439-2442`), then assigns `_managed_procs = procs` + (`supervisor.py:2455-2456`). + - Runtime `start_local` requests also append to `_managed_procs` + (`supervisor.py:1190-1192`). Both `handle_shutdown` and the watcher backstop + can sweep the local server. + +## Proposed Code Shape + +- Add `solstone/think/app_supervised.py`. + - `is_app_supervised(argv: Sequence[str] | None = None) -> bool` + - `resolve_parent_fd() -> int` + - Constants only: `FLAG`, `SELECTOR_ENV`, `PARENT_FD_ENV`. + - Keep it lightweight so `start.py` can scan argv/env without importing the + full supervisor module. +- Update `solstone/think/start.py`. + - Import `is_app_supervised`. + - Change `_refresh_for_version_marker(skip_reconcile: bool = False) -> None`. + - At current `start.py:53`, call `reconcile_installed_unit()` only when + `not skip_reconcile`. + - In `main()`, compute `app_supervised = is_app_supervised(sys.argv)`. + - At current `start.py:60`, call `reconcile_installed_unit()` only when + `not app_supervised`. + - At current `start.py:61`, call + `_refresh_for_version_marker(skip_reconcile=app_supervised)`. + - Do not argparse or consume args in `start.py`; `supervisor.main()` reparses + the same argv normally. +- Update `solstone/think/supervisor.py`. + - Import `stat`, likely `fcntl`, and the app-supervised constants/helpers. + - `parse_args()` adds `FLAG` as `action="store_true"` so + `setup_cli(parser).parse_args()` accepts it (`utils.py:902-921`, + `supervisor.py:2143-2191`). + - Add `HANDLE_SHUTDOWN_REAP_S = 3.0`. + - Add `APP_SUPERVISED_SHUTDOWN_CEILING_S = 10.0`. + - Add `APP_SUPERVISED_TASK_DRAIN_S = 2.0`. + - Add `APP_SUPERVISED_CHILD_STOP_S = 2.0`. + - Add `APP_SUPERVISED_CALLOSUM_JOIN_S = 2.0`. + - Add `PARENT_DEATH_POLL_INTERVAL_S = 1.0`. + - Add `app_supervised_graceful_budget_s() -> float`. + - Add `_parent_death_sigterm_sent = threading.Event()`. + - Add `_parent_fd_is_usable(fd: int) -> bool`: return true only for a readable + pipe/FIFO. Use `os.fstat` plus `stat.S_ISFIFO`; check access mode so a + write-only pipe fd falls back instead of becoming a read-error shutdown. + - Add `wait_until_parent_gone(parent_fd: int, *, poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S) -> str`. + - Add `enforce_parent_death_shutdown_deadline(reason: str, *, ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S, managed_procs: Iterable[RunnerManagedProcess] | None = None, sent_event: threading.Event | None = None, kill: Callable[[int, int], None] | None = None, exit_now: Callable[[int], NoReturn] | None = None, monotonic: Callable[[], float] | None = None, sleep: Callable[[float], None] | None = None) -> None`. + - Add `_parent_death_watcher_main(parent_fd: int, *, poll_interval: float, ceiling: float) -> None`. + - Add `start_parent_death_watcher(parent_fd: int | None = None, *, poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S, ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S) -> threading.Thread`. + - In `main()`, compute app-supervised selection with `is_app_supervised(sys.argv)` + and start the watcher only between current `signal_ready()` + (`supervisor.py:2501`) and `asyncio.run(...)` (`supervisor.py:2502`). + +## Implementation Sequence + +1. Add `app_supervised.py` constants and pure helper tests. +2. Update `supervisor.parse_args()` to accept `--app-supervised`. +3. Update `start.py` reconcile gating and marker-refresh signature. +4. Add supervisor watcher helper functions, isolated from `main()` first. +5. Wire watcher startup into `supervisor.main()` after `signal_ready()`. +6. Add tests in the order below. + +## Test Plan + +- `tests/test_app_supervised.py::test_is_app_supervised_uses_cli_flag_or_env_with_or_semantics` + - Covers no selector false, flag true, env `1` true, env `0` false, and flag + winning over env `0`. +- `tests/test_app_supervised.py::test_resolve_parent_fd_defaults_to_stdin_and_ignores_bad_env` + - Covers absent env -> `0`, parseable env -> int, bad env -> `0`. +- `tests/test_supervisor.py::test_parse_args_app_supervised_flag` + - Mirrors existing parse-arg tests around `tests/test_supervisor.py:132-149`; + asserts parser accepts `--app-supervised` without disturbing positional port + or `--no-*` flags. +- `tests/test_journal_start.py::test_app_supervised_start_skips_reconcile_but_refreshes_version_marker_artifacts` + - Mirrors reconcile-gating tests in `tests/test_journal_start.py`. + - With old marker and app-supervised selector, assert top-level reconcile and + marker-internal reconcile are skipped, wrappers/skills/marker still run, and + supervisor is invoked. +- `tests/test_supervisor.py::test_supervisor_starts_parent_watcher_only_after_ready_in_app_supervised_mode` + - Mirrors `test_graceful_shutdown_calls_stop_process_for_each_managed_proc` + (`tests/test_supervisor.py:211-279`). + - Capture event order: `_managed_procs` assigned, `signal_ready`, watcher + start, `asyncio.run`. Also assert default mode does not start the watcher. +- `tests/test_supervisor_shutdown.py::test_parent_watcher_waits_for_pipe_eof_without_ppid_poll` + - Mirrors shutdown helper style in `tests/test_supervisor_shutdown.py`. + - Pass an `os.pipe()` read fd, close the write fd, assert EOF reason and no + `getppid()` polling. +- `tests/test_supervisor_shutdown.py::test_parent_watcher_non_pipe_and_bad_fd_fall_back_to_ppid` + - Parametrize tempfile or `/dev/null` fd plus invalid fd. Assert fallback polls + until mocked `getppid()` returns `1`; no EOF read is attempted on non-pipe. +- `tests/test_supervisor_shutdown.py::test_parent_watcher_shutdown_deadline_sends_sigterm_once_kills_children_and_hard_exits` + - Patch `os.kill`, `os._exit`, clock, and sleep through injected callables. + - Assert one self-SIGTERM, SIGKILL for running fake managed processes after + the ceiling, and `exit_now(1)`. Repeat with pre-set `sent_event` to assert + no duplicate self-SIGTERM. +- `tests/test_supervisor_shutdown.py::test_app_supervised_graceful_budget_stays_under_hard_ceiling` + - New AC: assert `app_supervised_graceful_budget_s()` is strictly less than + `APP_SUPERVISED_SHUTDOWN_CEILING_S`. +- `tests/test_supervisor_shutdown.py::test_stop_process_applies_timeout_cap` + - Assert `_stop_process(..., timeout_cap=2.0)` applies `min(service_timeout, + timeout_cap)` and the default path still uses the service timeout. + +## Risks And Open Questions + +- No hard conflict found between pinned decisions and current code. +- Implementation note: `stat.S_ISFIFO` alone does not prove the fd is readable. + To satisfy the pinned "unreadable fd -> getppid fallback" rule, implement the + usability gate with a read-access check before entering the blocking + `os.read` loop. +- Backstop `os._exit(1)` intentionally bypasses existing cleanup, but normal + app-supervised shutdown should finish under the hard ceiling because the + `finally` block timeouts are compressed in that mode only. Tests should keep + `os._exit` behind an injected `exit_now` so unit tests never terminate the + pytest process.