From dee3ff89200d1380381d9bdc7ee064f850169521 Mon Sep 17 00:00:00 2001 From: ProtocolWarden Date: Mon, 13 Jul 2026 19:25:04 -0400 Subject: [PATCH 1/2] fix: repair OperationsCenter propose heartbeat watcher path --- scripts/operations-center.sh | 12 +--- .../entrypoints/pipeline_trigger/main.py | 63 +++++++++++++++- .../test_pipeline_trigger_heartbeat.py | 72 +++++++++++++++++++ 3 files changed, 135 insertions(+), 12 deletions(-) create mode 100644 tests/unit/entrypoints/test_pipeline_trigger_heartbeat.py diff --git a/scripts/operations-center.sh b/scripts/operations-center.sh index 5caa7b14c..b00cded9e 100755 --- a/scripts/operations-center.sh +++ b/scripts/operations-center.sh @@ -479,21 +479,14 @@ start_watch_role() { source '${ENV_PATH}' set +a _child_pid='' - _hb_pid='' - trap 'kill \$_hb_pid 2>/dev/null; kill \$_child_pid 2>/dev/null; exit 0' TERM INT - while [[ -f '${pid_file}' ]]; do - printf '{\"role\":\"propose\",\"at\":\"%s\",\"status\":\"idle\"}\n' \ - \$(date -u +%Y-%m-%dT%H:%M:%S+00:00) \ - > '${WATCH_DIR}/heartbeat_propose.json' - sleep 60 - done & - _hb_pid=\$! + trap 'kill \$_child_pid 2>/dev/null; exit 0' TERM INT while true; do set -a source '${ENV_PATH}' 2>/dev/null || true set +a '${VENV_DIR}/bin/python' -m operations_center.entrypoints.pipeline_trigger.main \ --config '${CONFIG_PATH}' \ + --status-dir '${WATCH_DIR}' \ --execute & _child_pid=\$! wait \$_child_pid @@ -502,7 +495,6 @@ start_watch_role() { echo \"{\\\"event\\\":\\\"watcher_restart\\\",\\\"role\\\":\\\"propose\\\",\\\"exit_code\\\":\$_exit}\" sleep 30 done - kill \$_hb_pid 2>/dev/null " >>"${log_file}" 2>&1 < /dev/null & else # goal, test, improve — Plane-polling board workers diff --git a/src/operations_center/entrypoints/pipeline_trigger/main.py b/src/operations_center/entrypoints/pipeline_trigger/main.py index 6a6b515d2..6504d6c3d 100644 --- a/src/operations_center/entrypoints/pipeline_trigger/main.py +++ b/src/operations_center/entrypoints/pipeline_trigger/main.py @@ -32,13 +32,17 @@ import os import subprocess import sys +import threading import time from datetime import UTC, datetime from pathlib import Path +from operations_center.entrypoints.heartbeat import touch_liveness, write_heartbeat + _TRIGGER_STATE_PATH = Path("state/pipeline_trigger_state.json") _DEFAULT_MIN_INTERVAL = 300 # 5 minutes between triggered runs _DEFAULT_POLL_INTERVAL = 30 # check every 30 seconds +_HEARTBEAT_INTERVAL_SECONDS = 30 logger = logging.getLogger(__name__) @@ -109,7 +113,26 @@ def _has_changed(old: dict[str, float], new: dict[str, float]) -> list[str]: return changed -def _run_pipeline(config_path: str, *, execute: bool) -> bool: +def _write_heartbeat( + status_dir: Path | None, + *, + success: bool, + status: str, + error: str | None = None, +) -> None: + if status_dir is None: + return + write_heartbeat(status_dir, "propose", status=status, success=success, error=error) + + +def _heartbeat_loop(status_dir: Path | None, stop_event: threading.Event) -> None: + if status_dir is None: + return + while not stop_event.wait(_HEARTBEAT_INTERVAL_SECONDS): + touch_liveness(status_dir, "propose", status="executing") + + +def _run_pipeline(config_path: str, *, execute: bool, status_dir: Path | None = None) -> bool: """Run autonomy-cycle. Returns True on success.""" cmd = [ sys.executable, @@ -131,9 +154,23 @@ def _run_pipeline(config_path: str, *, execute: bool) -> bool: ensure_ascii=False, ) ) + stop_event = threading.Event() + heartbeat_thread: threading.Thread | None = None try: + touch_liveness(status_dir, "propose", status="executing") + if status_dir is not None: + heartbeat_thread = threading.Thread( + target=_heartbeat_loop, args=(status_dir, stop_event), daemon=True + ) + heartbeat_thread.start() result = subprocess.run(cmd, timeout=600, capture_output=False) success = result.returncode == 0 + _write_heartbeat( + status_dir, + success=success, + status="idle" if success else "error", + error=None if success else f"autonomy_cycle_exit_{result.returncode}", + ) logger.info( json.dumps( { @@ -146,13 +183,24 @@ def _run_pipeline(config_path: str, *, execute: bool) -> bool: ) return success except subprocess.TimeoutExpired: + _write_heartbeat( + status_dir, + success=False, + status="error", + error="autonomy_cycle_timeout", + ) logger.warning(json.dumps({"event": "pipeline_trigger_timeout"}, ensure_ascii=False)) return False except Exception as exc: + _write_heartbeat(status_dir, success=False, status="error", error=str(exc)) logger.warning( json.dumps({"event": "pipeline_trigger_error", "error": str(exc)}, ensure_ascii=False) ) return False + finally: + stop_event.set() + if heartbeat_thread is not None: + heartbeat_thread.join(timeout=1) def run_trigger_loop( @@ -161,6 +209,7 @@ def run_trigger_loop( execute: bool = False, min_interval_seconds: int = _DEFAULT_MIN_INTERVAL, poll_interval_seconds: int = _DEFAULT_POLL_INTERVAL, + status_dir: Path | None = None, ) -> None: """Watch trigger sources and fire the pipeline on change. @@ -182,6 +231,7 @@ def run_trigger_loop( ensure_ascii=False, ) ) + _write_heartbeat(status_dir, success=True, status="idle") while True: time.sleep(poll_interval_seconds) @@ -190,6 +240,7 @@ def run_trigger_loop( changed = _has_changed(snapshot, new_snapshot) if not changed: + _write_heartbeat(status_dir, success=True, status="idle") continue now = time.time() @@ -207,6 +258,7 @@ def run_trigger_loop( ) ) snapshot = new_snapshot + _write_heartbeat(status_dir, success=True, status="idle") continue logger.info( @@ -220,7 +272,7 @@ def run_trigger_loop( ) ) - _run_pipeline(config_path, execute=execute) + _run_pipeline(config_path, execute=execute, status_dir=status_dir) last_run_at = time.time() state["last_run_at"] = last_run_at state["last_triggered_by"] = changed @@ -257,6 +309,12 @@ def main() -> None: dest="poll_interval", help=f"How often to check for trigger file changes in seconds (default: {_DEFAULT_POLL_INTERVAL}).", ) + parser.add_argument( + "--status-dir", + type=Path, + default=None, + help="Directory for heartbeat_propose.json", + ) args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(message)s") @@ -268,6 +326,7 @@ def main() -> None: execute=args.execute, min_interval_seconds=args.min_interval, poll_interval_seconds=args.poll_interval, + status_dir=args.status_dir, ) except KeyboardInterrupt: logger.info(json.dumps({"event": "pipeline_trigger_stopped"}, ensure_ascii=False)) diff --git a/tests/unit/entrypoints/test_pipeline_trigger_heartbeat.py b/tests/unit/entrypoints/test_pipeline_trigger_heartbeat.py new file mode 100644 index 000000000..31a207a01 --- /dev/null +++ b/tests/unit/entrypoints/test_pipeline_trigger_heartbeat.py @@ -0,0 +1,72 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 ProtocolWarden +"""Heartbeat coverage for the propose watcher pipeline trigger.""" + +from __future__ import annotations + +import subprocess +import time +from pathlib import Path + +from operations_center.entrypoints.heartbeat import read_heartbeat +from operations_center.entrypoints.pipeline_trigger import main as trigger_main + + +def test_run_pipeline_updates_propose_heartbeat_during_execution(monkeypatch, tmp_path: Path) -> None: + seen_midrun: dict[str, object] = {} + + def fake_run(cmd: list[str], timeout: int, capture_output: bool) -> subprocess.CompletedProcess[str]: + assert capture_output is False + assert timeout == 600 + time.sleep(0.03) + hb = read_heartbeat(tmp_path, "propose") + assert hb is not None + seen_midrun["status"] = hb["status"] + seen_midrun["last_success_at"] = hb["last_success_at"] + return subprocess.CompletedProcess(cmd, 0) + + monkeypatch.setattr(trigger_main, "_HEARTBEAT_INTERVAL_SECONDS", 0.01) + monkeypatch.setattr(trigger_main.subprocess, "run", fake_run) + + ok = trigger_main._run_pipeline("config.yaml", execute=True, status_dir=tmp_path) + + assert ok is True + assert seen_midrun == {"status": "executing", "last_success_at": None} + hb = read_heartbeat(tmp_path, "propose") + assert hb is not None + assert hb["status"] == "idle" + assert hb["last_success_at"] == hb["at"] + assert hb["consecutive_failures"] == 0 + + +def test_run_trigger_loop_marks_idle_heartbeat_on_quiet_cycles(monkeypatch, tmp_path: Path) -> None: + calls: list[tuple[bool, str]] = [] + sleep_calls = 0 + + def fake_write_heartbeat( + status_dir: Path | None, *, success: bool, status: str, error: str | None = None + ) -> None: + assert status_dir == tmp_path + assert error is None + calls.append((success, status)) + + def fake_sleep(_seconds: int) -> None: + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls >= 2: + raise KeyboardInterrupt + + monkeypatch.setattr(trigger_main, "_get_trigger_sources", lambda _config: []) + monkeypatch.setattr(trigger_main, "_load_state", lambda: {}) + monkeypatch.setattr(trigger_main, "_snapshot_mtimes", lambda _sources: {}) + monkeypatch.setattr(trigger_main, "_write_heartbeat", fake_write_heartbeat) + monkeypatch.setattr(trigger_main.time, "sleep", fake_sleep) + + try: + trigger_main.run_trigger_loop( + "config.yaml", execute=False, min_interval_seconds=300, poll_interval_seconds=1, status_dir=tmp_path + ) + except KeyboardInterrupt: + pass + + assert calls[:2] == [(True, "idle"), (True, "idle")] From d124eadacde14dc8ddb8e855e0fabb25c7e361c0 Mon Sep 17 00:00:00 2001 From: ProtocolWarden Date: Mon, 13 Jul 2026 21:35:57 -0400 Subject: [PATCH 2/2] fix: guard touch_liveness call with status_dir None check in pipeline_trigger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ty flagged Path | None passed where touch_liveness requires Path at main.py:160 — the unconditional call ran before the existing status_dir is not None guard that gates the heartbeat thread start. Root cause: OperationsCenter src/operations_center/entrypoints/pipeline_trigger/main.py. Fixes: Type check (ty) CI gate on PR #455. --- src/operations_center/entrypoints/pipeline_trigger/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/operations_center/entrypoints/pipeline_trigger/main.py b/src/operations_center/entrypoints/pipeline_trigger/main.py index 6504d6c3d..bf97daf95 100644 --- a/src/operations_center/entrypoints/pipeline_trigger/main.py +++ b/src/operations_center/entrypoints/pipeline_trigger/main.py @@ -157,8 +157,8 @@ def _run_pipeline(config_path: str, *, execute: bool, status_dir: Path | None = stop_event = threading.Event() heartbeat_thread: threading.Thread | None = None try: - touch_liveness(status_dir, "propose", status="executing") if status_dir is not None: + touch_liveness(status_dir, "propose", status="executing") heartbeat_thread = threading.Thread( target=_heartbeat_loop, args=(status_dir, stop_event), daemon=True )