Ir para o conteúdo

API reference

Generated from the apogee-ai-workflow source with mkdocstrings. Every symbol below is exported from apogee_ai_workflow, so it is part of the supported public surface.

Application · DTOs

CancelRunDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

run_id instance-attribute

Python
run_id: str

GateDecisionDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

gate_id instance-attribute

Python
gate_id: str

decided_by class-attribute instance-attribute

Python
decided_by: str | None = None

reason class-attribute instance-attribute

Python
reason: str | None = None

ResumeRunDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

run_id instance-attribute

Python
run_id: str

StartRunDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

workflow_name instance-attribute

Python
workflow_name: str

input class-attribute instance-attribute

Python
input: dict[str, Any] = Field(default_factory=dict)

tenant_id class-attribute instance-attribute

Python
tenant_id: str | None = None

user_id class-attribute instance-attribute

Python
user_id: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = Field(default_factory=dict)

Application · Use cases

ApproveGateUseCase

Python
ApproveGateUseCase(gates: IHumanGateGateway)
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
def __init__(self, gates: IHumanGateGateway) -> None:
    self._gates = gates

execute async

Python
execute(dto: GateDecisionDTO) -> HumanGate
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
async def execute(self, dto: GateDecisionDTO) -> HumanGate:
    gate = await self._gates.get(dto.gate_id)
    gate.approve(by=dto.decided_by, reason=dto.reason)
    return await self._gates.update(gate)

CancelRunUseCase

Python
CancelRunUseCase(runs: IRunRepository, engine: IWorkflowEngine)
Source code in apogee_ai_workflow/application/use_cases/cancel_run_use_case.py
Python
def __init__(self, runs: IRunRepository, engine: IWorkflowEngine) -> None:
    self._runs = runs
    self._engine = engine

execute async

Python
execute(run_id: str) -> bool
Source code in apogee_ai_workflow/application/use_cases/cancel_run_use_case.py
Python
async def execute(self, run_id: str) -> bool:
    run = await self._runs.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    ack = await self._engine.cancel(run_id)
    if run.is_terminal:
        return ack
    run.status = RunStatus.CANCELLED
    run.finished_at = datetime.now(timezone.utc)
    await self._runs.save(run)
    return ack

GetRunUseCase

Python
GetRunUseCase(runs: IRunRepository)
Source code in apogee_ai_workflow/application/use_cases/get_run_use_case.py
Python
def __init__(self, runs: IRunRepository) -> None:
    self._runs = runs

execute async

Python
execute(run_id: str) -> WorkflowRun
Source code in apogee_ai_workflow/application/use_cases/get_run_use_case.py
Python
async def execute(self, run_id: str) -> WorkflowRun:
    run = await self._runs.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    return run

ListGatesUseCase

Python
ListGatesUseCase(gates: IHumanGateGateway)
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
def __init__(self, gates: IHumanGateGateway) -> None:
    self._gates = gates

execute async

Python
execute(*, run_id: str | None = None, pending_only: bool = False) -> list[HumanGate]
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
async def execute(
    self,
    *,
    run_id: str | None = None,
    pending_only: bool = False,
) -> list[HumanGate]:
    return await self._gates.list(run_id=run_id, pending_only=pending_only)

ListRunsUseCase

Python
ListRunsUseCase(runs: IRunRepository)
Source code in apogee_ai_workflow/application/use_cases/list_runs_use_case.py
Python
def __init__(self, runs: IRunRepository) -> None:
    self._runs = runs

execute async

Python
execute(*, workflow_name: str | None = None, limit: int | None = None) -> list[WorkflowRun]
Source code in apogee_ai_workflow/application/use_cases/list_runs_use_case.py
Python
async def execute(
    self,
    *,
    workflow_name: str | None = None,
    limit: int | None = None,
) -> list[WorkflowRun]:
    return await self._runs.list(workflow_name=workflow_name, limit=limit)

RegisterWorkflowUseCase

Python
RegisterWorkflowUseCase(repository: IWorkflowRepository)
Source code in apogee_ai_workflow/application/use_cases/register_workflow_use_case.py
Python
def __init__(self, repository: IWorkflowRepository) -> None:
    self._repository = repository

execute async

Python
execute(definition: WorkflowDefinition, *, version: int = 1, replace: bool = False) -> Workflow
Source code in apogee_ai_workflow/application/use_cases/register_workflow_use_case.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    *,
    version: int = 1,
    replace: bool = False,
) -> Workflow:
    if not replace and await self._repository.exists(definition.name):
        raise WorkflowAlreadyRegistered(definition.name)
    workflow = Workflow(definition=definition, version=version)
    return await self._repository.save(workflow)

RejectGateUseCase

Python
RejectGateUseCase(gates: IHumanGateGateway)
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
def __init__(self, gates: IHumanGateGateway) -> None:
    self._gates = gates

execute async

Python
execute(dto: GateDecisionDTO) -> HumanGate
Source code in apogee_ai_workflow/application/use_cases/approve_gate_use_case.py
Python
async def execute(self, dto: GateDecisionDTO) -> HumanGate:
    gate = await self._gates.get(dto.gate_id)
    gate.reject(by=dto.decided_by, reason=dto.reason)
    return await self._gates.update(gate)

ResumeRunUseCase

Python
ResumeRunUseCase(workflows: IWorkflowRepository, runs: IRunRepository, engine: IWorkflowEngine)
Source code in apogee_ai_workflow/application/use_cases/resume_run_use_case.py
Python
def __init__(
    self,
    workflows: IWorkflowRepository,
    runs: IRunRepository,
    engine: IWorkflowEngine,
) -> None:
    self._workflows = workflows
    self._runs = runs
    self._engine = engine

execute async

Python
execute(run_id: str) -> WorkflowRun
Source code in apogee_ai_workflow/application/use_cases/resume_run_use_case.py
Python
async def execute(self, run_id: str) -> WorkflowRun:
    run = await self._runs.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    workflow = await self._workflows.find(run.workflow_name)
    if workflow is None:
        raise WorkflowNotFoundException(run.workflow_name)
    run = await self._engine.resume(workflow.definition, run)
    return await self._runs.save(run)

StartRunUseCase

Python
StartRunUseCase(workflows: IWorkflowRepository, runs: IRunRepository, engine: IWorkflowEngine)
Source code in apogee_ai_workflow/application/use_cases/start_run_use_case.py
Python
def __init__(
    self,
    workflows: IWorkflowRepository,
    runs: IRunRepository,
    engine: IWorkflowEngine,
) -> None:
    self._workflows = workflows
    self._runs = runs
    self._engine = engine

execute async

Python
execute(dto: StartRunDTO) -> WorkflowRun
Source code in apogee_ai_workflow/application/use_cases/start_run_use_case.py
Python
async def execute(self, dto: StartRunDTO) -> WorkflowRun:
    workflow = await self._workflows.find(dto.workflow_name)
    if workflow is None:
        raise WorkflowNotFoundException(dto.workflow_name)
    run = WorkflowRun(
        workflow_name=workflow.name,
        workflow_version=workflow.version,
        input=dict(dto.input),
        tenant_id=dto.tenant_id,
        user_id=dto.user_id,
        metadata=dict(dto.metadata),
    )
    await self._runs.save(run)
    run = await self._engine.execute(workflow.definition, run)
    return await self._runs.save(run)

Domain

Checkpoint dataclass

Python
Checkpoint(run_id: str, step_name: str, sequence: int, state: dict[str, Any], output: dict[str, Any] = dict(), created_at: datetime = (lambda: now(utc))())

Snapshot persisted after a step completes.

run_id instance-attribute

Python
run_id: str

step_name instance-attribute

Python
step_name: str

sequence instance-attribute

Python
sequence: int

state instance-attribute

Python
state: dict[str, Any]

output class-attribute instance-attribute

Python
output: dict[str, Any] = field(default_factory=dict)

created_at class-attribute instance-attribute

Python
created_at: datetime = field(default_factory=lambda: now(utc))

CompensationFailed

Python
CompensationFailed(step: str, message: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, step: str, message: str) -> None:
    super().__init__(f"Compensation for step {step!r} failed: {message}")
    self.step = step

step instance-attribute

Python
step = step

EngineConfig dataclass

Python
EngineConfig(max_concurrency: int = 8, default_step_timeout_seconds: float = 300.0, checkpoint_after_each_step: bool = True, persist_step_outputs: bool = True)

Tunables for the workflow engine.

max_concurrency class-attribute instance-attribute

Python
max_concurrency: int = 8

Max parallel step executions per run.

default_step_timeout_seconds class-attribute instance-attribute

Python
default_step_timeout_seconds: float = 300.0

checkpoint_after_each_step class-attribute instance-attribute

Python
checkpoint_after_each_step: bool = True

persist_step_outputs class-attribute instance-attribute

Python
persist_step_outputs: bool = True

EngineUnavailable

Python
EngineUnavailable(engine: str, reason: str = '')

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, engine: str, reason: str = "") -> None:
    super().__init__(f"Engine {engine!r} unavailable: {reason}".rstrip(": "))
    self.engine = engine

engine instance-attribute

Python
engine = engine

GateDecision

Bases: str, Enum

PENDING class-attribute instance-attribute

Python
PENDING = 'pending'

APPROVED class-attribute instance-attribute

Python
APPROVED = 'approved'

REJECTED class-attribute instance-attribute

Python
REJECTED = 'rejected'

EXPIRED class-attribute instance-attribute

Python
EXPIRED = 'expired'

GateNotFound

Python
GateNotFound(gate_id: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, gate_id: str) -> None:
    super().__init__(f"Gate {gate_id!r} not found")
    self.gate_id = gate_id

gate_id instance-attribute

Python
gate_id = gate_id

GateRejected

Python
GateRejected(gate_id: str, reason: str | None = None)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, gate_id: str, reason: str | None = None) -> None:
    super().__init__(
        f"Gate {gate_id!r} was rejected{f': {reason}' if reason else ''}"
    )
    self.gate_id = gate_id
    self.reason = reason

gate_id instance-attribute

Python
gate_id = gate_id

reason instance-attribute

Python
reason = reason

HumanGate dataclass

Python
HumanGate(id: str = (lambda: token_hex(8))(), run_id: str = '', step_name: str = '', label: str = '', decision: GateDecision = PENDING, reason: str | None = None, decided_at: datetime | None = None, decided_by: str | None = None, created_at: datetime = (lambda: now(utc))(), expires_at: datetime | None = None, metadata: dict[str, str] = dict())

Pause point that blocks the run until a decision is recorded.

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: token_hex(8))

run_id class-attribute instance-attribute

Python
run_id: str = ''

step_name class-attribute instance-attribute

Python
step_name: str = ''

label class-attribute instance-attribute

Python
label: str = ''

decision class-attribute instance-attribute

Python
decision: GateDecision = PENDING

reason class-attribute instance-attribute

Python
reason: str | None = None

decided_at class-attribute instance-attribute

Python
decided_at: datetime | None = None

decided_by class-attribute instance-attribute

Python
decided_by: str | None = None

created_at class-attribute instance-attribute

Python
created_at: datetime = field(default_factory=lambda: now(utc))

expires_at class-attribute instance-attribute

Python
expires_at: datetime | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

is_pending property

Python
is_pending: bool

approve

Python
approve(*, by: str | None = None, reason: str | None = None) -> None
Source code in apogee_ai_workflow/domain/entities/human_gate.py
Python
def approve(self, *, by: str | None = None, reason: str | None = None) -> None:
    self.decision = GateDecision.APPROVED
    self.decided_at = datetime.now(timezone.utc)
    self.decided_by = by
    self.reason = reason

reject

Python
reject(*, by: str | None = None, reason: str | None = None) -> None
Source code in apogee_ai_workflow/domain/entities/human_gate.py
Python
def reject(self, *, by: str | None = None, reason: str | None = None) -> None:
    self.decision = GateDecision.REJECTED
    self.decided_at = datetime.now(timezone.utc)
    self.decided_by = by
    self.reason = reason

RetryPolicy dataclass

Python
RetryPolicy(kind: RetryPolicyKind = NONE, max_attempts: int = 1, backoff_seconds: float = 1.0, max_backoff_seconds: float = 60.0, jitter: float = 0.1)

kind class-attribute instance-attribute

Python
kind: RetryPolicyKind = NONE

max_attempts class-attribute instance-attribute

Python
max_attempts: int = 1

backoff_seconds class-attribute instance-attribute

Python
backoff_seconds: float = 1.0

max_backoff_seconds class-attribute instance-attribute

Python
max_backoff_seconds: float = 60.0

jitter class-attribute instance-attribute

Python
jitter: float = 0.1

delay_for

Python
delay_for(attempt: int) -> float
Source code in apogee_ai_workflow/domain/value_objects/retry_policy.py
Python
def delay_for(self, attempt: int) -> float:
    if self.kind == RetryPolicyKind.NONE or attempt <= 0:
        return 0.0
    if self.kind == RetryPolicyKind.FIXED:
        return min(self.backoff_seconds, self.max_backoff_seconds)
    # EXPONENTIAL: backoff * 2^(attempt-1)
    delay = self.backoff_seconds * (2 ** (attempt - 1))
    return min(delay, self.max_backoff_seconds)

RunContext dataclass

Python
RunContext(run_id: str, workflow_name: str, tenant_id: str | None = None, user_id: str | None = None, correlation_id: str | None = None, metadata: dict[str, str] = dict())

Per-run identity + immutable metadata threaded through every step.

run_id instance-attribute

Python
run_id: str

workflow_name instance-attribute

Python
workflow_name: str

tenant_id class-attribute instance-attribute

Python
tenant_id: str | None = None

user_id class-attribute instance-attribute

Python
user_id: str | None = None

correlation_id class-attribute instance-attribute

Python
correlation_id: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

RunStatus

Bases: str, Enum

PENDING class-attribute instance-attribute

Python
PENDING = 'pending'

RUNNING class-attribute instance-attribute

Python
RUNNING = 'running'

WAITING_FOR_GATE class-attribute instance-attribute

Python
WAITING_FOR_GATE = 'waiting_for_gate'

SUCCEEDED class-attribute instance-attribute

Python
SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

Python
FAILED = 'failed'

COMPENSATED class-attribute instance-attribute

Python
COMPENSATED = 'compensated'

CANCELLED class-attribute instance-attribute

Python
CANCELLED = 'cancelled'

SagaCompensation dataclass

Python
SagaCompensation(step_name: str, succeeded: bool, error: str | None = None, started_at: datetime = (lambda: now(utc))(), finished_at: datetime | None = None)

Record of a compensating action invoked during rollback.

step_name instance-attribute

Python
step_name: str

succeeded instance-attribute

Python
succeeded: bool

error class-attribute instance-attribute

Python
error: str | None = None

started_at class-attribute instance-attribute

Python
started_at: datetime = field(default_factory=lambda: now(utc))

finished_at class-attribute instance-attribute

Python
finished_at: datetime | None = None

Step dataclass

Python
Step(name: str, handler: StepHandler | None = None, kind: StepKind = TASK, depends_on: tuple[str, ...] = tuple(), retry: RetryPolicy = RetryPolicy(), timeout_seconds: float | None = None, compensation: StepHandler | None = None, gate_label: str | None = None, description: str | None = None, metadata: dict[str, str] = dict())

Static definition of a single step within a Workflow.

name instance-attribute

Python
name: str

handler class-attribute instance-attribute

Python
handler: StepHandler | None = None

kind class-attribute instance-attribute

Python
kind: StepKind = TASK

depends_on class-attribute instance-attribute

Python
depends_on: tuple[str, ...] = field(default_factory=tuple)

retry class-attribute instance-attribute

Python
retry: RetryPolicy = field(default_factory=RetryPolicy)

timeout_seconds class-attribute instance-attribute

Python
timeout_seconds: float | None = None

compensation class-attribute instance-attribute

Python
compensation: StepHandler | None = None

Optional compensating action invoked when the run is rolled back.

gate_label class-attribute instance-attribute

Python
gate_label: str | None = None

Human-readable label shown to approvers when kind == HUMAN_GATE.

description class-attribute instance-attribute

Python
description: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

StepHandler module-attribute

Python
StepHandler = Callable[..., Awaitable[Any] | Any]

Signature: async def handler(io: StepIO) -> dict | Any.

StepIO dataclass

Python
StepIO(step_name: str, input: dict[str, Any], state: dict[str, Any], context: RunContext, attempt: int = 1)

Inputs + accumulated outputs visible to a single step handler.

input is the merge of upstream step outputs (by name) plus the initial run input. state is the read-write store the engine persists across checkpoints (handlers may add keys to it).

step_name instance-attribute

Python
step_name: str

input instance-attribute

Python
input: dict[str, Any]

state instance-attribute

Python
state: dict[str, Any]

context instance-attribute

Python
context: RunContext

attempt class-attribute instance-attribute

Python
attempt: int = 1

StepRun dataclass

Python
StepRun(step_name: str, status: StepStatus = PENDING, attempt: int = 0, started_at: datetime | None = None, finished_at: datetime | None = None, output: dict[str, Any] = dict(), error: str | None = None, gate_id: str | None = None, metadata: dict[str, str] = dict())

step_name instance-attribute

Python
step_name: str

status class-attribute instance-attribute

Python
status: StepStatus = PENDING

attempt class-attribute instance-attribute

Python
attempt: int = 0

started_at class-attribute instance-attribute

Python
started_at: datetime | None = None

finished_at class-attribute instance-attribute

Python
finished_at: datetime | None = None

output class-attribute instance-attribute

Python
output: dict[str, Any] = field(default_factory=dict)

error class-attribute instance-attribute

Python
error: str | None = None

gate_id class-attribute instance-attribute

Python
gate_id: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

begin

Python
begin() -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def begin(self) -> None:
    self.status = StepStatus.RUNNING
    self.started_at = datetime.now(timezone.utc)
    self.attempt += 1

succeed

Python
succeed(output: dict[str, Any] | None = None) -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def succeed(self, output: dict[str, Any] | None = None) -> None:
    self.status = StepStatus.SUCCEEDED
    self.finished_at = datetime.now(timezone.utc)
    if output is not None:
        self.output = dict(output)

fail

Python
fail(error: str) -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def fail(self, error: str) -> None:
    self.status = StepStatus.FAILED
    self.finished_at = datetime.now(timezone.utc)
    self.error = error

wait_gate

Python
wait_gate(gate_id: str) -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def wait_gate(self, gate_id: str) -> None:
    self.status = StepStatus.WAITING_FOR_GATE
    self.gate_id = gate_id

skip

Python
skip() -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def skip(self) -> None:
    self.status = StepStatus.SKIPPED
    self.finished_at = datetime.now(timezone.utc)

compensate

Python
compensate() -> None
Source code in apogee_ai_workflow/domain/entities/step_run.py
Python
def compensate(self) -> None:
    self.status = StepStatus.COMPENSATED
    self.finished_at = datetime.now(timezone.utc)

StepStatus

Bases: str, Enum

PENDING class-attribute instance-attribute

Python
PENDING = 'pending'

RUNNING class-attribute instance-attribute

Python
RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute

Python
SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

Python
FAILED = 'failed'

SKIPPED class-attribute instance-attribute

Python
SKIPPED = 'skipped'

COMPENSATED class-attribute instance-attribute

Python
COMPENSATED = 'compensated'

WAITING_FOR_GATE class-attribute instance-attribute

Python
WAITING_FOR_GATE = 'waiting_for_gate'

Workflow dataclass

Python
Workflow(definition: WorkflowDefinition, version: int = 1, registered_at: datetime = (lambda: now(utc))(), metadata: dict[str, str] = dict())

Persisted record of a registered workflow.

definition instance-attribute

Python
definition: WorkflowDefinition

version class-attribute instance-attribute

Python
version: int = 1

registered_at class-attribute instance-attribute

Python
registered_at: datetime = field(default_factory=lambda: now(utc))

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

name property

Python
name: str

WorkflowAlreadyRegistered

Python
WorkflowAlreadyRegistered(name: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, name: str) -> None:
    super().__init__(f"Workflow {name!r} already registered")
    self.name = name

name instance-attribute

Python
name = name

WorkflowDefinition dataclass

Python
WorkflowDefinition(name: str, steps: tuple[Step, ...] = tuple(), description: str | None = None, tags: tuple[str, ...] = tuple())

Logical workflow shape: name + steps DAG.

Validates non-cyclic dependencies at construction time.

name instance-attribute

Python
name: str

steps class-attribute instance-attribute

Python
steps: tuple[Step, ...] = field(default_factory=tuple)

description class-attribute instance-attribute

Python
description: str | None = None

tags class-attribute instance-attribute

Python
tags: tuple[str, ...] = field(default_factory=tuple)

step

Python
step(name: str) -> Step
Source code in apogee_ai_workflow/domain/entities/workflow_definition.py
Python
def step(self, name: str) -> Step:
    for step in self.steps:
        if step.name == name:
            return step
    raise KeyError(f"Step {name!r} not in workflow {self.name!r}")

topological_order

Python
topological_order() -> list[Step]

Return steps in a valid execution order (deps before dependents).

Source code in apogee_ai_workflow/domain/entities/workflow_definition.py
Python
def topological_order(self) -> list[Step]:
    """Return steps in a valid execution order (deps before dependents)."""
    order: list[Step] = []
    visited: set[str] = set()
    adjacency = {s.name: s for s in self.steps}

    def visit(name: str) -> None:
        if name in visited:
            return
        visited.add(name)
        step = adjacency[name]
        for dep in step.depends_on:
            visit(dep)
        order.append(step)

    for s in self.steps:
        visit(s.name)
    return order

WorkflowRun dataclass

Python
WorkflowRun(id: str = (lambda: token_hex(8))(), workflow_name: str = '', workflow_version: int = 1, status: RunStatus = PENDING, input: dict[str, Any] = dict(), state: dict[str, Any] = dict(), started_at: datetime = (lambda: now(utc))(), finished_at: datetime | None = None, steps: dict[str, StepRun] = dict(), error: str | None = None, pending_gate_id: str | None = None, tenant_id: str | None = None, user_id: str | None = None, metadata: dict[str, str] = dict())

Mutable record of a single workflow execution.

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: token_hex(8))

workflow_name class-attribute instance-attribute

Python
workflow_name: str = ''

workflow_version class-attribute instance-attribute

Python
workflow_version: int = 1

status class-attribute instance-attribute

Python
status: RunStatus = PENDING

input class-attribute instance-attribute

Python
input: dict[str, Any] = field(default_factory=dict)

state class-attribute instance-attribute

Python
state: dict[str, Any] = field(default_factory=dict)

started_at class-attribute instance-attribute

Python
started_at: datetime = field(default_factory=lambda: now(utc))

finished_at class-attribute instance-attribute

Python
finished_at: datetime | None = None

steps class-attribute instance-attribute

Python
steps: dict[str, StepRun] = field(default_factory=dict)

error class-attribute instance-attribute

Python
error: str | None = None

pending_gate_id class-attribute instance-attribute

Python
pending_gate_id: str | None = None

tenant_id class-attribute instance-attribute

Python
tenant_id: str | None = None

user_id class-attribute instance-attribute

Python
user_id: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

succeeded_steps property

Python
succeeded_steps: tuple[str, ...]

is_terminal property

Python
is_terminal: bool

step_run

Python
step_run(name: str) -> StepRun
Source code in apogee_ai_workflow/domain/entities/workflow_run.py
Python
def step_run(self, name: str) -> StepRun:
    if name not in self.steps:
        self.steps[name] = StepRun(step_name=name)
    return self.steps[name]

Domain · Enums

RetryPolicyKind

Bases: str, Enum

NONE class-attribute instance-attribute

Python
NONE = 'none'

FIXED class-attribute instance-attribute

Python
FIXED = 'fixed'

EXPONENTIAL class-attribute instance-attribute

Python
EXPONENTIAL = 'exponential'

StepKind

Bases: str, Enum

TASK class-attribute instance-attribute

Python
TASK = 'task'

BRANCH class-attribute instance-attribute

Python
BRANCH = 'branch'

HUMAN_GATE class-attribute instance-attribute

Python
HUMAN_GATE = 'human_gate'

PARALLEL class-attribute instance-attribute

Python
PARALLEL = 'parallel'

COMPENSATION class-attribute instance-attribute

Python
COMPENSATION = 'compensation'

Domain · Exceptions

RunNotFoundException

Python
RunNotFoundException(run_id: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, run_id: str) -> None:
    super().__init__(f"Run {run_id!r} not found")
    self.run_id = run_id

run_id instance-attribute

Python
run_id = run_id

StepFailedException

Python
StepFailedException(step: str, attempts: int, message: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, step: str, attempts: int, message: str) -> None:
    super().__init__(f"Step {step!r} failed after {attempts} attempt(s): {message}")
    self.step = step
    self.attempts = attempts

step instance-attribute

Python
step = step

attempts instance-attribute

Python
attempts = attempts

WorkflowError

Bases: Exception

Base for apogee-ai-workflow errors.

WorkflowNotFoundException

Python
WorkflowNotFoundException(name: str)

Bases: WorkflowError

Source code in apogee_ai_workflow/domain/exceptions/workflow_exceptions.py
Python
def __init__(self, name: str) -> None:
    super().__init__(f"Workflow {name!r} not found")
    self.name = name

name instance-attribute

Python
name = name

Domain · Protocols (ports)

ICheckpointer

Bases: Protocol

name instance-attribute

Python
name: str

save async

Python
save(checkpoint: Checkpoint) -> None
Source code in apogee_ai_workflow/domain/services/i_checkpointer.py
Python
async def save(self, checkpoint: Checkpoint) -> None:
    ...

list async

Python
list(run_id: str) -> list[Checkpoint]
Source code in apogee_ai_workflow/domain/services/i_checkpointer.py
Python
async def list(self, run_id: str) -> list[Checkpoint]:
    ...

latest async

Python
latest(run_id: str) -> Checkpoint | None
Source code in apogee_ai_workflow/domain/services/i_checkpointer.py
Python
async def latest(self, run_id: str) -> Checkpoint | None:
    ...

clear async

Python
clear(run_id: str) -> None
Source code in apogee_ai_workflow/domain/services/i_checkpointer.py
Python
async def clear(self, run_id: str) -> None:
    ...

IHumanGateGateway

Bases: Protocol

name instance-attribute

Python
name: str

open async

Python
open(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/domain/services/i_human_gate_gateway.py
Python
async def open(self, gate: HumanGate) -> HumanGate:
    ...

get async

Python
get(gate_id: str) -> HumanGate
Source code in apogee_ai_workflow/domain/services/i_human_gate_gateway.py
Python
async def get(self, gate_id: str) -> HumanGate:
    ...

find async

Python
find(gate_id: str) -> HumanGate | None
Source code in apogee_ai_workflow/domain/services/i_human_gate_gateway.py
Python
async def find(self, gate_id: str) -> HumanGate | None:
    ...

update async

Python
update(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/domain/services/i_human_gate_gateway.py
Python
async def update(self, gate: HumanGate) -> HumanGate:
    ...

list async

Python
list(*, run_id: str | None = None, pending_only: bool = False) -> list[HumanGate]
Source code in apogee_ai_workflow/domain/services/i_human_gate_gateway.py
Python
async def list(
    self,
    *,
    run_id: str | None = None,
    pending_only: bool = False,
) -> list[HumanGate]:
    ...

IRunRepository

Bases: Protocol

save async

Python
save(run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/domain/repositories/i_run_repository.py
Python
async def save(self, run: WorkflowRun) -> WorkflowRun:
    ...

get async

Python
get(run_id: str) -> WorkflowRun
Source code in apogee_ai_workflow/domain/repositories/i_run_repository.py
Python
async def get(self, run_id: str) -> WorkflowRun:
    ...

find async

Python
find(run_id: str) -> WorkflowRun | None
Source code in apogee_ai_workflow/domain/repositories/i_run_repository.py
Python
async def find(self, run_id: str) -> WorkflowRun | None:
    ...

list async

Python
list(*, workflow_name: str | None = None, limit: int | None = None) -> list[WorkflowRun]
Source code in apogee_ai_workflow/domain/repositories/i_run_repository.py
Python
async def list(
    self,
    *,
    workflow_name: str | None = None,
    limit: int | None = None,
) -> list[WorkflowRun]:
    ...

ISagaCompensator

Bases: Protocol

name instance-attribute

Python
name: str

compensate async

Python
compensate(definition: WorkflowDefinition, run: WorkflowRun) -> list[SagaCompensation]
Source code in apogee_ai_workflow/domain/services/i_saga_compensator.py
Python
async def compensate(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> list[SagaCompensation]:
    ...

IWorkflowEngine

Bases: Protocol

name instance-attribute

Python
name: str

supports_human_gates property

Python
supports_human_gates: bool

execute async

Python
execute(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/domain/services/i_workflow_engine.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    ...

resume async

Python
resume(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/domain/services/i_workflow_engine.py
Python
async def resume(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    ...

cancel async

Python
cancel(run_id: str) -> bool
Source code in apogee_ai_workflow/domain/services/i_workflow_engine.py
Python
async def cancel(self, run_id: str) -> bool:
    ...

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_workflow/domain/services/i_workflow_engine.py
Python
async def shutdown(self) -> None:
    ...

healthcheck async

Python
healthcheck() -> dict[str, Any]
Source code in apogee_ai_workflow/domain/services/i_workflow_engine.py
Python
async def healthcheck(self) -> dict[str, Any]:
    ...

IWorkflowRepository

Bases: Protocol

save async

Python
save(workflow: Workflow) -> Workflow
Source code in apogee_ai_workflow/domain/repositories/i_workflow_repository.py
Python
async def save(self, workflow: Workflow) -> Workflow:
    ...

get async

Python
get(name: str) -> Workflow
Source code in apogee_ai_workflow/domain/repositories/i_workflow_repository.py
Python
async def get(self, name: str) -> Workflow:
    ...

find async

Python
find(name: str) -> Workflow | None
Source code in apogee_ai_workflow/domain/repositories/i_workflow_repository.py
Python
async def find(self, name: str) -> Workflow | None:
    ...

list async

Python
list() -> list[Workflow]
Source code in apogee_ai_workflow/domain/repositories/i_workflow_repository.py
Python
async def list(self) -> list[Workflow]:
    ...

exists async

Python
exists(name: str) -> bool
Source code in apogee_ai_workflow/domain/repositories/i_workflow_repository.py
Python
async def exists(self, name: str) -> bool:
    ...

Infrastructure

DBOSAdapter

Python
DBOSAdapter(*, database_url: str | None = None)

Adapter for dbos-inc DBOS durable runtime.

Lazy import: install via pip install 'apogee-ai-workflow[dbos]'.

Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
def __init__(self, *, database_url: str | None = None) -> None:
    try:
        import dbos  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "DBOSAdapter requires `dbos`. "
            "Install with: pip install 'apogee-ai-workflow[dbos]'"
        ) from exc
    self._database_url = database_url

name class-attribute instance-attribute

Python
name = 'dbos'

supports_human_gates property

Python
supports_human_gates: bool

execute async

Python
execute(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    raise EngineUnavailable(
        self.name,
        "DBOSAdapter.execute expects DBOS application bootstrap; integrate via "
        "dbos.workflow decorators in your service code",
    )

resume async

Python
resume(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
async def resume(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    return run

cancel async

Python
cancel(run_id: str) -> bool
Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
async def cancel(self, run_id: str) -> bool:
    return False

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
async def shutdown(self) -> None:
    return None

healthcheck async

Python
healthcheck() -> dict[str, Any]
Source code in apogee_ai_workflow/infrastructure/engines/dbos_adapter.py
Python
async def healthcheck(self) -> dict[str, Any]:
    return {"engine": self.name, "database_url": self._database_url}

DefaultSagaCompensator

Walks succeeded steps in reverse and invokes compensation handlers.

Compensation failure does not abort the rollback — every record is captured so callers can decide what to do with the partial state.

name class-attribute instance-attribute

Python
name = 'default'

compensate async

Python
compensate(definition: WorkflowDefinition, run: WorkflowRun) -> list[SagaCompensation]
Source code in apogee_ai_workflow/infrastructure/saga/default_saga_compensator.py
Python
async def compensate(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> list[SagaCompensation]:
    order = list(definition.topological_order())
    # Reverse to process most-recent successful work first
    ordered_names = [step.name for step in order]
    succeeded = [n for n in reversed(ordered_names) if n in run.succeeded_steps]

    results: list[SagaCompensation] = []
    for name in succeeded:
        step = definition.step(name)
        if step.compensation is None:
            continue
        io = StepIO(
            step_name=step.name,
            input=dict(run.steps[name].output),
            state=run.state,
            context=RunContext(
                run_id=run.id,
                workflow_name=run.workflow_name,
                tenant_id=run.tenant_id,
                user_id=run.user_id,
            ),
            attempt=run.steps[name].attempt,
        )
        started = datetime.now(timezone.utc)
        try:
            result = step.compensation(io)
            if inspect.iscoroutine(result):
                await result
            run.steps[name].compensate()
            results.append(
                SagaCompensation(
                    step_name=step.name,
                    succeeded=True,
                    started_at=started,
                    finished_at=datetime.now(timezone.utc),
                )
            )
        except Exception as exc:  # noqa: BLE001 - compensation must not abort rollback
            results.append(
                SagaCompensation(
                    step_name=step.name,
                    succeeded=False,
                    error=str(exc),
                    started_at=started,
                    finished_at=datetime.now(timezone.utc),
                )
            )
        finally:
            # Cooperative yield so other compensations can interleave
            await asyncio.sleep(0)
    return results

has_run_succeeded staticmethod

Python
has_run_succeeded(run: WorkflowRun, step_name: str) -> bool
Source code in apogee_ai_workflow/infrastructure/saga/default_saga_compensator.py
Python
@staticmethod
def has_run_succeeded(run: WorkflowRun, step_name: str) -> bool:
    step = run.steps.get(step_name)
    return step is not None and step.status == StepStatus.SUCCEEDED

InMemoryCheckpointer

Python
InMemoryCheckpointer()
Source code in apogee_ai_workflow/infrastructure/checkpoints/in_memory_checkpointer.py
Python
def __init__(self) -> None:
    self._store: dict[str, list[Checkpoint]] = defaultdict(list)

name class-attribute instance-attribute

Python
name = 'in_memory'

save async

Python
save(checkpoint: Checkpoint) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/in_memory_checkpointer.py
Python
async def save(self, checkpoint: Checkpoint) -> None:
    self._store[checkpoint.run_id].append(deepcopy(checkpoint))

list async

Python
list(run_id: str) -> list[Checkpoint]
Source code in apogee_ai_workflow/infrastructure/checkpoints/in_memory_checkpointer.py
Python
async def list(self, run_id: str) -> list[Checkpoint]:
    return list(self._store.get(run_id, []))

latest async

Python
latest(run_id: str) -> Checkpoint | None
Source code in apogee_ai_workflow/infrastructure/checkpoints/in_memory_checkpointer.py
Python
async def latest(self, run_id: str) -> Checkpoint | None:
    items = self._store.get(run_id) or []
    return items[-1] if items else None

clear async

Python
clear(run_id: str) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/in_memory_checkpointer.py
Python
async def clear(self, run_id: str) -> None:
    self._store.pop(run_id, None)

InMemoryHumanGateGateway

Python
InMemoryHumanGateGateway()
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
def __init__(self) -> None:
    self._store: dict[str, HumanGate] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

open async

Python
open(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
async def open(self, gate: HumanGate) -> HumanGate:
    self._store[gate.id] = deepcopy(gate)
    return self._store[gate.id]

get async

Python
get(gate_id: str) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
async def get(self, gate_id: str) -> HumanGate:
    if gate_id not in self._store:
        raise GateNotFound(gate_id)
    return self._store[gate_id]

find async

Python
find(gate_id: str) -> HumanGate | None
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
async def find(self, gate_id: str) -> HumanGate | None:
    return self._store.get(gate_id)

update async

Python
update(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
async def update(self, gate: HumanGate) -> HumanGate:
    self._store[gate.id] = deepcopy(gate)
    return self._store[gate.id]

list async

Python
list(*, run_id: str | None = None, pending_only: bool = False) -> list[HumanGate]
Source code in apogee_ai_workflow/infrastructure/hil/in_memory_human_gate_gateway.py
Python
async def list(
    self,
    *,
    run_id: str | None = None,
    pending_only: bool = False,
) -> list[HumanGate]:
    items = list(self._store.values())
    if run_id is not None:
        items = [g for g in items if g.run_id == run_id]
    if pending_only:
        items = [g for g in items if g.is_pending]
    items.sort(key=lambda g: g.created_at)
    return items

InMemoryRunRepository

Python
InMemoryRunRepository()
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_run_repository.py
Python
def __init__(self) -> None:
    self._store: dict[str, WorkflowRun] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

save async

Python
save(run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_run_repository.py
Python
async def save(self, run: WorkflowRun) -> WorkflowRun:
    self._store[run.id] = deepcopy(run)
    return self._store[run.id]

get async

Python
get(run_id: str) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_run_repository.py
Python
async def get(self, run_id: str) -> WorkflowRun:
    if run_id not in self._store:
        raise RunNotFoundException(run_id)
    return self._store[run_id]

find async

Python
find(run_id: str) -> WorkflowRun | None
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_run_repository.py
Python
async def find(self, run_id: str) -> WorkflowRun | None:
    return self._store.get(run_id)

list async

Python
list(*, workflow_name: str | None = None, limit: int | None = None) -> list[WorkflowRun]
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_run_repository.py
Python
async def list(
    self,
    *,
    workflow_name: str | None = None,
    limit: int | None = None,
) -> list[WorkflowRun]:
    items = sorted(self._store.values(), key=lambda r: r.started_at, reverse=True)
    if workflow_name is not None:
        items = [r for r in items if r.workflow_name == workflow_name]
    if limit is not None:
        items = items[:limit]
    return items

InMemoryWorkflowRepository

Python
InMemoryWorkflowRepository()
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
def __init__(self) -> None:
    self._store: dict[str, Workflow] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

save async

Python
save(workflow: Workflow) -> Workflow
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
async def save(self, workflow: Workflow) -> Workflow:
    self._store[workflow.name] = deepcopy(workflow)
    return self._store[workflow.name]

get async

Python
get(name: str) -> Workflow
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
async def get(self, name: str) -> Workflow:
    if name not in self._store:
        raise WorkflowNotFoundException(name)
    return self._store[name]

find async

Python
find(name: str) -> Workflow | None
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
async def find(self, name: str) -> Workflow | None:
    return self._store.get(name)

list async

Python
list() -> list[Workflow]
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
async def list(self) -> list[Workflow]:
    return sorted(self._store.values(), key=lambda w: w.name)

exists async

Python
exists(name: str) -> bool
Source code in apogee_ai_workflow/infrastructure/registry/in_memory_workflow_repository.py
Python
async def exists(self, name: str) -> bool:
    return name in self._store

JsonCheckpointer

Python
JsonCheckpointer(root: str | Path)

One file per <root>/<run_id>.jsonl — append-only checkpoints.

Source code in apogee_ai_workflow/infrastructure/checkpoints/json_checkpointer.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

save async

Python
save(checkpoint: Checkpoint) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/json_checkpointer.py
Python
async def save(self, checkpoint: Checkpoint) -> None:
    await asyncio.to_thread(self._append, checkpoint)

list async

Python
list(run_id: str) -> list[Checkpoint]
Source code in apogee_ai_workflow/infrastructure/checkpoints/json_checkpointer.py
Python
async def list(self, run_id: str) -> list[Checkpoint]:
    return await asyncio.to_thread(self._read_all, run_id)

latest async

Python
latest(run_id: str) -> Checkpoint | None
Source code in apogee_ai_workflow/infrastructure/checkpoints/json_checkpointer.py
Python
async def latest(self, run_id: str) -> Checkpoint | None:
    items = await self.list(run_id)
    return items[-1] if items else None

clear async

Python
clear(run_id: str) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/json_checkpointer.py
Python
async def clear(self, run_id: str) -> None:
    path = self._path(run_id)
    if path.is_file():
        await asyncio.to_thread(path.unlink)

JsonHumanGateGateway

Python
JsonHumanGateGateway(root: str | Path)

File-backed gateway: <root>/<gate_id>.json.

Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

open async

Python
open(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
async def open(self, gate: HumanGate) -> HumanGate:
    await asyncio.to_thread(self._write, gate)
    return gate

get async

Python
get(gate_id: str) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
async def get(self, gate_id: str) -> HumanGate:
    gate = await self.find(gate_id)
    if gate is None:
        raise GateNotFound(gate_id)
    return gate

find async

Python
find(gate_id: str) -> HumanGate | None
Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
async def find(self, gate_id: str) -> HumanGate | None:
    return await asyncio.to_thread(self._read_one, gate_id)

update async

Python
update(gate: HumanGate) -> HumanGate
Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
async def update(self, gate: HumanGate) -> HumanGate:
    await asyncio.to_thread(self._write, gate)
    return gate

list async

Python
list(*, run_id: str | None = None, pending_only: bool = False) -> list[HumanGate]
Source code in apogee_ai_workflow/infrastructure/hil/json_human_gate_gateway.py
Python
async def list(
    self,
    *,
    run_id: str | None = None,
    pending_only: bool = False,
) -> list[HumanGate]:
    items = await asyncio.to_thread(self._read_all)
    if run_id is not None:
        items = [g for g in items if g.run_id == run_id]
    if pending_only:
        items = [g for g in items if g.is_pending]
    items.sort(key=lambda g: g.created_at)
    return items

JsonRunRepository

Python
JsonRunRepository(root: str | Path)

One file per <root>/<run_id>.json plus _index.json.

Source code in apogee_ai_workflow/infrastructure/registry/json_run_repository.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

save async

Python
save(run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/registry/json_run_repository.py
Python
async def save(self, run: WorkflowRun) -> WorkflowRun:
    await asyncio.to_thread(self._write, run)
    return run

get async

Python
get(run_id: str) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/registry/json_run_repository.py
Python
async def get(self, run_id: str) -> WorkflowRun:
    run = await self.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    return run

find async

Python
find(run_id: str) -> WorkflowRun | None
Source code in apogee_ai_workflow/infrastructure/registry/json_run_repository.py
Python
async def find(self, run_id: str) -> WorkflowRun | None:
    return await asyncio.to_thread(self._read_one, run_id)

list async

Python
list(*, workflow_name: str | None = None, limit: int | None = None) -> list[WorkflowRun]
Source code in apogee_ai_workflow/infrastructure/registry/json_run_repository.py
Python
async def list(
    self,
    *,
    workflow_name: str | None = None,
    limit: int | None = None,
) -> list[WorkflowRun]:
    return await asyncio.to_thread(self._read_index, workflow_name, limit)

NativeWorkflowEngine

Python
NativeWorkflowEngine(*, checkpointer: ICheckpointer | None = None, gates: IHumanGateGateway | None = None, compensator: ISagaCompensator | None = None, config: EngineConfig | None = None)

Async DAG executor with checkpointing, retries, saga, HIL.

Steps are executed respecting depends_on order; independent steps run concurrently up to EngineConfig.max_concurrency. After every step the engine writes a :class:Checkpoint so resumption from a crashed worker only repeats steps that didn't finish.

HUMAN_GATE steps pause the run, persist a :class:HumanGate via the gateway and return — the run state is WAITING_FOR_GATE until resume() is invoked after the gateway records a decision.

Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
def __init__(
    self,
    *,
    checkpointer: ICheckpointer | None = None,
    gates: IHumanGateGateway | None = None,
    compensator: ISagaCompensator | None = None,
    config: EngineConfig | None = None,
) -> None:
    self._checkpointer = checkpointer
    self._gates = gates or InMemoryHumanGateGateway()
    self._compensator = compensator or DefaultSagaCompensator()
    self._config = config or EngineConfig()
    self._cancelled: set[str] = set()

name class-attribute instance-attribute

Python
name = 'native'

supports_human_gates property

Python
supports_human_gates: bool

execute async

Python
execute(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    return await self._drive(definition, run)

resume async

Python
resume(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
async def resume(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    if run.pending_gate_id is not None:
        gate = await self._gates.find(run.pending_gate_id)
        if gate is None:
            raise GateRejected(run.pending_gate_id, "gate not found")
        if gate.is_pending:
            # Still pending — return the run unchanged so the caller
            # can show ``WAITING_FOR_GATE`` to the user.
            return run
        if gate.decision.value == "rejected":
            step_run = run.step_run(gate.step_name)
            step_run.fail(f"gate rejected: {gate.reason or ''}")
            run.error = f"gate rejected: {gate.reason or ''}"
            run.status = RunStatus.FAILED
            run.finished_at = datetime.now(timezone.utc)
            run.pending_gate_id = None
            return run
        # Approved → mark gate step as succeeded and continue
        step_run = run.step_run(gate.step_name)
        step_run.succeed(output={"gate": "approved"})
        run.pending_gate_id = None

    return await self._drive(definition, run)

cancel async

Python
cancel(run_id: str) -> bool
Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
async def cancel(self, run_id: str) -> bool:
    self._cancelled.add(run_id)
    return True

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
async def shutdown(self) -> None:
    return None

healthcheck async

Python
healthcheck() -> dict[str, Any]
Source code in apogee_ai_workflow/infrastructure/engines/native_engine.py
Python
async def healthcheck(self) -> dict[str, Any]:
    return {"engine": self.name, "ok": True}

RestateAdapter

Python
RestateAdapter(*, ingress_url: str = 'http://localhost:8080')

Adapter for restate.dev durable orchestration.

Lazy import: install via pip install 'apogee-ai-workflow[restate]'.

Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
def __init__(self, *, ingress_url: str = "http://localhost:8080") -> None:
    try:
        import restate  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "RestateAdapter requires `restate-sdk`. "
            "Install with: pip install 'apogee-ai-workflow[restate]'"
        ) from exc
    self._ingress_url = ingress_url

name class-attribute instance-attribute

Python
name = 'restate'

supports_human_gates property

Python
supports_human_gates: bool

execute async

Python
execute(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    # Real implementation would invoke a Restate service via ingress.
    # We surface a clear error so the caller knows infra is missing.
    raise EngineUnavailable(
        self.name,
        "RestateAdapter.execute requires a deployed Restate service mapped to "
        f"workflow {definition.name!r}",
    )

resume async

Python
resume(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
async def resume(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    return run

cancel async

Python
cancel(run_id: str) -> bool
Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
async def cancel(self, run_id: str) -> bool:
    return False

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
async def shutdown(self) -> None:
    return None

healthcheck async

Python
healthcheck() -> dict[str, Any]
Source code in apogee_ai_workflow/infrastructure/engines/restate_adapter.py
Python
async def healthcheck(self) -> dict[str, Any]:
    return {"engine": self.name, "ingress": self._ingress_url}

SqlCheckpointer

Python
SqlCheckpointer(session_factory: Any, *, table_name: str = 'apogee_workflow_checkpoints')

SQLAlchemy 2.x async checkpointer.

Lazy import via [sql] extra. Schema is one row per checkpoint; composite PK = (run_id, sequence).

Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
def __init__(
    self,
    session_factory: Any,
    *,
    table_name: str = "apogee_workflow_checkpoints",
) -> None:
    try:
        import sqlalchemy  # type: ignore  # noqa: F401
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "SqlCheckpointer requires SQLAlchemy. "
            "Install with: pip install 'apogee-ai-workflow[sql]'"
        ) from exc
    self._session_factory = session_factory
    self._table_name = table_name
    self._table = self._build_table()

name class-attribute instance-attribute

Python
name = 'sql'

ensure_schema async

Python
ensure_schema(engine: Any) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
async def ensure_schema(self, engine: Any) -> None:
    async with engine.begin() as conn:
        await conn.run_sync(self._table.metadata.create_all)

save async

Python
save(checkpoint: Checkpoint) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
async def save(self, checkpoint: Checkpoint) -> None:
    from sqlalchemy import insert

    async with self._session_factory() as session:
        await session.execute(
            insert(self._table).values(
                run_id=checkpoint.run_id,
                sequence=checkpoint.sequence,
                step_name=checkpoint.step_name,
                state=dict(checkpoint.state),
                output=dict(checkpoint.output),
                created_at=checkpoint.created_at,
            )
        )
        await session.commit()

list async

Python
list(run_id: str) -> list[Checkpoint]
Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
async def list(self, run_id: str) -> list[Checkpoint]:
    from sqlalchemy import select

    async with self._session_factory() as session:
        stmt = (
            select(self._table)
            .where(self._table.c.run_id == run_id)
            .order_by(self._table.c.sequence)
        )
        rows = (await session.execute(stmt)).all()
    return [self._from_row(r._mapping) for r in rows]

latest async

Python
latest(run_id: str) -> Checkpoint | None
Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
async def latest(self, run_id: str) -> Checkpoint | None:
    from sqlalchemy import desc, select

    async with self._session_factory() as session:
        stmt = (
            select(self._table)
            .where(self._table.c.run_id == run_id)
            .order_by(desc(self._table.c.sequence))
            .limit(1)
        )
        row = (await session.execute(stmt)).first()
    if row is None:
        return None
    return self._from_row(row._mapping)

clear async

Python
clear(run_id: str) -> None
Source code in apogee_ai_workflow/infrastructure/checkpoints/sql_checkpointer.py
Python
async def clear(self, run_id: str) -> None:
    from sqlalchemy import delete

    async with self._session_factory() as session:
        await session.execute(delete(self._table).where(self._table.c.run_id == run_id))
        await session.commit()

TemporalAdapter

Python
TemporalAdapter(*, target: str = 'localhost:7233', namespace: str = 'default', task_queue: str = 'apogee-workflow')

Adapter for Temporal.io durable workflow engine.

Lazy import: install via pip install 'apogee-ai-workflow[temporal]'.

The native Apogee step model maps onto Temporal activities; full code-gen of Temporal workflows is out of scope for this MVP. The adapter exposes the contract surface so callers can opt into Temporal when the surrounding org already runs it.

Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
def __init__(
    self,
    *,
    target: str = "localhost:7233",
    namespace: str = "default",
    task_queue: str = "apogee-workflow",
) -> None:
    try:
        import temporalio  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "TemporalAdapter requires `temporalio`. "
            "Install with: pip install 'apogee-ai-workflow[temporal]'"
        ) from exc
    self._target = target
    self._namespace = namespace
    self._task_queue = task_queue

name class-attribute instance-attribute

Python
name = 'temporal'

supports_human_gates property

Python
supports_human_gates: bool

execute async

Python
execute(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
async def execute(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    try:
        from temporalio.client import Client  # type: ignore
    except ImportError as exc:  # pragma: no cover
        raise EngineUnavailable(self.name, str(exc)) from exc
    try:
        client = await Client.connect(self._target, namespace=self._namespace)
        await client.start_workflow(
            definition.name,
            args=[run.input],
            id=run.id,
            task_queue=self._task_queue,
        )
    except Exception as exc:  # noqa: BLE001
        raise EngineUnavailable(self.name, str(exc)) from exc
    return run

resume async

Python
resume(definition: WorkflowDefinition, run: WorkflowRun) -> WorkflowRun
Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
async def resume(
    self,
    definition: WorkflowDefinition,
    run: WorkflowRun,
) -> WorkflowRun:
    # Temporal handles its own resume after worker restart; nothing to do.
    return run

cancel async

Python
cancel(run_id: str) -> bool
Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
async def cancel(self, run_id: str) -> bool:
    try:
        from temporalio.client import Client  # type: ignore
    except ImportError:
        return False
    try:
        client = await Client.connect(self._target, namespace=self._namespace)
        handle = client.get_workflow_handle(run_id)
        await handle.cancel()
        return True
    except Exception:  # noqa: BLE001
        return False

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
async def shutdown(self) -> None:
    return None

healthcheck async

Python
healthcheck() -> dict[str, Any]
Source code in apogee_ai_workflow/infrastructure/engines/temporal_adapter.py
Python
async def healthcheck(self) -> dict[str, Any]:
    return {"engine": self.name, "target": self._target, "namespace": self._namespace}

WorkflowEngineRegistry

Python
WorkflowEngineRegistry(engines: Mapping[str, IWorkflowEngine] | None = None)
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def __init__(self, engines: Mapping[str, IWorkflowEngine] | None = None) -> None:
    self._engines: dict[str, IWorkflowEngine] = dict(engines or {})

name class-attribute instance-attribute

Python
name = 'registry'

register

Python
register(engine: IWorkflowEngine) -> None
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def register(self, engine: IWorkflowEngine) -> None:
    self._engines[engine.name] = engine

unregister

Python
unregister(name: str) -> None
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def unregister(self, name: str) -> None:
    self._engines.pop(name, None)

get

Python
get(name: str) -> IWorkflowEngine
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def get(self, name: str) -> IWorkflowEngine:
    if name not in self._engines:
        raise EngineUnavailable(name, "not registered")
    return self._engines[name]

find

Python
find(name: str) -> IWorkflowEngine | None
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def find(self, name: str) -> IWorkflowEngine | None:
    return self._engines.get(name)

list

Python
list() -> list[str]
Source code in apogee_ai_workflow/infrastructure/registry/engine_registry.py
Python
def list(self) -> list[str]:
    return sorted(self._engines)