nodus-retry

Retry policies, backoff execution, and idempotency gates for AI execution systems.

Named retry policies for each execution type, sync and async execution with exponential backoff, non-retryable error classification, and an EXACTLY_ONCE idempotency gate via EffectStore. No required external dependencies — pure stdlib.

Note: nodus-retry is a required dependency of nodus-lang>=4.1.0. The EffectStore and InMemoryEffectStore are also available from nodus-lang scripts via std:effects.

Status: v0.1.0 — published on PyPI.

Install

pip install nodus-retry

What it provides

RetryPolicy

from nodus_retry import RetryPolicy

policy = RetryPolicy(
    max_attempts=3,
    backoff_ms=200,
    exponential_backoff=True,   # 200ms, 400ms, 800ms
    high_risk_immediate_fail=False,
    execution_guarantee="AT_LEAST_ONCE",  # or "EXACTLY_ONCE"
)

Named policies

from nodus_retry import (
    FLOW_NODE_DEFAULT,       # 3 attempts, 500ms exponential, AT_LEAST_ONCE
    AGENT_LOW_MEDIUM,        # 3 attempts, 200ms exponential, AT_LEAST_ONCE
    AGENT_HIGH_RISK,         # 1 attempt, high_risk_immediate_fail=True
    ASYNC_JOB_DEFAULT,       # 5 attempts, 1000ms exponential
    NODUS_SCHEDULED_DEFAULT, # 3 attempts, 300ms exponential
    NO_RETRY,                # max_attempts=1
)

Resolve by name

from nodus_retry import resolve_retry_policy

policy = resolve_retry_policy("FLOW_NODE_DEFAULT")
policy = resolve_retry_policy("AGENT_LOW_MEDIUM")

Execution

from nodus_retry import execute_with_retry, execute_with_retry_async, FLOW_NODE_DEFAULT

# Sync
result = execute_with_retry(
    lambda: call_external_api(),
    policy=FLOW_NODE_DEFAULT,
)

# Async
result = await execute_with_retry_async(
    my_async_fn,
    policy=AGENT_LOW_MEDIUM,
)

Retries on any exception unless is_retryable_error classifies the error string as non-retryable (e.g. auth errors, validation errors).

Error classification

from nodus_retry import is_retryable_error

is_retryable_error("connection timeout")    # True
is_retryable_error("rate limit exceeded")   # True
is_retryable_error("unauthorized")          # False
is_retryable_error("invalid input")         # False

EffectStore — EXACTLY_ONCE idempotency

from nodus_retry import InMemoryEffectStore, compute_action_id

store = InMemoryEffectStore()

# Compute a deterministic action ID
action_id = compute_action_id("memory.write", {"key": "k", "value": "v"}, scope="run-001")

# Check before executing
already_done, cached = store.resolve(action_id)
if already_done:
    return cached   # idempotent return

store.pending(action_id, input_hash="h")
try:
    result = do_the_work()
    store.complete(action_id, "success", result)
    return result
except Exception:
    store.complete(action_id, "failed", None)
    raise

EffectStore is a @runtime_checkable Protocol — any class with resolve, pending, and complete methods satisfies it.

Durable EXACTLY_ONCE across restarts

InMemoryEffectStore is the default and is per-process — a crash loses its records. For @exactly_once that survives process restarts, use the drop-in SqliteEffectStore (stdlib sqlite3, still zero third-party dependencies):

from nodus_retry import SqliteEffectStore

rt.set_effect_store(SqliteEffectStore("effects.db"))

It commits on every write (WAL journal), so a completed effect is durable before complete() returns; a fresh process pointed at the same file resolves prior effects idempotently. Pass ":memory:" for a non-durable in-process store. This does not change the default — durability stays the host's opt-in choice.

Design

• No required dependencies. Pure stdlib (hashlib, json, sqlite3, threading, asyncio, dataclasses).
• nodus-lang>=4.1.0 requires this package. InMemoryEffectStore is wired into the VM as self.effect_store and exposed via std:effects.

Development

pip install -e ".[dev]"
pytest tests/ -q

License

MIT — see LICENSE.