Memory events

• Pattern-based subscriptions -- watch specific keys, wildcard ranges, or entire namespaces with glob patterns.
• Scoped listeners -- subscribe at global, session, actor, or workflow scope for precise targeting.
• MemoryChangeEvent -- rich event object with key, new value, old value, scope, timestamp, and source agent.
• Cross-agent reactivity -- one agent writes, another agent's handler fires immediately.
• No polling -- push-based delivery via the control plane's internal event bus.
• Multiple patterns -- TypeScript supports arrays of patterns in a single subscription.

Every handler receives a MemoryChangeEvent with the following fields:

@app.on_change("orders.*")
async def on_order(event):
    print(event.key)            # "orders.ord-123"
    print(event.data)           # {"status": "shipped", "tracking": "1Z999..."}
    print(event.previous_data)  # {"status": "processing"}
    print(event.old_value)      # {"status": "processing"} (alias for previous_data)
    print(event.scope)          # "global"
    print(event.timestamp)      # "2026-03-24T10:30:00Z"
    print(event.action)         # "set"

Memory event patterns use a glob-like syntax to match keys.

Rules

• * is converted to .* regex internally, so it crosses dot boundaries. Unlike filesystem globs, * and ** are functionally identical here -- both match across dot-separated segments.
• Exact strings match literally.
• Patterns are case-sensitive.

Multiple Patterns (TypeScript)

// Watch multiple patterns in one subscription
agent.watchMemory(
  ["orders.*", "payments.*", "refunds.*"],
  async (event) => {
    console.log(`Transaction event: ${event.key}`);
  }
);

In Python, register multiple decorators:

for pattern in ["orders.*", "payments.*", "refunds.*"]:
    @app.on_change(pattern)
    async def on_transaction(event):
        print(f"Transaction event: {event.key}")

Subscribe to events within a specific memory scope for precise targeting.

# Global scope — watch shared config changes
@app.memory.global_scope.on_change("config.*")
async def on_config_change(event):
    print(f"Global config changed: {event.key}")

# Session scope — watch conversation state for a specific session
@app.memory.session("session-abc").on_change("conversation.*")
async def on_conversation(event):
    print(f"Session conversation updated: {event.key}")

# Actor scope — watch user preference changes
@app.memory.actor("user-42").on_change("preferences.*")
async def on_prefs_change(event):
    print(f"User preferences changed: {event.key}")

# Workflow scope — watch intermediate results
@app.memory.workflow("wf-789").on_change("step.*")
async def on_step_complete(event):
    print(f"Workflow step completed: {event.key}")

Event-Driven Pipeline

Chain agents reactively -- each writes results that trigger the next stage.

# Stage 1: Intake writes classification
@app.reasoner()
async def intake(document: str) -> dict:
    classification = await app.ai(user=document, schema=DocType)
    await app.memory.global_scope.set(f"pipeline.classified.{doc_id}", classification.dict())
    return classification.dict()

# Stage 2: Triggered when classification lands
@app.memory.global_scope.on_change("pipeline.classified.*")
async def on_classified(event):
    doc_id = event.key.split(".")[-1]
    result = await app.call(
        f"{event.data['type']}-analyzer.analyze",
        doc_id=doc_id,
        classification=event.data,
    )
    await app.memory.global_scope.set(f"pipeline.analyzed.{doc_id}", result)

# Stage 3: Triggered when analysis completes
@app.memory.global_scope.on_change("pipeline.analyzed.*")
async def on_analyzed(event):
    doc_id = event.key.split(".")[-1]
    await app.call("reporter.generate", doc_id=doc_id, analysis=event.data)

Cross-Agent Notifications

Decouple producers from consumers -- agents write to known keys, any agent can subscribe.

# Producer agent: writes alerts
await app.memory.global_scope.set("alerts.security.sql-injection", {
    "severity": "critical",
    "source": "waf-agent",
    "details": "SQL injection attempt blocked from 192.168.1.50",
})

# Consumer agent: reacts to any security alert
@app.memory.global_scope.on_change("alerts.security.*")
async def on_security_alert(event):
    if event.data.get("severity") == "critical":
        await app.call("incident.create", **event.data)

Saga Coordination

Coordinate distributed transactions across agents using memory events as signals.

@app.memory.global_scope.on_change("saga.payment.*")
async def on_payment_event(event):
    saga_id = event.key.split(".")[-1]
    status = event.data.get("status")

    if status == "completed":
        await app.call("shipping.dispatch", saga_id=saga_id)
        await app.memory.global_scope.set(f"saga.shipping.{saga_id}", {"status": "dispatching"})
    elif status == "failed":
        await app.call("orders.rollback", saga_id=saga_id)

Event Subscription Methods

MemoryChangeEvent Fields

Go Alternative

Go does not support in-process memory events. Use the REST SSE endpoint instead:

GET /api/v1/memory/events/sse
GET /api/v1/memory/events/ws

These stream memory change events over Server-Sent Events or WebSocket for any client, including Go agents.

For simple cases, a Go agent can also poll the current value directly with a.Memory().Get(ctx, key) (or a scoped variant such as a.Memory().GlobalScope().Get(ctx, key)) on an interval instead of subscribing to events.