Metadata-Version: 2.4
Name: ogha
Version: 0.3.0
Summary: Durable execution for Python — typed functions that survive crashes, restarts, and month-long waits.
Project-URL: Homepage, https://coding2fun.in/ogha
Project-URL: Documentation, https://coding2fun.in/ogha/python
Project-URL: Source, https://github.com/vedhlabs/sdk-python
Project-URL: Issues, https://github.com/vedhlabs/sdk-python/issues
Author: Kishore Karunakaran
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: background-jobs,distributed-systems,durable-execution,orchestration,resilience,saga,workflow,workflows
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: cbor2>=5.6
Requires-Dist: msgspec>=0.19
Description-Content-Type: text/markdown

# Ogha — durable execution for Python

Write ordinary typed functions that survive crashes, restarts, deploys, and
month-long waits. One `App` owns registration, connection, and worker lifecycle;
Ogha records every durable boundary and reuses its result during recovery.

> **Breaking in 0.3.0.** The explicit workflow Context, module-level decorators,
> and manual Worker authoring API were removed. Applications must use one `App`,
> `@app.step`, `@app.workflow`, and direct durable function calls. See
> [Migrating to 0.3](#migrating-to-03) before upgrading a running deployment.

```python
from dataclasses import dataclass

import ogha

app = ogha.App("checkout")

@dataclass
class Order:
    id: str
    card: str
    amount: int

@dataclass
class Receipt:
    id: str
    approved: bool

@app.step(retry=ogha.RetryPolicy(max_attempts=5), pivot=True)
def charge(order: Order) -> Receipt:
    result = payments.charge(order.card, order.amount)
    return Receipt(result.id, result.approved)

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

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

The original function signatures and annotations are retained. At a durable JSON
boundary, Ogha restores supported resolved annotations—such as dataclasses,
containers, enums, dates, UUIDs, and optional Pydantic models—so live execution and
recovery see the same declared type. Unannotated and unresolved types remain plain
decoded values rather than being guessed.

## Install

```bash
pip install ogha          # Python 3.10+
```

## Run the engine

Ogha ships as one container carrying the engine, storage, and dashboard. Pull it
and open `http://localhost:8080`:

```bash
docker run -p 8080:8080 public.ecr.aws/h6s8i3h8/ogha:0.1.2-obf \
  --storage=postgres --dsn="postgres://…" --auth-disabled
```

A ready-to-use Compose stack is in the getting-started guide linked below.

## Sync, Async, and Async Distributed

The three common experiences use the same durable Run:

| Experience | Code | Caller | Worker placement |
| :--- | :--- | :--- | :--- |
| **Sync** | `receipt = checkout.options(run_id=order.id).run(order)` | waits for the typed result | default sticky, or the workflow's declared placement |
| **Async** | `run = checkout.options(run_id=order.id).start(order)` | receives a durable `RunHandle` immediately | default sticky placement |
| **Async Distributed** | `run = distributed_checkout.options(run_id=order.id).start(order)` | receives a durable `RunHandle` immediately | each step is independently dispatched and fenced |

Declare distributed placement on the workflow:

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

Sync versus Async is a caller completion choice. Async versus Async Distributed
is a worker-placement choice. Waiting can be applied to either placement; there is
no `execution="sync"`. The default public execution value is `"async"`, which
uses Ogha's sticky placement.

For custom caller wait settings, start first and then read the result:

```python
run = checkout.options(run_id="order-42", timeout=60 * 60).start(
    Order("order-42", "card-token", 1250),
)

receipt = run.result(timeout=30)       # caller wait only; the run keeps going
# receipt is a Receipt, recovered from its declared return annotation
```

`await run` is the non-blocking asyncio caller form. It moves status polling to a
worker thread; ordinary Ogha workflow bodies accept only Ogha durable awaitables.

## Eager handles and structured concurrency

Calling a step or child workflow inside a workflow creates it immediately and
returns an awaitable Handle. Work does not wait for the Handle to be awaited.

```python
@app.workflow
async def price_order(order: Order) -> list[Quote]:
    handles = [quote.options(name=carrier)(order, carrier) for carrier in CARRIERS]
    return await ogha.gather(*handles)
```

If a workflow returns while owned children are unfinished, Ogha automatically
joins them in deterministic creation order. A child workflow can outlive its
parent only when detach is explicit:

```python
audit.detach(order)                       # an independent child Run
audit.options(detached=True)(order)       # equivalent immutable view
```

Ogha does not offer detached steps inside the parent's Run. On user-code failure,
the facade requests cancellation of unfinished owned children and preserves the
original failure even if cancellation cannot be delivered.

## Immutable per-call options

Options never consume names from the function's keyword arguments and never mutate
the registered function or another held view:

```python
fast = charge.options(timeout=5, name="fast-charge")
slow = charge.options(timeout=30, name="slow-charge")

fast(order)       # eager Handle; charge itself is unchanged
slow(order)
```

Step views support `name`, `timeout`, and `target`. Workflow views support
`run_id`, `timeout`, `target`, and `detached`.

## Durable control helpers

The context-free helpers lower to Ogha's existing durable operations:

| Helper | Meaning |
| :--- | :--- |
| `await ogha.sleep(seconds)` | durable timer; no worker held while sleeping |
| `await ogha.signal(name, timeout=...)` | externally settled webhook/event |
| `await ogha.approval(action, evidence=..., timeout=...)` | auditable, fail-closed approval |
| `await ogha.gather(*handles)` | wait for all owned work |
| `await ogha.race(*handles)` | first durable settlement |
| `await ogha.quorum(n, *handles)` | first durable quorum |
| `ogha.event(name, payload)` | replay-suppressed run fact |
| `ogha.cancel(handle, reason)` | cooperative child cancellation |

`approval` is one externally settled Promise with evidence and a timeout whose
expiry denies. It does not add a separate server state machine.

`race` and `quorum` select results but do not cancel their remaining Handles.
Cancel every nonwinner idempotently on every replay. Do not guard that cleanup with
`if not handle.settled`: a loser may already be terminal when the workflow replays,
but explicit cancellation also releases it from structured child ownership.

## Typed service boundaries

Declare a remote function stub to keep normal Python call signatures across an RPC
boundary. The stub body is never executed by the caller.

```python
@app.remote("billing", name="capture")
def capture(order: Order, *, idempotency_key: str) -> Receipt: ...

@app.workflow
async def checkout(order: Order) -> Receipt:
    return await capture(order, idempotency_key=order.id)
```

The billing service registers a local `@billing.step(name="capture")` with the
same argument contract. The envelope is JSON and language-neutral; future SDKs can
produce the same wire shape without copying Python's ambient execution mechanism.

## Cron schedules

Engine-owned schedules keep firing and deduplicate occurrences while every worker
is down. Stack the App-owned schedule declaration over a workflow:

```python
@app.schedule("0 6 * * *", input={"report": "kpi"})
@app.workflow(name="reports.daily")
async def reports_daily(param: dict[str, str]) -> Report:
    occurrence = ogha.scheduled_time()
    return await build_report(param["report"], occurrence.isoformat())
```

To change a live declaration, change its fields and bump `revision=`. Worker
startup converges to the highest revision.

`@app.schedule` is not the same as `@app.step`. A schedule asks the engine to
create root workflow Runs on cron time, including while workers are offline. A Step
creates child work only when an executing workflow calls it.

## Migrating to 0.3

Version 0.3.0 intentionally removes the old public Context authoring surface and
renames public `execution="async_sticky"` to `execution="async"`. It also replaces
the module-level schedule decorator's `context=` payload with
`@app.schedule(..., input=...)`.

The complete mechanical mapping is:

| Before 0.3 | Python 0.3 |
| :--- | :--- |
| `@ogha.step` / `@ogha.workflow` | `@app.step` / `@app.workflow` |
| `ctx.call(step, value)` | `step(value)` and `await` its Handle |
| `ctx.spawn(workflow, value)` | `workflow(value)`; use `workflow.detach(value)` only for independent work |
| `ctx.join`, `ctx.sleep`, `ctx.wait`, `ctx.gate`, `ctx.emit`, `ctx.cancel` | `ogha.gather/race/quorum`, `ogha.sleep`, `ogha.signal`, `ogha.approval`, `ogha.event`, `ogha.cancel` |
| `Worker(...)` and global `registry` | `app.worker()` or `app.serve()` |
| `@ogha.scheduled(..., context=value)` | `@app.schedule(..., input=value)` |
| `Client.create_schedule/update_schedule(..., context=value)` | `Client.create_schedule/update_schedule(..., input=value)` |
| `scheduled_time(ctx)` / `ctx.info` | `scheduled_time()` / `ogha.info()` |
| `execution="async_sticky"` | `execution="async"` |

Do not roll new code onto the same target while old Runs are active. Current worker
definition routing is by workflow name, not by `(name, version)`. Deploy 0.3 workers
to a new target or workflow name, send new Runs there, and keep the old worker fleet
until its Runs drain. This is an application cutover rule; the server and wire
protocol are unchanged.

Run metadata is available as an immutable value through `ogha.info()`. For example,
`ogha.info().run_id` replaces reading execution identity from a Context object.

## Documentation

- **Overview:** https://coding2fun.in/ogha
- **Getting started:** https://coding2fun.in/ogha/python
- **Source:** https://github.com/vedhlabs/sdk-python

## License

Apache License 2.0. See [`LICENSE`](LICENSE).
