Chapter 32 — Distributed Agent Execution

Ten thousand order exceptions hit ShopOps before breakfast — O-1001 among them. Worker W1 leases the case, Outreach drafts Jordan’s email, then W1 crashes before acking the queue item. W7 redelivers the same work.

queue → lease(case_id) → worker runs → idempotency key on send → ack (or redeliver on expiry)

Without leases and idempotency, redelivery becomes a second customer message. Exactly-once is not available across crash domains.

This chapter adds distributed execution assumptions: per-case leases, partition by case_id, durable idempotency keys, DLQ for poison messages, and backpressure when the gateway saturates.

First principles

1. Exactly-once is not available across crash domains. Design for duplicates.
2. Lease ≠ lock forever. Expiry redelivers; handlers must tolerate it.
3. Partition by case_id. Two workers should not mutate the same case concurrently under a healthy lease.
4. Poison messages go to DLQ, not into infinite retry storms.
5. Backpressure is a feature. Shedding beats melting the model gateway.
6. Checkpoints and idempotency keys make redelivery safe (Ch 6, 8).

Concrete example — burst day

Queue depth spikes to 10k intake_build items. Workers W1–W20 compete. W1 leases O-1001. W1 dies after drafting email but before ack. Lease expires. W7 redelivers. Idempotency key O-1001:email:3 makes the channel adapter return the prior message_id — no double email. A poison payload with bad schema fails 5 times → DLQ → human replay after fix.

Diagram — competing consumers and leases

                  ┌── Worker W1 (lease O-1001 until T+30s)
 ready ──► queue ─┼── Worker W2 (lease A-1002 …)
                  └── Worker W3 (idle)

 lease expiry: item returns to ready ──► possible double delivery
 ack: item done
 nack x N: DLQ

stateDiagram-v2
  [*] --> Ready
  Ready --> Leased: acquire
  Leased --> Done: ack
  Leased --> Ready: lease expiry / nack
  Leased --> DLQ: max attempts
  DLQ --> Ready: manual replay

Caption: Notice expiry transitions back to Ready — that edge is why idempotency is mandatory.

Implementation

Module: shopops/worker/queue.py.

from shopops.worker.queue import InMemoryQueue

q = InMemoryQueue(lease_seconds=30, max_attempts=5)
q.enqueue(case_id="O-1001", tenant_id="store_northline",
          kind="contact_send", payload={"idem_key": "O-1001:email:3"})

def handler(item):
    # Must be safe if called twice
    send_email_idempotent(item.payload["idem_key"], ...)

assert q.process_one("worker-1", handler) == "ok"

# Double-delivery sketch
lease = q.acquire("worker-1")
# crash: no ack; freeze time past lease
q._expire_leases(now=lease.expires_at + 1)
lease2 = q.acquire("worker-2")
assert lease2.case_id == "O-1001"

DLQ replay:

# after poison fix
for item in q.dlq_items():
    q.replay_dlq(item.id)

Ordering sketch: InMemoryQueue refuses a second lease for the same case_id while one is held. That is weaker than a broker partition but teaches the invariant workers need.

Heartbeats for long model calls

lease = q.acquire("worker-1")
while generating:
    if not q.heartbeat(lease):
        raise RuntimeError("lost lease — stop side effects")
    # ... stream tokens / await tool ...
q.ack(lease)

If you skip heartbeats, another worker steals the case mid-send. Idempotency saves you from doubles; heartbeats save you from wasted work and confusing races.

Backpressure sketch

When gateway_p95_ms > budget or queue depth exceeds soft limit N, workers sleep or reduce concurrency instead of amplifying retries. Prefer shedding new bulk outreach over dropping approval resumes.

Failure modes

Production considerations

• Heartbeat during long model/tool calls so leases do not expire mid-flight.
• Separate queues by priority (approvals vs bulk outreach).
• Metrics: time-in-queue, attempts histogram, DLQ age, redelivery rate.
• Chaos: kill workers mid-lease in staging; assert no double customer contact.
• Backpressure: when gateway latency spikes, slow acquire rate.

Chapter summary

• Distributed agents need queues, leases, DLQs.
• At-least-once + idempotency is the real contract.
• Partition/lease per case_id for sane concurrency.
• Poison messages must stop retrying.
• Lease expiry implies duplicate delivery.
• Checkpoints and idempotency keys complete the story.
• Brokers are transport; correctness is application design.
• Burst capacity is a queue + worker pool problem, not a bigger prompt.

Exercises

1. Mechanical. Force lease expiry; show handler invoked twice; prove idempotent adapter sends once.
2. DLQ. Fail a handler until DLQ; replay; assert success path.
3. Design. Choose lease TTL for a worker that may call a 120s model — include heartbeat policy.

References

• Kleppmann, 2017. Designing Data-Intensive Applications — at-least-once, consumers.
• Queue product docs (SQS / Redis / Kafka) — lease/visibility timeouts. [VERIFY]
• Chapters 6, 8, 31