agent-tx

agent-tx ⚡

Deterministic ACID Transaction Engine for Autonomous AI Agents in <1µs
Built in Rust with native Python bindings. Part of the next-generation agent runtime ecosystem.

---

🎯 The Problem

When production AI agents execute complex multi-step workflows (30+ tool invocations, code generation, external API calls):

1. Dirty State on Failure: If a tool fails at step 28, external resources (files, database records, webhooks) remain corrupted.
2. Punishing Token Costs: Standard frameworks abort the run and re-prefill the entire 50k–100k token context from step 1, burning latency and GPU budget.
3. Slow State Duplication: Cloning deep context arrays in Python takes 15ms–50ms and triggers stop-the-world Garbage Collection.

🚀 The Solution: agent-tx

agent-tx brings classical ACID semantics and Copy-on-Write branching directly into the agent execution loop:

• Sub-microsecond Checkpoints (~830 ns): Structural persistence (im::Vector) allows instantaneous context branching without copying memory.
• Saga Pattern Rollback (LIFO): Register compensatory actions for tools and unwind external side effects in reverse order.
• Durable Write-Ahead Log (WAL): Zero data loss across process crashes with CRC32-verified binary replay.
• Native Python Ergonomics: Drop-in with tx.transaction(): context managers and @transactional_tool decorators.

---

📊 Benchmarks (Criterion, Linux x86_64)

---

📦 Installation

pip install agent-tx

---

🛠️ Quickstart

1. Automatic Rollback with Context Manager

from agent_tx import AgentTx

tx = AgentTx()

tx.append("system", "You are a cloud orchestration agent.", 12)
tx.append("user", "Resize cluster node pool.", 15)

try:
    with tx.transaction() as tx_id:
        tx.append("assistant", "Calling scale_nodes(pool='prod', count=10)...", 20)
        
        # Register compensating undo action
        tx.register_undo(lambda: print("-> Compensating: Reverting node pool back to 3"))
        
        # Simulate business failure / validation rejection
        raise RuntimeError("Cloud quota exceeded!")

except RuntimeError:
    print("Execution failed. Context rolled back.")

# Context automatically pruned back to safe state (system + user only)
assert len(tx.export_context()) == 2

2. Durable Recovery via Write-Ahead Log (WAL)

from agent_tx import AgentTx

# Session persists across process restarts
tx = AgentTx.open("agent_session.wal")
tx.append("system", "Mission critical agent", 10)
tx.append("user", "Execute batch job", 12)

# If process dies here, re-opening "agent_session.wal" restores exact state
del tx

recovered_tx = AgentTx.open("agent_session.wal")
assert len(recovered_tx.export_context()) == 2

---

📄 License

Dual-licensed under MIT or Apache 2.0.