angee.workflows.engine

Runtime engine for workflow runs.

This module is the single owner of workflow advancement. It creates and replays the step-run journal, evaluates join rules, routes outcomes, claims work, and records cancellation. Step implementations run only through execute(), never inside advance().

DecisionAttemptResult

@dataclass(frozen=True, slots=True)
class DecisionAttemptResult()

Outcome of one authorized decision attempt and any shape validation failure.

start

def start(workflow: Any,
          subject: Any,
          actor: Any,
          *,
          trigger: Any = None,
          parent_step_run: Any = None,
          dedup_key: str | None = None) -> Any

Start the current published version after validating its subject declaration.

An empty subject declaration accepts any subject for backwards compatibility. A declared workflow raises ValidationError before creating a run when the subject's concrete model differs.

deliver

def deliver(run_id: int, *, now: datetime | None = None) -> dict[str, int]

Deliver an external event by waking this run's parked journal rows.

This is the workflow engine's event-delivery seam. Every delivery advances the run-scoped generation under the same short row lock as :func:advance, even when no row is waiting. A step that parks after observing an older generation is made immediately due by :func:execute; the existing advance/execute tasks still own claiming and running implementations.

advance

def advance(run_id: int, *, now: datetime | None = None) -> dict[str, int]

Advance one workflow run under a short row lock, without running impls.

execute

def execute(step_run_id: int,
            *,
            now: datetime | None = None) -> dict[str, int]

Run one claimed StepRun outside any advance lock and enqueue replay.

cancel

def cancel(run: Any) -> None

Cancel a run, its durable waits, scheduled rows, and child runs.

expire_pending_decisions

def expire_pending_decisions(run: Any, *, resolved_by: str) -> int

Expire every pending decision for run through the engine owner.

sweep

def sweep(*, now: datetime | None = None) -> dict[str, int]

Advance runs whose durable wake time is due.

reap

def reap(*, now: datetime | None = None) -> dict[str, int]

Fail started step-runs whose heartbeat is past the configured deadline.

decide

def decide(decision: Any,
           verdict: str,
           *,
           payload: Any = None,
           actor: Any = None) -> DecisionAttemptResult

Attempt one actor-authorized resolution and return its validation outcome.

escalate_decision

def escalate_decision(decision_id: int,
                      attempt: int,
                      *,
                      now: datetime | None = None) -> dict[str, int]

Resolve a pending decision as escalated when its timer is still current.

expire_decision

def expire_decision(decision_id: int,
                    attempt: int,
                    *,
                    now: datetime | None = None) -> dict[str, int]

Resolve a pending decision as expired when its timer is still current.

sweep_decisions

def sweep_decisions(*, now: datetime | None = None) -> dict[str, int]

Resolve pending decisions whose durable deadlines are due.

override_run

def override_run(run: Any, next_steps: Iterable[Any], *, actor: Any) -> Any

Cancel active rows, insert an override journal row, and schedule next steps.

enqueue_advance

def enqueue_advance(run_id: int) -> None

Enqueue an advance job.

enqueue_advance_at

def enqueue_advance_at(run_id: int, when: datetime) -> None

Enqueue a deferred advance job for a durable timer wake.

enqueue_decision_escalation_at

def enqueue_decision_escalation_at(decision_id: int, attempt: int,
                                   when: datetime) -> None

Enqueue a deferred escalation timer for one decision attempt.

enqueue_decision_expiry_at

def enqueue_decision_expiry_at(decision_id: int, attempt: int,
                               when: datetime) -> None

Enqueue a deferred expiry timer for one decision attempt.

enqueue_execute

def enqueue_execute(step_run_id: int) -> None

Enqueue one step execution job.