Idempotent Actions

Production agents retry, resume, re-plan, race other workers, and replay queue events. For read-only work that's harmless. For side-effectful tool calls — sending an email, issuing a refund, updating a CRM, creating a ticket — repeating the action is expensive or dangerous.

@side_effect has two modes:

• Default (reentrant caching): skip already-done work when a handler re-runs from the top on resume. Best-effort, single-process.
• Claim mode (claim=True): atomic, concurrency-safe execute-once for real actions, returning a structured action receipt.

The guarantee

Within the TTL window, a business action executes at most once, and duplicate callers get a receipt describing what happened.

This is at-most-once with best-effort reconciliation — not true exactly-once. If a worker dies after the external call succeeds but before the receipt is written, RAK cannot know the action happened. To get true exactly-once, forward the action's id to the downstream system as its own idempotency key (e.g. Stripe's Idempotency-Key) — see Exactly-once downstream.

Basic usage

from datetime import timedelta
from redis_agent_kit import side_effect, ActionStatus

@side_effect(
    claim=True,
    key_fields=["tenant_id", "order_id"],   # the business identity
    ttl=timedelta(hours=24),                 # dedupe window
    redis_client=client,
)
async def issue_refund(tenant_id: str, order_id: str, amount: float) -> dict:
    return await payments.refund(order_id, amount)

receipt = await issue_refund("t1", "o1", 49.99)
if receipt.status == ActionStatus.DONE:
    use(receipt.result)
elif receipt.status == ActionStatus.IN_PROGRESS:
    ...  # another worker is handling it right now

In claim mode the wrapped function returns an ActionReceipt, not the raw value. Read receipt.result for the value and receipt.status to branch.

Business keys

key_fields builds the action's identity from only the named arguments, so irrelevant arguments (or float jitter on an unrelated field) don't break deduplication. This differs from the default key policy, which hashes the function source and all inputs.

await issue_refund("t1", "o1", 49.99)   # runs
await issue_refund("t1", "o1", 10.00)   # SAME action (amount not in key) → deduped
await issue_refund("t1", "o2", 49.99)   # different order → runs

For identity that isn't a plain argument (e.g. the current agent/tenant), use the existing key=<callable> escape hatch instead of key_fields.

How it works

1. Claim the action atomically with SET NX EX before running it. Exactly one concurrent caller wins.
2. The winner runs the function, then writes a single done receipt carrying the result (status + result together — no "done but missing result" gap).
3. Losing/duplicate callers read the existing receipt and follow the policies below.

Policies

Option	Values	Meaning
on_duplicate	RETURN_RECEIPT (default), RAISE	What a completed duplicate gets. A concurrent in-flight duplicate always gets a non-blocking in_progress receipt.
on_failure	RELEASE (default), HOLD, MARK_FAILED	What happens to the claim when the action raises. RELEASE allows retry; HOLD blocks retries until the window expires (use for irreversible actions); MARK_FAILED records a failed receipt.

from redis_agent_kit import OnDuplicate, OnFailure

@side_effect(
    claim=True,
    key_fields=["order_id"],
    on_duplicate=OnDuplicate.RAISE,     # raise DuplicateActionError on a repeat
    on_failure=OnFailure.HOLD,          # don't auto-retry a high-stakes action
    redis_client=client,
)
async def issue_refund(order_id: str) -> dict: ...

Crash-window safety

If a worker dies mid-action, its claim is left in_progress. A later caller that finds a claim older than lease (default 5 minutes) treats it as abandoned. The default is safe: it does not silently re-run the action — it raises StaleActionError. Provide a reconcile hook to resolve it instead:

async def check_refund_status(action_id: str) -> dict:
    # ask the payment provider whether this refund actually went through
    return await payments.lookup(action_id)

@side_effect(
    claim=True,
    key_fields=["order_id"],
    reconcile=check_refund_status,   # called for abandoned claims
    redis_client=client,
)
async def issue_refund(order_id: str) -> dict: ...

Exactly-once downstream

The crash window above is unavoidable with Redis alone — "do the external thing" and "record that we did" can't be made atomic. The way to close it is to push deduplication onto the system of record: pass it a stable idempotency key so a retried call is a no-op on its side.

RAK already computes such a key — the action id — deterministically from key_fields, so every worker (in any process) derives the same value. Read it inside the action with current_action_id() and forward it:

from redis_agent_kit import side_effect, current_action_id

@side_effect(claim=True, key_fields=["tenant_id", "order_id"], redis_client=client)
async def issue_refund(tenant_id: str, order_id: str, amount: float) -> dict:
    return await stripe.refunds.create(
        amount=amount,
        idempotency_key=current_action_id(),  # e.g. "issue_refund:t1:o1"
    )

Now even if a crash causes RAK to re-execute, Stripe sees the same key and won't refund twice. current_action_id() returns None outside a claimed action.

Note

The id rides on a ContextVar: it is isolated per asyncio task and per thread and visible across awaits, but is not auto-propagated into work offloaded to a thread via loop.run_in_executor unless you copy the context explicitly (contextvars.copy_context().run(...)).

Performance

Claim mode adds one or two sub-millisecond Redis round trips per call — negligible next to the external action it guards. The claim is a key with a TTL, not a held lock, so non-duplicate calls never contend. The main lever at scale is memory: each claim is a receipt bounded by ttl; keep stored results small and the window no longer than you need. Default mode (claim=False) is unchanged and adds nothing.