diff --git a/CHANGELOG.md b/CHANGELOG.md index e25f1f04ef..857ce93e03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), third-party provider plugins published before the rename are still discovered, and it is not a CLI surface. +### Fixed + +- **Supervisor watchdog restarts are resumable and bounded.** A wedged + coordinator receives SIGHUP, preserving the interrupted phase segment without + creating a session outcome; repeated wedges become terminal after three + restart attempts, counted durably before the signal goes out. Inline role + turns now share an explicit total wall-clock timeout, independent of backend + streamed-message idle timeouts and retries. Reactor-stage boundaries refresh + supervisor progress, while a stage cancelled at the total timeout counts toward + the crash emergency stop. A failed restart-counter write refuses the resumable + restart and sends SIGTERM, and a cleanly ended leg can resume as soon as its + owner pid is gone instead of being held alive by the final state write. +- **The robustness monitor reads the real stop-reason vocabulary.** It imported + a module that does not exist and silently fell back to a subset missing 16 + terminal reasons, so a finished session could be relaunched. + ## [v1.1.0] - 2026-09-09 Current packaged version (`pyproject.toml`). See [release notes](docs/release-notes.md) and the diff --git a/docs/reference/environment-variables.md b/docs/reference/environment-variables.md index eaf16db5e0..5c2ae81dea 100644 --- a/docs/reference/environment-variables.md +++ b/docs/reference/environment-variables.md @@ -681,9 +681,10 @@ guarantee that varies silently by host is not a guarantee. | Variable | Default | Description | |---|---|---| | `HYPERLOOM_REAP_BACKEND` | `process_group` | Which unit ends a bring-up round's processes: `process_group`, `cgroup` or `container`. Only `cgroup` and `container` produce a reap that is *proof* the tree is gone — the kernel (or the container runtime) owns the membership list, so nothing can leave it by forking or re-parenting. `process_group` reaches only what it could enumerate from procfs before it signalled. A unit that cannot run on this host falls back to `process_group`, which weakens the claim rather than faking it. | -| `HYPERLOOM_SUPERVISOR` | `1` | Whether the optimizer starts an out-of-band supervisor process that watches for a coordinator that died or whose tick stopped advancing. | -| `HYPERLOOM_SUPERVISOR_ENFORCE` | unset (off) | Whether the supervisor may end a process tree. A wedged coordinator is sent SIGTERM regardless — that is the channel its signal drain reads while the loop is busy. Off by default, a coordinator that does not answer that stop, and a dead one's leftovers, are left alone and the refusal is recorded in `runtime/supervisor/status.json`. Ending a tree additionally requires a reap backend whose success is proof, so enforcement with the default `process_group` unit will refuse to kill and say so. | -| `HYPERLOOM_SUPERVISOR_TICK_STALL_SEC` | half of `--max-hours`, capped at `3600` and floored at `1800` | How long the coordinator's tick may go without advancing before the supervisor calls it wedged. Derived from the session budget so the window always fits inside the run it watches; setting this overrides the derivation. | +| `HYPERLOOM_SUPERVISOR` | `1` | Whether the optimizer starts an out-of-band supervisor process that watches for a coordinator that died or whose tick stopped making progress. | +| `HYPERLOOM_SUPERVISOR_ENFORCE` | unset (off) | Whether the supervisor may end a process tree. A wedged coordinator receives SIGHUP for each resumable restart and SIGTERM after the restart limit. Off by default, a coordinator that does not answer that stop, and a dead one's leftovers, are left alone and the refusal is recorded in `runtime/supervisor/status.json`. Ending a tree additionally requires a reap backend whose success is proof, so enforcement with the default `process_group` unit will refuse to kill and say so. | +| `INFERENCE_OPTIMIZER_REACTOR_TURN_TIMEOUT_SEC` | `1800` | Total wall-clock limit for each reactor stage, including backend startup, streamed output, retries, backoff, and cleanup. This is independent of backend `*_CALL_TIMEOUT_SEC` settings: for streamed Claude turns those settings bound idle time between SDK messages, and activity resets that idle timer. Reaching this total limit cancels the stage and records a crash; a shorter remaining session bound still ends the stage without recording a crash. | +| `HYPERLOOM_SUPERVISOR_TICK_STALL_SEC` | half of `--max-hours`, capped at the larger of `3600` or the reactor timeout plus `30`, and floored at the reactor timeout plus `30` | How long coordinator progress may remain unchanged before the supervisor calls it wedged. Progress is refreshed at every reactor-stage boundary, so the derived window covers one total reactor timeout plus one supervisor poll rather than the sum of all sequential roles. Raising `INFERENCE_OPTIMIZER_REACTOR_TURN_TIMEOUT_SEC` automatically raises the derived floor and default; an explicit value overrides the derivation and is the operator's responsibility. | The supervisor never opens `coordinator.db` — it sits on a network filesystem where a second writer risks the message bus and the task registry — and never @@ -693,6 +694,13 @@ writes `reports/final.json` itself, marked `producer: "supervisor"`; that record never replaces a full report, and the coordinator's own crash-safe fallback never replaces it. +A successful watchdog stop records a leg boundary but no session `stop_reason` +or final artifacts, allowing the monitor to resume it. The restart count is +persisted in `runtime/supervisor/status.json`; after three resumable restarts the +next watchdog request uses SIGTERM and the coordinator follows its normal +terminal path. The monitor treats `final.json`, `final.md`, a completed CLOSE +sequence, or a vocabulary stop reason as terminal. + --- ## Framework / source-tree discovery diff --git a/src/hyperloom/agents/robustness/role/envelope.py b/src/hyperloom/agents/robustness/role/envelope.py index 6fef0b69b3..8f52c1d558 100644 --- a/src/hyperloom/agents/robustness/role/envelope.py +++ b/src/hyperloom/agents/robustness/role/envelope.py @@ -68,6 +68,7 @@ class IntentType(str, Enum): "current_best", "stop_reason", "stop_ts", + "leg_ended_ts", "last_tick_exception", "cumulative_gain_validated", "cumulative_gain_validated_ts", diff --git a/src/hyperloom/inference_optimizer/SKILL.md b/src/hyperloom/inference_optimizer/SKILL.md index 971fa24f97..ca184d715f 100644 --- a/src/hyperloom/inference_optimizer/SKILL.md +++ b/src/hyperloom/inference_optimizer/SKILL.md @@ -761,6 +761,11 @@ for the kernel dispatch and artifact layout. ### Recovery +The out-of-band supervisor uses SIGHUP for a resumable watchdog restart. This +records the interrupted leg boundary without producing a stop reason or final +report; the monitor resumes the same session with `--resume-from`. After three +watchdog restarts, the next watchdog stop is terminal instead. + If the CLI exits with `Claude SDK exit code 1` or `Primus.00009 token not present`, the gateway rejected the request. Check that `OPENAI_BASE_URL` / `OPENAI_API_KEY` are set in `.env` (or the calling shell) and that the gateway is reachable: @@ -1217,10 +1222,10 @@ Three exceptions: For runs > 5 min, start a monitor in its own `setsid nohup` process. It polls `state.json` every 5 min, exits without resuming when the session is terminal -(any `stop_reason` in `STOP_REASON_VOCAB`, `phase=CLOSE`, or -`reports/final.md` present — including failure sentinels like -`baseline_failed`), and resumes via `--resume-from` only when the optimizer -dies without those markers (unexpected crash). +(any `stop_reason` in `STOP_REASON_VOCAB`, a completed CLOSE sequence, or a +`reports/final.json` / `final.md` artifact — including failure sentinels like +`baseline_failed`), and resumes via `--resume-from` when the optimizer dies +without those markers, including a resumable watchdog restart. ```bash export RUN_DIR="${USER_DATA_PATH:-/workspace/hyperloom}/optimizer_runs" diff --git a/src/hyperloom/inference_optimizer/cli/__init__.py b/src/hyperloom/inference_optimizer/cli/__init__.py index e20581f2d3..6245ce1670 100644 --- a/src/hyperloom/inference_optimizer/cli/__init__.py +++ b/src/hyperloom/inference_optimizer/cli/__init__.py @@ -102,6 +102,7 @@ default_enabled_actions, ) from hyperloom.orchestrator.supervisor import spawn_supervisor, stop_supervisor +from hyperloom.orchestrator.supervisor.watch import SUPERVISOR_RESTART_REASON from ..session.lock import SessionAlreadyRunning, SessionLock from ..session.paths import ( @@ -1304,6 +1305,65 @@ def _exit_code_for_stop_reason(stop_reason: str | None) -> int: return 0 if (stop_reason or "") in SUCCESS_STOP_REASONS else 1 +def _write_cli_terminal_artifacts(session_dir: Path, state: SharedState, stop_reason: str | None) -> None: + """Write the session's terminal artifacts in their established order. + + A watchdog restart ends a leg, not the session, so it produces none of them + -- including the close-out package, which speaks for a finished run. + """ + if stop_reason == SUPERVISOR_RESTART_REASON: + return + try: + from ..breakdown import write_minimal_final_json + + with timed_teardown_step(state, "final_json"): + final_json = write_minimal_final_json(session_dir) + print(f"Final summary : {final_json}") + except Exception: # noqa: BLE001 + log.exception("crash-safe final.json write failed (non-fatal)") + if state.close_sequence_done: + print("Session breakdown : (already written by CLOSE phase sequencer; skipping cli.finally safety-net write)") + else: + try: + from ..breakdown import write_breakdown_json + + with timed_teardown_step(state, "session_breakdown"): + breakdown_path = write_breakdown_json(session_dir) + print(f"Session breakdown : {breakdown_path}") + except Exception: # noqa: BLE001 + log.exception("session_breakdown finalize failed (non-fatal)") + try: + from ..breakdown import write_minimal_final_report + + with timed_teardown_step(state, "final_md"): + final_md = write_minimal_final_report(session_dir) + print(f"Final report : {final_md}") + except Exception: # noqa: BLE001 + log.exception("emergency final report write failed (non-fatal)") + try: + from hyperloom.orchestrator.trace.langfuse_emitter import flush_session, record_session_breakdown + + with timed_teardown_step(state, "langfuse"): + flush_session(session_dir) + from ..breakdown import patch_breakdown_langfuse + + patch_breakdown_langfuse(session_dir) + record_session_breakdown(session_dir) + except Exception: # noqa: BLE001 + log.debug("langfuse flush_session failed", exc_info=True) + # Safety net for paths that leave close_sequence_done False and never run + # the sequencer; ordered after langfuse so the package carries its patch. + try: + from ..breakdown import package_session_artifacts + + with timed_teardown_step(state, "artifact_package"): + pkg_path = package_session_artifacts(session_dir, session_id=str(getattr(state, "session_id", "") or "")) + if pkg_path is not None: + print(f"Artifact package : {pkg_path}") + except Exception: # noqa: BLE001 + log.exception("session artifact package failed (non-fatal)") + + def _new_preflight_failure_session_dir( args: argparse.Namespace, *, @@ -2306,6 +2366,7 @@ def _backend_kind(role: str) -> str: log.warning("supervisor: could not start (%s); the run proceeds unwatched", exc) print(f"[supervisor] could not start ({exc}); the run proceeds unwatched", file=sys.stderr) + stop_reason: str | None = None try: stop_reason = await coordinator.run( objective=objective, @@ -2317,6 +2378,7 @@ def _backend_kind(role: str) -> str: ) finally: state = coordinator.shared_state + effective_stop_reason = stop_reason or coordinator.stop_classification # Stopping the leases and the agent subprocesses is itself a step that # can hang, so it happens while the supervisor is still watching; the # supervisor is stood down only once it has returned. @@ -2329,90 +2391,18 @@ def _backend_kind(role: str) -> str: # exit anyway; this just frees it promptly for an intentional resume. with timed_teardown_step(state, "session_lock"): session_lock.release() - # Crash-safe reports/final.json. - try: - from ..breakdown import write_minimal_final_json - - with timed_teardown_step(state, "final_json"): - final_json = write_minimal_final_json(session_dir) - print(f"Final summary : {final_json}") - except Exception: # noqa: BLE001 — safety net must never mask stop_reason - log.exception("crash-safe final.json write failed (non-fatal)") - # End-of-session safety net: always materialize session_breakdown.json (best-effort; never mask stop_reason). - sequencer_done = getattr(state, "close_sequence_done", False) - if sequencer_done: - print( - "Session breakdown : (already written by CLOSE phase sequencer; skipping cli.finally safety-net write)" - ) - # Re-run the Langfuse flush idempotently as a safety net. - try: - from hyperloom.orchestrator.trace.langfuse_emitter import ( - flush_session, - record_session_breakdown, - ) - - with timed_teardown_step(state, "langfuse"): - flush_session(session_dir) - from ..breakdown import patch_breakdown_langfuse - - patch_breakdown_langfuse(session_dir) - record_session_breakdown(session_dir) - except Exception: # noqa: BLE001 - log.debug("langfuse flush_session (post-sequencer) failed", exc_info=True) - else: - try: - from ..breakdown import write_breakdown_json - - with timed_teardown_step(state, "session_breakdown"): - breakdown_path = write_breakdown_json(session_dir) - print(f"Session breakdown : {breakdown_path}") - except Exception: # noqa: BLE001 - log.exception("session_breakdown finalize failed (non-fatal)") - # Safety-net reports/final.md write (no-op when the sequencer's final.md already exists). - try: - from ..breakdown import write_minimal_final_report - - with timed_teardown_step(state, "final_md"): - final_md = write_minimal_final_report(session_dir) - print(f"Final report : {final_md}") - except Exception: # noqa: BLE001 - log.exception("emergency final report write failed (non-fatal)") - # Live Langfuse push (opt-in, default off): reconcile + flush, then splice the post-flush receipt into the - # session_breakdown.json langfuse section. - try: - from hyperloom.orchestrator.trace.langfuse_emitter import ( - flush_session, - record_session_breakdown, - ) - - with timed_teardown_step(state, "langfuse"): - flush_session(session_dir) - from ..breakdown import patch_breakdown_langfuse - - patch_breakdown_langfuse(session_dir) - record_session_breakdown(session_dir) - except Exception: # noqa: BLE001 - log.debug("langfuse flush_session failed (non-fatal)", exc_info=True) - - # Safety-net artifact package -> /workspace, for paths that leave close_sequence_done False and never run the - # sequencer. - try: - from ..breakdown import package_session_artifacts - - with timed_teardown_step(state, "artifact_package"): - pkg_path = package_session_artifacts( - session_dir, - session_id=str(getattr(state, "session_id", "") or ""), - ) - if pkg_path is not None: - print(f"Artifact package : {pkg_path}") - except Exception: # noqa: BLE001 - log.exception("session artifact package failed (non-fatal)") + _write_cli_terminal_artifacts(session_dir, state, effective_stop_reason) try: state.save(session_dir) except Exception: # noqa: BLE001 log.exception("failed to persist teardown timings (non-fatal)") + if stop_reason == SUPERVISOR_RESTART_REASON: + # The leg is over, the session is not: the monitor resumes it, and a + # final summary here would speak for a run that has not ended. + print(f"Leg stopped for a supervisor restart; session {session_dir} stays resumable.") + return _exit_code_for_stop_reason(stop_reason) + _reconcile_crash_count(coordinator.shared_state, session_dir) # NOTE: conc_sweep is now a SWEEP-phase action auto-enqueued by the Coordinator, not a post-hook here. diff --git a/src/hyperloom/inference_optimizer/cli/bootstrap.py b/src/hyperloom/inference_optimizer/cli/bootstrap.py index 83781be8a3..5a8fe6867a 100644 --- a/src/hyperloom/inference_optimizer/cli/bootstrap.py +++ b/src/hyperloom/inference_optimizer/cli/bootstrap.py @@ -408,7 +408,8 @@ def _print_final_summary( def _bank_previous_leg_phase_segment(state: SharedState) -> None: """Bank the phase time the stopped leg spent but never recorded.""" - stop_unix = min(to_unix(state.stop_ts, 0.0) or 0.0, time.time()) + boundary = state.leg_ended_ts or state.stop_ts + stop_unix = min(to_unix(boundary, 0.0) or 0.0, time.time()) if stop_unix <= 0.0: return bank_phase_segment(state, until_unix=stop_unix) @@ -438,6 +439,7 @@ def _begin_resume_leg(state: SharedState) -> str: state.resumed_ts = now_iso() state.stop_reason = "" state.stop_ts = "" + state.leg_ended_ts = "" state.closing_phase = False state.closing_started_unix = 0.0 state.closing_report_task_id = "" diff --git a/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py b/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py index 436edadbde..5e22d9db8f 100644 --- a/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py +++ b/src/hyperloom/inference_optimizer/tests/test_cli_bootstrap.py @@ -569,6 +569,20 @@ def test_a_resume_banks_what_the_stopped_leg_spent_in_its_phase() -> None: assert state.stop_ts == "" +def test_a_resume_banks_a_resumable_legs_own_boundary() -> None: + state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") + state.phase = "PRELUDE" + state.phase_started_ts = "2026-08-01T00:00:00+00:00" + state.phase_started_unix = 1785_542_400.0 + state.leg_ended_ts = "2026-08-01T00:30:00+00:00" + + cb._begin_resume_leg(state) + + assert state.phase_elapsed_totals == {"PRELUDE": 1800.0} + assert state.leg_ended_ts == "" + assert state.stop_ts == "" + + def test_a_second_resume_banks_only_the_leg_that_just_stopped() -> None: """The first leg's segment is already durable; re-banking it would double-charge the phase.""" state = SharedState(session_id="s", start_ts="2026-08-01T00:00:00+00:00") diff --git a/src/hyperloom/inference_optimizer/tests/test_coordinator_runtime.py b/src/hyperloom/inference_optimizer/tests/test_coordinator_runtime.py index b46c543862..dc1b91a07a 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coordinator_runtime.py +++ b/src/hyperloom/inference_optimizer/tests/test_coordinator_runtime.py @@ -6,6 +6,7 @@ from __future__ import annotations import json +import signal from pathlib import Path from types import SimpleNamespace from typing import Any @@ -45,7 +46,7 @@ SqliteLeaseBackend, ) from hyperloom.orchestrator.loop.dispatcher import DispatcherCollaborator -from hyperloom.inference_optimizer.session.session_paths import target_baseline_json +from hyperloom.inference_optimizer.session.session_paths import supervisor_status_path, target_baseline_json from hyperloom.orchestrator.bus.storage import SqliteConnection @@ -70,6 +71,135 @@ def _build_backends(scripts: dict[str, ScriptedPlan]) -> dict[str, Backend]: return backends +def test_signal_stop_reason_comes_from_the_received_signal(session_dir): + c = Coordinator(session_dir, backends=_build_backends({})) + try: + c._signals = SimpleNamespace(received={signal.SIGHUP}) + assert c._signal_stop_reason() == "supervisor_restart_requested" + + status = supervisor_status_path(session_dir) + status.parent.mkdir(parents=True, exist_ok=True) + status.write_text( + json.dumps({"stop_asked": ["supervisor_tick_stalled: stale"]}), + encoding="utf-8", + ) + c._signals = SimpleNamespace(received={signal.SIGTERM}) + assert c._signal_stop_reason() == "signal" + + c._signals = SimpleNamespace(received={signal.SIGHUP, signal.SIGTERM}) + assert c._signal_stop_reason() == "signal" + + c._signals = SimpleNamespace(received={signal.SIGHUP, signal.SIGINT}) + assert c._signal_stop_reason() == "signal" + finally: + c.db.close() + + +@pytest.mark.asyncio +async def test_terminal_state_outweighs_a_supervisor_restart_signal(session_dir): + c = Coordinator(session_dir, backends=_build_backends({})) + c._signals = SimpleNamespace(received={signal.SIGHUP}, close=lambda: frozenset({signal.SIGHUP})) + c._stop.set() + c.shared_state.set_stop_reason("time_exhausted") + close_calls = 0 + finalize_calls = 0 + + async def _close(*, reason): + nonlocal close_calls + close_calls += 1 + + async def _finalize(): + nonlocal finalize_calls + finalize_calls += 1 + + c.ensure_close_sequence = _close + c._recipe_kb_t4_hook = _finalize + try: + reason = await c.run(max_ticks=1, tick_interval_sec=0.0) + assert ( + reason, + c.shared_state.stop_reason, + bool(c.shared_state.stop_ts), + bool(c.shared_state.leg_ended_ts), + close_calls, + finalize_calls, + ) == ( + "time_exhausted", + "time_exhausted", + True, + False, + 1, + 1, + ) + finally: + await c.stop() + + +@pytest.mark.asyncio +async def test_the_final_signal_snapshot_makes_a_late_sigterm_terminal(session_dir): + c = Coordinator(session_dir, backends=_build_backends({})) + c._signals = SimpleNamespace( + received={signal.SIGHUP}, + close=lambda: frozenset({signal.SIGHUP, signal.SIGTERM}), + ) + c._stop.set() + close_calls = 0 + + async def _close(*, reason): + nonlocal close_calls + close_calls += 1 + + c.ensure_close_sequence = _close + c._recipe_kb_t4_hook = _noop_finalize + try: + reason = await c.run(max_ticks=1, tick_interval_sec=0.0) + assert ( + reason, + c.shared_state.stop_reason, + c.shared_state.leg_ended_ts, + close_calls, + ) == ("signal", "signal", "", 1) + finally: + await c.stop() + + +async def _noop_finalize() -> None: + """Stand in for the recipe/journal finalize hook.""" + + +@pytest.mark.asyncio +async def test_sigterm_follows_the_normal_terminal_path(session_dir): + c = Coordinator(session_dir, backends=_build_backends({})) + c._signals = SimpleNamespace(received={signal.SIGTERM}, close=lambda: frozenset({signal.SIGTERM})) + c._stop.set() + close_calls = 0 + finalize_calls = 0 + + async def _close(*, reason): + nonlocal close_calls + assert reason == "signal" + close_calls += 1 + + async def _finalize(): + nonlocal finalize_calls + finalize_calls += 1 + + c.ensure_close_sequence = _close + c._recipe_kb_t4_hook = _finalize + try: + reason = await c.run(max_ticks=1, tick_interval_sec=0.0) + assert ( + reason, + c.shared_state.stop_reason, + bool(c.shared_state.stop_ts), + c.shared_state.leg_ended_ts, + close_calls, + finalize_calls, + ) == ("signal", "signal", True, "", 1, 1) + finally: + await c.stop() + + # MockBackend @pytest.mark.asyncio async def test_mock_backend_plays_scripted_turns(): diff --git a/src/hyperloom/inference_optimizer/tests/test_deadlines.py b/src/hyperloom/inference_optimizer/tests/test_deadlines.py index 1795f3fcb1..227c81ea91 100644 --- a/src/hyperloom/inference_optimizer/tests/test_deadlines.py +++ b/src/hyperloom/inference_optimizer/tests/test_deadlines.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio +import signal import threading import time from datetime import datetime, timezone @@ -320,30 +321,35 @@ def _boom() -> str: class TestAStopIsRecordedBeforeItIsDispatched: """A signal must land while the loop is busy, not once it is free.""" - def test_the_drain_publishes_to_both_audiences(self): + @pytest.mark.parametrize("received_signal", [signal.SIGTERM, signal.SIGHUP]) + def test_the_drain_publishes_the_received_signal_to_both_audiences(self, received_signal): import os - import signal as signal_module from hyperloom.orchestrator.loop.signals import SignalDrain - async def _exercise() -> tuple[bool, bool]: + async def _exercise() -> tuple[bool, bool, frozenset[int]]: stop = asyncio.Event() - drain = SignalDrain(loop=asyncio.get_running_loop(), stop_event=stop) + drain = SignalDrain( + loop=asyncio.get_running_loop(), + stop_event=stop, + signals=(received_signal,), + ) if not drain.armed: # pragma: no cover — no handlers off the main thread pytest.skip("signal handlers are unavailable here") try: - os.kill(os.getpid(), signal_module.SIGTERM) + os.kill(os.getpid(), received_signal) # The threading event is set by the reading thread; the asyncio # one only once the loop runs, which is the ordering under test. await asyncio.wait_for(stop.wait(), timeout=5.0) - return drain.requested.is_set(), stop.is_set() + return drain.requested.is_set(), stop.is_set(), drain.close() finally: drain.close() - requested, stopped = asyncio.run(_exercise()) + requested, stopped, received = asyncio.run(_exercise()) assert requested is True assert stopped is True + assert received == {received_signal} def test_closing_restores_the_previous_handler(self): import signal as signal_module diff --git a/src/hyperloom/inference_optimizer/tests/test_robustness_monitor.py b/src/hyperloom/inference_optimizer/tests/test_robustness_monitor.py index 35204c1253..5faeddb4fa 100644 --- a/src/hyperloom/inference_optimizer/tests/test_robustness_monitor.py +++ b/src/hyperloom/inference_optimizer/tests/test_robustness_monitor.py @@ -5,6 +5,7 @@ from __future__ import annotations +import importlib import json import os import re @@ -65,6 +66,112 @@ def _run_monitor_briefly(env, *, seconds=3.0): return out, err +def _run_terminal_check(session_dir: Path) -> subprocess.CompletedProcess[str]: + text = MONITOR.read_text(encoding="utf-8") + match = re.search(r"is_terminal_session\(\) \{\n.*?<<'PY'\n(.*?)\nPY\n\}", text, re.DOTALL) + assert match is not None + return subprocess.run( + ["python3", "-", str(session_dir)], + input=match.group(1), + text=True, + capture_output=True, + check=False, + ) + + +def _run_alive_check(session_dir: Path, wrapper_pid: str = "") -> subprocess.CompletedProcess[str]: + text = MONITOR.read_text(encoding="utf-8") + match = re.search(r"is_session_alive\(\) \{\n.*?<<'PY'\n(.*?)\nPY\n\}", text, re.DOTALL) + assert match is not None + env = os.environ.copy() + env["STATE_FRESH_SEC"] = "600" + return subprocess.run( + ["python3", "-", str(session_dir), wrapper_pid], + input=match.group(1), + text=True, + capture_output=True, + check=False, + env=env, + ) + + +def test_an_incomplete_close_phase_is_not_terminal(tmp_path): + (tmp_path / "state.json").write_text( + json.dumps( + { + "phase": "CLOSE", + "close_sequence_done": False, + "stop_reason": "", + } + ), + encoding="utf-8", + ) + + result = _run_terminal_check(tmp_path) + + assert result.returncode == 1 + + +def test_a_completed_close_sequence_is_terminal(tmp_path): + (tmp_path / "state.json").write_text( + json.dumps( + { + "phase": "CLOSE", + "close_sequence_done": True, + "stop_reason": "", + } + ), + encoding="utf-8", + ) + + result = _run_terminal_check(tmp_path) + + assert result.returncode == 0 + + +def test_a_canonical_only_stop_reason_is_terminal(tmp_path): + """The vocabulary the monitor gates on is the orchestrator's, not a copy of it. + + ``coordinator_exception`` is terminal and absent from the offline fallback, + so a monitor reading the fallback would resume a finished session. + """ + (tmp_path / "state.json").write_text( + json.dumps({"phase": "SWEEP", "stop_reason": "coordinator_exception"}), + encoding="utf-8", + ) + + result = _run_terminal_check(tmp_path) + + assert result.returncode == 0, result.stderr + + +def test_the_monitor_names_the_module_the_vocabulary_lives_in(): + """A renamed module must break here, not silently downgrade the live monitor.""" + importlib.import_module("hyperloom.orchestrator.phases.machine_state") + + assert "from hyperloom.orchestrator.phases.machine_state import STOP_REASON_VOCAB" in MONITOR.read_text( + encoding="utf-8" + ) + + +def test_a_supervisor_final_json_is_terminal(tmp_path): + reports = tmp_path / "reports" + reports.mkdir() + (reports / "final.json").write_text( + json.dumps( + { + "producer": "supervisor", + "stop_reason": "supervisor_tick_stalled", + } + ), + encoding="utf-8", + ) + + result = _run_terminal_check(tmp_path) + + assert result.returncode == 0 + + def test_monitor_waits_for_delayed_launch_info(tmp_path): """A delayed LAUNCH_INFO_FILE: the monitor polls (bounded), resolves it, and exits 0 — not exit 2 before launch-info flushed.""" sess = tmp_path / "sess" @@ -247,6 +354,50 @@ def test_monitor_resume_is_pinned_to_resolved_session_dir(): assert re.search(r"--resume(?!-from)", text) is None +@pytest.mark.skipif(not _HAS_PROC, reason="liveness probe reads /proc (Linux)") +def test_fresh_cleanly_ended_leg_with_dead_owner_is_not_alive(tmp_path): + runtime = tmp_path / "runtime" + runtime.mkdir() + dead_pid = 999_999_999 + assert not Path(f"/proc/{dead_pid}").exists() + (runtime / "optimizer.lock").write_text(json.dumps({"pid": dead_pid}), encoding="utf-8") + (tmp_path / "state.json").write_text( + json.dumps({"leg_ended_ts": "2026-09-11T00:00:00+00:00", "stop_reason": ""}), + encoding="utf-8", + ) + + result = _run_alive_check(tmp_path) + + assert result.returncode == 1 + + +@pytest.mark.skipif(not _HAS_PROC, reason="liveness probe reads /proc (Linux)") +def test_fresh_running_leg_without_end_marker_is_alive(tmp_path): + (tmp_path / "state.json").write_text( + json.dumps({"leg_ended_ts": "", "stop_reason": ""}), + encoding="utf-8", + ) + + result = _run_alive_check(tmp_path) + + assert result.returncode == 0 + + +@pytest.mark.skipif(not _HAS_PROC, reason="liveness probe reads /proc (Linux)") +def test_live_owner_keeps_a_cleanly_ended_leg_alive(tmp_path): + runtime = tmp_path / "runtime" + runtime.mkdir() + (runtime / "optimizer.lock").write_text(json.dumps({"pid": os.getpid()}), encoding="utf-8") + (tmp_path / "state.json").write_text( + json.dumps({"leg_ended_ts": "2026-09-11T00:00:00+00:00", "stop_reason": ""}), + encoding="utf-8", + ) + + result = _run_alive_check(tmp_path) + + assert result.returncode == 0 + + @pytest.mark.skipif(not _HAS_PROC, reason="liveness probe reads /proc (Linux)") def test_monitor_does_not_resume_during_startup_grace(tmp_path): """Within the cold-start grace window a not-yet-alive session is never resumed.""" diff --git a/src/hyperloom/inference_optimizer/tests/test_run_optimize_exit_code_lock.py b/src/hyperloom/inference_optimizer/tests/test_run_optimize_exit_code_lock.py index 6b0d73e949..9ccdbbc4ab 100644 --- a/src/hyperloom/inference_optimizer/tests/test_run_optimize_exit_code_lock.py +++ b/src/hyperloom/inference_optimizer/tests/test_run_optimize_exit_code_lock.py @@ -7,12 +7,149 @@ import argparse import asyncio +import signal from pathlib import Path +from types import SimpleNamespace import pytest import hyperloom.inference_optimizer.cli as ocli from hyperloom.inference_optimizer.session.lock import SessionLock +from hyperloom.inference_optimizer.protocol.intent import Intent, IntentType +from hyperloom.orchestrator.loop.coordinator import Coordinator +from hyperloom.orchestrator.roles import Backend, MockBackend, ScriptedPlan +from hyperloom.orchestrator.state.shared_state import SharedState + + +def _record_terminal_writes(monkeypatch) -> list[str]: + """Capture every terminal artifact the close-out writes, in order.""" + order: list[str] = [] + for name, label in ( + ("write_minimal_final_json", "final_json"), + ("write_breakdown_json", "breakdown"), + ("write_minimal_final_report", "final_md"), + ("package_session_artifacts", "package"), + ): + monkeypatch.setattr( + f"hyperloom.inference_optimizer.breakdown.{name}", + lambda *_a, _label=label, **_kw: order.append(_label), + ) + return order + + +def _backends() -> dict[str, Backend]: + heartbeat = Intent(type=IntentType.SEND_MESSAGE, payload={"topic": "heartbeat", "body_md": "ok"}) + plan = ScriptedPlan(turns=[], default_intent=heartbeat) + return {name: MockBackend(plan, name=name) for name in ("orchestration", "critic", "robustness")} + + +def test_resumable_restart_writes_no_terminal_artifacts(tmp_path: Path, monkeypatch) -> None: + order = _record_terminal_writes(monkeypatch) + + ocli._write_cli_terminal_artifacts( + tmp_path, + SharedState(session_id="s"), + "supervisor_restart_requested", + ) + + assert order == [] + + +def test_terminal_artifacts_keep_the_existing_write_order(tmp_path: Path, monkeypatch) -> None: + order = _record_terminal_writes(monkeypatch) + + ocli._write_cli_terminal_artifacts(tmp_path, SharedState(session_id="s"), "signal") + + assert order == ["final_json", "breakdown", "final_md", "package"] + + +def test_completed_close_still_gets_its_close_out_package(tmp_path: Path, monkeypatch) -> None: + """The sequencer wrote the reports; the package is the session's, not the sequencer's.""" + order = _record_terminal_writes(monkeypatch) + + state = SharedState(session_id="s", close_sequence_done=True) + ocli._write_cli_terminal_artifacts(tmp_path, state, "signal") + + assert order == ["final_json", "package"] + + +@pytest.mark.asyncio +async def test_resumable_classification_survives_state_save_failure(tmp_path: Path, monkeypatch) -> None: + session_dir = tmp_path / "session" + session_dir.mkdir() + SharedState(session_id="s").save(session_dir) + (session_dir / "manifest.json").write_text( + '{"schema_version": 4, "session_id": "s"}', + encoding="utf-8", + ) + monkeypatch.setenv("USER_DATA_PATH", str(tmp_path)) + # _run_optimize pins these process-wide for in-process executors; route them + # through monkeypatch so the run does not leak strict paths into later tests. + monkeypatch.setenv("INFERENCE_OPTIMIZER_STRICT_PATHS", "") + monkeypatch.setenv("INFERENCE_OPTIMIZER_CURRENT_SESSION_DIR", "") + monkeypatch.setattr( + ocli, + "clean_stale_aiter_locks", + lambda: {"dir": "", "deleted": 0, "skipped_fresh": 0, "errors": 0}, + ) + monkeypatch.setattr(ocli, "_preflight", lambda _args: ("", "")) + monkeypatch.setattr(ocli, "_resolve_models_for_run", lambda *_args, **_kwargs: None) + monkeypatch.setattr(ocli, "_build_backends", lambda **_kwargs: _backends()) + monkeypatch.setattr(ocli, "spawn_supervisor", lambda *_args, **_kwargs: None) + + original_run = Coordinator.run + + async def _run_then_fail_resumable_save(self, **kwargs): + self._resumed_from = {"is_resume": False, "rebuilt": True} + self._signals = SimpleNamespace( + received={signal.SIGHUP}, + close=lambda: frozenset({signal.SIGHUP}), + ) + self._stop.set() + real_save = self.shared_state.save + + def _fail_save(path) -> None: + if self.stop_classification == "supervisor_restart_requested": + raise OSError("disk full") + real_save(path) + + self.shared_state.save = _fail_save + kwargs["install_signal_handlers"] = False + return await original_run(self, **kwargs) + + monkeypatch.setattr(Coordinator, "run", _run_then_fail_resumable_save) + order = _record_terminal_writes(monkeypatch) + reasons: list[str | None] = [] + write_terminal = ocli._write_cli_terminal_artifacts + + def _capture_terminal_reason(path, state, stop_reason) -> None: + reasons.append(stop_reason) + write_terminal(path, state, stop_reason) + + monkeypatch.setattr(ocli, "_write_cli_terminal_artifacts", _capture_terminal_reason) + args = ocli._build_parser().parse_args( + [ + "optimize", + "--resume-from", + str(session_dir), + "--critic-mock", + "--robustness-mock", + "--no-kernel", + "--no-framework-agent", + "--degraded-kb", + "--degraded-pr", + "--research-lane-capacity", + "0", + "--max-ticks", + "1", + ] + ) + + with pytest.raises(OSError, match="disk full"): + await ocli._run_optimize(args) + + assert reasons == ["supervisor_restart_requested"] + assert order == [] def test_multinode_tp_exceeds_total_gpus_exits_2() -> None: diff --git a/src/hyperloom/inference_optimizer/tests/test_session_time_budget.py b/src/hyperloom/inference_optimizer/tests/test_session_time_budget.py index 49f26d8b68..4048b72b37 100644 --- a/src/hyperloom/inference_optimizer/tests/test_session_time_budget.py +++ b/src/hyperloom/inference_optimizer/tests/test_session_time_budget.py @@ -6,6 +6,9 @@ from __future__ import annotations import asyncio +import json +import os +import socket import sys import threading import time @@ -18,6 +21,7 @@ from hyperloom.inference_optimizer.protocol.action_surfaces import ACTION_CATALOGUE from hyperloom.inference_optimizer.protocol.intent import Intent, IntentType +from hyperloom.inference_optimizer.session.session_paths import optimizer_lock_path from hyperloom.orchestrator.actions.cancel_channel import ( CancelScope, current_cancel_scope, @@ -46,6 +50,8 @@ from hyperloom.orchestrator.roles import Backend, MockBackend, ScriptedPlan from hyperloom.orchestrator.state.shared_state import SharedState, effective_closing_grace_sec from hyperloom.orchestrator.state.task_registry import Task +from hyperloom.orchestrator.supervisor import store as supervisor_store +from hyperloom.orchestrator.supervisor.watch import ALIVE, WEDGED, Supervisor # The costliest action the catalogue prices, so a short budget cannot fit it. _EXPENSIVE_ACTION = "conc_sweep" @@ -93,6 +99,15 @@ def _set_budget(coord: Coordinator, *, minutes: float, elapsed_min: float = 0.0) coord.shared_state.elapsed_minutes = lambda **_kw: elapsed_min # type: ignore[method-assign] +def _record_current_owner(session_dir) -> None: + path = optimizer_lock_path(session_dir) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"pid": os.getpid(), "hostname": socket.gethostname()}), + encoding="utf-8", + ) + + class TestTheCostTheGateJudgesOn: """Where the expected cost comes from: the action catalogue, or nowhere.""" @@ -1268,6 +1283,194 @@ async def _ok() -> None: await coord._await_within_session_bound(_ok, stage="test") assert started == [True] + @pytest.mark.asyncio + async def test_a_reactor_turn_outlives_the_backends_idle_budget(self, coord: Coordinator): + """``call_timeout_s`` bounds the silence between streamed messages, not the turn. + + A turn that keeps streaming is healthy however long it runs, so reading + that field as a wall-clock cap cancels working agents mid-flight. + """ + coord._run_deadline = Deadline.after(60.0) + coord.backends["orchestration"].call_timeout_s = 0.01 + finished: list[bool] = [] + + async def _slower_than_the_idle_budget() -> None: + await asyncio.sleep(0.2) + finished.append(True) + + await coord._await_within_session_bound( + _slower_than_the_idle_budget, + stage="reactor:orchestration", + ) + + assert finished == [True] + + @pytest.mark.asyncio + async def test_an_active_reactor_turn_is_cancelled_at_its_total_timeout(self, coord: Coordinator): + coord.reactor_turn_timeout_sec = 0.03 + coord._run_deadline = Deadline.after(60.0) + activity: list[float] = [] + cancelled = asyncio.Event() + + async def _stay_active() -> None: + try: + while True: + activity.append(time.monotonic()) + await asyncio.sleep(0.005) + finally: + cancelled.set() + + await asyncio.wait_for( + coord._await_within_session_bound( + _stay_active, + stage="reactor:orchestration", + ), + timeout=0.2, + ) + + assert cancelled.is_set() + assert len(activity) >= 2 + + @pytest.mark.asyncio + async def test_a_turn_cancelled_at_its_total_timeout_counts_as_a_crash(self, coord: Coordinator): + """Cancelling the turn lets the tick advance, which hides the wedge from the watchdog. + + Nothing downstream would otherwise see it, and the session would spend + the rest of its budget re-cancelling the same turn. + """ + coord.reactor_turn_timeout_sec = 0.01 + coord._run_deadline = Deadline.after(60.0) + before = coord.shared_state.recent_crash_count(window_sec=3600.0) + + await asyncio.wait_for( + coord._await_within_session_bound(_hang_forever, stage="reactor:orchestration"), + timeout=0.5, + ) + + assert coord.shared_state.recent_crash_count(window_sec=3600.0) == before + 1 + + @pytest.mark.asyncio + async def test_a_step_the_session_bound_cut_short_is_not_a_crash(self, coord: Coordinator): + """A session running out is how a run ends, not a fault the emergency stop should count.""" + coord._run_deadline = Deadline.after(0.01) + before = coord.shared_state.recent_crash_count(window_sec=3600.0) + + await asyncio.wait_for( + coord._await_within_session_bound(_hang_forever, stage="reactor:orchestration"), + timeout=0.5, + ) + + assert coord.shared_state.recent_crash_count(window_sec=3600.0) == before + + def test_all_reactor_steps_share_one_total_timeout(self, coord: Coordinator): + coord.reactor_turn_timeout_sec = 777.0 + assert coord._stage_timeout_sec("advance_phase") is None + assert coord._stage_timeout_sec("reactor:orchestration") == 777.0 + assert coord._stage_timeout_sec("reactor:critic") == 777.0 + assert coord._stage_timeout_sec("reactor:robustness") == 777.0 + + def test_total_timeout_reads_the_coordinator_environment(self, session_dir, monkeypatch): + monkeypatch.setenv("INFERENCE_OPTIMIZER_REACTOR_TURN_TIMEOUT_SEC", "42.5") + configured = Coordinator(session_dir / "configured", backends=_backends()) + try: + assert configured.reactor_turn_timeout_sec == 42.5 + finally: + configured.db.close() + + @pytest.mark.asyncio + async def test_non_reactor_steps_are_not_capped(self, coord: Coordinator): + coord.reactor_turn_timeout_sec = 0.01 + finished: list[bool] = [] + + async def _finish_after_the_reactor_cap() -> None: + await asyncio.sleep(0.03) + finished.append(True) + + await coord._await_within_session_bound(_finish_after_the_reactor_cap, stage="advance_phase") + + assert finished == [True] + + @pytest.mark.asyncio + async def test_pre_stage_progress_write_failure_does_not_block_the_reactor( + self, + coord: Coordinator, + monkeypatch, + ): + calls: list[str] = [] + + def _fail_stamp(*_args, **_kwargs) -> None: + calls.append("stamp") + raise OSError("network filesystem unavailable") + + async def _factory() -> None: + calls.append("factory") + + monkeypatch.setattr(supervisor_store, "stamp_tick", _fail_stamp) + + await coord._await_within_session_bound(_factory, stage="reactor:orchestration") + + assert calls == ["stamp", "factory", "stamp"] + + @pytest.mark.asyncio + async def test_post_stage_progress_write_failure_does_not_override_success( + self, + coord: Coordinator, + monkeypatch, + ): + calls: list[str] = [] + + def _fail_second_stamp(*_args, **_kwargs) -> None: + calls.append("stamp") + if calls == ["stamp", "factory", "stamp"]: + raise OSError("network filesystem unavailable") + + async def _factory() -> None: + calls.append("factory") + + monkeypatch.setattr(supervisor_store, "stamp_tick", _fail_second_stamp) + + await coord._await_within_session_bound(_factory, stage="reactor:orchestration") + + assert calls == ["stamp", "factory", "stamp"] + + @pytest.mark.asyncio + async def test_three_near_budget_reactors_refresh_progress_between_stages( + self, + coord: Coordinator, + monkeypatch, + ): + now = [1_000_000.0] + monkeypatch.setattr("hyperloom.orchestrator.loop.coordinator.time.time", lambda: now[0]) + _record_current_owner(coord.session_dir) + supervisor_store.stamp_tick(coord.session_dir, tick=1, now_unix=now[0]) + coord.reactor_turn_timeout_sec = 0.1 + supervisor = Supervisor(coord.session_dir, tick_stall_sec=0.11, now=lambda: now[0]) + verdicts: list[str] = [] + + async def _near_budget() -> None: + now[0] += 0.09 + verdicts.append(supervisor.observe().verdict) + + for role in ("orchestration", "critic", "robustness"): + await coord._await_within_session_bound(_near_budget, stage=f"reactor:{role}") + + assert verdicts == [ALIVE, ALIVE, ALIVE] + + @pytest.mark.asyncio + async def test_one_reactor_without_progress_becomes_wedged(self, coord: Coordinator): + _record_current_owner(coord.session_dir) + coord.reactor_turn_timeout_sec = 0.2 + supervisor = Supervisor(coord.session_dir, tick_stall_sec=0.03, now=time.time) + verdicts: list[str] = [] + + async def _outlive_stall_window() -> None: + await asyncio.sleep(0.05) + verdicts.append(supervisor.observe().verdict) + + await coord._await_within_session_bound(_outlive_stall_window, stage="reactor:orchestration") + + assert verdicts == [WEDGED] + @pytest.mark.asyncio async def test_closing_uses_the_grace_bound_not_the_session_deadline(self, coord: Coordinator): coord._run_deadline = Deadline.after(-10.0) diff --git a/src/hyperloom/inference_optimizer/tests/test_shared_state_evolution.py b/src/hyperloom/inference_optimizer/tests/test_shared_state_evolution.py index 55ff5c11f8..3ef76ca78b 100644 --- a/src/hyperloom/inference_optimizer/tests/test_shared_state_evolution.py +++ b/src/hyperloom/inference_optimizer/tests/test_shared_state_evolution.py @@ -251,6 +251,21 @@ def test_core_state_fields_contains_v08_new_additions(): assert not missing, f"v0.8 §3.10 requires these to be CORE: {sorted(missing)}" +def test_a_run_leg_boundary_is_not_writable_by_update_state(): + """``leg_ended_ts`` decides where the stopped leg's phase segment ends. + + The next leg banks time up to it, so a forged value bills that phase for + time it never ran; the Coordinator owns it exactly as it owns ``stop_ts``. + """ + state = SharedState() + state.leg_ended_ts = "2026-08-01T00:00:00+00:00" + + applied = state.apply_changes({"leg_ended_ts": "2099-01-01T00:00:00+00:00"}, allow_core=False) + + assert applied == {} + assert state.leg_ended_ts == "2026-08-01T00:00:00+00:00" + + def test_policy_blocks_llm_phase_write(): """LLM ``update_state`` setting ``phase=KERNEL`` is denied.""" from hyperloom.orchestrator.roles.agent_role import ( diff --git a/src/hyperloom/inference_optimizer/tools/robustness_monitor.sh.example b/src/hyperloom/inference_optimizer/tools/robustness_monitor.sh.example index 0cc4d0073c..95f7c25090 100644 --- a/src/hyperloom/inference_optimizer/tools/robustness_monitor.sh.example +++ b/src/hyperloom/inference_optimizer/tools/robustness_monitor.sh.example @@ -5,8 +5,8 @@ # Robustness monitor for long-running ``python -m hyperloom.inference_optimizer.cli optimize`` runs. # # Polls ``$INFERENCE_OPTIMIZER_SESSION_DIR/state.json`` every 300s. On a -# terminal session (any ``stop_reason`` in ``STOP_REASON_VOCAB``, ``phase=CLOSE``, -# or ``reports/final.md`` present), exits 0 without resuming. Otherwise it only +# terminal session (a completed CLOSE sequence, a final artifact, or any +# ``stop_reason`` in ``STOP_REASON_VOCAB``), exits 0 without resuming. Otherwise it only # re-launches the session via ``optimize --resume-from "$session_dir"`` # when the optimizer is judged dead by a ROBUST, multi-signal liveness check # (issue #592). The session is considered ALIVE when ANY of the following hold: @@ -79,7 +79,7 @@ # # Edit the defaults above before copying if your run uses different # values. Terminal detection is aligned with -# ``hyperloom.orchestrator.phase_state.STOP_REASON_VOCAB`` (not a +# ``hyperloom.orchestrator.phases.machine_state.STOP_REASON_VOCAB`` (not a # hard-coded subset — ``baseline_failed`` and other failure sentinels must not # trigger auto-resume). @@ -135,7 +135,8 @@ read_stop() { } # Exit 0 when the session is finished (no auto-resume). Uses STOP_REASON_VOCAB -# from phase_state when importable; falls back to a minimal set for offline use. +# from machine_state when importable; falls back to a minimal subset for offline +# use, and says on stderr when it does. is_terminal_session() { python3 - "$session_dir" <<'PY' import json @@ -145,25 +146,29 @@ import sys sd = pathlib.Path(sys.argv[1]) state_path = sd / "state.json" final_report = sd / "reports" / "final.md" +final_summary = sd / "reports" / "final.json" -if final_report.is_file(): - print("final_report", file=sys.stderr) +if final_report.is_file() or final_summary.is_file(): + print("final_artifact", file=sys.stderr) raise SystemExit(0) if not state_path.is_file(): raise SystemExit(1) state = json.loads(state_path.read_text()) -phase = (state.get("phase") or "").strip() stop = (state.get("stop_reason") or "").strip() +close_done = state.get("close_sequence_done") is True -if phase == "CLOSE": - print(f"phase={phase}", file=sys.stderr) +if close_done: + print("close_sequence_done", file=sys.stderr) raise SystemExit(0) try: - from hyperloom.orchestrator.phase_state import STOP_REASON_VOCAB + from hyperloom.orchestrator.phases.machine_state import STOP_REASON_VOCAB except ImportError: + # Narrower than the real vocabulary, so say so: a session that stopped for a + # reason only the canonical set carries would be resumed from here. + print("warn: STOP_REASON_VOCAB unavailable; using the offline subset", file=sys.stderr) STOP_REASON_VOCAB = frozenset({ "target_reached", "global_converged", "time_exhausted", "max_ticks", "baseline_failed", "prelude_baseline_failed", "policy_loop", @@ -248,9 +253,17 @@ try: except ValueError: fresh = 600 state_path = sd / "state.json" +clean_leg_ended = False +try: + state = json.loads(state_path.read_text()) + clean_leg_ended = bool(str(state.get("leg_ended_ts") or "").strip()) and not str( + state.get("stop_reason") or "" + ).strip() +except Exception: + pass try: age = time.time() - state_path.stat().st_mtime - if age <= fresh: + if age <= fresh and not clean_leg_ended: print(f"state_mtime_age={int(age)}s", file=sys.stderr) raise SystemExit(0) except SystemExit: diff --git a/src/hyperloom/orchestrator/bringup/reconcile.py b/src/hyperloom/orchestrator/bringup/reconcile.py index 9bf0a960ea..6dadf28c38 100644 --- a/src/hyperloom/orchestrator/bringup/reconcile.py +++ b/src/hyperloom/orchestrator/bringup/reconcile.py @@ -212,14 +212,21 @@ async def run(self, now_unix: float) -> ReconcileReport: async def _stamp_tick(self, now_unix: float, report: ReconcileReport) -> None: """Record that a tick has started, for the process watching from outside.""" + await self.stamp_progress(now_unix) + + async def stamp_progress(self, now_unix: float) -> None: + """Refresh the coordinator progress timestamp.""" if self._session_dir is None: return - await asyncio.to_thread( - supervisor_store.stamp_tick, - self._session_dir, - tick=int(getattr(self._shared_state, "tick", 0)), - now_unix=now_unix, - ) + try: + await asyncio.to_thread( + supervisor_store.stamp_tick, + self._session_dir, + tick=int(getattr(self._shared_state, "tick", 0)), + now_unix=now_unix, + ) + except OSError as exc: + log.warning("reconcile: progress stamp failed: %s", exc) async def _fail_dead_tasks(self, now_unix: float, report: ReconcileReport) -> None: """Fail every running task whose process is provably gone.""" diff --git a/src/hyperloom/orchestrator/loop/coordinator.py b/src/hyperloom/orchestrator/loop/coordinator.py index c2c1fb882b..89f8e19a99 100644 --- a/src/hyperloom/orchestrator/loop/coordinator.py +++ b/src/hyperloom/orchestrator/loop/coordinator.py @@ -8,12 +8,13 @@ import asyncio import logging import os +import signal import time import traceback from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Awaitable, Callable +from typing import AbstractSet, Any, Awaitable, Callable from hyperloom.orchestrator.actions.executors._grid_server_args import ( tokenize_server_args_preserving_json, @@ -86,12 +87,14 @@ from hyperloom.common.deadline import Deadline from hyperloom.common.prompt_safety import defang_prompt_structure as _defang_prompt_structure from hyperloom.common.prompt_safety import flatten_for_prompt as _flatten_for_inbox +from hyperloom.orchestrator.supervisor.watch import SUPERVISOR_RESTART_REASON from ..trace.orchestration_trace import ( write_mcp_setup_once, ) from .coordinator_helpers import ( _infer_model_class_from_config, format_exc_brief, + resolve_reactor_turn_timeout_sec, serialize_verdict_advisory, ) @@ -534,6 +537,7 @@ def __init__( if name not in backends: raise ValueError(f"missing backend for role {name!r} (provide via Coordinator(backends={{...}}))") self.backends = dict(backends) + self.reactor_turn_timeout_sec = resolve_reactor_turn_timeout_sec() # Persistence layer db_path = db_path_for(self.session_dir) @@ -616,6 +620,7 @@ def __init__( ) self.state = CoordinatorState() self._stop = asyncio.Event() + self._stop_classification = "" self._tasks_running: list[asyncio.Task] = [] # Wall-clock stamp of the last maintenance pass (lease reaping + DB retention). @@ -1256,7 +1261,7 @@ def _pin_source_trees(self) -> None: # Lifecycle async def stop(self) -> None: - """Signal shutdown, cancel in-flight work, finalize, and close the DB.""" + """Signal shutdown, cancel in-flight work, and close the DB.""" self._stop.set() try: await self.dispatcher.cancel_inflight_actions(reason="coordinator_stop") @@ -1273,8 +1278,6 @@ async def stop(self) -> None: pass except Exception: # noqa: BLE001 log.exception("reactor task raised on shutdown") - # Safety net: recipe/journal finalize when CLOSE sequencer didn't run. - await self._recipe_kb_t4_hook() self.db.close() def _bind_session_deadline( @@ -1443,6 +1446,13 @@ def _seconds_until_session_bound(self) -> float | None: return None return bound.remaining() + def _stage_timeout_sec(self, stage: str) -> float | None: + """Return the total wall-clock ceiling for an inline reactor turn.""" + role = stage.removeprefix("reactor:") + if role == stage: + return None + return self.reactor_turn_timeout_sec + def _stop_requested(self) -> bool: """Whether an operator has asked this run to stop. @@ -1457,6 +1467,28 @@ def _stop_requested(self) -> bool: drain = self._signals return drain is not None and drain.requested.is_set() + def _signal_stop_reason(self) -> str: + """Classify the stop from the signal captured by the drain.""" + drain = self._signals + received = frozenset() if drain is None else frozenset(drain.received) + return self._classify_stop(received) or "signal" + + def _classify_stop(self, received: AbstractSet[int], *, pending: str = "") -> str: + """Classify final state from terminal outcome and captured signals.""" + if signal.SIGINT in received or signal.SIGTERM in received: + return "signal" + terminal = pending or self.shared_state.stop_reason + if terminal: + return terminal + if received == {signal.SIGHUP}: + return SUPERVISOR_RESTART_REASON + return "signal" if received else "" + + @property + def stop_classification(self) -> str: + """Authoritative in-process stop classification.""" + return self._stop_classification + async def _await_within_session_bound( self, factory: Callable[[], Awaitable[Any]], @@ -1468,15 +1500,35 @@ async def _await_within_session_bound( if remaining is not None and remaining <= 0.0: log.warning("Coordinator: skipping %s; session bound already elapsed", stage) return + stage_timeout = self._stage_timeout_sec(stage) + if stage_timeout is not None: + await self.reconciler.stamp_progress(time.time()) + timeout = ( + min(remaining, stage_timeout) + if remaining is not None and stage_timeout is not None + else remaining + if stage_timeout is None + else stage_timeout + ) + if timeout is None: + await factory() + return try: - # ``timeout=None`` waits until the step finishes (unbounded run). - await asyncio.wait_for(factory(), timeout=remaining) - except asyncio.TimeoutError: + await asyncio.wait_for(factory(), timeout=timeout) + except asyncio.TimeoutError as exc: log.warning( - "Coordinator: %s hit the session bound after %.1fs; cancelled so the tick can close", + "Coordinator: %s hit its %.1fs bound; cancelled so the tick can close", stage, - remaining, + timeout, ) + if stage_timeout is not None and timeout == stage_timeout: + # Cancelling the turn lets the tick advance, which reads as a + # healthy loop to the supervisor; the crash count is the only + # channel left that can end a session wedged on one role. + self._record_coordinator_exception(stage=stage, exc=exc, agent=stage.removeprefix("reactor:")) + finally: + if stage_timeout is not None: + await self.reconciler.stamp_progress(time.time()) # Long-run interface async def run( @@ -1600,7 +1652,7 @@ async def run( # check stop conditions if self._stop_requested(): - stop_reason = "signal" + stop_reason = self._signal_stop_reason() break if self.shared_state.stop_reason and not in_closing: stop_reason = self.shared_state.stop_reason @@ -1652,32 +1704,39 @@ async def run( if tick_interval_sec > 0: try: await asyncio.wait_for(self._stop.wait(), timeout=tick_interval_sec) - stop_reason = "signal" + stop_reason = self._signal_stop_reason() break except asyncio.TimeoutError: # Normal path: no stop signal within the tick interval. pass finally: + final_signals: AbstractSet[int] = frozenset() + if self._signals is not None: + final_signals = self._signals.close() + self._signals = None + stop_reason = self._classify_stop(final_signals, pending=stop_reason) + self._stop_classification = stop_reason + resumable_stop = stop_reason == SUPERVISOR_RESTART_REASON if self.shared_state.closing_phase: self.shared_state.closing_phase = False - # Resuming a terminal session can break out before stop_reason is set. - self.shared_state.set_stop_reason( - stop_reason - or self.shared_state.stop_reason - or ("coordinator_exception" if last_tick_exc is not None else "unknown") - ) + if resumable_stop: + self.shared_state.stop_reason = "" + self.shared_state.stop_ts = "" + self.shared_state.leg_ended_ts = now_iso() + else: + # Resuming a terminal session can break out before stop_reason is set. + self.shared_state.set_stop_reason( + stop_reason + or self.shared_state.stop_reason + or ("coordinator_exception" if last_tick_exc is not None else "unknown") + ) self.shared_state.save(self.session_dir) - # Every exit from the loop lands here, including those the phase - # machine never saw -- a signal, an exception, a resumed terminal - # session -- so the report is written even when nothing entered - # CLOSE. The sequencer bounds its own steps. - try: - await self.ensure_close_sequence(reason=self.shared_state.stop_reason) - except (asyncio.CancelledError, Exception): # noqa: BLE001 — the teardown below must still run - log.exception("Coordinator: terminal close sequence did not finish") - # Every graceful terminal path gets one idempotent Recipe finalize - # attempt, including stop-check exits that never enter PHASE_CLOSE. - await self._recipe_kb_t4_hook() + if not resumable_stop: + try: + await self.ensure_close_sequence(reason=self.shared_state.stop_reason) + except (asyncio.CancelledError, Exception): # noqa: BLE001 + log.exception("Coordinator: terminal close sequence did not finish") + await self._recipe_kb_t4_hook() log.info( "Coordinator.run: stopped tick=%d reason=%s baseline_tput=%.1f " "cumulative_gain_validated=%.2f%% max_minutes=%.0f", @@ -1687,9 +1746,6 @@ async def run( self.shared_state.cumulative_gain_validated, max_minutes_value, ) - if self._signals is not None: - self._signals.close() - self._signals = None with timed_teardown_step(self.shared_state, "close_backends"): await self._close_backends() # A server outliving the run holds every GPU it was given, so the @@ -1697,7 +1753,7 @@ async def run( with timed_teardown_step(self.shared_state, "reap_orphaned_servers"): await asyncio.to_thread(self._reap_orphaned_servers_best_effort, phase="shutdown") self.shared_state.save(self.session_dir) - return self.shared_state.stop_reason + return stop_reason or self.shared_state.stop_reason async def _close_backends(self) -> None: """Release every backend holding a live agent session.""" diff --git a/src/hyperloom/orchestrator/loop/coordinator_helpers.py b/src/hyperloom/orchestrator/loop/coordinator_helpers.py index 6ffc969327..d895602239 100644 --- a/src/hyperloom/orchestrator/loop/coordinator_helpers.py +++ b/src/hyperloom/orchestrator/loop/coordinator_helpers.py @@ -8,6 +8,7 @@ import hashlib import json import logging +import math import os import re import shlex @@ -36,6 +37,8 @@ # Constants below are read from other modules; listed here to mark them as intentionally exported. __all__ = [ + "DEFAULT_REACTOR_TURN_TIMEOUT_SEC", + "REACTOR_TURN_TIMEOUT_ENV", "TIME_BUDGET_EXEMPT_ACTIONS", "_GEAK_MEASUREMENT_DIVERGENCE_WARN_PCT", "_MIN_KERNEL_ENGAGED_GAIN_PCT", @@ -43,8 +46,33 @@ "coerce_needs_gpu", "expected_action_cost_minutes", "measured_baseline_runtime_sec", + "resolve_reactor_turn_timeout_sec", ] +REACTOR_TURN_TIMEOUT_ENV = "INFERENCE_OPTIMIZER_REACTOR_TURN_TIMEOUT_SEC" +DEFAULT_REACTOR_TURN_TIMEOUT_SEC = 1800.0 + + +def resolve_reactor_turn_timeout_sec(env: Mapping[str, str] | None = None) -> float: + """Resolve the reactor turn's total wall-clock timeout.""" + environ = os.environ if env is None else env + raw = environ.get(REACTOR_TURN_TIMEOUT_ENV, "").strip() + if not raw: + return DEFAULT_REACTOR_TURN_TIMEOUT_SEC + try: + value = float(raw) + except ValueError: + value = 0.0 + if value > 0.0 and math.isfinite(value): + return value + log.warning( + "%s=%r is not a positive finite number; using default %.1fs", + REACTOR_TURN_TIMEOUT_ENV, + raw, + DEFAULT_REACTOR_TURN_TIMEOUT_SEC, + ) + return DEFAULT_REACTOR_TURN_TIMEOUT_SEC + def coerce_needs_gpu(value: Any) -> bool: """Coerce a ``needs_gpu`` specialist parameter value to a Python bool.""" diff --git a/src/hyperloom/orchestrator/loop/signals.py b/src/hyperloom/orchestrator/loop/signals.py index 95057fb139..3658eccb78 100644 --- a/src/hyperloom/orchestrator/loop/signals.py +++ b/src/hyperloom/orchestrator/loop/signals.py @@ -17,8 +17,8 @@ __all__ = ["SignalDrain"] -#: Signals an operator uses to ask for a graceful stop. -STOP_SIGNALS: tuple[int, ...] = (signal.SIGINT, signal.SIGTERM) +#: Signals that ask the coordinator to stop. +STOP_SIGNALS: tuple[int, ...] = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP) class SignalDrain: @@ -43,6 +43,8 @@ def __init__( self.requested = threading.Event() """threading.Event: Set by the reading thread the instant a stop arrives, so synchronous code on the tick's stack can see it without the loop.""" + self.received: set[int] = set() + """Signal numbers observed by the drain thread.""" self._read_fd = -1 self._write_fd = -1 self._previous: dict[int, Any] = {} @@ -88,6 +90,7 @@ def _drain(self) -> None: return if not data: return + self.received.update(data) self.requested.set() try: self._loop.call_soon_threadsafe(self._stop_event.set) @@ -112,24 +115,26 @@ def _close_fd(self, fd: int, what: str) -> int: log.debug("SignalDrain: closing %s failed", what, exc_info=True) return -1 - def close(self) -> None: - """Restore the previous handlers and stop the draining thread.""" - if self._previous_wakeup != -1 or self._previous: - try: - signal.set_wakeup_fd(self._previous_wakeup if self._previous_wakeup != -1 else -1) - except (ValueError, OSError): - log.debug("SignalDrain: restoring the wakeup fd failed", exc_info=True) - self._previous_wakeup = -1 + def close(self) -> frozenset[int]: + """Stop the drain and return every signal captured before closure.""" + restore_wakeup = self._previous_wakeup != -1 or bool(self._previous) for sig, handler in list(self._previous.items()): try: signal.signal(sig, handler) except (ValueError, OSError, TypeError): log.debug("SignalDrain: restoring handler for signal %s failed", sig, exc_info=True) self._previous.clear() + if restore_wakeup: + try: + signal.set_wakeup_fd(self._previous_wakeup if self._previous_wakeup != -1 else -1) + except (ValueError, OSError): + log.debug("SignalDrain: restoring the wakeup fd failed", exc_info=True) + self._previous_wakeup = -1 # Closing the write end ends the reader's blocking read. self._write_fd = self._close_fd(self._write_fd, "the write end") - self._read_fd = self._close_fd(self._read_fd, "the read end") thread, self._thread = self._thread, None if thread is not None: - thread.join(timeout=1.0) + thread.join() + self._read_fd = self._close_fd(self._read_fd, "the read end") self.armed = False + return frozenset(self.received) diff --git a/src/hyperloom/orchestrator/policy/gate.py b/src/hyperloom/orchestrator/policy/gate.py index 9f6f8e0798..1c4a6f9971 100644 --- a/src/hyperloom/orchestrator/policy/gate.py +++ b/src/hyperloom/orchestrator/policy/gate.py @@ -429,6 +429,9 @@ def _trace_path_allowlist() -> tuple[str, ...]: # without the other lets an update_state move the session's end time # away from the reason it was stamped for. "stop_ts", + # Where the stopped leg's phase segment ends; the next leg banks time up + # to it, so a forged value bills a phase for time it never ran. + "leg_ended_ts", "last_tick_exception", "cumulative_gain_validated", "cumulative_gain_validated_ts", diff --git a/src/hyperloom/orchestrator/roles/codex_agent.py b/src/hyperloom/orchestrator/roles/codex_agent.py index a94eed5aae..94731ae24a 100644 --- a/src/hyperloom/orchestrator/roles/codex_agent.py +++ b/src/hyperloom/orchestrator/roles/codex_agent.py @@ -25,7 +25,13 @@ from ..trace.llm_trace import new_call_id from .agent_role import DEFAULT_CODEX_MODEL -from .base import BackendError, BackendTurnResult, LLMCallFailed, parse_call_timeout_env, safe_int +from .base import ( + BackendError, + BackendTurnResult, + LLMCallFailed, + parse_call_timeout_env, + safe_int, +) _BARE_INTENTS_RE = re.compile(r'(\{.*?"intents".*\})', re.DOTALL) diff --git a/src/hyperloom/orchestrator/state/shared_state.py b/src/hyperloom/orchestrator/state/shared_state.py index 165f75fb40..71f7af8bfe 100644 --- a/src/hyperloom/orchestrator/state/shared_state.py +++ b/src/hyperloom/orchestrator/state/shared_state.py @@ -605,6 +605,8 @@ class SharedState(_RenderMixin, _ExploreStateMixin): stop_reason: str = "" # When the session first stopped, and therefore its end time for consumers. stop_ts: str = "" + # When the current run leg ended without ending the session. + leg_ended_ts: str = "" # When the current run leg began, i.e. the most recent ``--resume``; empty for a session that has only ever run # once. resumed_ts: str = "" diff --git a/src/hyperloom/orchestrator/supervisor/launcher.py b/src/hyperloom/orchestrator/supervisor/launcher.py index 915ba2f98c..c68c6a5897 100644 --- a/src/hyperloom/orchestrator/supervisor/launcher.py +++ b/src/hyperloom/orchestrator/supervisor/launcher.py @@ -15,7 +15,11 @@ from hyperloom.common.env_safety import scrub_benchmark_process_env from hyperloom.common.proctree import running from hyperloom.inference_optimizer.session.session_paths import supervisor_log_path -from hyperloom.orchestrator.supervisor.watch import DEFAULT_TICK_STALL_SEC +from hyperloom.orchestrator.loop.coordinator_helpers import ( + DEFAULT_REACTOR_TURN_TIMEOUT_SEC, + resolve_reactor_turn_timeout_sec, +) +from hyperloom.orchestrator.supervisor.watch import DEFAULT_POLL_SEC, DEFAULT_TICK_STALL_SEC log = logging.getLogger(__name__) @@ -28,9 +32,8 @@ #: How long the supervisor is given to exit after being asked to. _STOP_GRACE_SEC: float = 5.0 -#: The shortest stall window worth arming. One tick can legitimately spend two -#: role turns at their five-minute cap, plus a retry each. -_TICK_STALL_FLOOR_SEC: float = 1800.0 +#: The shortest default stall window covers one reactor turn and one poll. +_TICK_STALL_FLOOR_SEC: float = DEFAULT_REACTOR_TURN_TIMEOUT_SEC + DEFAULT_POLL_SEC __all__ = [ "SUPERVISOR_ENABLE_ENV", @@ -46,7 +49,7 @@ def _truthy(value: str) -> bool: return value.strip().lower() in {"1", "true", "yes", "on"} -def tick_stall_sec(session_sec: float) -> float: +def tick_stall_sec(session_sec: float, *, env: dict[str, str] | None = None) -> float: """Return the stall window a session of ``session_sec`` should be watched with. Args: @@ -55,9 +58,11 @@ def tick_stall_sec(session_sec: float) -> float: Returns: float: Seconds a tick may go without advancing. """ + floor = resolve_reactor_turn_timeout_sec(env) + DEFAULT_POLL_SEC + default = max(DEFAULT_TICK_STALL_SEC, floor) if session_sec <= 0.0: - return DEFAULT_TICK_STALL_SEC - return max(_TICK_STALL_FLOOR_SEC, min(DEFAULT_TICK_STALL_SEC, session_sec / 2.0)) + return default + return max(floor, min(default, session_sec / 2.0)) def spawn_supervisor( @@ -96,7 +101,7 @@ def spawn_supervisor( str(session_dir), ] override = environ.get(SUPERVISOR_STALL_ENV, "").strip() - argv += ["--tick-stall-sec", override or f"{tick_stall_sec(session_sec):.0f}"] + argv += ["--tick-stall-sec", override or f"{tick_stall_sec(session_sec, env=environ):.0f}"] log_path = supervisor_log_path(Path(session_dir)) log_path.parent.mkdir(parents=True, exist_ok=True) # The child dups this descriptor at spawn, so the parent's copy is done the diff --git a/src/hyperloom/orchestrator/supervisor/store.py b/src/hyperloom/orchestrator/supervisor/store.py index 6f2b2411b0..b19e4befa2 100644 --- a/src/hyperloom/orchestrator/supervisor/store.py +++ b/src/hyperloom/orchestrator/supervisor/store.py @@ -19,6 +19,7 @@ __all__ = [ "TickStamp", + "read_status", "read_tick", "stamp_tick", "write_status", @@ -102,6 +103,14 @@ def write_status(session_dir: Path | str, payload: dict) -> None: _write(supervisor_status_path(Path(session_dir)), payload) +def read_status(session_dir: Path | str) -> dict | None: + """Read the last supervisor status, if one exists.""" + try: + return read_json(supervisor_status_path(Path(session_dir)), strict=True, require_dict=True) + except FileNotFoundError: + return None + + def _write(path: Path, payload: dict) -> None: """Replace a JSON file in one step, durably enough for another host to read.""" atomic_write_json(path, payload, trailing_newline=True, fsync=True, fsync_dir=True) diff --git a/src/hyperloom/orchestrator/supervisor/tests/test_supervisor.py b/src/hyperloom/orchestrator/supervisor/tests/test_supervisor.py index bfd4bfd230..1501fcec75 100644 --- a/src/hyperloom/orchestrator/supervisor/tests/test_supervisor.py +++ b/src/hyperloom/orchestrator/supervisor/tests/test_supervisor.py @@ -7,6 +7,7 @@ import json import os +import signal import socket import subprocess # nosec B404 - spawns a process purely so its pid can be signalled import sys @@ -22,15 +23,18 @@ supervisor_status_path, ) from hyperloom.orchestrator.supervisor import store, tick_stall_sec +from hyperloom.orchestrator.supervisor.launcher import _TICK_STALL_FLOOR_SEC from hyperloom.orchestrator.supervisor.watch import ( ALIVE, DEAD, + DEFAULT_POLL_SEC, DIED_STOP_REASON, UNKNOWN, WEDGED, WEDGED_STOP_REASON, Supervisor, ) +from hyperloom.orchestrator.loop.coordinator_helpers import REACTOR_TURN_TIMEOUT_ENV _NOW = 1_000_000.0 @@ -66,7 +70,7 @@ def _a_pid_that_is_running(request, *, deaf: bool = False) -> int: Returns: int: The child's pid, once it has announced it is ready to be signalled. """ - ignore = "signal.signal(signal.SIGTERM, signal.SIG_IGN); " if deaf else "" + ignore = "signal.signal(signal.SIGHUP, signal.SIG_IGN); " if deaf else "" program = f"import signal, sys, time; {ignore}sys.stdout.write('r'); sys.stdout.flush(); time.sleep(300)" child = subprocess.Popen([sys.executable, "-c", program], stdout=subprocess.PIPE) # nosec B603 request.addfinalizer(child.kill) @@ -81,13 +85,25 @@ def _supervisor(session_dir, **kw) -> Supervisor: return Supervisor(session_dir, **kw) -def test_the_stall_window_always_fits_inside_the_session_it_watches(): - """A window the session cannot outlast is a watch that never fires.""" - two_hours = 2 * 3600.0 - assert tick_stall_sec(two_hours) <= two_hours / 2 - assert tick_stall_sec(3600.0) <= 3600.0 / 2 +def test_the_stall_window_fits_inside_any_session_long_enough_to_hold_a_tick(): + """A window the session cannot outlast is a watch that never fires. + + The floor is one legal tick, so only a session with room for two of them + can carry a window that is both armed and inside its own budget; a shorter + session is watched by a window it cannot outlast, which is the honest + trade for never calling a legal tick wedged. + """ + shortest = 2 * _TICK_STALL_FLOOR_SEC + assert tick_stall_sec(shortest) <= shortest / 2 + assert tick_stall_sec(4 * 3600.0) <= 4 * 3600.0 / 2 # And never so short that a slow tick reads as a stopped one. - assert tick_stall_sec(60.0) == tick_stall_sec(3600.0) + assert tick_stall_sec(60.0) == _TICK_STALL_FLOOR_SEC + + +def test_the_default_stall_window_follows_a_raised_reactor_timeout(): + timeout = 7200.0 + + assert tick_stall_sec(0.0, env={REACTOR_TURN_TIMEOUT_ENV: str(timeout)}) == timeout + DEFAULT_POLL_SEC def test_a_ticking_coordinator_is_left_alone(tmp_path): @@ -154,6 +170,104 @@ async def test_a_wedged_coordinator_is_asked_to_stop_on_the_one_channel_that_rea _wait_for_exit(pid) +def _a_wedged_reading(**kw) -> mock.Mock: + """A complete wedged reading, so a status write over it round-trips as JSON.""" + fields = {"verdict": WEDGED, "pid": 123, "detail": "stalled", "tick": 7, "tick_age_sec": 999.0} + fields.update(kw) + return mock.Mock(**fields) + + +def test_a_wedged_coordinator_is_asked_to_restart_with_sighup(tmp_path): + """The restart intent is carried by the signal itself.""" + supervisor = _supervisor(tmp_path) + + with mock.patch.object(os, "kill") as kill: + supervisor._ask_to_stop(_a_wedged_reading()) + + kill.assert_called_once_with(123, signal.SIGHUP) + + +def test_an_unreadable_restart_count_refuses_another_resumable_restart(tmp_path): + status = supervisor_status_path(tmp_path) + status.parent.mkdir(parents=True) + status.write_text("{", encoding="utf-8") + supervisor = _supervisor(tmp_path, max_restarts=3) + + with mock.patch.object(os, "kill") as kill: + supervisor._ask_to_stop(_a_wedged_reading()) + + kill.assert_called_once_with(123, signal.SIGTERM) + + +def test_the_restart_is_spent_on_disk_before_the_signal_goes_out(tmp_path): + """The coordinator's stop path can outlive this supervisor. + + A count banked only after the signal is a restart the next leg reads as + never spent, which is how a bounded limit becomes an unbounded loop. + """ + store.write_status(tmp_path, {"restart_count": 2}) + supervisor = _supervisor(tmp_path, max_restarts=3) + banked: list[int] = [] + + def _read_status_at_signal_time(pid, sig): + banked.append(json.loads(supervisor_status_path(tmp_path).read_text(encoding="utf-8"))["restart_count"]) + + with mock.patch.object(os, "kill", _read_status_at_signal_time): + supervisor._ask_to_stop(_a_wedged_reading()) + + assert banked == [3] + + +def test_restart_counter_write_failure_sends_terminal_sigterm(tmp_path): + supervisor = _supervisor(tmp_path, max_restarts=3) + + with ( + mock.patch.object(store, "write_status", side_effect=OSError("disk full")), + mock.patch.object(os, "kill") as kill, + ): + supervisor._ask_to_stop(_a_wedged_reading()) + + kill.assert_called_once_with(123, signal.SIGTERM) + assert supervisor.report.refusals == ["restart counter write failed; resumable restart refused: disk full"] + + +@pytest.mark.asyncio +async def test_a_restart_within_the_limit_stays_resumable(tmp_path): + store.write_status(tmp_path, {"restart_count": 2}) + supervisor = _supervisor(tmp_path, max_restarts=3) + wedged = mock.Mock(verdict=WEDGED, pid=123, detail="stalled", tick=7, tick_age_sec=999.0) + dead = mock.Mock(verdict=DEAD, pid=123, detail="gone", tick=7, tick_age_sec=1000.0) + + with mock.patch.object(os, "kill") as kill: + await supervisor.act(wedged) + kill.assert_called_once_with(123, signal.SIGHUP) + done = await supervisor.act(dead) + + status = json.loads(supervisor_status_path(tmp_path).read_text(encoding="utf-8")) + assert done is True + assert status["restart_count"] == 3 + assert not (reports_dir(tmp_path) / "final.json").exists() + + +@pytest.mark.asyncio +async def test_a_restart_past_the_limit_becomes_terminal(tmp_path): + store.write_status(tmp_path, {"restart_count": 3}) + supervisor = _supervisor(tmp_path, max_restarts=3) + wedged = mock.Mock(verdict=WEDGED, pid=123, detail="stalled", tick=7, tick_age_sec=999.0) + dead = mock.Mock(verdict=DEAD, pid=123, detail="gone", tick=7, tick_age_sec=1000.0) + + with mock.patch.object(os, "kill") as kill: + await supervisor.act(wedged) + kill.assert_called_once_with(123, signal.SIGTERM) + done = await supervisor.act(dead) + + final = json.loads((reports_dir(tmp_path) / "final.json").read_text(encoding="utf-8")) + assert done is True + assert final["stop_reason"] == WEDGED_STOP_REASON + # The terminal ask is not a restart, so the count stays at the three spent. + assert final["supervisor"]["restart_count"] == 3 + + @pytest.mark.asyncio async def test_the_stop_is_asked_for_once_and_then_waited_on(tmp_path, request): """A signal per poll would be a stop the coordinator never gets to run.""" diff --git a/src/hyperloom/orchestrator/supervisor/watch.py b/src/hyperloom/orchestrator/supervisor/watch.py index 0cc79345ba..d01897a4fd 100644 --- a/src/hyperloom/orchestrator/supervisor/watch.py +++ b/src/hyperloom/orchestrator/supervisor/watch.py @@ -34,11 +34,8 @@ #: Nothing about the coordinator could be established. Never escalated on. UNKNOWN = "unknown" -#: How long a tick may go without advancing before it counts as wedged. Well -#: above a legitimately slow tick -- role turns are capped at five minutes each -#: and long actions run as dispatched tasks the tick does not wait on -- and -#: well inside the default session, which a window it cannot fit in would make -#: unreachable. +#: Base stall window when the reactor timeout does not require a larger one. +#: The launcher enforces one reactor turn plus one supervisor poll. DEFAULT_TICK_STALL_SEC: float = 3600.0 #: How long the coordinator is given to act on the stop it was asked for before @@ -49,18 +46,26 @@ #: How often the watcher looks. DEFAULT_POLL_SEC: float = 30.0 +#: Maximum resumable watchdog restarts before the session becomes terminal. +DEFAULT_MAX_RESTARTS: int = 3 + #: The stop reason recorded for a coordinator that stopped running its loop. WEDGED_STOP_REASON = "supervisor_tick_stalled" +#: Internal handoff telling the launcher that the session remains resumable. +SUPERVISOR_RESTART_REASON = "supervisor_restart_requested" + #: The stop reason recorded in the terminal artifact for a dead coordinator. DIED_STOP_REASON = "supervisor_coordinator_died" __all__ = [ "ALIVE", "DEAD", + "DEFAULT_MAX_RESTARTS", "DEFAULT_POLL_SEC", "DEFAULT_TICK_STALL_SEC", "DIED_STOP_REASON", + "SUPERVISOR_RESTART_REASON", "UNKNOWN", "WEDGED", "WEDGED_STOP_REASON", @@ -120,6 +125,7 @@ def __init__( tick_stall_sec: float = DEFAULT_TICK_STALL_SEC, stop_grace_sec: float = DEFAULT_STOP_GRACE_SEC, poll_sec: float = DEFAULT_POLL_SEC, + max_restarts: int = DEFAULT_MAX_RESTARTS, reaper: ReapBackend | None = None, now: Callable[[], float] | None = None, ): @@ -130,6 +136,7 @@ def __init__( tick_stall_sec: How long a tick may go without advancing. stop_grace_sec: How long the coordinator is given to act on a stop. poll_sec: Seconds between readings. + max_restarts: Resumable restart attempts allowed for this session. reaper: The reap unit; defaults to whatever this host selects. now: Wall-clock source, for tests. """ @@ -137,12 +144,24 @@ def __init__( self.tick_stall_sec = max(0.0, tick_stall_sec) self.stop_grace_sec = max(0.0, stop_grace_sec) self.poll_sec = max(0.1, poll_sec) + self.max_restarts = max(0, max_restarts) self._reaper: ReapBackend = reaper if reaper is not None else select_reaper() self._now: Callable[[], float] = now if now is not None else time.time self._report = SupervisorReport() self._ask_attempted = False self._asked_unix = 0.0 self._end_attempted = False + self._terminal_stop = False + self._restart_count = self._load_restart_count() + + def _load_restart_count(self) -> int: + """Load the durable restart count, failing closed on corrupt status.""" + try: + status = store.read_status(self.session_dir) + return max(0, int((status or {}).get("restart_count") or 0)) + except (OSError, TypeError, ValueError): + log.error("SUPERVISOR: restart count is unreadable; refusing another resumable restart") + return self.max_restarts @property def unit(self) -> ReapBackend: @@ -208,7 +227,11 @@ async def act(self, observation: Observation) -> bool: if observation.verdict == WEDGED: over = await self._escalate_wedged(observation) elif observation.verdict == DEAD: - over = await self._end(observation, DIED_STOP_REASON) + if self._asked_unix and not self._terminal_stop: + over = True + else: + reason = WEDGED_STOP_REASON if self._asked_unix else DIED_STOP_REASON + over = await self._end(observation, reason) self._write_status(observation) return over @@ -239,8 +262,25 @@ async def _escalate_wedged(self, observation: Observation) -> bool: def _ask_to_stop(self, observation: Observation) -> None: """Send the coordinator the stop signal its drain thread is waiting on.""" reason = f"{WEDGED_STOP_REASON}: {observation.detail}" + resumable = self._restart_count < self.max_restarts + self._terminal_stop = not resumable + if resumable: + # Banked before the signal, not after: the coordinator's stop path + # can outlive this supervisor, and a restart spent only afterwards + # is one the next leg reads as never spent. A signal that then fails + # leaves the budget short, which is the side to be wrong on. + self._restart_count += 1 + try: + self._write_status(observation) + except OSError as exc: + self._restart_count -= 1 + resumable = False + self._terminal_stop = True + refusal = f"restart counter write failed; resumable restart refused: {exc}" + self._report.refusals.append(refusal) + log.error("SUPERVISOR: %s", refusal) try: - os.kill(observation.pid, signal.SIGTERM) + os.kill(observation.pid, signal.SIGHUP if resumable else signal.SIGTERM) except ProcessLookupError: # It exited between the reading and the signal; the next reading # sees a dead coordinator and ends the session on that. @@ -294,6 +334,7 @@ def _write_terminal(self, observation: Observation, stop_reason: str) -> None: "coordinator_pid": observation.pid, "last_tick": observation.tick, "tick_age_sec": observation.tick_age_sec, + "restart_count": self._restart_count, "refused": list(self._report.refusals), }, }, @@ -313,6 +354,8 @@ def _write_status(self, observation: Observation) -> None: "tick_age_sec": observation.tick_age_sec, "observed_unix": self._now(), "tick_stall_sec": self.tick_stall_sec, + "restart_count": self._restart_count, + "max_restarts": self.max_restarts, "stop_asked": list(self._report.asked), "refused": list(self._report.refusals), "terminal_path": self._report.terminal_path,