angee.workflows.tasks

Celery task wrappers for the workflow engine.

advance_workflow_run

@shared_task(
    bind=True,
    name="workflows.advance",
    autoretry_for=(Exception, ),
    retry_backoff=15,
    retry_kwargs={"max_retries": 5},
)
def advance_workflow_run(self: Any, run_id: int) -> None

Run one short orchestration pass for a workflow run.

execute_workflow_step

@shared_task(bind=True, name="workflows.execute")
def execute_workflow_step(self: Any, step_run_id: int) -> None

Execute one claimed step-run outside the advance lock.

escalate_workflow_decision

@shared_task(
    bind=True,
    name="workflows.decision_escalate",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def escalate_workflow_decision(self: Any, decision_id: int,
                               attempt: int) -> None

Resolve a decision escalation timer if it still matches the attempt.

expire_workflow_decision

@shared_task(
    bind=True,
    name="workflows.decision_expire",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def expire_workflow_decision(self: Any, decision_id: int,
                             attempt: int) -> None

Resolve a decision expiry timer if it still matches the attempt.

sweep_workflow_decisions

@shared_task(
    bind=True,
    name="workflows.decisions",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def sweep_workflow_decisions(self: Any, timestamp: int | None = None) -> None

Resolve workflow decisions whose durable deadlines are due.

sweep_workflow_runs

@shared_task(
    bind=True,
    name="workflows.sweep",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def sweep_workflow_runs(self: Any, timestamp: int | None = None) -> None

Advance workflow runs whose durable wake time is due.

reap_workflow_step_runs

@shared_task(
    bind=True,
    name="workflows.reap",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def reap_workflow_step_runs(self: Any, timestamp: int | None = None) -> None

Fail started step-runs whose heartbeat has expired.

run_workflow_schedule_triggers

@shared_task(
    bind=True,
    name="workflows.schedule_triggers",
    autoretry_for=(Exception, ),
    retry_backoff=30,
    retry_kwargs={"max_retries": 3},
)
def run_workflow_schedule_triggers(self: Any,
                                   timestamp: int | None = None) -> None

Start schedule triggers due at the injected periodic timestamp.