checkpoint
Checkpoint decorator for durable work boundaries.
A checkpoint is a unit of work inside a flow whose outcome is persisted. Successful outputs become artifacts; failures are recorded for retry.
funccheckpoint(func=None, *, retries=0, type=None, runtime=None) -> _CheckpointDefinition | Callable[[Callable[..., Any]], _CheckpointDefinition]Mark a function as a durable checkpoint.
Can be used as a bare decorator or with arguments:
from kitaru import checkpoint
@checkpoint
def my_step(): ...
@checkpoint(retries=3, type="llm_call")
def my_step(): ...
@checkpoint(runtime="isolated")
def heavy_step(): ...paramfuncCallable[..., Any] | None= NoneOptional function for bare decorator use.
paramretriesint= 0Number of checkpoint-level retries on failure.
paramtypestr | None= NoneCheckpoint type for dashboard visualization.
paramruntimestr | None= NoneExecution runtime for this checkpoint. Accepts "inline"
or "isolated" (case-insensitive). When set to "isolated",
the checkpoint runs in its own container on remote orchestrators
that support it. None (the default) lets the orchestrator
decide.
Returns
_CheckpointDefinition | Callable[[Callable[..., Any]], _CheckpointDefinition]The wrapped checkpoint object or a decorator that returns it.