cayu.workflows.journal

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.

canonical_workflow_step_completion_ids

from cayu.workflows import canonical_workflow_step_completion_ids

Build compact replay evidence from attempt-fenced completed events.

This is the public canonicalization boundary for custom workflow journals. The journal owns attempt-fenced selection; workflow core independently validates the returned snapshot shape before trusting its identities.

canonical_workflow_step_completion_ids(events: Iterable[Event], *, session_id: str, workflow_name: str, attempt_id: str) -> WorkflowStepCompletionSnapshot

Release source · Runtime guarantees · Capability guides

copy_workflow_step_completion_snapshot

from cayu.workflows import copy_workflow_step_completion_snapshot

copy_workflow_step_completion_snapshot(snapshot: WorkflowStepCompletionSnapshot, *, session_id: str, workflow_name: str, attempt_id: str) -> WorkflowStepCompletionSnapshot

Release source · Runtime guarantees · Capability guides

EventStoreJournal

from cayu.workflows import EventStoreJournal

Default WorkflowJournal backed by cayu’s event store.

init (method)

EventStoreJournal.__init__(session_store: SessionStore, session_id: str, workflow_name: str, *, event_emitter: EventEmitter | None=None, step_event_reserver: StepEventReserver | None=None) -> None

session_id (property)

EventStoreJournal.session_id() -> str

append (method)

async EventStoreJournal.append(event: Event) -> None

append_current_attempt (method)

async EventStoreJournal.append_current_attempt(event: Event, *, attempt_id: str) -> bool

append_step_started (method)

async EventStoreJournal.append_step_started(event: Event, *, attempt_id: str) -> bool

completed_step_ids (method)

async EventStoreJournal.completed_step_ids(*, attempt_id: str) -> set[str]

completed_step_snapshot (method)

async EventStoreJournal.completed_step_snapshot(*, attempt_id: str) -> WorkflowStepCompletionSnapshot

latest_step_child_session_id (method)

async EventStoreJournal.latest_step_child_session_id(*, step_id: str, event_type: EventType) -> str | None

step_replay_ids (method)

async EventStoreJournal.step_replay_ids(*, step_id: str, attempt_id: str) -> tuple[str | None, str | None]

latest_attempt_id (method)

async EventStoreJournal.latest_attempt_id() -> str | None

Release source · Runtime guarantees · Capability guides

WORKFLOW_JOURNAL_MODEL

from cayu.workflows import WORKFLOW_JOURNAL_MODEL

WORKFLOW_JOURNAL_MODEL = 'cayu.workflow'

Release source · Runtime guarantees · Capability guides

WORKFLOW_JOURNAL_PROVIDER

from cayu.workflows import WORKFLOW_JOURNAL_PROVIDER

WORKFLOW_JOURNAL_PROVIDER = 'cayu.workflow'

Release source · Runtime guarantees · Capability guides

WorkflowJournal

from cayu.workflows import WorkflowJournal

Durable record of one workflow run’s progress.

Decorators: @runtime_checkable.

Bases: Protocol.

append (method)

async WorkflowJournal.append(event: Event) -> None

Durably record one workflow/custom event for this run.

append_current_attempt (method)

async WorkflowJournal.append_current_attempt(event: Event, *, attempt_id: str) -> bool

Append only when attempt_id is still the latest attempt.

append_step_started (method)

async WorkflowJournal.append_step_started(event: Event, *, attempt_id: str) -> bool

Reserve one workflow.step.started event before child execution.

completed_step_ids (method)

async WorkflowJournal.completed_step_ids(*, attempt_id: str) -> set[str]

Legacy ID-only inspection of completed steps.

gated_loop authorizes replay from a completion snapshot instead; an ID-only set cannot prove whether a v2-shaped ID is modern evidence or a byte-identical legacy collision. Existing implementations may retain their former raw-ID projection for compatibility.

step_replay_ids (method)

async WorkflowJournal.step_replay_ids(*, step_id: str, attempt_id: str) -> tuple[str | None, str | None]

Latest (completed, started) child session ids for a step.

latest_attempt_id (method)

async WorkflowJournal.latest_attempt_id() -> str | None

Newest journaled workflow-attempt id, if any.

Release source · Runtime guarantees · Capability guides

WorkflowJournalContext

from cayu.workflows import WorkflowJournalContext

Factory input for journals bound to one workflow run.

Decorators: @dataclass(frozen=True, slots=True).

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
session_store	SessionStore	required
session_id	str	required
workflow_name	str	required
emit_events	EventEmitter	required
reserve_step_started	StepEventReserver | None	None

Release source · Runtime guarantees · Capability guides

WorkflowJournalReplayEvidence

from cayu.workflows import WorkflowJournalReplayEvidence

Optional journal capability for evidence-safe gated-loop replay.

Decorators: @runtime_checkable.

Bases: Protocol.

completed_step_snapshot (method)

async WorkflowJournalReplayEvidence.completed_step_snapshot(*, attempt_id: str) -> WorkflowStepCompletionSnapshot

Helper-built identities from attempt-fenced durable completions.

Release source · Runtime guarantees · Capability guides

WorkflowStepCompletionSnapshot

from cayu.workflows import WorkflowStepCompletionSnapshot

Opaque, transient completion identities derived from durable evidence.

Instances cannot be constructed directly. Custom journals create them through canonical_workflow_step_completion_ids so their scope and identities remain bound to the evidence that authorized them.

step_ids (property)

WorkflowStepCompletionSnapshot.step_ids() -> frozenset[str]

Release source · Runtime guarantees · Capability guides