L0 Memory Output Pipeline

Mems can publish every L0 memory write as an event to a Redis Stream, so you can consume memories with your own code: mirror them into another store, feed a real-time dashboard, or trigger downstream workflows.

The pipeline is disabled by default. It writes nothing and creates no stream until you enable it, so existing deployments are unaffected.

Enable

Set the following in .env:

L0_PIPELINE_ENABLED=true
L0_PIPELINE_STREAM=mems:l0:events
L0_PIPELINE_MAXLEN=10000

Config	Default	Description
L0_PIPELINE_ENABLED	false	Master switch for the output pipeline
L0_PIPELINE_STREAM	mems:l0:events	Redis stream key used for memory events
L0_PIPELINE_MAXLEN	10000	Approximate max stream length (old entries are trimmed)

No new infrastructure is required: the events are written to the same Redis instance that backs L0.

Event Format

Every event is a flat field-value entry in the stream:

Field	Description
event_id	Stable id generated by Mems, usable for idempotency
event_type	write (fresh L0 snapshot) or append (buffer continuation)
timestamp	Publication time, UTC ISO-8601
tenant_id / user_id / agent_id / session_id / scope	Identity context; unset fields are empty strings
content	L0 memory text snapshot after this event
messages	JSON-encoded short-term buffer
expires_at	When the L0 snapshot expires (TTL end)

Consume With Your Own Code

Any Redis client works. Read from the beginning:

redis-cli XREAD COUNT 10 STREAMS mems:l0:events 0

Block and tail new events:

redis-cli XREAD BLOCK 5000 STREAMS mems:l0:events $

For parallel workers with acknowledged consumption, create a consumer group once, then read with XREADGROUP:

redis-cli XGROUP CREATE mems:l0:events consumers 0 MKSTREAM
redis-cli XREADGROUP GROUP consumers worker1 COUNT 10 BLOCK 5000 \
  STREAMS mems:l0:events >

Example consumer in Python:

import asyncio, json
import redis.asyncio as redis

async def main():
    client = redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)
    stream = "mems:l0:events"
    group = "consumers"
    try:
        await client.xgroup_create(stream, group, id="0", mkstream=True)
    except redis.ResponseError:
        pass  # group already exists

    while True:
        entries = await client.xreadgroup(group, "worker1", {stream: ">"}, count=10)
        for _, events in entries:
            for event_id, fields in events:
                print(event_id, fields["event_type"], fields["agent_id"],
                      fields["content"], json.loads(fields["messages"]))
                await client.xack(stream, group, event_id)

asyncio.run(main())

Guarantees

• Best-effort publication: if the stream write fails, Mems logs a warning but the memory write itself still succeeds. The pipeline never blocks POST /v1/mems/write.
• Bounded stream: entries are trimmed to approximately L0_PIPELINE_MAXLEN. Consumers that fall too far behind should use consumer groups or catch up from an earlier offset.
• Snapshot semantics: each event carries the full L0 snapshot after the write/append, so consumers can treat every entry as the current state of that agent session's hot memory.
• Same trust domain as L0: events contain the full memory content and live in the same Redis instance/database as the L0 snapshots. Access control and encryption are the same as for the rest of L0.
• No delete events: deletions are not published in v1. The natural L0 lifecycle is snapshot TTL expiration; there are no tombstone entries in the stream.