`aaa.runtime`¶

Engine ¶

Engine(
    store: EventStore | None = None,
    bus: EventBus | None = None,
    schemas: SchemaRegistry | None = None,
    default_budget: Budget | None = None,
    blobs: BlobStore | None = None,
    claim_threshold: int = 16384,
    sweep_interval_s: float | None = None,
    store_retry: RetryPolicy | None = None,
    is_transient: Callable[[BaseException], bool]
    | None = None,
    redactor: Redactor | None = None,
    require_tenant: bool = False,
    retention: Sequence[RetentionPolicy] = (),
    memory: Sequence[Any] = (),
)

Owns store + bus + schemas (+ optional blob store). Everything persisted is appended to the log BEFORE it is delivered, so the log is the source of truth and delivery replayable.

Cluster-ready: per-run state in _runs is only a cache of a fold of the log, rebuilt on demand ("adopted") by whichever process the bus routes a run's events to. With a distributed bus, several Engine processes sharing one store cooperate on the same runs.

abort `async` ¶

abort() -> None

Hard stop without draining or completing anything — what a crash looks like to the log. Runs stay open in the store; another process (or recover()) continues them. Sets crashed so cleanup code (lease release, ...) behaves as if the process vanished.

spawn ¶

spawn(coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]

Run background work (e.g. a long tool call) tracked for clean shutdown.

tenant ¶

tenant(name: str) -> TenantScope

A view of this engine bound to one tenant: namespaced run ids, ownership-checked lookups.

erase `async` ¶

erase(
    run_id: str,
    *,
    delete: bool = False,
    reason: str = "erasure",
    force: bool = False,
) -> ErasureReport

Erase a finished run: scrub its payloads (keeping the audit skeleton) or, with delete=True, remove it. Also removes its snapshots, blobs and bus copies.

purge `async` ¶

purge(
    *,
    older_than: timedelta,
    tenant: str | None = None,
    delete: bool = False,
    reason: str = "retention",
) -> PurgeReport

Erase every finished run (of tenant, if given) that completed more than older_than ago.

forget_memory `async` ¶

forget_memory(scope: Any) -> MemoryReport

Erase everything the registered memory backends hold for scope (an aaa.memory.Scope: a tenant, or one user of it). Backends without a forget capability are reported, not skipped silently.

apply_retention `async` ¶

apply_retention() -> PurgeReport

Evaluate the configured retention policies now (each is rate-limited by its interval_s).

handle ¶

handle(run_id: str) -> RunHandle

A handle to any run by id (started here, in another process, or before a restart).

retry `async` ¶

retry(
    fn: Callable[[], Awaitable[Any]],
    what: str = "store operation",
    policy: RetryPolicy | None = None,
) -> Any

Run a store/blob operation, riding out transient outages per store_retry (or policy). Raises StoreUnavailable if the outage outlasts the budget; other errors pass through.

hydrate `async` ¶

hydrate(event: Event) -> Event

Restore a claim-checked payload (no-op for ordinary events).

subscribe(
    pattern: str,
    handler: Handler,
    *,
    group: str | None = None,
    **kw: Any,
) -> None

Subscribe a handler whose returned events are persisted+published by the engine. Handlers receive hydrated events. Handler exceptions fail the run (as an event), never the bus. A handler that fails because the STORE is unavailable (beyond store_retry) is not a failed run: the engine raises Redeliver and the bus delivers the event again later. Events for runs that already finished are not delivered to handlers.

group names a logical consumer. With a distributed bus, processes subscribing with the same group compete (each event is handled by exactly one of them, in order per run); group=None means broadcast to every process. Ignored by the in-process bus.

register ¶

register(algo: Algorithm) -> None

Subscribe an algorithm. It only ever sees events of runs it was started for, so several algorithms (and nested child runs) coexist on one bus. With a distributed bus, algorithm names must be unique and identical across processes.

emit `async` ¶

emit(
    event: Event, *, retry: RetryPolicy | None = None
) -> Event | None

Validate -> claim-check -> append (assign seq) -> account budget -> deliver. Returns the stored event, or None if dropped (duplicate terminal event).

A store outage is ridden out per retry (default store_retry); if it outlasts that budget StoreUnavailable is raised and NOTHING has been recorded or marked done locally.

await_event `async` ¶

await_event(
    stream: str,
    match: Callable[[Event], bool],
    *,
    from_seq: int = 0,
) -> Event

Wait for the first event on stream (from from_seq on) that match accepts. A dropped store connection mid-read is ridden out like any other store call: resubscribe from the last event seen, backing off per store_retry (the budget restarts on progress).

start `async` ¶

start(
    algo: Algorithm,
    input: dict[str, Any],
    *,
    budget: Budget | None = None,
    run_id: str | None = None,
    parent: str | None = None,
    tag: str | None = None,
    tenant: str | None = None,
    meta: dict[str, str | int | float | bool] | None = None,
) -> RunHandle

Start a run (idempotent for a given run_id). With parent, the child's spend rolls up into the parent's budget, its completion is announced on the parent stream, and it is cancelled if the parent finishes first. tenant tags the run and everything derived from it (children inherit their parent's); see Engine.tenant() for the safe, namespaced way. meta is trusted application data (e.g. {"user": "ann"}) stamped on the run's events and inherited by everything derived from them; memory scopes are built from it, never from model output.

recover `async` ¶

recover(algos: list[Algorithm]) -> dict[str, RunHandle]

Continue every run left open in the store (single-process restart). See recovery.py.

sweep `async` ¶

sweep() -> Any

Stateless, idempotent cluster maintenance: expire deadlines of dead processes, cancel orphaned children, repair lost child announcements. Safe to run anywhere, any time.

run `async` ¶

run(
    algo: Algorithm, input: dict[str, Any], **kw: Any
) -> Outcome[Any]

Facade: start and await the outcome (no event knowledge needed).

RunHandle ¶

RunHandle(engine: Engine, run_id: str)

result `async` ¶

result() -> Outcome[Any]

Outcome of the run, wherever (whichever process) it was hosted or finished.

events `async` ¶

events(*, hydrate: bool = False) -> AsyncIterator[Event]

Replay the run's stream from the start, then tail until the run completes.

signal `async` ¶

signal(
    type: str, data: dict[str, Any] | None = None
) -> Event | None

Inject an external event (human approval, webhook, ...) into the run.

Budget ¶

Bases: BaseModel

Hard limits for a run. None = unlimited. Enforced by the Engine, not by algorithms.

Ledger ¶

Bases: BaseModel

fold ¶

fold(e: Event) -> Ledger

Pure fold; also usable as a Projection to rebuild spend from the log. Child-run spend rolls up into the parent via aaa.child.completed.

Ctx ¶

Ctx(engine: Engine, run_id: str | None)

Per-event context handed to algorithms. Algorithms hold no run state of their own: everything is derived from the run's event stream, so runs resume, replay and migrate between processes.

history `async` ¶

history(from_seq: int = 0) -> list[Event]

The run's events with any claim-checked payloads restored.

project `async` ¶

project(projection: Projection[S]) -> S

Current state of projection for this run: latest snapshot + tail (O(tail)).

started `async` ¶

started() -> Event

The run's aaa.run.started event (input, budget, parent).

Algorithm ¶

Bases: Protocol

subscriptions ¶

subscriptions() -> list[str]

Event-type glob patterns this algorithm reacts to.

Tenancy and retention¶

TenantScope ¶

TenantScope(engine: Engine, tenant: str)

run_id ¶

run_id(run_id: str) -> str

The namespaced stream id for one of this tenant's own run ids (idempotent).

handle `async` ¶

handle(run_id: str) -> RunHandle

A handle to one of this tenant's runs. KeyError if it does not exist OR belongs to someone else.

purge `async` ¶

purge(*, older_than: timedelta, **kw: Any) -> PurgeReport

Erase this tenant's finished runs older than older_than (never another tenant's).

erase_all `async` ¶

erase_all(*, memory: bool = True, **kw: Any) -> PurgeReport

Right-to-erasure for the whole tenant: every FINISHED run now, and (unless memory=False) everything the engine's memory backends hold for the tenant. Open runs are reported in skipped_open; cancel them first, then call again.

forget_memory `async` ¶

forget_memory(
    *, user: str | None = None, agent: str | None = None
) -> MemoryReport

Erase this tenant's long-term memories (or one user's / agent's) from every backend.

RetentionPolicy `dataclass` ¶

RetentionPolicy(
    older_than: timedelta,
    delete: bool = False,
    tenant: str | None = None,
    interval_s: float = 3600.0,
)

Erase finished runs older than older_than. tenant=None matches runs of any tenant (and untenanted ones). With several policies the first one whose tenant matches a run applies, so list tenant-specific policies before a catch-all.

ErasureReport `dataclass` ¶

ErasureReport(
    stream: str,
    mode: str,
    events: int = 0,
    blobs: int = 0,
    provider_state: int = 0,
)

PurgeReport `dataclass` ¶

PurgeReport(
    erased: list[ErasureReport] = list(),
    skipped_open: int = 0,
    memory: MemoryReport | None = None,
)

Redaction (`aaa.privacy`)¶

Redactor ¶

Bases: Protocol

redact ¶

redact(event: Event) -> Event

Return the event with sensitive values replaced. Must be pure and idempotent.

PatternRedactor ¶

PatternRedactor(
    patterns: Iterable[Pattern] = DEFAULT_PATTERNS,
    *,
    only: Iterable[str] = ("*",),
    skip: Iterable[str] = (),
)

Bases: _Base

Replace matches of regular expressions inside any string value with [REDACTED:<name>].

PatternRedactor() covers e-mail addresses, payment-card numbers (Luhn-checked), US SSNs, common API keys/tokens, JWTs and bearer tokens. Pass your own Patterns for anything else (national ids, phone formats, account numbers): redaction is only as good as its patterns, so treat this as a safety net and keep sensitive data out of events where you can.

KeyRedactor ¶

KeyRedactor(
    keys: Iterable[str] = SECRET_KEYS,
    *,
    replacement: str = "[REDACTED]",
    only: Iterable[str] = ("*",),
    skip: Iterable[str] = (),
)

Bases: _Base

Replace the VALUE of well-known secret-bearing keys (password, api_key, authorization, cookie...) wherever they appear, however deeply nested. Keys match whole (case- and punctuation-insensitively), so accounting fields such as prompt_tokens are never touched.

ChainRedactor ¶

ChainRedactor(*redactors: Redactor)

Apply several redactors in order.

Pattern ¶

Pattern(
    name: str,
    regex: str | Pattern[str],
    check: Callable[[Match[str]], bool] | None = None,
)

A named regex, with an optional check that can veto a match (e.g. a Luhn check).

standard_redactor ¶

standard_redactor(
    *,
    only: Iterable[str] = ("*",),
    skip: Iterable[str] = (),
) -> ChainRedactor

Secret-bearing keys + the default patterns: a sensible starting point.

aaa.runtime¶

Engine ¶

abort async ¶

spawn ¶

tenant ¶

erase async ¶

purge async ¶

forget_memory async ¶

apply_retention async ¶

handle ¶

retry async ¶

hydrate async ¶

subscribe ¶

register ¶

emit async ¶

await_event async ¶

start async ¶

recover async ¶

sweep async ¶

run async ¶

RunHandle ¶

result async ¶

events async ¶

signal async ¶

Budget ¶

Ledger ¶

fold ¶

Ctx ¶

history async ¶

project async ¶

started async ¶

Algorithm ¶

subscriptions ¶

Tenancy and retention¶

TenantScope ¶

run_id ¶

handle async ¶

purge async ¶

erase_all async ¶

forget_memory async ¶

RetentionPolicy dataclass ¶

ErasureReport dataclass ¶

PurgeReport dataclass ¶

Redaction (aaa.privacy)¶

Redactor ¶

redact ¶

PatternRedactor ¶

KeyRedactor ¶

ChainRedactor ¶

Pattern ¶

standard_redactor ¶

`aaa.runtime`¶

abort `async` ¶

erase `async` ¶

purge `async` ¶

forget_memory `async` ¶

apply_retention `async` ¶

retry `async` ¶

hydrate `async` ¶

emit `async` ¶

await_event `async` ¶

start `async` ¶

recover `async` ¶

sweep `async` ¶

run `async` ¶

result `async` ¶

events `async` ¶

signal `async` ¶

history `async` ¶

project `async` ¶

started `async` ¶

handle `async` ¶

purge `async` ¶

erase_all `async` ¶

forget_memory `async` ¶

RetentionPolicy `dataclass` ¶

ErasureReport `dataclass` ¶

PurgeReport `dataclass` ¶

Redaction (`aaa.privacy`)¶