Advanced Patterns

Deep dives into the three advanced coordination patterns. Each pattern handles specific failure modes that simple OCC and CRDT can't address alone.

Prerequisites

bash

# Start the gateway
cd packages/gateway && pnpm dev

Pattern 1: Task Claiming

Location: examples/patterns/task-claiming/

When to Use It

Use task claiming when multiple agents compete for work items and you need exactly-once processing.

Examples:

Job queues with multiple workers
Research topics with multiple researchers
Files to process with multiple processors
Orders to fulfill with multiple handlers

The Problem

Simple "check if available, then take it" doesn't work with concurrent agents:

plaintext

❌ Race condition:
Agent A: Is task-1 available? → Yes
Agent B: Is task-1 available? → Yes
Agent A: I'll take task-1! → Success
Agent B: I'll take task-1! → Success  ← Both think they own it!
 
Result: Task processed twice (or more)

The Solution: Write-Wait-Reread

The protocol ensures exactly one winner:

plaintext

1. WRITE: Optimistically write "I claim this task"
2. WAIT:  Brief pause for sync (network round-trip)
3. READ:  Check who actually owns the task
4. DECIDE: If it's me → I own it. If not → I lost, try another.

How It Works

plaintext

Time    Agent A                   Agent B                   Task State
─────────────────────────────────────────────────────────────────────────
T1      claim("task-1")           claim("task-1")           pending
        └─ write: claimedBy=A     └─ write: claimedBy=B
T2      wait (sync)               wait (sync)
T3      read: claimedBy=B         read: claimedBy=B         claimed by B
        └─ "I lost!"              └─ "I won!"
T4      claim("task-2")           process task-1

Agent B's write happened to be the last one, so B wins. Agent A sees B's claim on reread and moves on.

Run It

bash

cd examples/patterns/task-claiming
python main.py

Implementation

Manual (understanding the protocol):

python

async def manual_claim_task(client, shard_id, task_id, agent_id):
    """
    Manual write-wait-reread implementation.
    Use SDK helpers in production code.
    """
    SYNC_DELAY = 0.05  # 50ms wait for sync
 
    # Step 1: READ — Get current state
    shard = await client.get_shard(shard_id)
    tasks = shard.value or {}
 
    if task_id not in tasks:
        return False, "TASK_NOT_FOUND"
 
    task = tasks[task_id]
    if task.get("status") == "claimed":
        return False, task["claimedBy"]  # Already claimed
 
    # Step 2: WRITE — Optimistically claim
    new_tasks = {**tasks}
    new_tasks[task_id] = {
        **task,
        "status": "claimed",
        "claimedBy": agent_id,
    }
 
    try:
        await client.commit_shard(shard_id, new_tasks, shard.version)
    except StaleCommitError:
        return False, "STALE_VERSION"
 
    # Step 3: WAIT — Allow sync
    await asyncio.sleep(SYNC_DELAY)
 
    # Step 4: REREAD — Verify we still own it
    shard = await client.get_shard(shard_id)
    tasks = shard.value or {}
 
    current_owner = tasks.get(task_id, {}).get("claimedBy")
 
    if current_owner == agent_id:
        return True, agent_id  # We won!
    else:
        return False, current_owner  # We lost

Using SDK helpers (recommended):

python

from zerolock_sdk.task_claim import claim_task, complete_task, get_pending_tasks
 
async def worker(agent_id: str):
    async with ZeroLockClient(GATEWAY, agent_id=agent_id) as client:
        while True:
            # Get available tasks
            pending = await get_pending_tasks(client, "my-queue")
            if not pending:
                break  # All done
 
            # Try to claim one
            result = await claim_task(client, "my-queue", pending[0].id, agent_id)
 
            if result.claimed:
                # Process it (we own it exclusively)
                await process(pending[0])
                await complete_task(client, "my-queue", pending[0].id, agent_id)
            else:
                # Lost the race — try another task
                continue

Batch Claiming (Efficiency)

For high-throughput scenarios, claim multiple tasks before processing:

python

async def batch_worker(agent_id: str, batch_size: int = 5):
    """Claim a batch of tasks, then process all of them."""
 
    async with ZeroLockClient(GATEWAY, agent_id=agent_id) as client:
        while True:
            # Claim up to batch_size tasks
            claimed_tasks = []
            for _ in range(batch_size):
                pending = await get_pending_tasks(client, "my-queue")
                if not pending:
                    break
 
                result = await claim_task(client, "my-queue", pending[0].id, agent_id)
                if result.claimed:
                    claimed_tasks.append(pending[0])
 
            if not claimed_tasks:
                break  # No more tasks
 
            # Process entire batch
            for task in claimed_tasks:
                await process(task)
                await complete_task(client, "my-queue", task.id, agent_id)

Key Takeaways

What	Why
Use `claim_task()` helper	Handles the protocol correctly
Check `result.claimed`	You might have lost the race
Move on when you lose	Don't retry the same task
Call `complete_task()` when done	Otherwise task stays claimed

Pattern 2: File Locking

Location: examples/patterns/file-locking/

When to Use It

Use file locking when agents need exclusive access to a resource before modifying it.

Examples:

Editing the same source file
Updating a shared config
Modifying database records
Accessing external APIs with rate limits

The Problem

Even with OCC, multiple agents might edit the same file simultaneously:

plaintext

❌ Without locking:
Agent A: Read config.py version 5
Agent B: Read config.py version 5
Agent A: Add function foo(), commit version 6 ✓
Agent B: Add function bar(), commit version 6 ✗ (StaleCommitError)
Agent B: Retry... Read version 6, add bar(), commit version 7 ✓
 
Result: Both functions added, but Agent B had to retry.

That's actually fine for simple cases! But what if:

Processing takes a long time (LLM call)
You want to avoid conflicts entirely
You need predictable execution order

The Solution: OCC-Based Locks

Use a separate "locks" shard to track who holds each resource:

plaintext

✓ With locking:
Agent A: Acquire lock on config.py ✓
Agent A: Edit config.py (has exclusive access)
Agent B: Acquire lock on config.py... waiting...
Agent A: Release lock on config.py
Agent B: Acquire lock on config.py ✓
Agent B: Edit config.py
 
Result: No conflicts, no retries, predictable order

Run It

bash

cd examples/patterns/file-locking
python main.py

Implementation

FileLock Class:

python

class FileLock:
    """OCC-based file lock with timeout and retry."""
 
    def __init__(self, client, locks_shard, agent_id):
        self.client = client
        self.locks_shard = locks_shard
        self.agent_id = agent_id
 
    async def acquire(self, file_path, timeout_seconds=30.0):
        """Acquire exclusive lock on a file."""
        start_time = time.time()
        retry_delay = 0.1
 
        while True:
            # Check timeout
            elapsed = time.time() - start_time
            if elapsed > timeout_seconds:
                return False  # Timed out
 
            # Read current locks
            shard = await self.client.get_shard(self.locks_shard)
            locks = shard.value or {}
 
            # Check if file is locked by someone else
            existing = locks.get(file_path)
            if existing and existing["owner"] != self.agent_id:
                # Check if lock expired
                lock_time = datetime.fromisoformat(existing["acquired_at"])
                lock_timeout = existing.get("timeout_seconds", 60)
                if datetime.now() - lock_time < timedelta(seconds=lock_timeout):
                    # Lock still valid, wait and retry
                    await asyncio.sleep(retry_delay)
                    retry_delay = min(retry_delay * 1.5, 1.0)  # Backoff
                    continue
 
            # Try to acquire lock
            new_locks = {**locks}
            new_locks[file_path] = {
                "owner": self.agent_id,
                "acquired_at": datetime.now().isoformat(),
                "timeout_seconds": timeout_seconds,
            }
 
            try:
                await self.client.commit_shard(self.locks_shard, new_locks, shard.version)
                return True  # Lock acquired!
            except StaleCommitError:
                # Someone else modified locks, retry
                await asyncio.sleep(0.05)
                continue
 
    async def release(self, file_path):
        """Release lock on a file."""
        for _ in range(10):  # Retry a few times
            shard = await self.client.get_shard(self.locks_shard)
            locks = shard.value or {}
 
            if file_path not in locks:
                return True  # Already unlocked
 
            if locks[file_path]["owner"] != self.agent_id:
                return False  # We don't own this lock
 
            # Remove our lock
            new_locks = {k: v for k, v in locks.items() if k != file_path}
 
            try:
                await self.client.commit_shard(self.locks_shard, new_locks, shard.version)
                return True
            except StaleCommitError:
                await asyncio.sleep(0.05)
                continue
 
        return False

Using the lock:

python

async def edit_file_safely(agent_id: str, file_path: str, new_content: str):
    async with ZeroLockClient(GATEWAY, agent_id=agent_id) as client:
        lock = FileLock(client, "file-locks", agent_id)
 
        # Step 1: Acquire lock
        acquired = await lock.acquire(file_path, timeout_seconds=10.0)
        if not acquired:
            return False, "Failed to acquire lock"
 
        try:
            # Step 2: Read and edit
            shard = await client.get_shard("files")
            files = shard.value or {}
 
            new_files = {**files}
            new_files[file_path] = new_content
 
            # Step 3: Commit
            await client.commit_shard("files", new_files, shard.version)
            return True, "Success"
 
        finally:
            # Step 4: ALWAYS release lock
            await lock.release(file_path)

Deadlock Prevention

When acquiring multiple locks, always acquire in sorted order:

python

async def acquire_multiple(agent_id: str, files: list[str]):
    """Acquire locks on multiple files without deadlock."""
 
    # CRITICAL: Sort to prevent circular wait
    sorted_files = sorted(files)
    acquired = []
 
    try:
        for file_path in sorted_files:
            if await lock.acquire(file_path):
                acquired.append(file_path)
            else:
                # Failed to get one lock, release all
                for f in acquired:
                    await lock.release(f)
                return False, []
 
        return True, acquired
 
    except Exception:
        for f in acquired:
            await lock.release(f)
        raise

Context Manager Pattern

For cleaner code:

python

class FileLocker:
    def __init__(self, gateway, agent_id, locks_shard, file_path, timeout=30.0):
        self.gateway = gateway
        self.agent_id = agent_id
        self.locks_shard = locks_shard
        self.file_path = file_path
        self.timeout = timeout
 
    async def __aenter__(self):
        self._client = ZeroLockClient(self.gateway, agent_id=self.agent_id)
        await self._client.__aenter__()
 
        self._lock = FileLock(self._client, self.locks_shard, self.agent_id)
        acquired = await self._lock.acquire(self.file_path, timeout_seconds=self.timeout)
 
        if not acquired:
            await self._client.__aexit__(None, None, None)
            raise TimeoutError(f"Failed to acquire lock on {self.file_path}")
 
        return self
 
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._lock.release(self.file_path)
        await self._client.__aexit__(exc_type, exc_val, exc_tb)
 
 
# Usage
async with FileLocker(GATEWAY, "agent-1", "locks", "config.py") as locker:
    # Lock acquired
    # ... do work ...
    # Lock automatically released on exit

Key Takeaways

What	Why
Always set lock timeout	Prevents orphaned locks if agent crashes
Use exponential backoff	Reduces contention when waiting
Sort before multi-lock	Prevents deadlock
Use try/finally for release	Ensures lock is always freed

Pattern 3: Approval Gates (Layer 4)

Location: examples/patterns/approval-gates/

When to Use It

Use approval gates when an operation is irreversible and needs human review:

Financial transactions
Production deployments
Sending emails/messages
Deleting data
Changing security settings

The Problem

Some operations should never auto-execute, even if the AI agent "decides" to:

plaintext

❌ Dangerous:
Agent: "Based on my analysis, I will now transfer $50,000 to..."
[Transfer executes immediately]
 
Even if the agent is wrong, the money is gone.

The Solution: Protocol-Level Approval

Layer 4 puts the gate in the protocol itself. The gateway refuses to execute commits on irreversible shards until an approver signs off:

plaintext

✓ With approval gate:
Agent: "I recommend transferring $50,000 to..."
Agent: → requests approval (cannot commit directly)
Gateway: → stores pending approval, notifies approvers
Risk Officer: → reviews, approves or rejects
Gateway: → commits (if approved) or blocks (if rejected)

Run It

bash

cd examples/patterns/approval-gates
python main.py

Implementation

Step 1: Create an irreversible shard

python

from zerolock_sdk import ZeroLockClient
from zerolock_sdk.types import CreateShardOptions
 
async with ZeroLockClient(GATEWAY, agent_id="admin") as client:
    await client.create_shard("transactions", CreateShardOptions(
        type="decision",
        initial_value={
            "balance": 100000,
            "transactions": []
        },
        reversible=False  # ← Layer 4: Requires approval
    ))

Step 2: Direct commits are blocked

python

# This will FAIL!
try:
    await client.commit_shard(
        "transactions",
        {"balance": 50000, ...},
        version
    )
except IrreversibleShardError:
    print("Cannot directly modify irreversible shard")

Step 3: Request approval instead

python

from zerolock_sdk.approvals import ApprovalClient
 
async def execute_transaction(transaction):
    async with ApprovalClient(GATEWAY, agent_id="executor") as approvals:
        # Get current state
        shard = await approvals.client.get_shard("transactions")
        current = shard.value
 
        # Prepare proposed new state
        new_state = {
            "balance": current["balance"] - transaction["amount"],
            "transactions": current["transactions"] + [transaction]
        }
 
        # Request approval (cannot commit directly!)
        approval = await approvals.request_approval(
            shard_id="transactions",
            proposed_value=new_state,
            expected_version=shard.version,
            description=f"Transfer ${transaction['amount']:,.2f} to {transaction['recipient']}",
            ttl_seconds=3600,  # 1 hour timeout
        )
 
        print(f"Approval requested: {approval.id}")
        print(f"Status: {approval.status}")  # "pending"
 
        # Wait for decision
        result = await approvals.wait_for_approval(
            approval.id,
            poll_interval_seconds=1,
            timeout_seconds=300,  # 5 minute max wait
        )
 
        if result.status == "approved":
            print("Transaction executed!")
            return True
        elif result.status == "rejected":
            print(f"Rejected: {result.rejection_reason}")
            return False
        else:
            print(f"Expired or error: {result.status}")
            return False

Step 4: Supervisor reviews and decides

python

async def supervisor_review():
    async with ApprovalClient(GATEWAY, agent_id="risk-officer") as approvals:
        # List pending approvals
        pending = await approvals.list_pending_approvals()
 
        for approval in pending:
            print(f"Reviewing: {approval.description}")
            print(f"  Proposed: {approval.proposed_value}")
 
            # Make decision (could involve human, could use another LLM)
            decision = await human_or_ai_review(approval)
 
            if decision == "approve":
                await approvals.approve(
                    approval.id,
                    reason="Approved by risk officer"
                )
                print("  ✓ Approved")
            else:
                await approvals.reject(
                    approval.id,
                    reason="Risk too high"
                )
                print("  ✗ Rejected")

Workflow Diagram

plaintext

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│  Executor   │         │   Gateway   │         │  Approver   │
└──────┬──────┘         └──────┬──────┘         └──────┬──────┘
       │                       │                       │
       │ request_approval()    │                       │
       │──────────────────────►│                       │
       │                       │                       │
       │   pending             │                       │
       │◄──────────────────────│                       │
       │                       │                       │
       │                       │  list_pending()       │
       │                       │◄──────────────────────│
       │                       │                       │
       │                       │  [approval details]   │
       │                       │──────────────────────►│
       │                       │                       │
       │                       │     approve()         │
       │                       │◄──────────────────────│
       │                       │                       │
       │  wait_for_approval()  │                       │
       │──────────────────────►│                       │
       │                       │                       │
       │  approved + committed │                       │
       │◄──────────────────────│                       │

Key Insight

The approval gate is in the protocol, not in agent code. The gateway enforces it. Even if an agent tries to bypass it:

python

# Agent tries to bypass approval...
await client.commit_shard("transactions", new_value, version)

The gateway rejects the commit with IrreversibleShardError. The agent cannot skip the approval process because it's enforced at the infrastructure level.

Key Takeaways

What	Why
`reversible=False`	Makes the shard require approvals
Use `request_approval()`	Only way to propose changes
Set `ttl_seconds`	Prevents stale approvals
Gateway enforces	Can't be bypassed by agent code

Summary

Pattern	Use Case	Core Mechanism
Task Claiming	One winner per task	Write-wait-reread
File Locking	Exclusive resource access	OCC lock shard
Approval Gates	Irreversible operations	Protocol-level gate

Next Steps

Quickstart Examples — Core OCC and CRDT patterns
Framework Integrations — Use patterns with LangGraph, CrewAI, etc.
Real-World Use Cases — See patterns combined in production scenarios