SignalStopper

gepa.utils.stop_condition.SignalStopper(signals=None)

Bases: StopperProtocol

Stop callback that stops when a signal is received.

Source code in gepa/utils/stop_condition.py

def __init__(self, signals=None):
    self.signals = signals or [signal.SIGINT, signal.SIGTERM]
    self._stop_requested = False
    self._original_handlers = {}
    self._setup_signal_handlers()

Attributes

signals = signals or [signal.SIGINT, signal.SIGTERM] instance-attribute

Functions

__call__(gepa_state: GEPAState) -> bool

Source code in gepa/utils/stop_condition.py

def __call__(self, gepa_state: GEPAState) -> bool:
    # return true if a signal was received
    return self._stop_requested

cleanup()

Restore original signal handlers.

Source code in gepa/utils/stop_condition.py

def cleanup(self):
    """Restore original signal handlers."""
    for sig, handler in self._original_handlers.items():
        try:
            signal.signal(sig, handler)
        except (OSError, ValueError):
            pass