Skip to content

Termination

Composable stop conditions — combine with | (OR) and & (AND):

from locus.core.termination import MaxIterations, ToolCalled, ConfidenceMet

termination = (
    ToolCalled("submit") & ConfidenceMet(0.9)
) | MaxIterations(15)

Base

TerminationCondition

Bases: ABC

Base class for composable termination conditions.

Subclasses implement should_terminate(state) which returns (should_stop: bool, reason: str | None).

Conditions are composable: - a | b — stop if either condition is met (OR) - a & b — stop if both conditions are met (AND)

check abstractmethod 

check(state: AgentState, **context: Any) -> tuple[bool, str | None]

Check if termination condition is met.

Parameters:

Name Type Description Default
state AgentState

Current agent state.

 required
**context Any

Additional context (last_message, tool_names, etc.)

 {}

Returns:

Type Description
tuple[bool, str | None]

Tuple of (should_stop, reason).
Source code in src/locus/core/termination.py 
50
51
52
53
54
55
56
57
58
59
60
61
@abstractmethod
def check(self, state: AgentState, **context: Any) -> tuple[bool, str | None]:
    """Check if termination condition is met.

    Args:
        state: Current agent state.
        **context: Additional context (last_message, tool_names, etc.)

    Returns:
        Tuple of (should_stop, reason).
    """
    ...

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

Conditions

MaxIterations 

MaxIterations(max_iterations: int)

Bases: TerminationCondition

Stop after N iterations.

Source code in src/locus/core/termination.py 
122
123
def __init__(self, max_iterations: int) -> None:
    self._max = max_iterations

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

TokenLimit 

TokenLimit(max_tokens: int)

Bases: TerminationCondition

Stop when token usage exceeds a limit.

Source code in src/locus/core/termination.py 
134
135
def __init__(self, max_tokens: int) -> None:
    self._max = max_tokens

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

TextMention 

TextMention(text: str, case_sensitive: bool = False)

Bases: TerminationCondition

Stop when specific text appears in the last message.

Source code in src/locus/core/termination.py 
164
165
166
def __init__(self, text: str, case_sensitive: bool = False) -> None:
    self._text = text
    self._case_sensitive = case_sensitive

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

TimeLimit 

TimeLimit(seconds: float)

Bases: TerminationCondition

Stop after a time duration.

Source code in src/locus/core/termination.py 
146
147
148
def __init__(self, seconds: float) -> None:
    self._seconds = seconds
    self._start: float | None = None

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

ToolCalled 

ToolCalled(tool_name: str)

Bases: TerminationCondition

Stop when a specific tool is called.

Source code in src/locus/core/termination.py 
192
193
def __init__(self, tool_name: str) -> None:
    self._tool_name = tool_name

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

ConfidenceMet 

ConfidenceMet(threshold: float = 0.9)

Bases: TerminationCondition

Stop when confidence threshold is reached.

Source code in src/locus/core/termination.py 
205
206
def __init__(self, threshold: float = 0.9) -> None:
    self._threshold = threshold

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

NoToolCalls

Bases: TerminationCondition

Stop when the model produces no tool calls.

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

CustomCondition 

CustomCondition(fn: Callable[..., tuple[bool, str | None]])

Bases: TerminationCondition

Stop based on a custom function.

Example

condition = CustomCondition( lambda state, **ctx: (state.iteration > 3 and "error" in str(state.messages[-1].content), "custom") )

Source code in src/locus/core/termination.py 
233
234
def __init__(self, fn: Callable[..., tuple[bool, str | None]]) -> None:
    self._fn = fn

reset 

reset() -> None

Reset any internal state. Called between runs.

Source code in src/locus/core/termination.py 
63
64
def reset(self) -> None:
    """Reset any internal state. Called between runs."""

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

Composition operators

OrCondition 

OrCondition(*conditions: TerminationCondition)

Bases: TerminationCondition

Stop if ANY child condition is met.

Source code in src/locus/core/termination.py 
78
79
def __init__(self, *conditions: TerminationCondition) -> None:
    self._conditions = list(conditions)

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)

AndCondition 

AndCondition(*conditions: TerminationCondition)

Bases: TerminationCondition

Stop only if ALL child conditions are met.

Source code in src/locus/core/termination.py 
96
97
def __init__(self, *conditions: TerminationCondition) -> None:
    self._conditions = list(conditions)

__or__ 

__or__(other: TerminationCondition) -> TerminationCondition

Combine with OR: stop if either condition met.

Source code in src/locus/core/termination.py 
66
67
68
def __or__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with OR: stop if either condition met."""
    return OrCondition(self, other)

__and__ 

__and__(other: TerminationCondition) -> TerminationCondition

Combine with AND: stop if both conditions met.

Source code in src/locus/core/termination.py 
70
71
72
def __and__(self, other: TerminationCondition) -> TerminationCondition:
    """Combine with AND: stop if both conditions met."""
    return AndCondition(self, other)