haive.core.utils.interrupt_utils

Utilities for detecting whether a callable uses pause_for_human(…) to pause execution.

This wraps LangGraph’s interrupt(…) signal and provides AST-based static analysis to detect if a function or callable object may yield control for human input.

Functions

is_interruptible(obj)

Determine whether an object will trigger an interrupt via pause_for_human().

pause_for_human(payload)

Pause execution for human input and return a resume value of the same type.

uses_pause(fn)

Detect whether a function contains a call to pause_for_human(...).

Module Contents

haive.core.utils.interrupt_utils.is_interruptible(obj)[source]

Determine whether an object will trigger an interrupt via pause_for_human().

Works for: - Plain functions and methods - Callable objects (e.g., classes with __call__)

Parameters:

obj (object) – The object to test for interruptibility.

Returns:

True if the object contains or delegates to a function that uses pause_for_human(…), False otherwise.

Return type:

bool

haive.core.utils.interrupt_utils.pause_for_human(payload)[source]

Pause execution for human input and return a resume value of the same type.

This is a wrapper around langgraph.types.interrupt(…) and should be used to signal an interruptible pause in a LangGraph node.

Parameters:

payload (T) – The value to pass along with the interrupt signal.

Returns:

The same payload after human resumption.

Return type:

T

haive.core.utils.interrupt_utils.uses_pause(fn)[source]

Detect whether a function contains a call to pause_for_human(…).

Parameters:

fn (collections.abc.Callable[Ellipsis, Any]) – A top-level function or method object.

Returns:

True if the function body invokes pause_for_human(…). False if the source cannot be retrieved or does not contain such a call.

Return type:

bool