Skip to content

File

file

FileStopper for stopping evolution when a file exists.

This module provides a stopper that terminates evolution when a specified file exists, enabling external orchestration and graceful shutdown signals.

ATTRIBUTE DESCRIPTION
FileStopper

Stop evolution when a stop file exists.

TYPE: class

Examples:

Basic usage:

from gepa_adk.adapters.stoppers import FileStopper

stopper = FileStopper("/tmp/gepa_stop")

With automatic file removal:

stopper = FileStopper("/tmp/gepa_stop", remove_on_stop=True)

External orchestration (from another process):

touch /tmp/gepa_stop  # Signal evolution to stop
Note

This stopper is particularly useful for CI/CD pipelines, job schedulers, and external monitoring systems that cannot easily send process signals but can create files.

FileStopper

Stop evolution when a specified file exists.

Useful for external orchestration where CI/CD pipelines, job schedulers, or monitoring tools can signal graceful termination by creating a file.

ATTRIBUTE DESCRIPTION
stop_file_path

Path to the stop signal file.

TYPE: Path

remove_on_stop

If True, remove the file after triggering stop.

TYPE: bool

Examples:

Stop when file exists:

stopper = FileStopper("/tmp/gepa_stop")

Auto-remove file after triggering:

stopper = FileStopper("/tmp/gepa_stop", remove_on_stop=True)
Note

Any path that doesn't exist simply won't trigger a stop. Invalid paths are handled gracefully without raising errors.

Source code in src/gepa_adk/adapters/stoppers/file.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class FileStopper:
    """Stop evolution when a specified file exists.

    Useful for external orchestration where CI/CD pipelines, job schedulers,
    or monitoring tools can signal graceful termination by creating a file.

    Attributes:
        stop_file_path (Path): Path to the stop signal file.
        remove_on_stop (bool): If True, remove the file after triggering stop.

    Examples:
        Stop when file exists:

        ```python
        stopper = FileStopper("/tmp/gepa_stop")
        ```

        Auto-remove file after triggering:

        ```python
        stopper = FileStopper("/tmp/gepa_stop", remove_on_stop=True)
        ```

    Note:
        Any path that doesn't exist simply won't trigger a stop. Invalid paths
        are handled gracefully without raising errors.
    """

    def __init__(
        self, stop_file_path: str | Path, remove_on_stop: bool = False
    ) -> None:
        """Initialize the stopper with a stop file path.

        Args:
            stop_file_path: Path to the stop signal file. Can be a string
                or Path object. Will be converted to Path internally.
            remove_on_stop: If True, automatically remove the stop file
                after triggering a stop. Defaults to False.

        Note:
            Configure the path based on your orchestration system. Common
            locations include /tmp/, /var/run/, or project-specific directories.
        """
        self.stop_file_path = Path(stop_file_path)
        self.remove_on_stop = remove_on_stop

    def __call__(self, state: StopperState) -> bool:
        """Check if evolution should stop based on file existence.

        Args:
            state: Current evolution state snapshot (not used for file-based
                stopping, but required by the StopperProtocol).

        Returns:
            True if the stop file exists, False otherwise.

        Note:
            Once this returns True, the stop file may be removed if
            remove_on_stop was enabled. Subsequent calls will return False.
        """
        if self.stop_file_path.exists():
            if self.remove_on_stop:
                self.stop_file_path.unlink(missing_ok=True)
            return True
        return False

    def remove_stop_file(self) -> None:
        """Manually remove the stop file.

        This is idempotent - safe to call even if the file doesn't exist.
        Useful for resetting the stop condition before starting a new
        evolution run.

        Note:
            Only call this when you explicitly want to remove the stop file.
            The file is automatically removed during __call__ if remove_on_stop=True.
        """
        self.stop_file_path.unlink(missing_ok=True)

__init__ 

__init__(
    stop_file_path: str | Path, remove_on_stop: bool = False
) -> None

Initialize the stopper with a stop file path.

PARAMETER DESCRIPTION
stop_file_path

Path to the stop signal file. Can be a string or Path object. Will be converted to Path internally.

TYPE: str | Path

remove_on_stop

If True, automatically remove the stop file after triggering a stop. Defaults to False.

TYPE: bool DEFAULT: False

Note

Configure the path based on your orchestration system. Common locations include /tmp/, /var/run/, or project-specific directories.

Source code in src/gepa_adk/adapters/stoppers/file.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self, stop_file_path: str | Path, remove_on_stop: bool = False
) -> None:
    """Initialize the stopper with a stop file path.

    Args:
        stop_file_path: Path to the stop signal file. Can be a string
            or Path object. Will be converted to Path internally.
        remove_on_stop: If True, automatically remove the stop file
            after triggering a stop. Defaults to False.

    Note:
        Configure the path based on your orchestration system. Common
        locations include /tmp/, /var/run/, or project-specific directories.
    """
    self.stop_file_path = Path(stop_file_path)
    self.remove_on_stop = remove_on_stop

__call__ 

__call__(state: StopperState) -> bool

Check if evolution should stop based on file existence.

PARAMETER DESCRIPTION
state

Current evolution state snapshot (not used for file-based stopping, but required by the StopperProtocol).

TYPE: StopperState

RETURNS DESCRIPTION
bool

True if the stop file exists, False otherwise.
Note

Once this returns True, the stop file may be removed if remove_on_stop was enabled. Subsequent calls will return False.

Source code in src/gepa_adk/adapters/stoppers/file.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __call__(self, state: StopperState) -> bool:
    """Check if evolution should stop based on file existence.

    Args:
        state: Current evolution state snapshot (not used for file-based
            stopping, but required by the StopperProtocol).

    Returns:
        True if the stop file exists, False otherwise.

    Note:
        Once this returns True, the stop file may be removed if
        remove_on_stop was enabled. Subsequent calls will return False.
    """
    if self.stop_file_path.exists():
        if self.remove_on_stop:
            self.stop_file_path.unlink(missing_ok=True)
        return True
    return False

remove_stop_file 

remove_stop_file() -> None

Manually remove the stop file.

This is idempotent - safe to call even if the file doesn't exist. Useful for resetting the stop condition before starting a new evolution run.

Note

Only call this when you explicitly want to remove the stop file. The file is automatically removed during call if remove_on_stop=True.

Source code in src/gepa_adk/adapters/stoppers/file.py 
107
108
109
110
111
112
113
114
115
116
117
118
def remove_stop_file(self) -> None:
    """Manually remove the stop file.

    This is idempotent - safe to call even if the file doesn't exist.
    Useful for resetting the stop condition before starting a new
    evolution run.

    Note:
        Only call this when you explicitly want to remove the stop file.
        The file is automatically removed during __call__ if remove_on_stop=True.
    """
    self.stop_file_path.unlink(missing_ok=True)