Aga — durable execution for Python

Write typed functions that survive crashes, restarts, deploys, and long waits. One App owns declarations, connection, Run creation, and worker lifecycle. Aga records each durable boundary and reuses its committed result on recovery.

Breaking 0.4 release. This version removes the equivalent spellings from 0.3 and intentionally provides one public form for each operation.

New in 0.4.1: calls made before a join can overlap in sticky and distributed workflows; caller waiting does not change placement. concurrency sizes separate bounded claim and sticky-function pools per worker (default four). Immediate awaits remain sequential. Checkpoints are written by the workflow thread; only committed results survive crashes. See the concurrent-task walkthrough. Use server 0.2.1 or later to inspect measured sticky function times in the dashboard.

Install

The predecessor was published as ogha. Its package name and API are incompatible with Aga 0.4, so upgrade as a hard cutover rather than installing both generations:

python -m pip install --upgrade "aga-runtime==0.4.1"

The distribution is aga-runtime; import it with the short, documented alias:

import aga_runtime as aga

App reads AGA_URL, AGA_TENANT, AGA_NAMESPACE, and AGA_API_KEY by default; constructor values override the connection location and scope.

Declare durable functions

Declarations belong to one App:

from dataclasses import dataclass

import aga_runtime as aga

app = aga.App("checkout")


@dataclass
class Order:
    id: str
    amount: int


@dataclass
class Receipt:
    id: str
    approved: bool


@app.step(retry=aga.RetryPolicy(max_attempts=5), pivot=True)
def charge(order: Order) -> Receipt:
    return payments.charge(order)


@app.workflow()
async def checkout(order: Order) -> Receipt:
    receipt = await charge(order)
    if order.amount > 10_000:
        await aga.signal(
            aga.Approval("large-charge", evidence=receipt),
            timeout=24 * 60 * 60,
        )
    aga.event("order.charged", {"order": order.id})
    return receipt


if __name__ == "__main__":
    app.serve()

Aga restores supported declared Python types at durable JSON boundaries, including dataclasses, containers, enums, dates, UUIDs, and Pydantic models.

Create and coordinate work

Calling a registered Step or Remote eagerly creates a typed Handle. Workflows have one creation form: pass the registered Workflow to app.start(...).

Every Step, Remote, root Run, and child Run uses the same Handle[T]. Inside a Workflow, await a Handle. Outside, a root Handle also supports blocking .result(timeout=...).

run = app.start(checkout.options(run_id=order.id), order)
receipt = run.result()


@app.workflow()
async def checkout_with_audit(order: Order) -> Receipt:
    audit = app.start(audit_order, order)  # owned child; auto-joined
    receipt = await charge(order)
    await audit
    return receipt

Only independent child work opts out of ownership:

app.start(audit_order.options(detached=True), order)

join is the single multi-Handle combinator:

all_values = await aga.join(*handles)
first_value = await aga.join(*handles, count=1)
two_values = await aga.join(*handles, count=2)

Sync, Async, and Async Distributed

These describe two separate choices:

• Sync versus Async is caller behavior. Sync waits for a root Handle's result; Async keeps the same Handle and lets the caller continue.
• Async versus Async Distributed is workflow placement. Default Async keeps ordinary Steps with the Workflow worker. Async Distributed independently dispatches and fences each Step.

Use Sync when a request or script needs the result now. Use Async for long work or responsive callers. Use Async Distributed when Steps need separate scaling, isolation, placement, or leases. There is no execution="sync"; waiting never changes placement.

# Sync caller
receipt = app.start(checkout, order).result()

# Async caller
run = app.start(checkout, order)
receipt = await run


# Distributed placement
@app.workflow(execution="async_distributed")
async def distributed_checkout(order: Order) -> Receipt:
    return await charge(order)

A caller timeout does not cancel the durable Run. Cancellation is explicit with aga.cancel(run, reason).

Waiting, services, and schedules

sleep completes through engine time. signal completes through external input. Approval is an immutable request passed to signal, not a third wait verb; it selects governed, fail-closed behavior and replay-bound evidence.

A Remote accepts one required request object so its envelope maps to Python, Go, Java, and TypeScript without language-specific argument rules. The serving App registers a Step with the same method name and portable request schema.

Schedules are engine-owned root-Run creation and continue while workers are offline:

@app.schedule("0 6 * * *", input={"report": "kpi"}, revision=2)
@app.workflow(name="reports.daily")
async def reports_daily(request: dict[str, str]) -> Report:
    return await build_report(request["report"])

Migrating from 0.3

This generation requires the lineage-aware server and wire contract. Freeze submission ingress and schedules for the complete database, settle or explicitly cancel every resumable 0.3 Run, archive its resettable hot origins, revoke the old worker credentials, and stop every old server and worker before starting the 0.4 fleet. A target or Workflow name is not a generation fence, and old and current workers must not coexist even on different targets. Once 0.4 accepts work, do not roll back to an older generation.

Documentation: overview · Python guide · source

Apache License 2.0. See LICENSE.