GitHub - rune0-dev/agent-ledger: The missing transactional layer for AI agents side effects. At-most-once execution, human-in-the-loop approvals, queryable audit trail.

AI agents retry. Side effects shouldn't.

git clone https://github.com/rune0-dev/agent-ledger-demo
cd agent-ledger-demo

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -e .

export OPENAI_API_KEY='sk-...'
agent

# To use in dev-mode with MemoryStore
pip install agent-ledger

# To use in production-mode with PostgresDB
pip install agent-ledger[postgres]

import asyncio
from psycopg_pool import AsyncConnectionPool
from agent_ledger import EffectLedger, EffectLedgerOptions, ToolCall
from agent_ledger.stores.postgres import PostgresStore, SCHEMA_SQL

async def charge_customer(effect):
    print(f"Charging {effect.args_canonical}...")
    return {"status": "charged", "id": "ch_123"}

async def main():
    # Connect to Postgres
    pool = AsyncConnectionPool(conninfo="postgresql://localhost/mydb")
    async with pool.connection() as conn:
        await conn.execute(SCHEMA_SQL)  # Create table if needed

    store = PostgresStore(pool=pool)
    ledger = EffectLedger(EffectLedgerOptions(store=store))

    # First call: executes the handler
    result = await ledger.run(
        ToolCall(
            workflow_id="order-123",
            tool="stripe.charge",
            args={"amount": 1000, "currency": "usd"},
        ),
        handler=charge_customer,
    )
    print(f"First call: {result}")

    # Second call: same inputs → returns recorded result (handler not called)
    result2 = await ledger.run(
        ToolCall(
            workflow_id="order-123",
            tool="stripe.charge",
            args={"amount": 1000, "currency": "usd"},
        ),
        handler=charge_customer,
    )
    print(f"Second call: {result2}")
    # Handler only executed once. No double charge.

asyncio.run(main())

# For quick prototyping without Postgres:
# from agent_ledger import MemoryStore
# store = MemoryStore()  # In-memory, not durable

Charging {"amount":1000,"currency":"usd"}...
First call: {'status': 'charged', 'id': 'ch_123'}
Second call: {'status': 'charged', 'id': 'ch_123'}

from agent_ledger import LedgerHooks, ToolCall

# Define a policy: which tool calls need approval?
def approval_policy(call: ToolCall) -> bool:
    # Large payments need approval
    if call.tool == "stripe.charge" and call.args.get("amount", 0) > 1000:
        return True
    # Production deploys need approval
    if call.tool == "k8s.deploy" and call.args.get("env") == "production":
        return True
    return False

# Notification hook: fires when approval is required
async def notify_slack(effect):
    await slack.post_message(
        channel="#approvals",
        text=f"Approval needed: {effect.tool}",
        blocks=[...],  # Include approve/deny buttons with effect.idem_key
    )

# Combine policy + notification in hooks
hooks = LedgerHooks(
    requires_approval=approval_policy,      # Policy: decides IF approval needed
    on_approval_required=notify_slack,      # Notification: fires WHEN approval needed
)

result = await ledger.run(
    ToolCall(
        workflow_id="deploy-prod",
        tool="k8s.deploy",
        args={"image": "app:v2", "env": "production"},
    ),
    handler=deploy_to_k8s,
    hooks=hooks,
)

# Slack bot side: handle button click
@slack_app.action("approve")
async def handle_approve(ack, body):
    await ack()
    idem_key = body["actions"][0]["value"]  # From button payload

    await ledger.approve(idem_key)
    await slack.post_message(channel="#deployments", text=f"✅ Approved: {idem_key}")

@slack_app.action("deny")
async def handle_deny(ack, body):
    await ack()
    idem_key = body["actions"][0]["value"]

    await ledger.deny(idem_key, reason="Denied by operator")
    await slack.post_message(channel="#deployments", text=f"❌ Denied: {idem_key}")

from agent_ledger import EffectLedger, EffectLedgerOptions, MemoryStore, ToolCall

ledger = EffectLedger(EffectLedgerOptions(store=MemoryStore()))

async def send_email(to: str, subject: str, body: str) -> str:
    # Your actual email-sending logic
    return f"Email sent to {to}"

async def execute_tool_safely(tool_name: str, args: dict, workflow_id: str):
    """Wrap any async function with idempotency."""
    async def handler(_):
        return await send_email(**args)

    return await ledger.run(
        ToolCall(workflow_id=workflow_id, tool=tool_name, args=args),
        handler=handler,
    )

# In your agent's tool execution loop:
result = await execute_tool_safely(
    tool_name="send_email",
    args={"to": "customer@example.com", "subject": "Order confirmed", "body": "..."},
    workflow_id="order-456",
)
# Retries won't send duplicate emails

ToolCall(workflow_id, tool, args)
              │
              ▼
    SHA256(workflow_id | tool | args) → idem_key
              │
              ▼
         ┌─────────┐
         │ LEDGER  │
         └────┬────┘
              │
   ┌──────────┼──────────┐
   │          │          │
 fresh    in-flight   terminal
   │          │          │
   ▼          ▼          ▼
execute     wait       replay
handler   for result   recorded

processing → succeeded
           → failed
           → requires_approval → ready → succeeded/failed
                               → denied
           → canceled

await ledger.run(
    ToolCall(
        workflow_id="ticket-456",
        tool="github.create_issue",
        args={"owner": "acme", "repo": "app", "title": "Bug", "body": "Details..."},
        idempotency_keys=["owner", "repo", "title"],  # body changes won't re-execute; would compute to the same hash.
    ),
    handler=create_issue,
)

from psycopg_pool import AsyncConnectionPool
from agent_ledger.stores.postgres import PostgresStore, SCHEMA_SQL

pool = AsyncConnectionPool(conninfo="postgresql://localhost/mydb")
async with pool.connection() as conn:
    await conn.execute(SCHEMA_SQL)

store = PostgresStore(pool=pool)
ledger = EffectLedger(EffectLedgerOptions(store=store))
# Query your audit trail with SQL

from agent_ledger import CommitSucceeded, CommitFailed, EffectError

begin_result = await ledger.begin(call)

if begin_result.cached:
    return begin_result.cached_result

try:
    result = await execute_tool(call.args)
    await ledger.commit(begin_result.effect.id, CommitSucceeded(result=result))
    return result
except Exception as e:
    await ledger.commit(
        begin_result.effect.id,
        CommitFailed(error=EffectError(message=str(e))),
    )
    raise

from agent_ledger import RunOptions, ConcurrencyOptions, StaleOptions

# Per-call configuration
await ledger.run(
    call,
    handler=my_handler,
    run_options=RunOptions(
        concurrency=ConcurrencyOptions(
            effect_timeout_s=30.0,       # How long to wait for another worker to finish (default: 30s)
            approval_timeout_s=None,     # How long to wait for human approval (default: None = indefinite)
            initial_interval_s=0.05,     # First poll interval (default: 50ms)
            max_interval_s=1.0,          # Max poll interval after backoff (default: 1s)
            backoff_multiplier=1.5,      # Poll interval multiplier each retry (default: 1.5x)
            jitter_factor=0.3,           # Random jitter to prevent thundering herd (default: 0.3)
        ),
        stale=StaleOptions(
            after_ms=60_000,             # Take over PROCESSING effects older than this (default: 0 = disabled)
        ),
    ),
)

# Global defaults (applied to all run() calls unless overridden)
from agent_ledger import EffectLedger, EffectLedgerOptions, LedgerDefaults

ledger = EffectLedger(
    EffectLedgerOptions(
        store=store,
        defaults=LedgerDefaults(
            run=RunOptions(
                concurrency=ConcurrencyOptions(effect_timeout_s=60.0),
            ),
        ),
    ),
)