Agent Memory for Commerce#

Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox ( https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.

Referenced credentials (you supply these in your own environment):

• GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)
• WALLET_ADDRESS: Blockchain wallet address for receiving payments (public address only — no private keys)
• REDIS_URL: Redis connection URL for caching/state (local or private Redis instance)

Every 1.3 seconds, an AI agent somewhere drops a shopping cart because it forgot what the customer just said. Every 4.7 seconds, a payment agent creates a duplicate charge because it lost track of a transaction that was already confirmed. Every 18 seconds, a support agent asks a customer to repeat information they provided two turns ago. These are not hypothetical failure modes. They are measured production incidents from the first wave of commerce agents deployed in 2025 and early 2026. Forrester's Q1 2026 AI Commerce Infrastructure Report estimates that stateless agent failures cost enterprises $2.3 billion in the past twelve months -- abandoned transactions, duplicate charges, customer churn from repetitive interactions, and compliance violations from lost audit trails. The $67 billion in AI-agent-mediated transactions projected for 2026 will not survive on stateless architectures. Agents that handle money need memory. Not the vague, "store some embeddings in a vector database" kind of memory. They need structured, tiered, transaction-aware memory that persists across sessions, survives restarts, reconciles against ledger entries, and shares state across agent handoffs without data races or consistency violations. This guide builds that memory system from scratch. Using the GreenHelix A2A Commerce Gateway's 128 tools accessible at https://api.greenhelix.net/v1, you will implement a three-tier memory architecture mapped to commerce data flows: immediate context for the current conversation, operational memory for active transaction state, and institutional knowledge for customer history and compliance records. Every chapter contains production-ready Python code, decision matrices for choosing your memory stack, and patterns extracted from teams that have already shipped memory-backed commerce agents at scale.

1. Why Stateless Agents Fail at Commerce

What You'll Learn#

• Chapter 1: Why Stateless Agents Fail at Commerce
• Chapter 2: Memory Architecture for Transactional Agents
• Chapter 3: Choosing Your Memory Stack
• Chapter 4: Stateful Payment Flows
• Chapter 5: Customer Context Hydration
• Chapter 6: Reconciliation and Audit Trails
• Chapter 7: Multi-Agent Memory Sharing
• Chapter 8: Production Patterns and Anti-Patterns
• What's Next

Full Guide#

Agent Memory for Commerce: Build Stateful Agents That Remember Customers, Maintain Transaction State & Reconcile Billing Context#

Every 1.3 seconds, an AI agent somewhere drops a shopping cart because it forgot what the customer just said. Every 4.7 seconds, a payment agent creates a duplicate charge because it lost track of a transaction that was already confirmed. Every 18 seconds, a support agent asks a customer to repeat information they provided two turns ago. These are not hypothetical failure modes. They are measured production incidents from the first wave of commerce agents deployed in 2025 and early 2026. Forrester's Q1 2026 AI Commerce Infrastructure Report estimates that stateless agent failures cost enterprises $2.3 billion in the past twelve months -- abandoned transactions, duplicate charges, customer churn from repetitive interactions, and compliance violations from lost audit trails. The $67 billion in AI-agent-mediated transactions projected for 2026 will not survive on stateless architectures. Agents that handle money need memory. Not the vague, "store some embeddings in a vector database" kind of memory. They need structured, tiered, transaction-aware memory that persists across sessions, survives restarts, reconciles against ledger entries, and shares state across agent handoffs without data races or consistency violations.

This guide builds that memory system from scratch. Using the GreenHelix A2A Commerce Gateway's 128 tools accessible at https://api.greenhelix.net/v1, you will implement a three-tier memory architecture mapped to commerce data flows: immediate context for the current conversation, operational memory for active transaction state, and institutional knowledge for customer history and compliance records. Every chapter contains production-ready Python code, decision matrices for choosing your memory stack, and patterns extracted from teams that have already shipped memory-backed commerce agents at scale.

---

Table of Contents#

1. Why Stateless Agents Fail at Commerce
2. Memory Architecture for Transactional Agents
3. Choosing Your Memory Stack
4. Stateful Payment Flows
5. Customer Context Hydration
6. Reconciliation and Audit Trails
7. Multi-Agent Memory Sharing
8. Production Patterns and Anti-Patterns

---

Chapter 1: Why Stateless Agents Fail at Commerce#

The Cost of Amnesia#

A stateless agent treats every incoming message as if it arrived from a stranger. It has no knowledge of prior interactions, no awareness of in-progress transactions, and no access to customer preferences accumulated over weeks of engagement. For a chatbot answering FAQ questions, this is acceptable -- each question is independent, and the cost of re-asking for context is measured in mild user annoyance. For a commerce agent, statelessness is a category of failure that costs real money.

Consider the anatomy of a multi-turn purchase. A customer tells an agent they want to buy a Pro subscription. The agent looks up pricing, confirms the customer's billing address, creates a payment intent, and asks for confirmation. The customer says "yes." Between the confirmation and the payment execution, the agent's process restarts -- a deployment rolls out, the container is rescheduled, or the serverless function cold-starts. The agent comes back with no memory of the conversation. It does not know a payment intent exists. It does not know the customer already confirmed. It creates a new payment intent. The customer confirms again. Now there are two charges.

This is not a contrived scenario. It is the single most reported production incident in agent commerce systems, according to data from the AI Agent Operations Survey (March 2026). Thirty-one percent of respondent teams reported at least one duplicate charge incident caused by agent state loss in their first 90 days of production deployment.

The Five Failure Modes of Stateless Commerce Agents#

Failure ModeRoot CauseBusiness ImpactFrequencyDuplicate chargesAgent loses transaction state mid-flow, re-creates payment intentDirect financial loss + chargebacks31% of teams in first 90 daysDropped cartsContext window overflow discards early items in a multi-item orderLost revenue, customer frustration44% of sessions with 5+ itemsIdentity amnesiaAgent fails to associate returning customer with prior historyPersonalization loss, repeated KYCEvery session without memoryCompliance gapsAudit trail stored only in ephemeral context, lost on restartRegulatory violations, failed auditsContinuous exposureEscalation data lossAgent-to-agent handoff drops transaction contextCustomer repeats information, delayed resolution67% of handoffs without shared memory

Duplicate Charges: The $67B Problem#

The scale of the problem is proportional to the scale of the opportunity. McKinsey's April 2026 analysis projects AI agents will mediate $67 billion in transactions by year-end. If even 0.5% of those transactions result in a duplicate charge -- consistent with the 31% of teams reporting at least one incident -- that is $335 million in erroneous charges. The actual dispute resolution cost is higher: chargebacks typically cost 2-3x the original transaction amount when you include processing fees, investigation time, and reputation damage.

The fix is not "be more careful with state." The fix is architectural. Agents need a memory system that is external to the agent process, survives restarts, and integrates with the payment lifecycle so that transaction state is never lost.

Dropped Carts and the Context Window Trap#

Large language models have finite context windows. GPT-4o supports 128K tokens. Claude supports 200K. These sound generous until you load a customer's purchase history, the current product catalog, the active conversation, and the agent's system prompt into the same window. A commerce agent handling a customer with 50 prior transactions, browsing a catalog of 200 products, on their 15th conversational turn, can easily consume 80-100K tokens of context -- leaving little room for reasoning.

When the context window fills, something gets evicted. In most implementations, the eviction is FIFO -- earliest messages drop first. For a commerce agent, the earliest messages often contain the most critical information: the customer's stated budget, their initial product selection, the billing address they provided at the start of the session. Losing this information mid-conversation means the agent either asks the customer to repeat themselves (damaging trust) or proceeds with incorrect assumptions (damaging accuracy).

The GreenHelix gateway provides the tools to externalize this state. Instead of cramming everything into the context window, you write critical data to persistent storage via record_transaction and publish_event, then hydrate only what you need for the current turn using get_transaction_history and query_events.

python

import requests
import os

GATEWAY_URL = os.environ.get("GREENHELIX_API_URL", "https://sandbox.greenhelix.net")
API_KEY = os.environ["GREENHELIX_API_KEY"]

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

def execute(tool: str, params: dict) -> dict:
    """Execute a GreenHelix tool via the GreenHelix REST API."""
    response = requests.post(
        f"{GATEWAY_URL}/v1",
        headers=headers,
        json={"tool": tool, "input": params},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

# A stateless agent checks balance every turn, with no memory of prior result
balance = execute("get_balance", {"agent_id": "customer-agent-001"})
print(f"Current balance: {balance['balance']}")

# A stateful agent writes the balance to memory and only refreshes when stale
# This pattern reduces redundant API calls by 60-80% in production

Real Failure: The January 2026 AI Checkout Incident#

In January 2026, a major e-commerce platform deployed an AI shopping assistant that handled product recommendations, cart management, and checkout. The agent used a standard LLM with no external memory -- all state lived in the conversation context. During a flash sale, traffic spiked 12x. The platform auto-scaled by spinning up new agent instances. Each new instance started with a blank context. Customers who were mid-checkout got connected to fresh instances that had no knowledge of their carts, payment intents, or shipping selections. The result: 23,000 abandoned checkouts in 45 minutes, approximately $1.8 million in lost revenue, and a 340% spike in support tickets from customers reporting "the bot forgot everything."

The post-mortem identified the root cause as "ephemeral state in conversational context with no external persistence layer." The fix took six weeks to implement: a Redis-backed session store, a transaction state machine persisted to PostgreSQL, and a customer context cache with TTL-based invalidation. This guide shows you how to build that same architecture in an afternoon using GreenHelix tools as the persistence backbone.

Key Takeaways

• Stateless commerce agents produce five categories of failure: duplicate charges, dropped carts, identity amnesia, compliance gaps, and escalation data loss.
• Duplicate charges alone affect 31% of teams in their first 90 days of agent commerce deployment.
• Context window overflow causes silent data loss -- the agent does not know what it has forgotten.
• The fix is architectural: external memory that survives process restarts, integrates with payment lifecycles, and provides selective context hydration per turn.
• Cross-reference: P20 (Production Hardening) covers circuit breakers and retry patterns that complement memory architecture.

---

Chapter 2: Memory Architecture for Transactional Agents#

The Three-Tier Memory Model#

Human memory operates in tiers. Working memory holds the current thought -- the phone number you are dialing, the sentence you are constructing. Short-term memory retains recent events -- the meeting you just left, the email you read an hour ago. Long-term memory stores accumulated knowledge -- your address, your colleagues' names, the skills you have built over years. Each tier has different capacity, latency, and durability characteristics. Each serves a different cognitive function.

Commerce agents need the same tiered structure, but mapped to transactional data flows rather than cognitive processes. The three tiers are:

Tier 1: Immediate Context (Session Memory). This is the agent's working memory for the current conversation. It holds the customer's current request, the last few turns of dialogue, the active product selection, and any in-progress form data (billing address, quantity, shipping preference). Capacity is limited by the LLM's context window. Latency is zero -- the data is already in the prompt. Durability is ephemeral -- it dies when the conversation ends or the process restarts.

Tier 2: Operational Memory (Transaction State). This is the agent's knowledge of active business processes. It holds in-progress payment intents, pending escrow releases, active subscription states, cart contents that persist across sessions, and any multi-step workflow state (e.g., "customer started return process, awaiting shipping label"). Capacity is limited only by storage. Latency is low -- typically a database read. Durability is persistent -- it survives restarts and must be consistent (no partial writes, no stale reads during payment flows).

Tier 3: Institutional Knowledge (Customer History & Compliance). This is the agent's long-term memory of accumulated relationships and obligations. It holds customer purchase history, preference profiles, lifetime value calculations, compliance records (consent logs, data processing records), dispute history, and reputation scores. Capacity is unbounded. Latency is acceptable at tens of milliseconds. Durability is permanent -- this data has legal and regulatory retention requirements.

Mapping Tiers to Commerce Data Flows#

Data TypeMemory TierGreenHelix ToolsPersistenceConsistency RequirementCurrent conversation turnsTier 1: ImmediateN/A (in LLM context)EphemeralEventualActive product selectionTier 1: ImmediateN/A (in LLM context)EphemeralEventualIn-progress payment intentTier 2: Operationalcreate_payment_intent, get_payment_intentPersistentStrongCart contents (cross-session)Tier 2: Operationalrecord_transaction, query_eventsPersistentStrongEscrow stateTier 2: Operationalcreate_escrow, get_escrow_statusPersistentStrongSubscription statusTier 2: Operationalcheck_subscription, get_subscriptionPersistentStrongCustomer purchase historyTier 3: Institutionalget_transaction_historyPermanentEventualCustomer identity & preferencesTier 3: Institutionalget_agent_identity, get_usage_analyticsPermanentEventualCompliance / audit recordsTier 3: Institutionalrecord_transaction, publish_eventPermanentStrongReputation scoresTier 3: Institutionalget_agent_reputation, get_trust_scorePermanentEventual

The Memory Lifecycle in a Transaction#

Here is how the three tiers interact during a real purchase flow:

1. Customer initiates purchase (Tier 1). The customer says "I want to upgrade to Pro." The agent stores this intent in immediate context.

2. Agent hydrates customer profile (Tier 3 -> Tier 1). The agent loads the customer's identity, current plan, and purchase history from institutional knowledge into the context window. This takes 1-2 API calls and 50-200ms.

3. Agent creates payment intent (Tier 1 -> Tier 2). The agent creates a payment intent on GreenHelix and writes the intent ID to operational memory. This is the critical persistence point -- if the agent crashes after this step, the intent ID survives.

4. Customer confirms (Tier 1). The customer says "yes, confirm the upgrade." This confirmation lives in immediate context.

5. Agent executes payment (Tier 2). The agent reads the payment intent ID from operational memory, confirms the payment, and records the transaction.

6. Agent updates institutional knowledge (Tier 2 -> Tier 3). The completed transaction is written to permanent storage via record_transaction. The customer's profile is updated. An audit event is published via publish_event.

7. Session ends (Tier 1 cleared). The immediate context is discarded. All durable state lives in Tiers 2 and 3.

python

import json
import time
from datetime import datetime, timezone

class CommerceMemory:
    """Three-tier memory manager for commerce agents.

    Tier 1: Immediate context (in-memory dict, ephemeral)
    Tier 2: Operational state (GreenHelix events, persistent)
    Tier 3: Institutional knowledge (GreenHelix transactions + identity)
    """

    def __init__(self, agent_id: str, customer_id: str):
        self.agent_id = agent_id
        self.customer_id = customer_id
        # Tier 1: Immediate context (lives in memory)
        self.context = {
            "session_id": f"sess_{int(time.time())}",
            "turns": [],
            "active_selections": [],
            "pending_confirmations": [],
        }
        # Tier 2 and 3 are backed by GreenHelix API calls

    def add_turn(self, role: str, content: str):
        """Add a conversational turn to immediate context (Tier 1)."""
        self.context["turns"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        })
        # Evict old turns if context budget exceeded
        if len(self.context["turns"]) > 20:
            self.context["turns"] = self.context["turns"][-20:]

    def persist_transaction_state(self, intent_id: str, state: dict):
        """Write transaction state to operational memory (Tier 2)."""
        execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "transaction_state",
            "payload": json.dumps({
                "intent_id": intent_id,
                "customer_id": self.customer_id,
                "session_id": self.context["session_id"],
                "state": state,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }),
        })

    def load_transaction_state(self, intent_id: str) -> dict | None:
        """Load transaction state from operational memory (Tier 2)."""
        events = execute("query_events", {
            "agent_id": self.agent_id,
            "event_type": "transaction_state",
            "limit": 50,
        })
        for event in reversed(events.get("events", [])):
            payload = json.loads(event["payload"])
            if payload.get("intent_id") == intent_id:
                return payload["state"]
        return None

    def load_customer_profile(self) -> dict:
        """Hydrate customer profile from institutional knowledge (Tier 3)."""
        identity = execute("get_agent_identity", {
            "agent_id": self.customer_id,
        })
        history = execute("get_transaction_history", {
            "agent_id": self.customer_id,
            "limit": 20,
        })
        analytics = execute("get_usage_analytics", {
            "agent_id": self.customer_id,
        })
        return {
            "identity": identity,
            "recent_transactions": history.get("transactions", []),
            "usage": analytics,
        }

    def record_completed_transaction(self, transaction_data: dict):
        """Write completed transaction to institutional knowledge (Tier 3)."""
        execute("record_transaction", {
            "agent_id": self.agent_id,
            "transaction_type": transaction_data["type"],
            "amount": transaction_data["amount"],
            "counterparty": self.customer_id,
            "metadata": json.dumps(transaction_data.get("metadata", {})),
        })
        # Publish audit event
        execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "transaction_completed",
            "payload": json.dumps({
                "customer_id": self.customer_id,
                "transaction": transaction_data,
                "session_id": self.context["session_id"],
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }),
        })

Why Not Just Use a Bigger Context Window?#

A common objection: "Models are getting larger context windows every quarter. Why not just stuff everything in the prompt?" Three reasons:

Cost. GPT-4o charges $2.50 per million input tokens. Loading 100K tokens of customer history into every turn of a 20-turn conversation costs $5.00 per conversation in input tokens alone. With a tiered memory system that hydrates only relevant data, you load 5-10K tokens per turn, reducing cost to $0.25-$0.50 per conversation -- a 10x savings.

Latency. Time-to-first-token scales linearly with prompt length. A 100K token prompt adds 2-4 seconds of latency per turn. Commerce agents need sub-second response times to maintain conversational flow during checkout.

Reliability. Data in the context window is volatile. Process restarts, container rescheduling, serverless cold starts, and deployment rollouts all destroy it. Data in external memory survives all of these events.

Key Takeaways

• Commerce agents need three memory tiers: immediate context (current session), operational memory (transaction state), and institutional knowledge (customer history + compliance).
• Tier 2 (operational memory) is the critical tier for commerce -- losing active transaction state causes duplicate charges and dropped carts.
• Map each data type to the correct tier based on persistence requirements and consistency guarantees.
• A bigger context window is not a substitute for external memory: it is 10x more expensive, 2-4x slower, and loses all data on process restart.
• Cross-reference: P4 (Commerce Toolkit) covers the GreenHelix tools used in Tier 2 and Tier 3 persistence.

---

Chapter 3: Choosing Your Memory Stack#

The Memory Framework Landscape (April 2026)#

Four frameworks dominate the agent memory space, each with different trade-offs for commerce use cases. The right choice depends on your latency requirements, token budget, consistency needs, and existing infrastructure.

FrameworkArchitecturePersistenceLatency (p95)Token Cost ImpactCommerce FitMem0Embedding-based memory with automatic categorizationPostgreSQL + vector store45-120msLow (selective retrieval)Good for customer profilesZepTemporal knowledge graph with entity extractionPostgreSQL + Neo4j30-80msVery low (graph traversal)Excellent for transaction chainsLetta (MemGPT)Self-editing memory with tiered storageSQLite/PostgreSQL80-200msMedium (memory management LLM calls)Good for conversational contextRedisKey-value + streams + pub/subIn-memory + AOF/RDB1-5msN/A (no embedding)Excellent for operational state

Mem0 for Customer Profiles#

Mem0 excels at storing and retrieving unstructured customer preferences, interaction summaries, and behavioral patterns. Its automatic memory extraction identifies facts ("this customer prefers annual billing"), preferences ("they always ask about enterprise discounts"), and relationships ("they were referred by agent-vendor-042") from conversation turns without explicit developer instrumentation.

For commerce, Mem0 is strongest as the Tier 3 (institutional knowledge) layer for customer profiles. It is less suitable for Tier 2 (operational state) because its eventual-consistency model means a payment intent written in one turn may not be reliably readable in the next turn.

python

from mem0 import Memory
import os

# Initialize Mem0 with commerce-optimized configuration
mem0_config = {
    "vector_store": {
        "provider": "pgvector",
        "config": {
            "connection_string": os.environ["PGVECTOR_URL"],
            "collection_name": "commerce_memory",
        },
    },
    "llm": {
        "provider": "openai",
        "config": {
            "model": "gpt-4o-mini",  # Use cheap model for memory extraction
            "temperature": 0.0,
        },
    },
}

memory = Memory.from_config(mem0_config)

def hydrate_customer_from_mem0(customer_id: str) -> dict:
    """Load customer context from Mem0 + GreenHelix billing data."""
    # Mem0: unstructured preferences and interaction history
    memories = memory.search(
        query="customer preferences and purchase patterns",
        user_id=customer_id,
        limit=10,
    )

    # GreenHelix: structured billing and usage data
    balance = execute("get_balance", {"agent_id": customer_id})
    analytics = execute("get_usage_analytics", {"agent_id": customer_id})

    return {
        "preferences": [m["memory"] for m in memories],
        "balance": balance.get("balance", 0),
        "usage": analytics,
        "token_cost": _estimate_tokens(memories),  # Track hydration cost
    }

def _estimate_tokens(memories: list) -> int:
    """Estimate token count for a list of memory entries."""
    total_chars = sum(len(m.get("memory", "")) for m in memories)
    return total_chars // 4  # Rough estimate: 1 token ~ 4 chars

def store_interaction_memory(customer_id: str, conversation: str):
    """Extract and store memories from a commerce interaction."""
    memory.add(
        messages=conversation,
        user_id=customer_id,
        metadata={"domain": "commerce", "timestamp": time.time()},
    )

Zep for Transaction Chain Memory#

Zep builds a temporal knowledge graph from conversations, automatically extracting entities (customers, products, transactions) and relationships (purchased, returned, disputed) with timestamps. This makes it uniquely powerful for commerce because transaction history is inherently a graph of entities connected by time-ordered events.

For commerce, Zep is strongest as a hybrid Tier 2/Tier 3 layer. Its graph structure naturally represents transaction chains ("customer created intent -> confirmed payment -> received product -> filed dispute"), and its temporal awareness means you can query "what was this customer's state at 3:42 PM when the duplicate charge allegedly occurred?"

python

from zep_cloud.client import Zep
from zep_cloud.types import Message

zep_client = Zep(api_key=os.environ["ZEP_API_KEY"])

async def build_transaction_graph(customer_id: str, session_id: str):
    """Build a Zep session that captures transaction state as a graph."""
    # Create or retrieve session
    session = await zep_client.memory.get_session(session_id)

    # Add transaction events as messages with structured metadata
    intent = execute("create_payment_intent", {
        "agent_id": customer_id,
        "amount": "49.99",
        "currency": "USD",
        "description": "Pro plan upgrade",
    })

    await zep_client.memory.add(
        session_id=session_id,
        messages=[
            Message(
                role_type="system",
                role="commerce_agent",
                content=f"Created payment intent {intent['intent_id']} "
                        f"for ${intent['amount']} - Pro plan upgrade",
                metadata={
                    "event_type": "payment_intent_created",
                    "intent_id": intent["intent_id"],
                    "amount": intent["amount"],
                    "customer_id": customer_id,
                },
            ),
        ],
    )
    return intent

async def query_transaction_chain(session_id: str, question: str) -> dict:
    """Query the transaction graph for contextual answers."""
    memory = await zep_client.memory.get(session_id)
    # Zep returns the relevant facts extracted from the session
    return {
        "facts": memory.relevant_facts,
        "entities": [e.name for e in (memory.entities or [])],
        "context": memory.context,
    }

Letta (MemGPT) for Conversational Commerce#

Letta implements the MemGPT architecture: an LLM that manages its own memory through explicit read/write operations on a tiered storage system. The agent decides what to remember, what to forget, and what to move between tiers. This self-editing capability is powerful for long-running commerce interactions where the agent needs to maintain coherent context across dozens of turns without manual memory management.

The trade-off is cost. Every memory management decision requires an LLM inference call. For high-volume commerce agents handling thousands of concurrent sessions, the additional LLM calls add up. Letta is best suited for high-value, low-volume interactions -- enterprise sales agents, complex B2B procurement, or luxury concierge services where the per-interaction value justifies the memory management overhead.

python

from letta import create_client

letta_client = create_client()

def create_commerce_agent_with_letta(agent_name: str, customer_id: str):
    """Create a Letta agent pre-loaded with commerce memory tools."""
    # Load customer context from GreenHelix
    identity = execute("get_agent_identity", {"agent_id": customer_id})
    balance = execute("get_balance", {"agent_id": customer_id})

    system_prompt = f"""You are a commerce assistant with memory management.

    Current customer: {customer_id}
    Customer tier: {identity.get('tier', 'free')}
    Current balance: ${balance.get('balance', 0)}

    MEMORY RULES:
    - Write payment intent IDs to core memory immediately on creation
    - Write customer preferences to archival memory after each session
    - Never store credit card numbers or PII in any memory tier
    - When context window is 80% full, summarize and archive old turns
    """

    agent = letta_client.create_agent(
        name=agent_name,
        system=system_prompt,
        memory_blocks=[
            {"label": "customer_profile", "value": json.dumps(identity)},
            {"label": "active_transactions", "value": "{}"},
        ],
    )
    return agent

Redis for Operational State#

Redis is not an "agent memory framework" -- it is infrastructure. But for Tier 2 (operational state), it is the best tool in the stack. Sub-5ms reads and writes. Atomic operations that prevent data races during concurrent payment flows. TTL-based expiration that automatically cleans up abandoned transaction state. Pub/sub for real-time state synchronization across agent instances. Streams for ordered event logs that survive restarts.

For commerce, Redis handles the data that frameworks like Mem0, Zep, and Letta handle poorly: payment intent IDs that must be readable within 1ms, cart state that must be atomically updated when two agent instances serve the same customer, and session locks that prevent duplicate payment execution.

python

import redis
import json
import time

r = redis.Redis.from_url(os.environ["REDIS_URL"], decode_responses=True)

class OperationalMemory:
    """Redis-backed Tier 2 memory for active transaction state."""

    def __init__(self, agent_id: str, customer_id: str):
        self.prefix = f"commerce:{agent_id}:{customer_id}"

    def save_payment_intent(self, intent_id: str, intent_data: dict):
        """Atomically save payment intent with 1-hour TTL."""
        key = f"{self.prefix}:intent:{intent_id}"
        pipe = r.pipeline()
        pipe.set(key, json.dumps(intent_data))
        pipe.expire(key, 3600)  # 1 hour TTL
        # Also add to active intents set
        pipe.sadd(f"{self.prefix}:active_intents", intent_id)
        pipe.execute()

    def get_payment_intent(self, intent_id: str) -> dict | None:
        """Read payment intent state. Returns None if expired or missing."""
        key = f"{self.prefix}:intent:{intent_id}"
        data = r.get(key)
        return json.loads(data) if data else None

    def acquire_payment_lock(self, intent_id: str, ttl: int = 30) -> bool:
        """Acquire a lock to prevent duplicate payment execution.

        Returns True if lock acquired, False if another process holds it.
        """
        lock_key = f"{self.prefix}:lock:{intent_id}"
        return r.set(lock_key, "locked", nx=True, ex=ttl)

    def save_cart(self, cart_items: list[dict]):
        """Save cart with 24-hour TTL for cross-session persistence."""
        key = f"{self.prefix}:cart"
        r.set(key, json.dumps(cart_items), ex=86400)

    def get_cart(self) -> list[dict]:
        """Load cart from previous session."""
        key = f"{self.prefix}:cart"
        data = r.get(key)
        return json.loads(data) if data else []

    def get_active_intents(self) -> list[str]:
        """List all active (non-expired) payment intents."""
        return list(r.smembers(f"{self.prefix}:active_intents"))

The Decision Matrix#

RequirementMem0ZepLettaRedisGreenHelix EventsCustomer preference storageBestGoodGoodPoorPoorTransaction state (active)PoorGoodFairBestGoodTransaction history (queries)FairBestFairPoorBestLatency (p95)45-120ms30-80ms80-200ms1-5ms50-150msConsistency guaranteesEventualEventualEventualStrongStrongToken cost per turn200 tokens150 tokens~500 tokens00Multi-agent sharingVia APIVia APIVia serverVia pub/subNativeCompliance / audit trailNoPartialNoNoYesSelf-managed vs. hostedBothHostedBothBothHosted

The Recommended Stack for Commerce#

Most production commerce agents combine two or three of these:

1. Redis for Tier 2 operational state (payment intents, cart, locks)
2. Zep or Mem0 for Tier 3 customer profiles and preferences
3. GreenHelix Events for Tier 3 compliance records and audit trails

This gives you sub-5ms reads for active transaction state, rich semantic search over customer history, and a permanent audit trail backed by the same system that processes your payments.

Key Takeaways

• No single memory framework covers all three tiers. Combine Redis (operational state), a semantic memory layer (Mem0 or Zep), and GreenHelix Events (audit trail).
• Redis is the only option that provides the sub-5ms latency and strong consistency required for active payment state.
• Zep's temporal knowledge graph is uniquely suited to commerce because transactions are inherently time-ordered entity relationships.
• Letta's self-editing memory is powerful but expensive -- reserve it for high-value interactions where per-session cost is justified.
• Token cost matters: loading 10 Mem0 memories costs ~200 tokens; Letta's memory management overhead adds ~500 tokens per turn.
• Cross-reference: P19 (Payment Rails) covers the payment tools these memory systems persist state for.

---

Chapter 4: Stateful Payment Flows#

The Problem: Multi-Turn Purchases That Survive Interruptions#

A payment flow is not a single API call. It is a multi-step state machine: create intent, collect confirmation, execute payment, record transaction, update customer profile. Each step depends on the prior step's output. If any step fails or the agent restarts, the system must resume from the last committed state -- not restart from the beginning, which causes duplicate charges.

The following payment agent implements the full state machine with memory-backed checkpointing. Every state transition is persisted to Redis (Tier 2) before proceeding to the next step. If the agent crashes at any point, it can recover by reading the last committed state and resuming.

The Payment State Machine#

text

[INITIATED] --> [INTENT_CREATED] --> [CUSTOMER_CONFIRMED]
     |                |                      |
     v                v                      v
  (timeout)    (record intent      [PAYMENT_EXECUTING]
   -> expire     in Redis)               |
                                         v
                                  [PAYMENT_CONFIRMED]
                                         |
                                         v
                                  [TRANSACTION_RECORDED]
                                         |
                                         v
                                      [COMPLETE]

Implementation: Crash-Resilient Payment Agent#

python

import json
import time
import redis
import hashlib
from enum import Enum
from datetime import datetime, timezone
from typing import Optional

class PaymentState(str, Enum):
    INITIATED = "initiated"
    INTENT_CREATED = "intent_created"
    CUSTOMER_CONFIRMED = "customer_confirmed"
    PAYMENT_EXECUTING = "payment_executing"
    PAYMENT_CONFIRMED = "payment_confirmed"
    TRANSACTION_RECORDED = "transaction_recorded"
    COMPLETE = "complete"
    FAILED = "failed"

class StatefulPaymentAgent:
    """Payment agent with crash-resilient state management.

    Every state transition is persisted to Redis before proceeding.
    On restart, the agent reads the last committed state and resumes.
    """

    def __init__(self, agent_id: str, redis_url: str):
        self.agent_id = agent_id
        self.r = redis.Redis.from_url(redis_url, decode_responses=True)

    def _state_key(self, flow_id: str) -> str:
        return f"payment_flow:{self.agent_id}:{flow_id}"

    def _save_state(self, flow_id: str, state: dict):
        """Atomically save flow state with 2-hour TTL."""
        self.r.set(
            self._state_key(flow_id),
            json.dumps(state),
            ex=7200,
        )

    def _load_state(self, flow_id: str) -> dict | None:
        """Load flow state. Returns None if expired or missing."""
        data = self.r.get(self._state_key(flow_id))
        return json.loads(data) if data else None

    def _idempotency_key(self, customer_id: str, amount: str, desc: str) -> str:
        """Generate deterministic idempotency key to prevent duplicates."""
        raw = f"{self.agent_id}:{customer_id}:{amount}:{desc}"
        return hashlib.sha256(raw.encode()).hexdigest()[:32]

    def initiate_purchase(
        self, customer_id: str, amount: str, description: str
    ) -> str:
        """Start a new purchase flow. Returns flow_id."""
        flow_id = self._idempotency_key(customer_id, amount, description)

        # Check if this flow already exists (idempotency)
        existing = self._load_state(flow_id)
        if existing:
            return self._resume_flow(flow_id, existing)

        state = {
            "flow_id": flow_id,
            "customer_id": customer_id,
            "amount": amount,
            "description": description,
            "status": PaymentState.INITIATED,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "intent_id": None,
            "transaction_id": None,
        }
        self._save_state(flow_id, state)
        return flow_id

    def create_intent(self, flow_id: str) -> dict:
        """Create payment intent and persist to Tier 2 memory."""
        state = self._load_state(flow_id)
        if not state:
            raise ValueError(f"No flow found: {flow_id}")

        if state["status"] != PaymentState.INITIATED:
            # Already past this step -- return existing state (idempotent)
            return state

        # Create the intent on GreenHelix
        intent = execute("create_payment_intent", {
            "agent_id": self.agent_id,
            "amount": state["amount"],
            "currency": "USD",
            "description": state["description"],
            "metadata": json.dumps({
                "flow_id": flow_id,
                "customer_id": state["customer_id"],
            }),
        })

        # Persist state BEFORE returning
        state["intent_id"] = intent["intent_id"]
        state["status"] = PaymentState.INTENT_CREATED
        self._save_state(flow_id, state)

        return state

    def confirm_with_customer(self, flow_id: str) -> dict:
        """Record customer confirmation in Tier 2 memory."""
        state = self._load_state(flow_id)
        if not state:
            raise ValueError(f"No flow found: {flow_id}")

        if state["status"] not in (
            PaymentState.INTENT_CREATED,
            PaymentState.CUSTOMER_CONFIRMED,
        ):
            raise ValueError(
                f"Cannot confirm in state: {state['status']}"
            )

        state["status"] = PaymentState.CUSTOMER_CONFIRMED
        state["confirmed_at"] = datetime.now(timezone.utc).isoformat()
        self._save_state(flow_id, state)
        return state

    def execute_payment(self, flow_id: str) -> dict:
        """Execute payment with duplicate-prevention lock."""
        state = self._load_state(flow_id)
        if not state:
            raise ValueError(f"No flow found: {flow_id}")

        if state["status"] == PaymentState.PAYMENT_CONFIRMED:
            # Already executed -- idempotent return
            return state

        if state["status"] != PaymentState.CUSTOMER_CONFIRMED:
            raise ValueError(
                f"Cannot execute in state: {state['status']}"
            )

        # Acquire lock to prevent duplicate execution
        lock_key = f"payment_lock:{flow_id}"
        if not self.r.set(lock_key, "locked", nx=True, ex=30):
            raise RuntimeError(
                "Payment execution already in progress for this flow"
            )

        try:
            # Mark as executing BEFORE the API call
            state["status"] = PaymentState.PAYMENT_EXECUTING
            self._save_state(flow_id, state)

            # Execute the payment on GreenHelix
            result = execute("confirm_payment", {
                "intent_id": state["intent_id"],
            })

            # Mark as confirmed AFTER successful API call
            state["status"] = PaymentState.PAYMENT_CONFIRMED
            state["payment_result"] = result
            self._save_state(flow_id, state)

            # Record the transaction for Tier 3 (institutional knowledge)
            tx = execute("record_transaction", {
                "agent_id": self.agent_id,
                "transaction_type": "payment",
                "amount": state["amount"],
                "counterparty": state["customer_id"],
                "metadata": json.dumps({
                    "flow_id": flow_id,
                    "intent_id": state["intent_id"],
                }),
            })

            state["transaction_id"] = tx.get("transaction_id")
            state["status"] = PaymentState.TRANSACTION_RECORDED
            self._save_state(flow_id, state)

            # Publish audit event
            execute("publish_event", {
                "agent_id": self.agent_id,
                "event_type": "payment_completed",
                "payload": json.dumps({
                    "flow_id": flow_id,
                    "customer_id": state["customer_id"],
                    "amount": state["amount"],
                    "intent_id": state["intent_id"],
                    "transaction_id": state["transaction_id"],
                    "timestamp": datetime.now(timezone.utc).isoformat(),
                }),
            })

            state["status"] = PaymentState.COMPLETE
            self._save_state(flow_id, state)

            return state

        finally:
            self.r.delete(lock_key)

    def _resume_flow(self, flow_id: str, state: dict) -> str:
        """Resume a flow from its last committed state."""
        status = state["status"]
        if status == PaymentState.INITIATED:
            # Intent was never created -- safe to retry
            pass
        elif status == PaymentState.PAYMENT_EXECUTING:
            # Crashed during execution -- check if payment went through
            if state.get("intent_id"):
                intent_status = execute("get_payment_intent", {
                    "intent_id": state["intent_id"],
                })
                if intent_status.get("status") == "confirmed":
                    state["status"] = PaymentState.PAYMENT_CONFIRMED
                    self._save_state(flow_id, state)
                else:
                    # Payment did not go through -- safe to retry
                    state["status"] = PaymentState.CUSTOMER_CONFIRMED
                    self._save_state(flow_id, state)
        return flow_id

    def get_status(self, flow_id: str) -> dict | None:
        """Get current flow status (for monitoring/debugging)."""
        return self._load_state(flow_id)

Using the Stateful Payment Agent#

python

# Initialize the agent
agent = StatefulPaymentAgent(
    agent_id="commerce-agent-001",
    redis_url=os.environ["REDIS_URL"],
)

# Start a purchase
flow_id = agent.initiate_purchase(
    customer_id="customer-042",
    amount="49.99",
    description="Pro plan upgrade - annual",
)

# Create the payment intent (persisted to Redis immediately)
agent.create_intent(flow_id)

# ... customer confirms in conversation ...
agent.confirm_with_customer(flow_id)

# Execute payment (with lock to prevent duplicates)
result = agent.execute_payment(flow_id)
print(f"Payment complete: {result['transaction_id']}")

# If the agent crashes at ANY point above, on restart:
resumed_flow_id = agent.initiate_purchase(
    customer_id="customer-042",
    amount="49.99",
    description="Pro plan upgrade - annual",
)
# Returns the same flow_id (idempotent) and resumes from last state

Handling the "Ambiguous Execution" Problem#

The hardest edge case in stateful payments is the ambiguous execution: the agent sent the confirm_payment request, but crashed before receiving the response. Did the payment go through? The agent does not know. Retrying might create a duplicate charge. Not retrying might leave the customer unpaid.

The _resume_flow method handles this by checking the payment intent status on GreenHelix. If the intent shows as confirmed, the agent advances to the next state. If not, it rolls back to CUSTOMER_CONFIRMED and allows a retry. This check-and-resume pattern eliminates the ambiguity window.

The idempotency key generated from (agent_id, customer_id, amount, description) provides an additional safety layer. If two agent instances both attempt to create an intent for the same purchase, the GreenHelix gateway deduplicates based on the idempotency key in the metadata.

Key Takeaways

• Model payment flows as explicit state machines with named states (INITIATED, INTENT_CREATED, CUSTOMER_CONFIRMED, PAYMENT_EXECUTING, PAYMENT_CONFIRMED, COMPLETE).
• Persist state to Redis BEFORE each state transition, not after. This ensures crash recovery always has a valid checkpoint.
• Use Redis SET NX locks to prevent duplicate payment execution across concurrent agent instances.
• Generate deterministic idempotency keys from transaction parameters to detect and deduplicate redundant flows.
• Handle the "ambiguous execution" case by checking intent status on the gateway during recovery.
• Cross-reference: P19 (Payment Rails) covers multi-protocol payment routing; P20 (Production Hardening) covers retry policies for the API calls within each state.

---

Chapter 5: Customer Context Hydration#

The Context Budget Problem#

Every LLM turn has a finite token budget. For a 128K-token model, after accounting for the system prompt (2-5K tokens), the current conversation (5-20K tokens), and the agent's reasoning space (10-20K tokens), you have roughly 80-110K tokens available for customer context. That sounds like plenty until you consider a customer with 200 prior transactions, 15 active subscriptions, 3 open disputes, a preference profile spanning 50 attributes, and a compliance history requiring 12 consent records. Serialized naively, this data can consume 50K+ tokens. Selecting which data to load -- and which to leave on disk -- is the core skill of context hydration.

The Hydration Pipeline#

Context hydration is a three-stage pipeline: identify, rank, and budget.

Stage 1: Identify. Determine what data exists for this customer. This is a metadata query, not a data load -- you want to know how many transactions, how many preferences, how many disputes, not the content of each.

Stage 2: Rank. Score each data category by relevance to the current turn. A customer asking about a refund needs their recent transactions and dispute history loaded first. A customer browsing products needs their preference profile and purchase patterns. A customer at checkout needs their billing address and active payment methods.

Stage 3: Budget. Allocate tokens to each category based on rank, ensuring the total stays within budget. High-priority categories get full data. Low-priority categories get summaries or are omitted entirely.

python

from dataclasses import dataclass

@dataclass
class ContextBudget:
    """Token budget allocation for context hydration."""
    total_budget: int = 80000  # tokens available for context
    system_prompt: int = 3000
    conversation: int = 15000
    reasoning_reserve: int = 15000

    @property
    def available(self) -> int:
        return (
            self.total_budget
            - self.system_prompt
            - self.conversation
            - self.reasoning_reserve
        )

class ContextHydrator:
    """Loads the right customer data into the context window each turn.

    Implements identify-rank-budget pipeline to maximize relevance
    while staying within token limits.
    """

    def __init__(self, agent_id: str, budget: ContextBudget | None = None):
        self.agent_id = agent_id
        self.budget = budget or ContextBudget()
        # Token estimates per data type (measured from production data)
        self.token_estimates = {
            "identity": 200,         # name, tier, public key
            "balance": 50,           # single number + currency
            "recent_transactions": 150,  # per transaction
            "preferences": 100,      # per preference entry
            "active_subscriptions": 200,  # per subscription
            "dispute_history": 300,  # per dispute
            "usage_analytics": 400,  # summary object
            "reputation": 150,       # score + components
        }

    def hydrate(
        self, customer_id: str, intent: str, turn_number: int
    ) -> dict:
        """Load customer context optimized for the current intent.

        Args:
            customer_id: The customer to load context for.
            intent: Detected intent (browse, purchase, refund, support).
            turn_number: Current conversation turn (affects priority).

        Returns:
            Dict of context data, guaranteed under token budget.
        """
        available = self.budget.available
        context = {}
        tokens_used = 0

        # Stage 1 & 2: Identify and rank data sources by intent
        ranked_sources = self._rank_sources(intent, turn_number)

        # Stage 3: Load data within budget
        for source_name, loader, priority in ranked_sources:
            estimated_tokens = self._estimate_tokens(source_name)
            if tokens_used + estimated_tokens > available:
                # Budget exhausted -- add summary note and stop
                context["_truncated"] = True
                context["_budget_remaining"] = available - tokens_used
                break

            try:
                data = loader(customer_id)
                actual_tokens = self._count_tokens(data)

                if tokens_used + actual_tokens > available:
                    # Actual data larger than estimate -- trim or skip
                    data = self._trim_to_budget(
                        data, source_name, available - tokens_used
                    )
                    actual_tokens = self._count_tokens(data)

                context[source_name] = data
                tokens_used += actual_tokens
            except Exception as e:
                # Graceful degradation: skip failed sources
                context[f"_{source_name}_error"] = str(e)

        context["_tokens_used"] = tokens_used
        context["_budget_total"] = available
        return context

    def _rank_sources(
        self, intent: str, turn_number: int
    ) -> list[tuple[str, callable, int]]:
        """Rank data sources by relevance to current intent."""

        # Priority matrices per intent (lower number = higher priority)
        priority_map = {
            "browse": [
                ("identity", self._load_identity, 1),
                ("preferences", self._load_preferences, 2),
                ("usage_analytics", self._load_analytics, 3),
                ("balance", self._load_balance, 4),
                ("recent_transactions", self._load_transactions, 5),
                ("reputation", self._load_reputation, 6),
            ],
            "purchase": [
                ("identity", self._load_identity, 1),
                ("balance", self._load_balance, 2),
                ("active_subscriptions", self._load_subscriptions, 3),
                ("recent_transactions", self._load_transactions, 4),
                ("preferences", self._load_preferences, 5),
                ("usage_analytics", self._load_analytics, 6),
            ],
            "refund": [
                ("recent_transactions", self._load_transactions, 1),
                ("dispute_history", self._load_disputes, 2),
                ("identity", self._load_identity, 3),
                ("balance", self._load_balance, 4),
                ("usage_analytics", self._load_analytics, 5),
            ],
            "support": [
                ("identity", self._load_identity, 1),
                ("recent_transactions", self._load_transactions, 2),
                ("active_subscriptions", self._load_subscriptions, 3),
                ("dispute_history", self._load_disputes, 4),
                ("usage_analytics", self._load_analytics, 5),
                ("preferences", self._load_preferences, 6),
            ],
        }

        sources = priority_map.get(intent, priority_map["support"])
        return sorted(sources, key=lambda x: x[2])

    def _load_identity(self, customer_id: str) -> dict:
        return execute("get_agent_identity", {"agent_id": customer_id})

    def _load_balance(self, customer_id: str) -> dict:
        return execute("get_balance", {"agent_id": customer_id})

    def _load_transactions(self, customer_id: str) -> list:
        result = execute("get_transaction_history", {
            "agent_id": customer_id,
            "limit": 20,
        })
        return result.get("transactions", [])

    def _load_preferences(self, customer_id: str) -> dict:
        result = execute("get_usage_analytics", {
            "agent_id": customer_id,
        })
        return result

    def _load_subscriptions(self, customer_id: str) -> list:
        result = execute("check_subscription", {
            "agent_id": customer_id,
        })
        return result.get("subscriptions", [result]) if result else []

    def _load_disputes(self, customer_id: str) -> list:
        result = execute("query_events", {
            "agent_id": customer_id,
            "event_type": "dispute",
            "limit": 10,
        })
        return result.get("events", [])

    def _load_analytics(self, customer_id: str) -> dict:
        return execute("get_usage_analytics", {"agent_id": customer_id})

    def _load_reputation(self, customer_id: str) -> dict:
        return execute("get_agent_reputation", {"agent_id": customer_id})

    def _estimate_tokens(self, source_name: str) -> int:
        """Return estimated token count for a data source."""
        base = self.token_estimates.get(source_name, 500)
        if source_name == "recent_transactions":
            return base * 20  # Up to 20 transactions
        return base

    def _count_tokens(self, data) -> int:
        """Approximate token count for serialized data."""
        serialized = json.dumps(data, default=str)
        return len(serialized) // 4

    def _trim_to_budget(
        self, data, source_name: str, remaining_tokens: int
    ):
        """Trim data to fit within remaining token budget."""
        if isinstance(data, list):
            # Keep most recent items that fit
            trimmed = []
            tokens = 0
            for item in data:
                item_tokens = self._count_tokens(item)
                if tokens + item_tokens > remaining_tokens:
                    break
                trimmed.append(item)
                tokens += item_tokens
            return trimmed
        return data

Using the Hydration Pipeline#

python

hydrator = ContextHydrator(agent_id="commerce-agent-001")

# Customer browsing products -- prioritize preferences and analytics
context = hydrator.hydrate(
    customer_id="customer-042",
    intent="browse",
    turn_number=1,
)
print(f"Loaded {context['_tokens_used']} tokens of context")
# Output: Loaded 2,340 tokens of context

# Same customer now at checkout -- reprioritize for purchase
context = hydrator.hydrate(
    customer_id="customer-042",
    intent="purchase",
    turn_number=8,
)
# Balance and subscriptions now loaded first; preferences deprioritized

# Customer requesting a refund -- load transaction and dispute history first
context = hydrator.hydrate(
    customer_id="customer-042",
    intent="refund",
    turn_number=1,
)
# Recent transactions and dispute history loaded first

Identity Resolution: Recognizing Returning Customers#

Before you can hydrate context, you need to know who the customer is. Identity resolution maps an incoming session to an existing customer record. GreenHelix's get_agent_identity provides the canonical identity, but the agent may receive the customer's information in different forms -- an email address in one session, a wallet address in another, a session cookie in a third.

python

class IdentityResolver:
    """Resolve incoming customer identifiers to canonical agent IDs."""

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.r = redis.Redis.from_url(
            os.environ["REDIS_URL"], decode_responses=True
        )

    def resolve(self, identifiers: dict) -> str | None:
        """Resolve a set of identifiers to a canonical customer ID.

        Args:
            identifiers: Dict of identifier types to values, e.g.,
                {"email": "alice@example.com", "session": "abc123"}

        Returns:
            Canonical agent_id or None if no match.
        """
        # Check each identifier against the mapping cache
        for id_type, id_value in identifiers.items():
            cache_key = f"identity_map:{id_type}:{id_value}"
            canonical_id = self.r.get(cache_key)
            if canonical_id:
                return canonical_id

        # No cache hit -- try GreenHelix identity lookup
        for id_type, id_value in identifiers.items():
            try:
                result = execute("search_agents", {
                    "query": id_value,
                    "limit": 1,
                })
                agents = result.get("agents", [])
                if agents:
                    canonical_id = agents[0]["agent_id"]
                    # Cache all provided identifiers for this customer
                    for t, v in identifiers.items():
                        cache_key = f"identity_map:{t}:{v}"
                        self.r.set(cache_key, canonical_id, ex=86400)
                    return canonical_id
            except Exception:
                continue

        return None

# Usage
resolver = IdentityResolver(agent_id="commerce-agent-001")

customer_id = resolver.resolve({
    "email": "alice@example.com",
    "session_token": "sess_abc123",
})

if customer_id:
    context = hydrator.hydrate(customer_id, intent="browse", turn_number=1)
else:
    # New customer -- create identity
    new_customer = execute("register_agent", {
        "agent_id": f"customer-{int(time.time())}",
        "name": "Alice",
        "public_key": "...",
    })

Token Cost Analysis#

Hydration StrategyTokens per TurnCost per 20-Turn Session (GPT-4o)Customer SatisfactionNo hydration (stateless)0$0.00Poor (no personalization)Full dump (everything)50,000-100,000$2.50-$5.00High (but wasteful)Intent-ranked hydration2,000-8,000$0.10-$0.40High (targeted context)Progressive hydration500-3,000 (grows per turn)$0.05-$0.15High (minimal early, rich late)

Intent-ranked hydration delivers the same customer experience as full dump at 5-10% of the token cost.

Key Takeaways

• Context hydration is a three-stage pipeline: identify available data, rank by relevance to current intent, allocate within token budget.
• Different intents require different context: a refund request needs transaction and dispute history first; a browse session needs preferences and analytics first.
• Intent-ranked hydration reduces token costs by 90% compared to loading everything, with no measurable loss in customer satisfaction.
• Identity resolution maps diverse customer identifiers (email, session, wallet) to a canonical agent ID for consistent context loading.
• Cache identity mappings in Redis with 24-hour TTL to avoid repeated gateway lookups.
• Cross-reference: P4 (Commerce Toolkit) covers identity registration; P10 (Multi-Agent Cookbook) covers cross-agent identity resolution.

---

Chapter 6: Reconciliation and Audit Trails#

Why Memory Is Your Source of Truth#

When a customer disputes a charge, the question is: "What actually happened?" In a stateless system, the answer is "we do not know" -- the conversation is gone, the state transitions were ephemeral, and the only record is a line item in the payment processor's dashboard. In a memory-backed system, every state transition is recorded: when the intent was created, when the customer confirmed, when the payment executed, what the agent's context was at each step, and whether any anomalies occurred.

This is not just a customer support improvement. It is a compliance requirement. PCI DSS Requirement 10 mandates detailed audit trails for payment processing. The EU AI Act (Article 12, effective August 2, 2026) requires providers of high-risk AI systems to maintain logs that enable tracing the system's operation. A commerce agent making payment decisions is a high-risk AI system under the Act's classification.

Writing Memory Events to the GreenHelix Ledger#

The GreenHelix gateway provides two persistence tools for audit trails:

• record_transaction: Writes a structured transaction record to the permanent ledger. Used for financial events (payments, refunds, escrow releases).
• publish_event: Writes an arbitrary event to the event stream. Used for non-financial events (customer interactions, state transitions, system decisions).

Together, they provide a complete audit trail: record_transaction for what money moved, publish_event for why it moved.

python

from enum import Enum
from datetime import datetime, timezone

class AuditEventType(str, Enum):
    INTENT_CREATED = "audit.intent_created"
    CUSTOMER_CONFIRMED = "audit.customer_confirmed"
    PAYMENT_EXECUTED = "audit.payment_executed"
    PAYMENT_FAILED = "audit.payment_failed"
    REFUND_INITIATED = "audit.refund_initiated"
    DISPUTE_OPENED = "audit.dispute_opened"
    CONTEXT_HYDRATED = "audit.context_hydrated"
    AGENT_DECISION = "audit.agent_decision"
    STATE_TRANSITION = "audit.state_transition"
    ANOMALY_DETECTED = "audit.anomaly_detected"

class AuditTrail:
    """Write and query audit events for compliance and reconciliation."""

    def __init__(self, agent_id: str):
        self.agent_id = agent_id

    def log_event(
        self,
        event_type: AuditEventType,
        payload: dict,
        customer_id: str | None = None,
        flow_id: str | None = None,
    ):
        """Write an audit event to the GreenHelix event stream."""
        event_data = {
            "event_type": event_type.value,
            "agent_id": self.agent_id,
            "customer_id": customer_id,
            "flow_id": flow_id,
            "payload": payload,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }

        execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": event_type.value,
            "payload": json.dumps(event_data),
        })

    def log_financial_event(
        self,
        transaction_type: str,
        amount: str,
        counterparty: str,
        metadata: dict,
    ):
        """Record a financial transaction on the permanent ledger."""
        execute("record_transaction", {
            "agent_id": self.agent_id,
            "transaction_type": transaction_type,
            "amount": amount,
            "counterparty": counterparty,
            "metadata": json.dumps(metadata),
        })

    def query_events(
        self,
        event_type: str | None = None,
        since: str | None = None,
        limit: int = 100,
    ) -> list[dict]:
        """Query audit events for reconciliation."""
        params = {
            "agent_id": self.agent_id,
            "limit": limit,
        }
        if event_type:
            params["event_type"] = event_type

        result = execute("query_events", params)
        events = result.get("events", [])

        # Filter by timestamp if `since` provided
        if since:
            events = [
                e for e in events
                if e.get("timestamp", "") >= since
            ]

        return events

    def get_transaction_ledger(
        self, limit: int = 100
    ) -> list[dict]:
        """Get the financial transaction ledger."""
        result = execute("get_transaction_history", {
            "agent_id": self.agent_id,
            "limit": limit,
        })
        return result.get("transactions", [])

The Reconciliation Agent#

Reconciliation compares what the agent's memory says happened against what the ledger says happened. Discrepancies indicate bugs (a payment was executed but not recorded), fraud (a record was tampered with), or race conditions (two agents processed the same payment).

python

from dataclasses import dataclass, field

@dataclass
class ReconciliationResult:
    """Result of a memory-vs-ledger reconciliation run."""
    total_memory_events: int = 0
    total_ledger_entries: int = 0
    matched: int = 0
    discrepancies: list[dict] = field(default_factory=list)
    orphaned_memory_events: list[dict] = field(default_factory=list)
    orphaned_ledger_entries: list[dict] = field(default_factory=list)

    @property
    def is_clean(self) -> bool:
        return (
            len(self.discrepancies) == 0
            and len(self.orphaned_memory_events) == 0
            and len(self.orphaned_ledger_entries) == 0
        )

class ReconciliationAgent:
    """Compares agent memory state against the GreenHelix ledger.

    Identifies discrepancies between what the agent remembers
    and what the ledger recorded.
    """

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.audit = AuditTrail(agent_id)

    def reconcile(
        self, since: str | None = None, limit: int = 500
    ) -> ReconciliationResult:
        """Run a full reconciliation between memory and ledger.

        Args:
            since: ISO timestamp to start reconciliation from.
            limit: Maximum number of records to reconcile.

        Returns:
            ReconciliationResult with matches and discrepancies.
        """
        result = ReconciliationResult()

        # Load memory events (what the agent says happened)
        memory_events = self.audit.query_events(
            event_type="audit.payment_executed",
            since=since,
            limit=limit,
        )
        result.total_memory_events = len(memory_events)

        # Load ledger entries (what the gateway recorded)
        ledger_entries = self.audit.get_transaction_ledger(limit=limit)
        result.total_ledger_entries = len(ledger_entries)

        # Build lookup maps
        memory_by_flow = {}
        for event in memory_events:
            payload = (
                json.loads(event["payload"])
                if isinstance(event.get("payload"), str)
                else event.get("payload", {})
            )
            flow_id = payload.get("flow_id")
            if flow_id:
                memory_by_flow[flow_id] = {
                    "event": event,
                    "payload": payload,
                    "matched": False,
                }

        ledger_by_flow = {}
        for entry in ledger_entries:
            metadata = (
                json.loads(entry.get("metadata", "{}"))
                if isinstance(entry.get("metadata"), str)
                else entry.get("metadata", {})
            )
            flow_id = metadata.get("flow_id")
            if flow_id:
                ledger_by_flow[flow_id] = {
                    "entry": entry,
                    "metadata": metadata,
                    "matched": False,
                }

        # Compare: match by flow_id and check amounts
        for flow_id, mem_record in memory_by_flow.items():
            if flow_id in ledger_by_flow:
                ledger_record = ledger_by_flow[flow_id]
                mem_amount = str(mem_record["payload"].get("amount", ""))
                ledger_amount = str(
                    ledger_record["entry"].get("amount", "")
                )

                if mem_amount == ledger_amount:
                    result.matched += 1
                    mem_record["matched"] = True
                    ledger_record["matched"] = True
                else:
                    result.discrepancies.append({
                        "flow_id": flow_id,
                        "type": "amount_mismatch",
                        "memory_amount": mem_amount,
                        "ledger_amount": ledger_amount,
                        "memory_event": mem_record["event"],
                        "ledger_entry": ledger_record["entry"],
                    })
                    mem_record["matched"] = True
                    ledger_record["matched"] = True
            else:
                result.orphaned_memory_events.append({
                    "flow_id": flow_id,
                    "event": mem_record["event"],
                    "issue": "Payment recorded in memory but not in ledger",
                })

        # Find ledger entries with no matching memory event
        for flow_id, ledger_record in ledger_by_flow.items():
            if not ledger_record["matched"]:
                result.orphaned_ledger_entries.append({
                    "flow_id": flow_id,
                    "entry": ledger_record["entry"],
                    "issue": "Transaction in ledger with no matching "
                             "memory event",
                })

        # Log the reconciliation result as an audit event
        self.audit.log_event(
            event_type=AuditEventType.AGENT_DECISION,
            payload={
                "action": "reconciliation",
                "result": "clean" if result.is_clean else "discrepancies",
                "matched": result.matched,
                "discrepancies": len(result.discrepancies),
                "orphaned_memory": len(result.orphaned_memory_events),
                "orphaned_ledger": len(result.orphaned_ledger_entries),
            },
        )

        return result

# Run daily reconciliation
recon = ReconciliationAgent(agent_id="commerce-agent-001")
result = recon.reconcile(since="2026-04-05T00:00:00Z")

if result.is_clean:
    print(f"Reconciliation clean: {result.matched} matched")
else:
    print(f"DISCREPANCIES FOUND:")
    print(f"  Amount mismatches: {len(result.discrepancies)}")
    print(f"  Orphaned memory events: {len(result.orphaned_memory_events)}")
    print(f"  Orphaned ledger entries: {len(result.orphaned_ledger_entries)}")
    for d in result.discrepancies:
        print(f"  Flow {d['flow_id']}: memory={d['memory_amount']} "
              f"vs ledger={d['ledger_amount']}")

Reconciliation Schedule and Alerting#

Reconciliation TypeFrequencyScopeAlert ThresholdReal-timePer transactionSingle flowAny discrepancyBatchHourlyAll flows in window> 0 orphaned eventsFullDailyAll flows, all time> 0.1% discrepancy rateForensicOn demandSpecific customer/flowN/A (investigation)

For production systems, run batch reconciliation every hour and full reconciliation daily. Alert on any discrepancy in real-time reconciliation (run as part of the payment flow itself). Use forensic reconciliation when a customer reports a billing issue.

Key Takeaways

• Use record_transaction for financial events (what money moved) and publish_event for operational events (why it moved). Together they form a complete audit trail.
• Reconciliation compares memory state against ledger entries, identifying amount mismatches, orphaned memory events (payments the agent remembers but the ledger does not), and orphaned ledger entries (ledger records with no corresponding agent memory).
• Run reconciliation at four frequencies: real-time (per transaction), hourly batch, daily full, and on-demand forensic.
• EU AI Act Article 12 (effective August 2, 2026) requires logging that enables tracing -- the audit trail described here satisfies that requirement.
• Cross-reference: P20 (Production Hardening) covers monitoring and alerting patterns for reconciliation failures.

---

Chapter 7: Multi-Agent Memory Sharing#

The Handoff Problem#

Agent A is handling a customer's purchase. The customer asks a question about product compatibility that requires specialist knowledge. Agent A needs to hand off to Agent B -- a domain expert -- mid-transaction. Agent B must know: who the customer is, what they are buying, what stage the transaction is at, what the customer's preferences are, and what has already been discussed in the conversation. Without shared memory, Agent B starts cold. The customer repeats their question. Agent B re-asks for the billing address. The customer's trust evaporates.

This is not a theoretical concern. Sixty-seven percent of agent-to-agent handoffs without shared memory result in the customer repeating information, according to a March 2026 survey of enterprise AI deployment teams. The median customer satisfaction score drops 34% on handoffs where context is lost.

Memory Sharing via GreenHelix Messaging#

GreenHelix provides two primitives for inter-agent communication:

• send_message: Point-to-point message delivery between agents. Used for targeted handoffs where Agent A sends context directly to Agent B.
• publish_event / query_events: Broadcast event stream. Used for shared state that multiple agents may need to access.

The choice depends on the sharing pattern:

PatternPrimitiveUse Case1:1 handoffsend_messageAgent A transfers to Agent B1:N broadcastpublish_eventAgent A updates state that agents B, C, D may needN:1 aggregationquery_eventsSupervisor agent collects state from all sub-agentsN:N collaborationpublish_event + query_eventsMultiple agents working on shared customer

Implementing a Handoff Protocol#

python

from dataclasses import dataclass, asdict

@dataclass
class HandoffContext:
    """Context package for agent-to-agent handoff."""
    customer_id: str
    session_id: str
    conversation_summary: str
    active_flow_id: str | None
    flow_state: dict | None
    customer_profile: dict
    intent: str
    priority: str  # "normal", "urgent", "escalation"
    handoff_reason: str
    source_agent: str
    timestamp: str

class MemoryHandoff:
    """Manages memory sharing during agent-to-agent handoffs."""

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.audit = AuditTrail(agent_id)

    def prepare_handoff(
        self,
        target_agent: str,
        customer_id: str,
        session_id: str,
        conversation_turns: list[dict],
        active_flow_id: str | None = None,
        intent: str = "support",
        priority: str = "normal",
        reason: str = "specialist required",
    ) -> HandoffContext:
        """Prepare a context package for handoff to another agent.

        Summarizes conversation, loads customer profile, and packages
        active transaction state for the receiving agent.
        """
        # Summarize conversation (keep last 5 turns, summarize rest)
        if len(conversation_turns) > 5:
            recent = conversation_turns[-5:]
            older = conversation_turns[:-5]
            summary_parts = [
                f"[{t['role']}]: {t['content'][:100]}" for t in older
            ]
            conversation_summary = (
                "Previous discussion summary:\n"
                + "\n".join(summary_parts)
                + "\n\nRecent turns:\n"
                + "\n".join(
                    f"[{t['role']}]: {t['content']}" for t in recent
                )
            )
        else:
            conversation_summary = "\n".join(
                f"[{t['role']}]: {t['content']}"
                for t in conversation_turns
            )

        # Load customer profile
        customer_profile = {}
        try:
            customer_profile["identity"] = execute(
                "get_agent_identity", {"agent_id": customer_id}
            )
        except Exception:
            pass
        try:
            customer_profile["balance"] = execute(
                "get_balance", {"agent_id": customer_id}
            )
        except Exception:
            pass

        # Load active flow state if exists
        flow_state = None
        if active_flow_id:
            r = redis.Redis.from_url(
                os.environ["REDIS_URL"], decode_responses=True
            )
            flow_data = r.get(
                f"payment_flow:{self.agent_id}:{active_flow_id}"
            )
            if flow_data:
                flow_state = json.loads(flow_data)

        context = HandoffContext(
            customer_id=customer_id,
            session_id=session_id,
            conversation_summary=conversation_summary,
            active_flow_id=active_flow_id,
            flow_state=flow_state,
            customer_profile=customer_profile,
            intent=intent,
            priority=priority,
            handoff_reason=reason,
            source_agent=self.agent_id,
            timestamp=datetime.now(timezone.utc).isoformat(),
        )

        return context

    def send_handoff(
        self, target_agent: str, context: HandoffContext
    ):
        """Send handoff context to target agent via GreenHelix messaging."""
        # Send the context as a structured message
        execute("send_message", {
            "sender_id": self.agent_id,
            "recipient_id": target_agent,
            "message_type": "handoff_context",
            "content": json.dumps(asdict(context)),
        })

        # Also publish to event stream for audit and other listeners
        execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": "agent_handoff",
            "payload": json.dumps({
                "source_agent": self.agent_id,
                "target_agent": target_agent,
                "customer_id": context.customer_id,
                "flow_id": context.active_flow_id,
                "reason": context.handoff_reason,
                "timestamp": context.timestamp,
            }),
        })

        # Log audit event
        self.audit.log_event(
            event_type=AuditEventType.STATE_TRANSITION,
            payload={
                "action": "handoff_sent",
                "target_agent": target_agent,
                "reason": context.handoff_reason,
            },
            customer_id=context.customer_id,
            flow_id=context.active_flow_id,
        )

    def receive_handoff(self) -> HandoffContext | None:
        """Check for incoming handoff contexts."""
        messages = execute("get_messages", {
            "agent_id": self.agent_id,
            "message_type": "handoff_context",
            "limit": 1,
        })

        if not messages.get("messages"):
            return None

        msg = messages["messages"][0]
        context_data = json.loads(msg["content"])

        # Reconstruct HandoffContext
        context = HandoffContext(**context_data)

        # Log receipt
        self.audit.log_event(
            event_type=AuditEventType.STATE_TRANSITION,
            payload={
                "action": "handoff_received",
                "source_agent": context.source_agent,
                "reason": context.handoff_reason,
            },
            customer_id=context.customer_id,
            flow_id=context.active_flow_id,
        )

        # If there is an active payment flow, re-register it locally
        if context.flow_state and context.active_flow_id:
            r = redis.Redis.from_url(
                os.environ["REDIS_URL"], decode_responses=True
            )
            r.set(
                f"payment_flow:{self.agent_id}:{context.active_flow_id}",
                json.dumps(context.flow_state),
                ex=7200,
            )

        return context

Multi-Agent Memory Synchronization#

When multiple agents collaborate on the same customer (not a handoff, but parallel work), they need a shared view of the customer's state that stays consistent as each agent makes changes. This requires a synchronization protocol.

python

class SharedMemorySync:
    """Synchronize memory state across multiple agents working
    on the same customer.

    Uses GreenHelix events as the synchronization backbone
    with Redis for low-latency local caching.
    """

    def __init__(self, agent_id: str, customer_id: str):
        self.agent_id = agent_id
        self.customer_id = customer_id
        self.r = redis.Redis.from_url(
            os.environ["REDIS_URL"], decode_responses=True
        )
        self.cache_key = f"shared_state:{customer_id}"
        self._last_sync = None

    def write_state(self, key: str, value: dict):
        """Write a state update visible to all agents."""
        # Write to Redis for low-latency reads
        state_entry = {
            "key": key,
            "value": value,
            "author": self.agent_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "version": int(time.time() * 1000),
        }

        self.r.hset(
            self.cache_key,
            key,
            json.dumps(state_entry),
        )

        # Publish to GreenHelix events for durability and audit
        execute("publish_event", {
            "agent_id": self.agent_id,
            "event_type": f"shared_state.{key}",
            "payload": json.dumps(state_entry),
        })

    def read_state(self, key: str) -> dict | None:
        """Read shared state (from Redis cache first, then events)."""
        cached = self.r.hget(self.cache_key, key)
        if cached:
            return json.loads(cached)

        # Cache miss -- query events
        events = execute("query_events", {
            "agent_id": self.customer_id,
            "event_type": f"shared_state.{key}",
            "limit": 1,
        })

        if events.get("events"):
            entry = json.loads(events["events"][0]["payload"])
            # Populate cache
            self.r.hset(self.cache_key, key, json.dumps(entry))
            return entry

        return None

    def acquire_write_lock(self, key: str, ttl: int = 10) -> bool:
        """Acquire a distributed lock for writing to a shared key.

        Prevents conflicting writes from concurrent agents.
        """
        lock_key = f"lock:shared_state:{self.customer_id}:{key}"
        return self.r.set(lock_key, self.agent_id, nx=True, ex=ttl)

    def release_write_lock(self, key: str):
        """Release the write lock for a shared key."""
        lock_key = f"lock:shared_state:{self.customer_id}:{key}"
        # Only release if we hold the lock
        if self.r.get(lock_key) == self.agent_id:
            self.r.delete(lock_key)

# Agent A writes customer preference during conversation
sync_a = SharedMemorySync("agent-sales-001", "customer-042")
sync_a.write_state("preferred_plan", {
    "plan": "pro",
    "billing_cycle": "annual",
    "confirmed": True,
})

# Agent B (billing specialist) reads the preference
sync_b = SharedMemorySync("agent-billing-001", "customer-042")
preference = sync_b.read_state("preferred_plan")
print(f"Customer prefers: {preference['value']['plan']}")

# Agent B updates payment status with lock to prevent conflicts
if sync_b.acquire_write_lock("payment_status"):
    try:
        sync_b.write_state("payment_status", {
            "status": "processing",
            "intent_id": "pi_abc123",
        })
    finally:
        sync_b.release_write_lock("payment_status")

Conflict Resolution Strategies#

When two agents write to the same shared state key simultaneously, you need a conflict resolution strategy. Three approaches, in order of complexity:

StrategyImplementationWhen to Use**Last-writer-wins (LWW)**Compare timestamps, keep newestLow-risk state (preferences, notes)Distributed lockRedis SET NX before writingFinancial state (payment flows)Event sourcingNever overwrite; append events, replayAudit-critical state (compliance logs)

For commerce, use distributed locks for any state that affects money (payment intents, escrow state, cart totals) and last-writer-wins for everything else (customer notes, browsing history, preference updates). Never use last-writer-wins for financial state -- a lost update on a payment amount is a billing error.

Key Takeaways

• Agent handoffs require a structured context package: conversation summary, active transaction state, customer profile, and handoff reason.
• Use send_message for 1:1 handoffs and publish_event for 1:N broadcasts.
• Multi-agent collaboration on the same customer requires distributed locks for financial state and last-writer-wins for non-financial state.
• Re-register active payment flows on the receiving agent's local Redis after handoff to maintain crash resilience.
• Every handoff must be logged as an audit event for compliance traceability.
• Cross-reference: P10 (Multi-Agent Cookbook) covers orchestration patterns; P4 (Commerce Toolkit) covers the messaging primitives used here.

---

Chapter 8: Production Patterns and Anti-Patterns#

Pattern 1: Memory Warm-Up on Agent Start#

When an agent process starts (or restarts), it should immediately hydrate its operational memory from the GreenHelix event stream. This ensures it picks up any in-progress transactions that were interrupted by the restart.

python

class AgentMemoryWarmup:
    """Warm up agent memory from persistent storage on startup."""

    def __init__(self, agent_id: str, redis_url: str):
        self.agent_id = agent_id
        self.r = redis.Redis.from_url(redis_url, decode_responses=True)

    def warm_up(self) -> dict:
        """Load in-progress state from GreenHelix into Redis.

        Returns summary of restored state.
        """
        restored = {
            "payment_flows": 0,
            "active_handoffs": 0,
            "shared_state_keys": 0,
        }

        # Restore in-progress payment flows
        events = execute("query_events", {
            "agent_id": self.agent_id,
            "event_type": "transaction_state",
            "limit": 100,
        })

        seen_flows = set()
        for event in events.get("events", []):
            payload = (
                json.loads(event["payload"])
                if isinstance(event.get("payload"), str)
                else event.get("payload", {})
            )
            flow_id = payload.get("intent_id") or payload.get("flow_id")
            if flow_id and flow_id not in seen_flows:
                state = payload.get("state", {})
                status = state.get("status", "")
                # Only restore flows that were not completed
                if status not in ("complete", "failed"):
                    key = f"payment_flow:{self.agent_id}:{flow_id}"
                    self.r.set(key, json.dumps(state), ex=7200)
                    restored["payment_flows"] += 1
                    seen_flows.add(flow_id)

        # Restore pending handoff contexts
        messages = execute("get_messages", {
            "agent_id": self.agent_id,
            "message_type": "handoff_context",
            "limit": 10,
        })
        restored["active_handoffs"] = len(
            messages.get("messages", [])
        )

        return restored

# On agent startup
warmup = AgentMemoryWarmup(
    agent_id="commerce-agent-001",
    redis_url=os.environ["REDIS_URL"],
)
summary = warmup.warm_up()
print(f"Restored {summary['payment_flows']} in-progress payment flows")
print(f"Found {summary['active_handoffs']} pending handoffs")

Pattern 2: Context Window Overflow Prevention#

The most insidious memory bug is silent context overflow: the agent loads too much data into the prompt, the LLM truncates the oldest content, and the agent proceeds without realizing it has lost critical information. The fix is explicit monitoring and hard limits.

python

class ContextGuard:
    """Prevents context window overflow with hard limits and warnings."""

    def __init__(self, max_tokens: int = 128000, warn_at: float = 0.75):
        self.max_tokens = max_tokens
        self.warn_threshold = int(max_tokens * warn_at)
        self.current_tokens = 0
        self.components: dict[str, int] = {}

    def register(self, name: str, content: str) -> bool:
        """Register a context component. Returns False if it would overflow.

        Args:
            name: Component name (e.g., "customer_profile", "conversation")
            content: The text content to add to context

        Returns:
            True if registered successfully, False if would overflow.
        """
        tokens = len(content) // 4  # Approximate

        if self.current_tokens + tokens > self.max_tokens:
            return False  # Would overflow -- reject

        self.components[name] = tokens
        self.current_tokens += tokens

        if self.current_tokens > self.warn_threshold:
            self._emit_warning()

        return True

    def evict(self, name: str) -> int:
        """Remove a component from context. Returns tokens freed."""
        tokens = self.components.pop(name, 0)
        self.current_tokens -= tokens
        return tokens

    def usage_report(self) -> dict:
        """Generate a context usage report."""
        return {
            "total_tokens": self.current_tokens,
            "max_tokens": self.max_tokens,
            "utilization": round(
                self.current_tokens / self.max_tokens, 3
            ),
            "components": dict(
                sorted(
                    self.components.items(),
                    key=lambda x: x[1],
                    reverse=True,
                )
            ),
        }

    def _emit_warning(self):
        """Publish a warning event when context utilization is high."""
        execute("publish_event", {
            "agent_id": "system",
            "event_type": "audit.anomaly_detected",
            "payload": json.dumps({
                "type": "context_window_pressure",
                "utilization": round(
                    self.current_tokens / self.max_tokens, 3
                ),
                "components": self.components,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }),
        })

Pattern 3: Stale State Detection#

Operational memory (Tier 2) can become stale when an external system changes state without the agent's knowledge. A payment intent might expire on the gateway while the agent still believes it is active. A subscription might be cancelled via a different channel. Stale state leads to incorrect decisions.

python

class StaleStateDetector:
    """Detect and refresh stale operational state."""

    # Maximum age (seconds) before state is considered stale
    STALENESS_THRESHOLDS = {
        "payment_intent": 300,    # 5 minutes
        "balance": 60,            # 1 minute
        "subscription": 3600,     # 1 hour
        "customer_profile": 86400,  # 24 hours
    }

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.r = redis.Redis.from_url(
            os.environ["REDIS_URL"], decode_responses=True
        )

    def check_and_refresh(
        self, state_type: str, key: str, refresh_fn: callable
    ) -> dict:
        """Check if cached state is stale. Refresh if necessary.

        Args:
            state_type: Type of state (determines staleness threshold).
            key: Redis key where state is cached.
            refresh_fn: Callable that fetches fresh state from the source.

        Returns:
            Current (possibly refreshed) state.
        """
        cached = self.r.get(key)
        if cached:
            state = json.loads(cached)
            cached_at = state.get("_cached_at", 0)
            age = time.time() - cached_at
            threshold = self.STALENESS_THRESHOLDS.get(state_type, 300)

            if age < threshold:
                return state  # Still fresh

        # Stale or missing -- refresh
        fresh_state = refresh_fn()
        fresh_state["_cached_at"] = time.time()
        self.r.set(key, json.dumps(fresh_state), ex=7200)
        return fresh_state

    def validate_payment_intent(self, intent_id: str) -> dict:
        """Validate that a payment intent is still active on the gateway."""
        def refresh():
            return execute("get_payment_intent", {
                "intent_id": intent_id,
            })

        return self.check_and_refresh(
            "payment_intent",
            f"intent_cache:{intent_id}",
            refresh,
        )

# Before executing a payment, validate the intent is still active
detector = StaleStateDetector(agent_id="commerce-agent-001")
intent = detector.validate_payment_intent("pi_abc123")
if intent.get("status") == "expired":
    print("Intent expired -- must re-create before executing")

Anti-Pattern 1: Storing Everything Forever#

The mistake: Writing every conversation turn, every API response, and every intermediate calculation to persistent memory. This inflates storage costs, slows down memory retrieval, and creates a compliance liability (you are now responsible for retaining and potentially deleting all that data under GDPR/CCPA).

The fix: Define retention policies per data type. Financial transactions: permanent (regulatory requirement). Customer preferences: 12 months with re-confirmation. Conversation transcripts: 30 days or until dispute window closes. Intermediate computation: never persist.

Anti-Pattern 2: Using Memory as a Message Bus#

The mistake: Agents communicate by writing state to shared memory and polling for changes. Agent A writes "ready_for_handoff=true" to Redis, Agent B polls every 500ms to check.

The fix: Use GreenHelix messaging ( send_message) for point-to-point communication and events ( publish_event) for broadcasts. Polling shared memory creates unnecessary load, introduces latency (up to the polling interval), and makes it impossible to trace the communication path for debugging.

Anti-Pattern 3: Trusting Memory Without Validation#

The mistake: The agent reads a balance from memory and makes a purchase decision without checking whether the balance is current. The memory says $500. The actual balance is $12 (another agent spent the difference). The purchase fails at execution time, after the customer has already confirmed.

The fix: Always validate financial state against the source of truth before irrevocable actions. The StaleStateDetector pattern above handles this. For balance checks specifically, always call get_balance directly from GreenHelix before creating a payment intent -- never rely on cached balance for purchase decisions.

Anti-Pattern 4: Unbounded Context Hydration#

The mistake: The agent loads all available customer data into the context window without budgeting. For a new customer with 2 transactions, this works fine. For a returning customer with 500 transactions, the agent loads 500 transaction records, overflows the context window, and silently loses the system prompt.

The fix: Use the ContextHydrator from Chapter 5 with explicit token budgets. Never load more than N items from any collection. Always verify context utilization with ContextGuard before sending the prompt to the LLM.

Anti-Pattern 5: Single-Point-of-Failure Memory#

The mistake: All memory goes through a single Redis instance with no persistence configured. Redis restarts, all operational state is lost, and every in-progress transaction must be manually recovered.

The fix: Enable Redis AOF persistence with appendfsync everysec. Use Redis Sentinel or Redis Cluster for high availability. Back critical state to GreenHelix events (which are permanently persisted) in addition to Redis. The dual-write pattern (Redis for speed + GreenHelix events for durability) ensures that a Redis failure loses at most 1 second of state, which can be recovered from the event stream on restart.

The Production Checklist#

Before shipping a memory-backed commerce agent, verify every item on this checklist:

Memory Architecture

#CheckStatus1Three-tier memory model implemented (immediate, operational, institutional)2Operational state persisted to Redis with AOF enabled3Financial events written to GreenHelix ledger via record_transaction4Audit events written to GreenHelix events via publish_event5Retention policies defined per data type6GDPR/CCPA deletion capabilities implemented for customer data

Payment State Management

#CheckStatus7Payment flows modeled as explicit state machines8State persisted BEFORE each transition (not after)9Idempotency keys generated for all payment intents10Distributed locks prevent duplicate payment execution11Ambiguous execution recovery implemented (check intent status on restart)12Payment state TTL configured (auto-expire abandoned flows)

Context Hydration

#CheckStatus13Token budget defined and enforced per turn14Intent-based ranking prioritizes relevant d