openswarm/backend/apps/workflows/restart_loop_guard.py

"""Scheduled-run restart-loop breaker (lifted from hermes-agent gateway/restart_loop_guard.py,
MIT; their incident #30719: a session that restarts its own gateway gets auto-resumed on boot,
which restarts the gateway again, SIGTERM every ~10s forever).

Our version of that loop: a scheduled workflow whose run kills the backend is overdue again the
moment the backend boots, so the scheduler re-fires it, which kills the backend again. This
breaker persists a rolling window of boots plus which workflow was MID-FIRE at each death; a
workflow implicated in consecutive dirty deaths during a boot storm gets its schedule paused
with a note instead of fired, which breaks the cycle and puts a human back in the loop.

Everything here fails OPEN (hermes's own rule: any read/write failure means no trip), because
a broken breaker must never wedge a healthy scheduler."""

import json
import logging
import os
import time
from typing import Dict, List, Optional

from typeguard import typechecked

from backend.apps.workflows.storage import DATA_DIR

logger = logging.getLogger(__name__)

# Hermes defaults: a legitimate operator restart or two never trips; a tight respawn loop does.
MAX_BOOTS = 3
WINDOW_SECONDS = 600
IMPLICATED_TRIP_COUNT = 2


def p_state_path() -> str:
    return os.path.join(DATA_DIR, "restart_loop.json")


@typechecked
def p_load() -> Dict:
    try:
        with open(p_state_path(), encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):
            return {}
        return data
    except (OSError, ValueError):
        return {}


@typechecked
def p_save(state: Dict) -> None:
    try:
        os.makedirs(DATA_DIR, exist_ok=True)
        with open(p_state_path(), "w", encoding="utf-8") as f:
            json.dump(state, f)
    except OSError:
        pass


@typechecked
def record_boot(now: Optional[float] = None) -> List[float]:
    """Called once at scheduler start: prune the boot window, append this boot, and convert any
    stale mid-fire marker into an implication (the last life died while that workflow ran)."""
    ts = time.time() if now is None else now
    state = p_load()
    boots = [float(t) for t in state.get("boots", []) if isinstance(t, (int, float)) and t >= ts - WINDOW_SECONDS]
    boots.append(ts)
    implicated = dict(state.get("implicated", {}))
    for wf_id in list(state.get("firing", {}) or {}):
        implicated[wf_id] = int(implicated.get(wf_id, 0)) + 1
        logger.warning(f"restart-loop guard: workflow {wf_id} was mid-fire when the last backend life died (implication #{implicated[wf_id]})")
    p_save({"boots": boots, "implicated": implicated, "firing": {}})
    return boots


@typechecked
def mark_firing(workflow_id: str) -> None:
    state = p_load()
    firing = dict(state.get("firing", {}))
    firing[workflow_id] = time.time()
    state["firing"] = firing
    p_save(state)


@typechecked
def clear_firing(workflow_id: str) -> None:
    """A run that ENDS (success or failure) proves the workflow doesn't kill the process; its
    implication history is forgiven so one old crash can never combine with a later one."""
    state = p_load()
    state.get("firing", {}).pop(workflow_id, None)
    if workflow_id in state.get("implicated", {}):
        state["implicated"].pop(workflow_id, None)
    p_save(state)


@typechecked
def is_tripped(workflow_id: str, now: Optional[float] = None) -> bool:
    """True when the process is boot-storming AND this workflow died mid-fire in consecutive
    lives. Both legs required: a crashy workflow on a stable backend surfaces through the normal
    failed-run path, and a boot storm with no implicated workflow is not the scheduler's fault."""
    try:
        ts = time.time() if now is None else now
        state = p_load()
        boots = [t for t in state.get("boots", []) if isinstance(t, (int, float)) and t >= ts - WINDOW_SECONDS]
        if len(boots) < MAX_BOOTS:
            return False
        return int(state.get("implicated", {}).get(workflow_id, 0)) >= IMPLICATED_TRIP_COUNT
    except Exception:
        return False


@typechecked
def clear() -> None:
    try:
        os.unlink(p_state_path())
    except OSError:
        pass