cayu.workflows.workflow

Public Cayu API. Definitions are grouped by their public source module, and definitions from private implementation modules appear on their package’s page; adding a symbol does not move existing links. The import shown at each entry is a supported public namespace. Read the Python API overview and install the required extra.

gated_loop

from cayu.workflows import gated_loop

Process items sequentially with resumable pass/fail journaling.

Use stable item keys when the item list can change across resumes. Hooks run before the item completion is journaled, so on_pass and on_fail must tolerate at-least-once execution after a crash.

async gated_loop(ctx: WorkflowContext, items: Iterable[Any], *, do: Callable[[Any], Awaitable[StepResult]], gate: Callable[[Any, StepResult], Awaitable[GateOutcome | bool]], on_pass: Callable[[Any, StepResult, GateOutcome], Awaitable[None]] | None=None, on_fail: Callable[[Any, StepResult, GateOutcome], Awaitable[None]] | None=None, key: Callable[[Any], str] | None=None, name: str | None=None)

Release source · Runtime guarantees · Capability guides

JournalFactory

from cayu.workflows import JournalFactory

JournalFactory = Callable[[WorkflowJournalContext], WorkflowJournal]

Release source · Runtime guarantees · Capability guides

parallel

from cayu.workflows import parallel

Run steps concurrently and collect outcomes in submission order.

Branch failures become StepFailure, but duplicate-step and takeover signals propagate because the whole workflow must stop.

async parallel(steps: Iterable[Awaitable[StepResult]]) -> ParallelResult

Release source · Runtime guarantees · Capability guides

pipeline

from cayu.workflows import pipeline

Chain stages, feeding each StepResult into the next stage.

async pipeline(initial: Any, stages: Iterable[Callable[[Any], Awaitable[StepResult]]]) -> StepResult

Release source · Runtime guarantees · Capability guides

step

from cayu.workflows import step

Run one child agent session and return a typed StepResult.

Progress is journaled under the workflow run id. Child failures and interruptions raise StepError; takeover raises WorkflowSupersededError.

step(ctx: WorkflowContext, *, agent: str, step_id: str, prompt: str | None=None, messages: list[Message] | None=None, schema: dict[str, Any] | None=None, structured_output: StructuredOutputSpec | None=None, session_id: str | None=None, run_options: StepRunOptions | None=None) -> Coroutine[Any, Any, StepResult]

Release source · Runtime guarantees · Capability guides

StepRunOptions

from cayu.workflows import StepRunOptions

Runtime controls forwarded from step to the child RunRequest.

Workflow-owned lineage fields are derived from the journal anchor rather than exposed here, so parent/child budget roll-ups stay coherent.

Bases: BaseModel.

Fields and attributes

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name	Type	Default
execution_deadline	ExecutionDeadline	Field(default_factory=ExecutionDeadline, exclude_if=lambda boundary: boundary.expires_at is None)
target	ModelTarget | None	None
failover	ModelFailoverPolicy | None	Field(default=None, exclude_if=lambda value: value is None)
environment_name	str | None	None
labels	dict[str, str]	Field(default_factory=dict)
metadata	dict[str, Any]	Field(default_factory=dict)
max_steps	StrictInt	Field(default=DEFAULT_MAX_STEPS, ge=1, le=MAX_STEPS)
limits	RunLimits	Field(default_factory=RunLimits)
budget_limits	tuple[BudgetLimit, ...]	Field(default_factory=tuple)
retry_policy	RetryPolicy | None	None
thinking	ThinkingConfig | None	None
task_id	str | None	None
task_worker_id	str | None	None
task_lease_expires_at	datetime | None	None

copy_failover (method)

StepRunOptions.copy_failover(value: object) -> ModelFailoverPolicy | None

validate_optional_nonblank (method)

StepRunOptions.validate_optional_nonblank(value: str | None, info) -> str | None

normalize_task_lease_expires_at (method)

StepRunOptions.normalize_task_lease_expires_at(value: datetime | None) -> datetime | None

copy_labels (method)

StepRunOptions.copy_labels(value) -> dict[str, str]

copy_metadata (method)

StepRunOptions.copy_metadata(value: dict[str, Any]) -> dict[str, Any]

copy_limits (method)

StepRunOptions.copy_limits(value: RunLimits) -> RunLimits

copy_budget_limits (method)

StepRunOptions.copy_budget_limits(value) -> tuple[BudgetLimit, ...]

copy_retry_policy (method)

StepRunOptions.copy_retry_policy(value: RetryPolicy | None) -> RetryPolicy | None

validate_thinking (method)

StepRunOptions.validate_thinking(value: ThinkingConfig | None) -> ThinkingConfig | None

validate_task_worker_handoff (method)

StepRunOptions.validate_task_worker_handoff() -> StepRunOptions

Release source · Runtime guarantees · Capability guides

WorkflowBase

from cayu.workflows import WorkflowBase

Base class for workflows authored as async generators.

Bases: Workflow.

Fields and attributes

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name	Type	Default
spec	WorkflowSpec	required

run (method)

WorkflowBase.run(session_id: str) -> AsyncIterator[Event]

Run the workflow and return a stream of structured events.

Declared non-async on purpose: implementations are expected to be async generators (async def run(...): yield ...), which are plain callables returning an AsyncIterator. Keeping the abstract method non-async gives every implementation the same calling convention — async for event in workflow.run(session_id) — instead of some callers needing await workflow.run(...) first.

init (method)

WorkflowBase.__init__(app: CayuApp, *, spec: WorkflowSpec | None=None, journal_factory: JournalFactory | None=None) -> None

execute (method)

async WorkflowBase.execute(session_id: str, *, execution_deadline: ExecutionDeadline | None=None)

Execute with Runtime cancellation and the original durable expiry on resume.

context (method)

WorkflowBase.context(session_id: str) -> WorkflowContext

Build the per-run context (and its journal) for one workflow run.

Release source · Runtime guarantees · Capability guides

WorkflowContext

from cayu.workflows import WorkflowContext

Per-run binding for helper state, journal access, and attempt fencing.

init (method)

WorkflowContext.__init__(*, app: CayuApp, spec: WorkflowSpec, session_id: str, journal: WorkflowJournal) -> None

execution_deadline (method)

async WorkflowContext.execution_deadline() -> ExecutionDeadline

Live boundary, including the durable anchor on recovery.

remaining_seconds (method)

async WorkflowContext.remaining_seconds() -> float | None

workflow_name (property)

WorkflowContext.workflow_name() -> str

next_gated_loop_name (method)

WorkflowContext.next_gated_loop_name() -> str

Return the one safe automatic loop namespace.

Additional loops need explicit names because resume replays run from the top, and call-order-derived names can shift when control flow shifts.

event (method)

WorkflowContext.event(event_type: EventType | str, *, payload: dict[str, Any] | None=None, agent_name: str | None=None) -> Event

Build a workflow/custom event stamped with this context’s attempt id.

emit_custom_event (method)

async WorkflowContext.emit_custom_event(event_type: str, *, payload: dict[str, Any] | None=None, agent_name: str | None=None) -> Event

Journal + return one custom workflow event.

custom.cayu. is reserved so user events cannot forge or mask attempt fence markers.

start (method)

async WorkflowContext.start(payload: dict[str, Any] | None=None) -> Event

Journal + return the workflow.started event. Use yield await.

completed (method)

async WorkflowContext.completed(payload: dict[str, Any] | None=None) -> Event

Journal + return the workflow.completed event. Use yield await.

Release source · Runtime guarantees · Capability guides

WorkflowSupersededError

from cayu.workflows import WorkflowSupersededError

Raised when a newer attempt has taken over the same workflow run id.

Bases: RuntimeError.

This type’s implementation and inherited behavior are available in the linked release source.

Release source · Runtime guarantees · Capability guides