diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index 83528e3..c6ce85d 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -1,6 +1,9 @@ from __future__ import annotations +import contextlib import json +import signal +import threading import time from pathlib import Path @@ -28,6 +31,26 @@ from .triage import build_triage_queue from .stream_profiles import parse_profiles +@contextlib.contextmanager +def _cycle_timeout(seconds: int): + if seconds <= 0 or threading.current_thread() is not threading.main_thread() or not hasattr(signal, "SIGALRM"): + yield + return + previous_handler = signal.getsignal(signal.SIGALRM) + previous_timer = signal.getitimer(signal.ITIMER_REAL) + + def _raise_timeout(_signum, _frame): + raise TimeoutError(f"status_cycle_timeout_{seconds}s") + + signal.signal(signal.SIGALRM, _raise_timeout) + signal.setitimer(signal.ITIMER_REAL, seconds) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, previous_timer[0], previous_timer[1]) + signal.signal(signal.SIGALRM, previous_handler) + + def _stream_titles(config: dict[str, object]) -> dict[str, str]: return { str(item.get("id", "")): str(item.get("title", "") or item.get("id", "")) @@ -604,8 +627,8 @@ def monitor_loop( history_path: str | None = None, status_cache_path: str | None = None, ) -> None: - print(f"Monitoring {log_path}") - print(f"Writing status to {output}") + print(f"Monitoring {log_path}", flush=True) + print(f"Writing status to {output}", flush=True) last_llm_at = 0 last_llm_text: str | None = None while True: @@ -617,11 +640,13 @@ def monitor_loop( if runtime.get("log_source") == "graylog_mcp": write_refreshing_status(output, cache_path=status_cache_path, call_timeout_seconds=mcp_call_timeout, poll_timeout_seconds=mcp_poll_timeout, runtime_values=runtime) try: - status = build_status( - log_path, policy_path=policy_path, anomaly_limit=anomaly_limit, - baseline_path=baseline_path, config_path=config_path, history_path=history_path, - status_cache_path=status_cache_path, - ) + status_timeout = mcp_poll_timeout + max(30, mcp_call_timeout * 2) + with _cycle_timeout(status_timeout if runtime.get("log_source") == "graylog_mcp" else 0): + status = build_status( + log_path, policy_path=policy_path, anomaly_limit=anomaly_limit, + baseline_path=baseline_path, config_path=config_path, history_path=history_path, + status_cache_path=status_cache_path, + ) except Exception as exc: error_status = {"status": "error", "error": f"monitor_error: {exc}", "call_timeout_seconds": mcp_call_timeout, "poll_timeout_seconds": mcp_poll_timeout} status = cached_status_with_error(status_cache_path, error_status) if status_cache_path else None