OpenRath v2.0.0 API Reference

The Stable 2.0.0 release adds public packages for workflow definition, durable execution, governed adapters, artifacts, deployment revisions, evaluation, observability, security, and the Agent Server. Stability classifications and compatibility commitments are governed by the v2 API policy.

Package map

Package	Primary contracts
rath.definition	step, router, WorkflowCompiler, ExecutionPlan, EffectClass, RetryPolicy
rath.context	RunContext, TraceContext, deadline propagation
rath.errors	Stable ErrorCode values and serializable RathError
rath.events	Immutable Event, EventKind, ProducerRef, and SessionEventLog
rath.adapters	Governed Provider, Tool, Sandbox, and Memory executors and resource specifications
rath.artifacts	Tenant-scoped LocalArtifactStore and S3ArtifactStore
rath.deployment	Content-identified Revision, DeploymentManifest, and revision stores
rath.eval	Versioned datasets, evaluators, experiments, persistence, and regression gates
rath.observability	Telemetry spans/counters, OpenTelemetry bridge, structured logs, and redaction
rath.runtime	LocalRuntime, Run/Event/Checkpoint/Interrupt models, SQLite/PostgreSQL stores, effect ledgers, Redis signals
rath.security	Principal and tenant context, policies, authorization, secret references, structured audit
rath.server	AgentServer, authentication, Assistant/Session resource stores, HTTP/SSE application
rath.client	Synchronous RemoteClient and native-async AsyncRemoteClient

The retained v1-compatible packages remain documented in the general API Reference.

Definition and plan compilation

Surface	Contract
@step(entry=..., successors=..., effects=..., retry=..., timeout_seconds=...)	Marks a checkpointable execution method.
@router(successors=..., entry=...)	Marks a pure router with an explicit successor allowlist.
Workflow.compile_plan(revision_id=...)	Compiles explicit v2 nodes into an ExecutionPlan.
WorkflowCompiler.compile(...)	Validates entrypoint, successors, schemas, effect declarations, and durable compatibility.
EffectClass	Classifies no-effect, read-only, idempotent, or non-idempotent work for retry and reconciliation.
ResourceManifestV2	Declares the Provider, Tool, Sandbox, and Memory resources required by a plan.

Non-idempotent retries require an explicit idempotency key. For timeout-controlled execution, use an async handler or isolated executor.

Request context

Surface	Contract
SecurityContext	Principal, tenant, optional project, action grants, and trusted attributes.
RunContext	Security, revision, request, trace, and optional deadline identity propagated through a Run.
TraceContext.new(...)	Creates W3C-compatible trace and span identifiers.
RunContext.remaining_seconds(...)	Returns the bounded time remaining for a deadline.
RunContext.ensure_active(...)	Raises DeadlineExceededError after the deadline.

Durable runtime

Method	Behavior
LocalRuntime.register(workflow, revision_id=...)	Compiles and registers the executable plan for a revision.
LocalRuntime.submit(...)	Creates a durable queued Run with session, context, state, priority, and optional idempotency key.
LocalRuntime.work_once(worker_id=..., lease_seconds=..., max_steps=...)	Claims one Run under a lease/fencing token and advances it from the latest committed checkpoint.
SQLiteRunStore(path)	Persists local Run state, events, checkpoints, interrupts, and leases.
PostgresRunStore(dsn, ..., auto_migrate=False)	Provides the production durable store; migrations are a separate operator action.
SQLiteEffectLedger / PostgresEffectLedger	Records dispatch and outcome state for external side effects.
InMemorySignalBus / RedisSignalBus	Carries wake, cancel, and fanout signals without replacing durable Run state.
reconcile_stale_effects(...)	Classifies stale Tool invocations for retry, failure, or operator review.

The durable state model includes Run, RunEvent, Checkpoint, Interrupt, ResourceLease, ToolInvocation, and their explicit status enums.

Governed adapters

Surface	Contract
AdapterRequestContext	Reduced Run, security, trace, deadline, and policy context supplied to an adapter call.
ProviderExecutor	Applies policy and schema constraints around Provider calls.
ToolExecutor	Applies approval, output-size, policy, and schema checks around Tool calls.
SandboxExecutor	Applies isolation and policy constraints around Sandbox calls.
MemoryExecutor	Applies namespace and policy constraints around Memory operations.
ProviderSpec / ToolSpec / SandboxSpec / MemoryNamespace	Explicit resource declarations for governed calls.
ApprovalGrant / ApprovalValidator	Carries and validates an approval before controlled work.

Adapter handlers receive only the context and resource specification needed for their operation. See the threat model for the corresponding trust boundaries.

Supporting runtime planes

Package	Contract
rath.artifacts	Content-addressed, tenant-scoped payload storage with bounded reads and SHA-256 verification.
rath.deployment	Immutable revision identity derived from code and a canonical deployment manifest.
rath.eval	Offline dataset execution, per-evaluator results, revision-bound experiments, and regression decisions.
rath.observability	Trace-correlated spans, counters, redacted structured logs, and an OpenTelemetry SDK bridge.

See Evaluation, Observability, and Artifacts for executable examples.

Security

Surface	Contract
PolicyEngine / authorize(...)	Makes explicit allow, deny, or approval decisions over actions and resources.
SecretRef / SecretResolver	References credentials without persisting raw secrets in Run state.
AuditSink	Receives redacted, correlated security and control-plane events.
StructuredAuditSink	Emits redacted newline-delimited JSON audit records.
LocalTrustedPolicy	Allows explicitly trusted embedded workflows.
DenyAllPolicy	Provides a fail-closed policy default.

Authentication alone grants no action. Service tokens need explicit grants, and object access is filtered by tenant and project scope.

Agent Server and clients

Surface	Contract
AgentServer(...)	Creates the tenant-scoped HTTP/SSE service over a RunStore and LocalRuntime.
AgentServer.register_assistant(...)	Registers a workflow template under an immutable revision ID.
StaticTokenAuth	Reference bearer-token authentication for self-hosted deployments and tests.
AuthProvider	Async authentication protocol for integrated identity systems.
RemoteClient	Synchronous Session, Assistant, Run, Interrupt, Feedback, Event, and Memory operations.
AsyncRemoteClient	Native-async form of the same client surface.

RemoteClient and AsyncRemoteClient expose:

Operation group	Methods
Assistants	create_assistant(...), list_assistants()
Sessions	create_session()
Runs	create_run(...), get_run(...), list_runs(...), cancel_run(...), resume_run(...)
Events	events(...)
Interrupts	list_interrupts(...), decide_interrupt(...)
Feedback	create_feedback(...)
Governed Memory	`store(“put”
Lifecycle	close() or aclose()

See Agent Server and Remote Client for a complete executable path and the OpenAPI document for HTTP request and response schemas.

Events

OpenRath exposes two related event contracts:

Surface	Purpose
rath.runtime.RunEvent	Durable, monotonically sequenced events attached to one Run and replayed by the Agent Server.
rath.events.Event	Deeply immutable Session/lineage event with producer, trust, provenance, and trace identity.
rath.events.SessionEventLog	Immutable ordered Session event view with parent Session identities.

Machine-readable errors

RathError carries a stable code, message, retryable flag, and JSON-safe details. RathError.to_dict() returns the transport representation.

ErrorCode value	Meaning
request.invalid_argument	A request or argument failed validation.
security.unauthenticated	Authentication did not produce a principal.
security.forbidden	The principal lacks the required action or scope.
security.approval_required	The policy requires an explicit approval.
security.policy_error	Policy evaluation failed.
resource.conflict	The requested state change conflicts with durable state.
resource.not_found	The scoped resource is not visible or does not exist.
runtime.deadline_exceeded	The Run or operation exceeded its deadline.
runtime.cancelled	The operation was cancelled.
runtime.unavailable	A required runtime dependency is unavailable.
internal.error	An unexpected internal error occurred.

The Agent Server can additionally return resource.exhausted with HTTP 429 when a configured tenant queue or request bound is reached. The OpenAPI document is the contract source for endpoint response codes.

Command-line entry points

Command	Arguments and environment
openrath-server	--app, --host, --port, --workers; corresponding OPENRATH_APP, OPENRATH_HOST, OPENRATH_PORT, and OPENRATH_WEB_WORKERS.
openrath-worker	--app, --worker-id, --lease-seconds; corresponding OPENRATH_APP, OPENRATH_WORKER_ID, and OPENRATH_WORKER_LEASE_SECONDS.
openrath-migrate	--dsn, --schema, --check; corresponding OPENRATH_POSTGRES_DSN and OPENRATH_DB_SCHEMA.

openrath-server uses one process worker; service deployments scale through container or pod replicas. openrath-worker resolves its application reference to an AgentServer. openrath-migrate --check verifies the schema without applying changes.

Source

Module	Source
rath.definition	src/rath/definition/
rath.context	src/rath/context.py
rath.errors	src/rath/errors.py
rath.events	src/rath/events.py
rath.adapters	src/rath/adapters/
rath.artifacts	src/rath/artifacts/
rath.deployment	src/rath/deployment/
rath.eval	src/rath/eval/
rath.observability	src/rath/observability/
rath.runtime	src/rath/runtime/
rath.security	src/rath/security/
rath.server	src/rath/server/
rath.client	src/rath/client/

Autodoc

Public workflow definition, compiler, and execution-plan contracts.

exception rath.definition.DefinitionError

Workflow declaration cannot produce a safe deterministic plan.

class rath.definition.EdgeSpec(source: 'str', target: 'str')

source: str

target: str

to_dict() → dict[str, str]

class rath.definition.EffectClass(value)

An enumeration.

NONE = 'none'

READ_ONLY = 'read_only'

IDEMPOTENT = 'idempotent'

NON_IDEMPOTENT = 'non_idempotent'

class rath.definition.ExecutionPlan(id: 'UUID', definition_hash: 'str', revision_id: 'UUID', definition: 'WorkflowDefinition', nodes: 'tuple[NodeSpec, ...]', resources: 'ResourceManifestV2', policy_manifest: 'Mapping[str, JSONValue]', durable: 'bool', compatibility_issues: 'tuple[str, ...]' = ())

id: UUID

definition_hash: str

revision_id: UUID

definition: WorkflowDefinition

nodes: tuple[NodeSpec, ...]

resources: ResourceManifestV2

policy_manifest: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

durable: bool

compatibility_issues: tuple[str, ...]

to_dict() → dict[str, object]

canonical_json() → str

class rath.definition.NodeKind(value)

An enumeration.

STEP = 'step'

ROUTER = 'router'

OPAQUE = 'opaque'

class rath.definition.NodeSpec(id: 'str', kind: 'NodeKind', handler: 'str', is_async: 'bool', implementation_hash: 'str | None' = None, retry: 'RetryPolicy' = <factory>, effects: 'EffectClass' = <EffectClass.NON_IDEMPOTENT: 'non_idempotent'>, idempotency_key: 'str | None' = None, timeout_seconds: 'float | None' = None, checkpoint: 'bool' = True, successors: 'tuple[str, ...]' = ())

id: str

kind: NodeKind

handler: str

is_async: bool

implementation_hash: str | None

retry: RetryPolicy

effects: EffectClass

idempotency_key: str | None

timeout_seconds: float | None

checkpoint: bool

successors: tuple[str, ...]

to_dict() → dict[str, object]

class rath.definition.ProviderResource(path: 'str', provider_kind: 'str', model: 'str | None', has_memory: 'bool')

path: str

provider_kind: str

model: str | None

has_memory: bool

to_dict() → dict[str, object]

class rath.definition.ResourceManifestV2(providers: 'tuple[ProviderResource, ...]' = ())

providers: tuple[ProviderResource, ...]

to_dict() → dict[str, object]

class rath.definition.RetryPolicy(max_attempts: 'int' = 1, base_seconds: 'float' = 0.25, max_seconds: 'float' = 30.0)

max_attempts: int

base_seconds: float

max_seconds: float

to_dict() → dict[str, object]

rath.definition.router(*, successors: tuple[str, ...], entry: bool = False) → Callable[[F], F]

Mark a pure routing method with an explicit successor allowlist.

rath.definition.step(*, entry: bool = False, successors: tuple[str, ...] = (), retry: RetryPolicy | None = None, effects: EffectClass = EffectClass.NON_IDEMPOTENT, idempotency_key: str | None = None, timeout_seconds: float | None = None, checkpoint: bool = True) → Callable[[F], F]

Mark a method as a checkpointable execution step.

class rath.definition.WorkflowCompiler

Compile explicit Python step boundaries without executing workflow code.

compile(workflow: object, *, revision_id: UUID, production_durable: bool = False, input_schema: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]] | None = None, state_schema: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]] | None = None, policy_manifest: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]] | None = None) → ExecutionPlan

class rath.definition.WorkflowDefinition(id: 'UUID', name: 'str', version: 'str', entrypoint: 'str', nodes: 'tuple[NodeSpec, ...]', edges: 'tuple[EdgeSpec, ...]', input_schema: 'Mapping[str, JSONValue]' = <factory>, state_schema: 'Mapping[str, JSONValue]' = <factory>)

id: UUID

name: str

version: str

entrypoint: str

nodes: tuple[NodeSpec, ...]

edges: tuple[EdgeSpec, ...]

input_schema: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

state_schema: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

to_dict() → dict[str, object]

Request, trace, and durable run context contracts.

exception rath.context.DeadlineExceededError

class rath.context.RunContext(security: ~rath.security.context.SecurityContext, revision_id: ~uuid.UUID, request_id: ~uuid.UUID = <factory>, trace_context: ~rath.context.TraceContext = <factory>, deadline: ~datetime.datetime | None = None)

Explicit context propagated through runtime and adapter calls.

security: SecurityContext

revision_id: UUID

request_id: UUID

trace_context: TraceContext

deadline: datetime | None

classmethod local(*, revision_id: UUID, deadline: datetime | None = None) → RunContext

remaining_seconds(*, now: datetime | None = None) → float | None

ensure_active(*, now: datetime | None = None) → None

class rath.context.TraceContext(trace_id: str, span_id: str, sampled: bool = True)

Minimal W3C-compatible trace correlation identifiers.

trace_id: str

span_id: str

sampled: bool

classmethod new(*, sampled: bool = True) → TraceContext

Stable machine-readable errors for OpenRath v2 public contracts.

class rath.errors.ErrorCode(value)

Stable error identifiers; enum values are part of the public API.

INVALID_ARGUMENT = 'request.invalid_argument'

UNAUTHENTICATED = 'security.unauthenticated'

FORBIDDEN = 'security.forbidden'

APPROVAL_REQUIRED = 'security.approval_required'

POLICY_ERROR = 'security.policy_error'

CONFLICT = 'resource.conflict'

NOT_FOUND = 'resource.not_found'

DEADLINE_EXCEEDED = 'runtime.deadline_exceeded'

CANCELLED = 'runtime.cancelled'

UNAVAILABLE = 'runtime.unavailable'

INTERNAL = 'internal.error'

exception rath.errors.RathError(code: ErrorCode, message: str, *, retryable: bool = False, details: Mapping[str, object] | None = None)

Base exception with a stable code and serialization contract.

to_dict() → dict[str, Any]

Return the stable transport representation.

Immutable session event and lineage-friendly event-log contracts.

class rath.events.Event(id: UUID, session_id: UUID, sequence: int, kind: EventKind, payload: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]], producer: ProducerRef, trust: TrustLevel, provenance: Provenance, created_at: datetime, trace_context: TraceContext | None = None, schema_version: int = 1)

Deeply immutable event ordered within one Session.

id: UUID

session_id: UUID

sequence: int

kind: EventKind

payload: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

producer: ProducerRef

trust: TrustLevel

provenance: Provenance

created_at: datetime

trace_context: TraceContext | None

schema_version: int

classmethod create(*, session_id: UUID, sequence: int, kind: EventKind, payload: Mapping[str, object], producer: ProducerRef, trust: TrustLevel, provenance: Provenance, trace_context: TraceContext | None = None) → Event

class rath.events.EventKind(value)

An enumeration.

MESSAGE_CREATED = 'session.message.created'

SESSION_FORKED = 'session.forked'

SESSION_MERGED = 'session.merged'

RUN_STATE_CHANGED = 'run.state.changed'

NODE_STARTED = 'run.node.started'

NODE_COMPLETED = 'run.node.completed'

OUTPUT_DELTA = 'run.output.delta'

INTERRUPT_CREATED = 'run.interrupt.created'

TOOL_INVOCATION_CHANGED = 'run.tool_invocation.changed'

class rath.events.ProducerRef(kind: 'str', id: 'str', revision_id: 'UUID | None' = None)

kind: str

id: str

revision_id: UUID | None

class rath.events.SessionEventLog(id: ~uuid.UUID = <factory>, events: tuple[~rath.events.Event, ...] = (), parent_session_ids: tuple[~uuid.UUID, ...] = ())

Immutable ordered Event view; live runtime state is intentionally absent.

id: UUID

events: tuple[Event, ...]

parent_session_ids: tuple[UUID, ...]

append(*, kind: EventKind, payload: Mapping[str, object], producer: ProducerRef, trust: TrustLevel, provenance: Provenance, trace_context: TraceContext | None = None) → SessionEventLog

Shared v2 adapter contracts.

class rath.adapters.AdapterRequestContext(run_id: 'UUID', node_id: 'str', tenant_id: 'str', deadline: 'datetime | None', trace_context: 'TraceContext', idempotency_key: 'str | None', policy_constraints: 'PolicyConstraints', checkpoint_sequence: 'int | None' = None)

run_id: UUID

node_id: str

tenant_id: str

deadline: datetime | None

trace_context: TraceContext

idempotency_key: str | None

policy_constraints: PolicyConstraints

checkpoint_sequence: int | None

class rath.adapters.ApprovalGrant(decision_id: 'UUID', run_id: 'UUID', node_id: 'str', tenant_id: 'str', tool_id: 'str', arguments_digest: 'str', actor_id: 'str')

decision_id: UUID

run_id: UUID

node_id: str

tenant_id: str

tool_id: str

arguments_digest: str

actor_id: str

class rath.adapters.ApprovalValidator(*args, **kwargs)

class rath.adapters.MemoryNamespace(tenant_id: 'str', user_id: 'str | None' = None, agent_id: 'str | None' = None, session_id: 'str | None' = None, trust: 'TrustLevel' = <TrustLevel.UNTRUSTED: 'untrusted'>)

tenant_id: str

user_id: str | None

agent_id: str | None

session_id: str | None

trust: TrustLevel

class rath.adapters.MemoryExecutor(policy: PolicyEngine)

async execute(handler: MemoryHandler, operation: Literal['put', 'search', 'delete'], namespace: MemoryNamespace, payload: Mapping[str, object], *, adapter_context: AdapterRequestContext, run_context: RunContext, timeout_seconds: float = 30) → object

class rath.adapters.MemoryHandler(*args, **kwargs)

rath.adapters.merge_policy_constraints(declared: PolicyConstraints, decided: PolicyConstraints) → PolicyConstraints

Combine caller and policy limits without weakening either side.

class rath.adapters.ProviderCapability(value)

An enumeration.

CHAT = 'chat'

STREAM = 'stream'

TOOLS = 'tools'

STRUCTURED_OUTPUT = 'structured_output'

EMBEDDING = 'embedding'

VISION = 'vision'

class rath.adapters.ProviderExecutor(policy: PolicyEngine)

async execute(spec: ProviderSpec, handler: ProviderHandler, request: Mapping[str, object], *, capability: ProviderCapability, adapter_context: AdapterRequestContext, run_context: RunContext) → object

class rath.adapters.ProviderHandler(*args, **kwargs)

class rath.adapters.ProviderSpec(id: 'str', kind: 'str', model: 'str', credential: 'SecretRef | None' = None, capabilities: 'frozenset[ProviderCapability]' = <factory>, connect_timeout_seconds: 'float' = 10.0, read_timeout_seconds: 'float' = 60.0, total_timeout_seconds: 'float' = 120.0, max_concurrency: 'int' = 16)

id: str

kind: str

model: str

credential: SecretRef | None

capabilities: frozenset[ProviderCapability]

connect_timeout_seconds: float

read_timeout_seconds: float

total_timeout_seconds: float

max_concurrency: int

class rath.adapters.SandboxIsolation(value)

An enumeration.

TRUSTED_HOST = 'trusted_host'

LOCAL_CONTAINER = 'local_container'

REMOTE_CONTAINER = 'remote_container'

class rath.adapters.SandboxExecutor(policy: PolicyEngine)

async execute(spec: SandboxSpec, handler: SandboxHandler, operation: str, payload: Mapping[str, object], *, adapter_context: AdapterRequestContext, run_context: RunContext, timeout_seconds: float = 60) → object

class rath.adapters.SandboxHandler(*args, **kwargs)

class rath.adapters.SandboxSpec(id: 'str', isolation: 'SandboxIsolation', image_digest: 'str | None' = None, cpu_limit: 'float | None' = None, memory_bytes: 'int | None' = None, disk_bytes: 'int | None' = None, process_limit: 'int | None' = None, network: "Literal['deny', 'allowlist', 'unrestricted']" = 'deny', allowed_hosts: 'frozenset[str]' = <factory>, ttl_seconds: 'int' = 900)

id: str

isolation: SandboxIsolation

image_digest: str | None

cpu_limit: float | None

memory_bytes: int | None

disk_bytes: int | None

process_limit: int | None

network: Literal['deny', 'allowlist', 'unrestricted']

allowed_hosts: frozenset[str]

ttl_seconds: int

exception rath.adapters.SchemaValidationError

class rath.adapters.ToolExecutor(policy: PolicyEngine, *, effect_ledger: EffectLedger | None = None, artifact_store: ArtifactStore | None = None, approval_validator: ApprovalValidator | None = None)

async execute(spec: ToolSpec, handler: ToolHandler, arguments: Mapping[str, object], *, adapter_context: AdapterRequestContext, run_context: RunContext, approval: ApprovalGrant | None = None, run_id: UUID | None = None, idempotency_key: str | None = None) → object

class rath.adapters.ToolHandler(*args, **kwargs)

exception rath.adapters.ToolOutputTooLarge

class rath.adapters.ToolSpec(name: 'str', version: 'str', input_schema: 'dict[str, object]', output_schema: 'dict[str, object] | None' = None, effects: 'EffectClass' = <EffectClass.NON_IDEMPOTENT: 'non_idempotent'>, risk: "Literal['low', 'medium', 'high', 'critical']" = 'high', timeout_seconds: 'float' = 30.0, max_output_bytes: 'int' = 1048576, requires_approval: 'bool' = False)

name: str

version: str

input_schema: dict[str, object]

output_schema: dict[str, object] | None

effects: EffectClass

risk: Literal['low', 'medium', 'high', 'critical']

timeout_seconds: float

max_output_bytes: int

requires_approval: bool

rath.adapters.validate_json(value: object, schema: Mapping[str, object], *, path: str = '$') → None

rath.adapters.with_policy_constraints(context: AdapterRequestContext, constraints: PolicyConstraints) → AdapterRequestContext

Return the adapter context carrying the evaluated policy constraints.

Content-addressed artifact storage.

class rath.artifacts.Artifact(tenant_id: 'str', digest: 'str', size: 'int', media_type: 'str', created_at: 'datetime', metadata: 'Mapping[str, JSONValue]' = <factory>)

tenant_id: str

digest: str

size: int

media_type: str

created_at: datetime

metadata: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

property uri: str

exception rath.artifacts.ArtifactNotFound

Raised when an artifact does not exist in the requested tenant.

class rath.artifacts.ArtifactStore(*args, **kwargs)

put(tenant_id: str, content: bytes | BinaryIO, *, media_type: str = 'application/octet-stream', metadata: Mapping[str, object] | None = None) → Artifact

get(tenant_id: str, digest: str) → bytes

stat(tenant_id: str, digest: str) → Artifact

delete(tenant_id: str, digest: str) → bool

class rath.artifacts.LocalArtifactStore(root: str | Path, *, max_bytes: int = 134217728)

Atomic filesystem store intended for embedded and single-node operation.

put(tenant_id: str, content: bytes | BinaryIO, *, media_type: str = 'application/octet-stream', metadata: Mapping[str, object] | None = None) → Artifact

get(tenant_id: str, digest: str) → bytes

stat(tenant_id: str, digest: str) → Artifact

delete(tenant_id: str, digest: str) → bool

class rath.artifacts.S3ArtifactStore(bucket: str, *, prefix: str = 'openrath', client: object | None = None, max_bytes: int = 134217728, **client_options: object)

S3-compatible store; durable identity is the SHA-256 object key.

put(tenant_id: str, content: bytes | BinaryIO, *, media_type: str = 'application/octet-stream', metadata: Mapping[str, object] | None = None) → Artifact

get(tenant_id: str, digest: str) → bytes

stat(tenant_id: str, digest: str) → Artifact

delete(tenant_id: str, digest: str) → bool

Immutable deployment revision contracts.

class rath.deployment.DeploymentManifest(image_digest: 'str', plan_hash: 'str', python_version: 'str', dependencies_digest: 'str', resources: 'Mapping[str, JSONValue]' = <factory>)

image_digest: str

plan_hash: str

python_version: str

dependencies_digest: str

resources: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

canonical_json() → str

class rath.deployment.PostgresRevisionStore(run_store: PostgresRunStore)

put(revision: Revision) → Revision

get(revision_id: UUID) → Revision

class rath.deployment.Revision(id: 'UUID', code_digest: 'str', manifest: 'DeploymentManifest', created_at: 'datetime')

id: UUID

code_digest: str

manifest: DeploymentManifest

created_at: datetime

classmethod create(*, code_digest: str, manifest: DeploymentManifest) → Revision

property content_digest: str

SHA-256 identity covering executable code and deployment manifest.

exception rath.deployment.RevisionConflict

class rath.deployment.RevisionStore(*args, **kwargs)

put(revision: Revision) → Revision

get(revision_id: UUID) → Revision

class rath.deployment.SQLiteRevisionStore(run_store: SQLiteRunStore)

put(revision: Revision) → Revision

get(revision_id: UUID) → Revision

class rath.eval.Dataset(id: 'UUID', name: 'str', version: 'str', examples: 'tuple[Example, ...]')

id: UUID

name: str

version: str

examples: tuple[Example, ...]

class rath.eval.EvaluationResult(evaluator: 'str', score: 'float', passed: 'bool', reason: 'str', metadata: 'Mapping[str, JSONValue]' = <factory>)

evaluator: str

score: float

passed: bool

reason: str

metadata: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.eval.EvaluationRunner

async run(dataset: Dataset, *, revision_id: UUID, execute: Callable[[Example], Awaitable[Run]], evaluators: Sequence[Evaluator]) → Experiment

class rath.eval.EvaluationStore(*args, **kwargs)

save_dataset(dataset: Dataset) → Dataset

get_dataset(dataset_id: UUID) → Dataset

save_experiment(experiment: Experiment) → Experiment

get_experiment(experiment_id: UUID) → Experiment

class rath.eval.Evaluator(*args, **kwargs)

name: str

async evaluate(example: Example, run: Run) → EvaluationResult

class rath.eval.Example(id: 'UUID', inputs: 'Mapping[str, JSONValue]', expected: 'Mapping[str, JSONValue]')

id: UUID

inputs: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

expected: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

classmethod create(inputs: Mapping[str, object], expected: Mapping[str, object]) → Example

class rath.eval.Experiment(id: 'UUID', dataset_id: 'UUID', revision_id: 'UUID', results: 'tuple[EvaluationResult, ...]')

id: UUID

dataset_id: UUID

revision_id: UUID

results: tuple[EvaluationResult, ...]

property mean_score: float

class rath.eval.GateDecision(value)

An enumeration.

PASS = 'pass'

FAIL = 'fail'

class rath.eval.PostgresEvaluationStore(run_store: PostgresRunStore)

save_dataset(dataset: Dataset) → Dataset

get_dataset(dataset_id: UUID) → Dataset

save_experiment(experiment: Experiment) → Experiment

get_experiment(experiment_id: UUID) → Experiment

class rath.eval.SQLiteEvaluationStore(run_store: SQLiteRunStore)

save_dataset(dataset: Dataset) → Dataset

get_dataset(dataset_id: UUID) → Dataset

save_experiment(experiment: Experiment) → Experiment

get_experiment(experiment_id: UUID) → Experiment

rath.eval.regression_gate(candidate: Experiment, *, baseline: Experiment, maximum_regression: float = 0.02, minimum_score: float = 0.8) → GateDecision

class rath.observability.InMemoryTelemetry

Reference exporter used by tests and embedded diagnostics.

property spans: tuple[SpanRecord, ...]

property counters: Mapping[tuple[str, tuple[tuple[str, str], ...]], int]

span(name: str, *, context: TraceContext, attributes: Mapping[str, object] | None = None) → Iterator[None]

increment(name: str, value: int = 1, *, attributes: Mapping[str, str] | None = None) → None

class rath.observability.GuardedTelemetry(delegate: Telemetry)

Failure-isolating wrapper: exporter faults never change application results.

span(name: str, *, context: TraceContext, attributes: Mapping[str, object] | None = None) → Iterator[None]

increment(name: str, value: int = 1, *, attributes: Mapping[str, str] | None = None) → None

class rath.observability.NoOpTelemetry

span(name: str, *, context: TraceContext, attributes: Mapping[str, object] | None = None) → Iterator[None]

increment(name: str, value: int = 1, *, attributes: Mapping[str, str] | None = None) → None

class rath.observability.OpenTelemetry(*, service_name: str = 'openrath', tracer_provider: Any | None = None, meter_provider: Any | None = None)

Telemetry implementation backed by configured OpenTelemetry providers.

span(name: str, *, context: TraceContext, attributes: Mapping[str, object] | None = None) → Iterator[None]

increment(name: str, value: int = 1, *, attributes: Mapping[str, str] | None = None) → None

class rath.observability.SpanRecord(name: 'str', trace_id: 'str', span_id: 'str', parent_span_id: 'str | None', started_at: 'datetime', ended_at: 'datetime', duration_ms: 'float', status: 'str', attributes: 'Mapping[str, JSONValue]' = <factory>)

name: str

trace_id: str

span_id: str

parent_span_id: str | None

started_at: datetime

ended_at: datetime

duration_ms: float

status: str

attributes: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.observability.StructuredLogger(sink: Callable[[str], None] | None = None)

Emit stable JSON records without ever failing the application path.

emit(event: str, *, context: TraceContext | None = None, fields: Mapping[str, object] | None = None) → None

class rath.observability.Telemetry(*args, **kwargs)

span(name: str, *, context: TraceContext, attributes: Mapping[str, object] | None = None) → Iterator[None]

increment(name: str, value: int = 1, *, attributes: Mapping[str, str] | None = None) → None

rath.observability.redact(value: object) → object

Public durable runtime state and persistence contracts.

class rath.runtime.ApprovalDecision(kind: 'ApprovalDecisionKind', actor_id: 'str', reason: 'str', payload: 'Mapping[str, JSONValue]' = <factory>)

kind: ApprovalDecisionKind

actor_id: str

reason: str

payload: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.runtime.ApprovalDecisionKind(value)

An enumeration.

APPROVE = 'approve'

EDIT = 'edit'

REJECT = 'reject'

RESPOND = 'respond'

rath.runtime.assert_transition(source: RunStatus, target: RunStatus) → None

class rath.runtime.Checkpoint(id: 'UUID', run_id: 'UUID', sequence: 'int', plan_hash: 'str', state: 'Mapping[str, JSONValue]', next_nodes: 'tuple[str, ...]', pending_interrupts: 'tuple[UUID, ...]', effect_watermark: 'int', created_at: 'datetime')

id: UUID

run_id: UUID

sequence: int

plan_hash: str

state: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

next_nodes: tuple[str, ...]

pending_interrupts: tuple[UUID, ...]

effect_watermark: int

created_at: datetime

classmethod create(*, run_id: UUID, sequence: int, plan_hash: str, state: Mapping[str, object], next_nodes: tuple[str, ...], effect_watermark: int, pending_interrupts: tuple[UUID, ...] = ()) → Checkpoint

class rath.runtime.ClaimedRun(run: 'Run', lease: 'ResourceLease')

run: Run

lease: ResourceLease

exception rath.runtime.ConflictError(message: str, *, details: Mapping[str, object] | None = None)

class rath.runtime.EffectLedger(*args, **kwargs)

prepare(*, run_id: UUID, tool_name: str, effect_class: EffectClass, arguments_digest: str, idempotency_key: str | None, node_id: str | None = None, checkpoint_sequence: int | None = None) → ToolInvocation

get(invocation_id: UUID) → ToolInvocation

mark_dispatched(invocation_id: UUID) → ToolInvocation

complete(invocation_id: UUID, result: object) → ToolInvocation

fail(invocation_id: UUID, error: str) → ToolInvocation

reconcile_stale(*, older_than: datetime) → tuple[ToolInvocation, ...]

watermark(run_id: UUID) → int

class rath.runtime.ExecutionServices(policy: PolicyEngine, tools: ToolExecutor, providers: ProviderExecutor, sandboxes: SandboxExecutor, memory: MemoryExecutor, effects: EffectLedger, audit: AuditSink)

Governed capabilities made available to a workflow step.

policy: PolicyEngine

tools: ToolExecutor

providers: ProviderExecutor

sandboxes: SandboxExecutor

memory: MemoryExecutor

effects: EffectLedger

audit: AuditSink

class rath.runtime.Interrupt(id: 'UUID', run_id: 'UUID', kind: 'InterruptKind', request: 'Mapping[str, JSONValue]', created_at: 'datetime', expires_at: 'datetime | None' = None, decision: 'ApprovalDecision | None' = None, decided_at: 'datetime | None' = None)

id: UUID

run_id: UUID

kind: InterruptKind

request: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

created_at: datetime

expires_at: datetime | None

decision: ApprovalDecision | None

decided_at: datetime | None

classmethod create(*, run_id: UUID, kind: InterruptKind, request: Mapping[str, object], timeout_seconds: float | None = None) → Interrupt

class rath.runtime.InterruptKind(value)

An enumeration.

APPROVAL = 'approval'

INPUT = 'input'

REVIEW = 'review'

class rath.runtime.InvocationStatus(value)

An enumeration.

PREPARED = 'prepared'

DISPATCHED = 'dispatched'

SUCCEEDED = 'succeeded'

FAILED = 'failed'

AMBIGUOUS = 'ambiguous'

class rath.runtime.GuardedSignalBus(inner: SignalBus)

Best-effort wrapper preserving database success when Redis is unavailable.

publish(signal: RunSignal) → None

receive(*, timeout_seconds: float = 0) → RunSignal | None

class rath.runtime.InMemorySignalBus

publish(signal: RunSignal) → None

receive(*, timeout_seconds: float = 0) → RunSignal | None

exception rath.runtime.InvalidRunTransition(source: RunStatus, target: RunStatus)

class rath.runtime.LocalRuntime(store: RunStore, *, telemetry: Telemetry | None = None, structured_logger: StructuredLogger | None = None, effect_ledger: EffectLedger | None = None, production_mode: bool = False, execution_services: ExecutionServices | None = None, step_executor: StepExecutor | None = None)

Sync façade over the durable local worker engine.

register(workflow: object, *, revision_id: UUID) → ExecutionPlan

submit(workflow: object, *, session_id: UUID, context: RunContext, state: Mapping[str, object] | None = None, idempotency_key: str | None = None, priority: int = 0) → Run

work_once(*, worker_id: str, lease_seconds: float = 30.0, max_steps: int | None = None, now: datetime | None = None) → Run | None

reconcile_effects(*, grace_seconds: float = 30.0, now: datetime | None = None) → Reconciliation

exception rath.runtime.PlanMismatchError

A durable checkpoint does not belong to the registered executable plan.

class rath.runtime.PostgresRunStore(dsn: str, *, schema: str = 'openrath', auto_migrate: bool = False, pool_max_size: int = 20)

Transactional Postgres store using row locks and fencing tokens.

classmethod migrate(dsn: str, *, schema: str = 'openrath') → None

classmethod verify_schema(dsn: str, *, schema: str = 'openrath') → None

connection() → Iterator[Any]

Borrow a schema-configured pooled connection for related stores.

close() → None

create_run(run: Run) → Run

get_run(run_id: UUID) → Run

list_runs(*, tenant_id: str, after: UUID | None = None, limit: int | None = None, session_id: UUID | None = None, statuses: tuple[RunStatus, ...] | None = None) → tuple[Run, ...]

count_runs(*, status: RunStatus, tenant_id: str | None = None) → int

transition_run(run_id: UUID, *, expected_version: int, target: RunStatus, state: Mapping[str, object] | None = None, next_nodes: tuple[str, ...] | None = None) → Run

list_run_events(run_id: UUID, *, after_sequence: int = 0, limit: int | None = None) → tuple[RunEvent, ...]

append_run_event(run_id: UUID, type: str, data: Mapping[str, object]) → RunEvent

append_checkpoint(checkpoint: Checkpoint) → None

latest_checkpoint(run_id: UUID) → Checkpoint | None

list_checkpoints(run_id: UUID) → tuple[Checkpoint, ...]

commit_checkpoint(checkpoint: Checkpoint, *, worker_id: str, fencing_token: int, expected_run_version: int) → Run

finish_claim(run_id: UUID, *, worker_id: str, fencing_token: int, expected_run_version: int, target: RunStatus, event_type: str = 'run.execution.completed', event_data: Mapping[str, object] | None = None) → Run

create_interrupt(interrupt: Interrupt, *, expected_run_version: int) → Run

get_interrupt(interrupt_id: UUID) → Interrupt

list_interrupts(*, tenant_id: str, pending_only: bool = True) → tuple[Interrupt, ...]

expire_interrupts(*, now: datetime | None = None) → tuple[UUID, ...]

decide_interrupt(interrupt_id: UUID, *, decision: ApprovalDecision, expected_run_version: int) → Run

claim_next(*, worker_id: str, lease_seconds: float, now: datetime | None = None) → ClaimedRun | None

renew_lease(run_id: UUID, *, worker_id: str, fencing_token: int, lease_seconds: float, now: datetime | None = None) → ResourceLease

assert_fencing_token(run_id: UUID, *, worker_id: str, fencing_token: int) → None

requeue_expired_leases(*, now: datetime | None = None) → tuple[UUID, ...]

class rath.runtime.PythonStepExecutor(store: RunStore)

Reference in-process executor for embedded and async durable steps.

execute(*, handler: object, state: dict[str, object], context: StepContext, node: NodeSpec, run: Run) → object

class rath.runtime.PostgresEffectLedger(dsn: str, *, schema: str = 'openrath')

Effect ledger sharing a production Postgres Run schema.

prepare(*, run_id: UUID, tool_name: str, effect_class: EffectClass, arguments_digest: str, idempotency_key: str | None, node_id: str | None = None, checkpoint_sequence: int | None = None) → ToolInvocation

get(invocation_id: UUID) → ToolInvocation

mark_dispatched(invocation_id: UUID) → ToolInvocation

complete(invocation_id: UUID, result: object) → ToolInvocation

fail(invocation_id: UUID, error: str) → ToolInvocation

reconcile_stale(*, older_than: datetime) → tuple[ToolInvocation, ...]

watermark(run_id: UUID) → int

class rath.runtime.Reconciliation(retryable: 'tuple[UUID, ...]', needs_review: 'tuple[UUID, ...]')

retryable: tuple[UUID, ...]

needs_review: tuple[UUID, ...]

class rath.runtime.RedisSignalBus(url: str, *, namespace: str = 'openrath', client: object | None = None)

Redis list transport used only to reduce polling latency.

publish(signal: RunSignal) → None

receive(*, timeout_seconds: float = 0) → RunSignal | None

class rath.runtime.Run(id: 'UUID', plan_id: 'UUID', revision_id: 'UUID', session_id: 'UUID', tenant_id: 'str', status: 'RunStatus', state: 'Mapping[str, JSONValue]', next_nodes: 'tuple[str, ...]', created_at: 'datetime', updated_at: 'datetime', version: 'int' = 0, idempotency_key: 'str | None' = None, context: 'Mapping[str, JSONValue]' = <factory>, priority: 'int' = 0)

id: UUID

plan_id: UUID

revision_id: UUID

session_id: UUID

tenant_id: str

status: RunStatus

state: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

next_nodes: tuple[str, ...]

created_at: datetime

updated_at: datetime

version: int

idempotency_key: str | None

context: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

priority: int

classmethod create(*, plan_id: UUID, revision_id: UUID, session_id: UUID, tenant_id: str, status: RunStatus = RunStatus.QUEUED, state: Mapping[str, object] | None = None, next_nodes: tuple[str, ...] = (), idempotency_key: str | None = None, context: Mapping[str, object] | None = None, priority: int = 0, id: UUID | None = None) → Run

class rath.runtime.RunEvent(run_id: 'UUID', sequence: 'int', type: 'str', data: 'Mapping[str, JSONValue]', created_at: 'datetime')

run_id: UUID

sequence: int

type: str

data: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

created_at: datetime

class rath.runtime.RunStatus(value)

An enumeration.

QUEUED = 'queued'

RUNNING = 'running'

WAITING = 'waiting'

SUCCEEDED = 'succeeded'

FAILED = 'failed'

CANCELLED = 'cancelled'

TIMED_OUT = 'timed_out'

NEEDS_REVIEW = 'needs_review'

class rath.runtime.RunSignal(kind: 'SignalKind', run_id: 'UUID', tenant_id: 'str', created_at: 'datetime')

kind: SignalKind

run_id: UUID

tenant_id: str

created_at: datetime

class rath.runtime.ResourceLease(id: 'UUID', resource_type: 'str', resource_id: 'str', owner_run_id: 'UUID', holder_worker_id: 'str', expires_at: 'datetime', fencing_token: 'int', created_at: 'datetime', updated_at: 'datetime')

id: UUID

resource_type: str

resource_id: str

owner_run_id: UUID

holder_worker_id: str

expires_at: datetime

fencing_token: int

created_at: datetime

updated_at: datetime

class rath.runtime.RunStore(*args, **kwargs)

create_run(run: Run) → Run

get_run(run_id: UUID) → Run

list_runs(*, tenant_id: str, after: UUID | None = None, limit: int | None = None, session_id: UUID | None = None, statuses: tuple[RunStatus, ...] | None = None) → tuple[Run, ...]

count_runs(*, status: RunStatus, tenant_id: str | None = None) → int

transition_run(run_id: UUID, *, expected_version: int, target: RunStatus, state: Mapping[str, object] | None = None, next_nodes: tuple[str, ...] | None = None) → Run

list_run_events(run_id: UUID, *, after_sequence: int = 0, limit: int | None = None) → tuple[RunEvent, ...]

append_run_event(run_id: UUID, type: str, data: Mapping[str, object]) → RunEvent

append_checkpoint(checkpoint: Checkpoint) → None

latest_checkpoint(run_id: UUID) → Checkpoint | None

list_checkpoints(run_id: UUID) → tuple[Checkpoint, ...]

commit_checkpoint(checkpoint: Checkpoint, *, worker_id: str, fencing_token: int, expected_run_version: int) → Run

create_interrupt(interrupt: Interrupt, *, expected_run_version: int) → Run

get_interrupt(interrupt_id: UUID) → Interrupt

list_interrupts(*, tenant_id: str, pending_only: bool = True) → tuple[Interrupt, ...]

expire_interrupts(*, now: datetime | None = None) → tuple[UUID, ...]

decide_interrupt(interrupt_id: UUID, *, decision: ApprovalDecision, expected_run_version: int) → Run

claim_next(*, worker_id: str, lease_seconds: float, now: datetime | None = None) → ClaimedRun | None

renew_lease(run_id: UUID, *, worker_id: str, fencing_token: int, lease_seconds: float, now: datetime | None = None) → ResourceLease

assert_fencing_token(run_id: UUID, *, worker_id: str, fencing_token: int) → None

requeue_expired_leases(*, now: datetime | None = None) → tuple[UUID, ...]

finish_claim(run_id: UUID, *, worker_id: str, fencing_token: int, expected_run_version: int, target: RunStatus, event_type: str = 'run.execution.completed', event_data: Mapping[str, object] | None = None) → Run

close() → None

class rath.runtime.SQLiteRunStore(path: str | Path)

SQLite source of truth for local mode; every mutation is transactional.

close() → None

create_run(run: Run) → Run

get_run(run_id: UUID) → Run

list_runs(*, tenant_id: str, after: UUID | None = None, limit: int | None = None, session_id: UUID | None = None, statuses: tuple[RunStatus, ...] | None = None) → tuple[Run, ...]

count_runs(*, status: RunStatus, tenant_id: str | None = None) → int

transition_run(run_id: UUID, *, expected_version: int, target: RunStatus, state: Mapping[str, object] | None = None, next_nodes: tuple[str, ...] | None = None) → Run

list_run_events(run_id: UUID, *, after_sequence: int = 0, limit: int | None = None) → tuple[RunEvent, ...]

append_run_event(run_id: UUID, type: str, data: Mapping[str, object]) → RunEvent

append_checkpoint(checkpoint: Checkpoint) → None

latest_checkpoint(run_id: UUID) → Checkpoint | None

list_checkpoints(run_id: UUID) → tuple[Checkpoint, ...]

commit_checkpoint(checkpoint: Checkpoint, *, worker_id: str, fencing_token: int, expected_run_version: int) → Run

finish_claim(run_id: UUID, *, worker_id: str, fencing_token: int, expected_run_version: int, target: RunStatus, event_type: str = 'run.execution.completed', event_data: Mapping[str, object] | None = None) → Run

create_interrupt(interrupt: Interrupt, *, expected_run_version: int) → Run

get_interrupt(interrupt_id: UUID) → Interrupt

list_interrupts(*, tenant_id: str, pending_only: bool = True) → tuple[Interrupt, ...]

expire_interrupts(*, now: datetime | None = None) → tuple[UUID, ...]

decide_interrupt(interrupt_id: UUID, *, decision: ApprovalDecision, expected_run_version: int) → Run

claim_next(*, worker_id: str, lease_seconds: float, now: datetime | None = None) → ClaimedRun | None

renew_lease(run_id: UUID, *, worker_id: str, fencing_token: int, lease_seconds: float, now: datetime | None = None) → ResourceLease

assert_fencing_token(run_id: UUID, *, worker_id: str, fencing_token: int) → None

requeue_expired_leases(*, now: datetime | None = None) → tuple[UUID, ...]

class rath.runtime.SignalBus(*args, **kwargs)

publish(signal: RunSignal) → None

receive(*, timeout_seconds: float = 0) → RunSignal | None

class rath.runtime.SignalKind(value)

An enumeration.

WAKE = 'wake'

CANCEL = 'cancel'

class rath.runtime.SQLiteEffectLedger(path: str)

Effect ledger sharing the embedded runtime SQLite database.

prepare(*, run_id: UUID, tool_name: str, effect_class: EffectClass, arguments_digest: str, idempotency_key: str | None, node_id: str | None = None, checkpoint_sequence: int | None = None) → ToolInvocation

get(invocation_id: UUID) → ToolInvocation

mark_dispatched(invocation_id: UUID) → ToolInvocation

complete(invocation_id: UUID, result: object) → ToolInvocation

fail(invocation_id: UUID, error: str) → ToolInvocation

reconcile_stale(*, older_than: datetime) → tuple[ToolInvocation, ...]

watermark(run_id: UUID) → int

class rath.runtime.StepContext(run_id: 'UUID', request: 'RunContext', worker_id: 'str', fencing_token: 'int', policy_manifest: 'Mapping[str, JSONValue]', services: 'ExecutionServices | None', _interrupt_handler: 'Callable[[InterruptKind, Mapping[str, object], float | None], ApprovalDecision]')

run_id: UUID

request: RunContext

worker_id: str

fencing_token: int

policy_manifest: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

services: ExecutionServices | None

interrupt(kind: InterruptKind, request: Mapping[str, object], *, timeout_seconds: float | None = None) → ApprovalDecision

Suspend durably on first call and return the decision after resume.

class rath.runtime.StepExecutor(*args, **kwargs)

Execute one compiled node through the durable worker boundary.

execute(*, handler: object, state: dict[str, object], context: StepContext, node: NodeSpec, run: Run) → object

exception rath.runtime.StepSuspended

Internal control flow indicating a durable interrupt boundary.

class rath.runtime.ToolInvocation(id: 'UUID', run_id: 'UUID', tool_name: 'str', effect_class: 'EffectClass', arguments_digest: 'str', status: 'InvocationStatus', created_at: 'datetime', updated_at: 'datetime', node_id: 'str | None' = None, checkpoint_sequence: 'int | None' = None, invocation_sequence: 'int | None' = None, idempotency_key: 'str | None' = None, result: 'JSONValue | None' = None, error: 'str | None' = None)

id: UUID

run_id: UUID

tool_name: str

effect_class: EffectClass

arguments_digest: str

status: InvocationStatus

created_at: datetime

updated_at: datetime

node_id: str | None

checkpoint_sequence: int | None

invocation_sequence: int | None

idempotency_key: str | None

result: None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]

error: str | None

rath.runtime.arguments_digest(arguments: Mapping[str, object]) → str

rath.runtime.reconcile_stale_effects(ledger: EffectLedger, run_store: RunStore, *, grace_seconds: float = 30.0, now: datetime | None = None) → Reconciliation

Classify dispatched work after a worker crash.

Idempotent calls may be retried under their stable key. Non-idempotent calls are never replayed automatically and move their Run to NEEDS_REVIEW.

Public security contracts for identity, policy, secrets, and audit.

class rath.security.Action(name: 'str')

name: str

exception rath.security.ApprovalRequiredError(decision: PolicyDecision)

class rath.security.AuditEvent(id: 'UUID', kind: 'AuditKind', occurred_at: 'datetime', tenant_id: 'str', principal_id: 'str', request_id: 'UUID', trace_id: 'str', action: 'str', resource_kind: 'str', resource_id: 'str', outcome: 'str', reason: 'str', policy_id: 'str | None' = None, attributes: 'Mapping[str, JSONValue]' = <factory>)

id: UUID

kind: AuditKind

occurred_at: datetime

tenant_id: str

principal_id: str

request_id: UUID

trace_id: str

action: str

resource_kind: str

resource_id: str

outcome: str

reason: str

policy_id: str | None

attributes: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

classmethod for_policy_decision(*, kind: AuditKind, action: Action, resource: ResourceRef, context: RunContext, decision: PolicyDecision, attributes: Mapping[str, object] | None = None) → AuditEvent

class rath.security.AuditKind(value)

An enumeration.

AUTHENTICATION = 'authentication'

POLICY_DECISION = 'policy_decision'

SECRET_RESOLUTION = 'secret_resolution'

TOOL_ACCESS = 'tool_access'

SANDBOX_ACCESS = 'sandbox_access'

MEMORY_ACCESS = 'memory_access'

RUN_CONTROL = 'run_control'

OPERATOR_OVERRIDE = 'operator_override'

class rath.security.AuditSink(*args, **kwargs)

async emit(event: AuditEvent) → None

exception rath.security.AuthorizationError(decision: PolicyDecision)

async rath.security.authorize(engine: PolicyEngine, *, action: Action, resource: ResourceRef, context: RunContext) → PolicyDecision

Evaluate a policy and turn non-allow effects into stable exceptions.

class rath.security.DenyAllPolicy

Safe default for service and untrusted deployment profiles.

async evaluate(action: Action, resource: ResourceRef, context: RunContext) → PolicyDecision

class rath.security.InMemoryAuditSink

Deterministic reference sink for embedded mode and contract tests.

property events: tuple[AuditEvent, ...]

async emit(event: AuditEvent) → None

class rath.security.LocalTrustedPolicy

Explicit opt-in policy for the embedded trusted-process profile.

async evaluate(action: Action, resource: ResourceRef, context: RunContext) → PolicyDecision

class rath.security.PolicyConstraints(timeout_seconds: 'float | None' = None, max_output_bytes: 'int | None' = None, allowed_network_hosts: 'frozenset[str]' = <factory>, filesystem_root: 'str | None' = None, read_only: 'bool' = False, redactions: 'frozenset[str]' = <factory>)

timeout_seconds: float | None

max_output_bytes: int | None

allowed_network_hosts: frozenset[str]

filesystem_root: str | None

read_only: bool

redactions: frozenset[str]

class rath.security.PolicyDecision(effect: 'PolicyEffect', reason: 'str', policy_id: 'str', constraints: 'PolicyConstraints' = <factory>)

effect: PolicyEffect

reason: str

policy_id: str

constraints: PolicyConstraints

class rath.security.PolicyEffect(value)

An enumeration.

ALLOW = 'allow'

DENY = 'deny'

REQUIRE_APPROVAL = 'require_approval'

ALLOW_WITH_CONSTRAINTS = 'allow_with_constraints'

class rath.security.PolicyEngine(*args, **kwargs)

async evaluate(action: Action, resource: ResourceRef, context: RunContext) → PolicyDecision

exception rath.security.PolicyEvaluationError

class rath.security.Principal(id: str, kind: ~rath.security.context.PrincipalKind, claims: ~collections.abc.Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | ~collections.abc.Mapping[str, JSONValue]] = <factory>)

Authenticated caller identity detached from transport concerns.

id: str

kind: PrincipalKind

claims: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.security.PrincipalKind(value)

An enumeration.

USER = 'user'

SERVICE = 'service'

SYSTEM = 'system'

class rath.security.Provenance(source_type: str, source_id: str, producer: str | None = None, metadata: ~collections.abc.Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | ~collections.abc.Mapping[str, JSONValue]] = <factory>)

Origin metadata carried by untrusted and trusted content.

source_type: str

source_id: str

producer: str | None

metadata: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.security.ResolvedSecret(ref: SecretRef, value: str)

Short-lived secret value whose representation is always redacted.

ref: SecretRef

value: str

reveal() → str

Return the value at the adapter boundary; callers must not log it.

class rath.security.ResourceRef(kind: 'str', id: 'str', tenant_id: 'str | None' = None, attributes: 'Mapping[str, JSONValue]' = <factory>)

kind: str

id: str

tenant_id: str | None

attributes: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

class rath.security.SecretRef(provider: 'str', key: 'str', version: 'str | None' = None)

provider: str

key: str

version: str | None

class rath.security.SecretResolver(*args, **kwargs)

async resolve(ref: SecretRef, *, context: RunContext) → ResolvedSecret

class rath.security.SecurityContext(principal: ~rath.security.context.Principal, tenant_id: str, project_id: str | None = None, grants: frozenset[str] = <factory>, attributes: ~collections.abc.Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | ~collections.abc.Mapping[str, JSONValue]] = <factory>)

Run-scoped identity and tenant boundary.

principal: Principal

tenant_id: str

project_id: str | None

grants: frozenset[str]

attributes: Mapping[str, None | bool | int | float | str | tuple[JSONValue, ...] | Mapping[str, JSONValue]]

classmethod local(*, grants: Iterable[str] = ('trusted_host',)) → SecurityContext

Create the explicit trusted-process context for embedded local mode.

has_grant(grant: str) → bool

class rath.security.StructuredAuditSink(sink: Callable[[str], None] | None = None)

Emit redacted, newline-delimited JSON security audit records.

The default sink writes and flushes stdout so container log collectors can retain the security stream independently from diagnostic traces. Sink failures deliberately propagate: a production reference deployment must not silently discard an audit record.

async emit(event: AuditEvent) → None

class rath.security.TrustLevel(value)

An enumeration.

UNTRUSTED = 'untrusted'

TRUSTED = 'trusted'

SYSTEM = 'system'

class rath.server.AgentServer(store: RunStore, runtime: LocalRuntime, *, auth: AuthProvider, resources: ResourceStore | None = None, embedded_worker: bool = False, worker_id: str = 'agent-server', signals: SignalBus | None = None, worker_lease_seconds: float = 30.0, max_queued_runs_per_tenant: int = 1000, memory_executor: MemoryExecutor | None = None, memory_handler: MemoryHandler | None = None, audit_sink: AuditSink | None = None)

register_assistant(assistant_id: str, workflow: object, *, revision_id: UUID) → None

class rath.server.AuthProvider(*args, **kwargs)

async authenticate(authorization: str | None) → SecurityContext | None

class rath.server.AssistantRecord(id: 'str', tenant_id: 'str', template_id: 'str', revision_id: 'UUID', created_at: 'datetime')

id: str

tenant_id: str

template_id: str

revision_id: UUID

created_at: datetime

class rath.server.FeedbackRecord(id: 'UUID', tenant_id: 'str', run_id: 'UUID', key: 'str', score: 'float | None', value: 'str | None', created_at: 'datetime')

id: UUID

tenant_id: str

run_id: UUID

key: str

score: float | None

value: str | None

created_at: datetime

class rath.server.InMemoryResourceStore

create_assistant(*, tenant_id: str, id: str, template_id: str, revision_id: UUID) → AssistantRecord

get_assistant(tenant_id: str, id: str) → AssistantRecord

list_assistants(tenant_id: str) → tuple[AssistantRecord, ...]

create_session(tenant_id: str) → SessionRecord

get_session(session_id: UUID) → SessionRecord

ensure_session(session: SessionRecord) → SessionRecord

count_tenants() → tuple[str, ...]

create_feedback(*, tenant_id: str, run_id: UUID, key: str, score: float | None, value: str | None) → FeedbackRecord

class rath.server.PostgresResourceStore(run_store: PostgresRunStore)

Multi-replica resource store sharing the production Run schema.

create_assistant(*, tenant_id: str, id: str, template_id: str, revision_id: UUID) → AssistantRecord

get_assistant(tenant_id: str, id: str) → AssistantRecord

list_assistants(tenant_id: str) → tuple[AssistantRecord, ...]

create_session(tenant_id: str) → SessionRecord

get_session(session_id: UUID) → SessionRecord

ensure_session(session: SessionRecord) → SessionRecord

count_tenants() → tuple[str, ...]

create_feedback(*, tenant_id: str, run_id: UUID, key: str, score: float | None, value: str | None) → FeedbackRecord

class rath.server.ResourceStore(*args, **kwargs)

create_assistant(*, tenant_id: str, id: str, template_id: str, revision_id: UUID) → AssistantRecord

get_assistant(tenant_id: str, id: str) → AssistantRecord

list_assistants(tenant_id: str) → tuple[AssistantRecord, ...]

create_session(tenant_id: str) → SessionRecord

get_session(session_id: UUID) → SessionRecord

ensure_session(session: SessionRecord) → SessionRecord

count_tenants() → tuple[str, ...]

create_feedback(*, tenant_id: str, run_id: UUID, key: str, score: float | None, value: str | None) → FeedbackRecord

class rath.server.SQLiteResourceStore(run_store: SQLiteRunStore)

Resource store sharing the embedded Run database.

create_assistant(*, tenant_id: str, id: str, template_id: str, revision_id: UUID) → AssistantRecord

get_assistant(tenant_id: str, id: str) → AssistantRecord

list_assistants(tenant_id: str) → tuple[AssistantRecord, ...]

create_session(tenant_id: str) → SessionRecord

get_session(session_id: UUID) → SessionRecord

ensure_session(session: SessionRecord) → SessionRecord

count_tenants() → tuple[str, ...]

create_feedback(*, tenant_id: str, run_id: UUID, key: str, score: float | None, value: str | None) → FeedbackRecord

class rath.server.SessionRecord(id: 'UUID', tenant_id: 'str', created_at: 'datetime')

id: UUID

tenant_id: str

created_at: datetime

class rath.server.StaticTokenAuth(tokens: dict[str, SecurityContext])

Reference bearer-token provider for self-hosted deployments and tests.

async authenticate(authorization: str | None) → SecurityContext | None

rath.server.create_app(server: AgentServer) → Starlette

class rath.client.AsyncRemoteClient(base_url: str, *, token: str, timeout: float = 30.0)

async create_run(*, assistant_id: str, session_id: str, state: dict[str, object] | None = None, idempotency_key: str | None = None) → dict[str, Any]

async get_run(run_id: str) → dict[str, Any]

async create_session() → dict[str, Any]

async create_assistant(*, assistant_id: str, template_id: str) → dict[str, Any]

async list_assistants() → tuple[dict[str, Any], ...]

async store(operation: str, payload: dict[str, object], *, user_id: str | None = None, agent_id: str | None = None, session_id: str | None = None) → dict[str, Any]

async list_runs(*, limit: int = 50, after: str | None = None) → dict[str, Any]

async cancel_run(run_id: str) → dict[str, Any]

async resume_run(run_id: str) → dict[str, Any]

async list_interrupts(*, pending_only: bool = True, limit: int = 50) → tuple[dict[str, Any], ...]

async decide_interrupt(interrupt_id: str, *, kind: str, reason: str, payload: dict[str, object] | None = None) → dict[str, Any]

async create_feedback(run_id: str, *, key: str, score: float | None = None, value: str | None = None) → dict[str, Any]

async events(run_id: str, *, after: int = 0) → AsyncIterator[dict[str, Any]]

async aclose() → None

class rath.client.RemoteClient(base_url: str, *, token: str, timeout: float = 30.0)

create_run(*, assistant_id: str, session_id: str, state: dict[str, object] | None = None, idempotency_key: str | None = None) → dict[str, Any]

get_run(run_id: str) → dict[str, Any]

create_session() → dict[str, Any]

create_assistant(*, assistant_id: str, template_id: str) → dict[str, Any]

list_assistants() → tuple[dict[str, Any], ...]

store(operation: str, payload: dict[str, object], *, user_id: str | None = None, agent_id: str | None = None, session_id: str | None = None) → dict[str, Any]

list_runs(*, limit: int = 50, after: str | None = None) → dict[str, Any]

cancel_run(run_id: str) → dict[str, Any]

resume_run(run_id: str) → dict[str, Any]

list_interrupts(*, pending_only: bool = True, limit: int = 50) → tuple[dict[str, Any], ...]

decide_interrupt(interrupt_id: str, *, kind: str, reason: str, payload: dict[str, object] | None = None) → dict[str, Any]

events(run_id: str, *, after: int = 0) → tuple[dict[str, Any], ...]

create_feedback(run_id: str, *, key: str, score: float | None = None, value: str | None = None) → dict[str, Any]

close() → None

← OpenRath v2.0.0 overview