Redis¶

The reusable Redis layer drives hiredis from A11's libuv loop and presents binary-safe commands as native futures and Python awaitables. It maintains a multiplexed command connection and a separate Pub/Sub connection, so waiting for chunk-store changes does not block ordinary commands.

RedisChunkStore maps A11's ordered chunks, cursors, and terminal status onto Redis Streams, hashes, Lua transactions, and Pub/Sub. Persistence and multi-process access therefore use a widely deployed database and its existing operational tools. A11 does not add a separate storage service. Redis replication and scaling behavior still apply, including the Cluster routing constraints below.

Compose a client explicitly into a RedisChunkStore when the application owns connection policy:

from a11.redis import RedisClient
from a11.stores.redis_chunk_store import RedisChunkStore

client = RedisClient({"host": "redis.internal", "port": 6379})
await client.ready()
store = RedisChunkStore("agent-output", client=client)
await store.initialize()

Omitting client uses the process-global client. The global is created lazily from the environment and can be replaced for dependency injection with set_default_client. Existing stores retain the client they were constructed with; replacing the global affects only subsequently created stores.

Shared node identity¶

AsyncNode objects remain local to their processes. Two nodes share Redis data when their RedisChunkStore instances resolve to the same node ID and key prefix in the same Redis deployment. One process can append while another follows or replays the ordered records.

Finality and closure are stored state. Finalizing a Redis-backed node seals that named stream for every process, including readers that attach later. Use a new node ID or key prefix for an independent run.

Redis supplies persistence and cross-process cursors; it does not provide action discovery or connection-scoped dispatch. Use a session when peers need to discover and call actions over a live transport. Use Redis-backed nodes when named streams must outlive a process or accept records before a reader starts.

Configuration¶

A11_REDIS_URL takes precedence over all individual connection variables. Its supported form is redis://[user:password@]host[:port][/database]; credentials and hosts may use percent encoding, and IPv6 hosts use brackets.

Variable	Default	Meaning
`A11_REDIS_URL`	unset	Complete connection URL; overrides the variables below.
`A11_REDIS_HOST`	`127.0.0.1`	Redis host or routing endpoint.
`A11_REDIS_PORT`	`6379`	TCP port.
`A11_REDIS_USERNAME`	empty	ACL username.
`A11_REDIS_PASSWORD`	empty	Password, with or without an ACL username.
`A11_REDIS_DB`	`0`	Logical database selected on both connections.
`A11_REDIS_CLIENT_NAME`	`a11`	Base name reported by `CLIENT SETNAME`.
`A11_REDIS_CONNECT_TIMEOUT_MS`	`10000`	Finite connection timeout in milliseconds.
`A11_REDIS_COMMAND_TIMEOUT_MS`	`10000`	Default command and subscription-ack timeout.

Chunk-store layout has two additional settings:

Variable	Default	Meaning
`A11_REDIS_CHUNK_STORE_KEY_PREFIX`	`a11:`	Prefix placed before each per-node hash tag. Braces are rejected.
`A11_REDIS_CHUNK_STORE_INLINE_DATA_THRESHOLD_BYTES`	`262144`	Raw `Chunk.data` size above which the encoded chunk is stored in the blob hash.

An explicit RedisClientOptions or RedisChunkStoreOptions object takes the place of its corresponding environment configuration.

TLS and Redis Cluster endpoints

Native TLS is not enabled yet: rediss:// is rejected. Use a trusted TLS-terminating Redis proxy or tunnel and configure its plain Redis endpoint for A11.

Every key touched by one chunk-store script shares a Redis Cluster hash tag, so atomic Lua operations are cross-slot safe. The client does not yet follow MOVED or ASK redirects. For a sharded deployment, configure a cluster-aware proxy/routing endpoint, or connect only to the node that owns the relevant slot. Redis Cluster deployments must use database 0.

Chunk-store layout and atomicity¶

For node ID agent-output, the default base is a11:{chunk-store:agent-output}. A store owns six names under that base:

Suffix	Redis role
`:metadata`	Hash with ID, closure status, final sequence, size, cursors, and revision.
`:stream`	Redis Stream containing ordered chunk records and control history.
`:sequences`	Sequence-to-stream-entry index.
`:arrivals`	Arrival-order-to-sequence index.
`:blobs`	Encoded chunks whose raw data exceeds the inline threshold.
`:events`	Pub/Sub invalidation channel used by responsive getters.

Put, PutMany, closure, and data clearing each run as one Lua state-machine operation. Validation happens before mutation, so a rejected batch cannot leave partial indexes, blobs, metadata, or stream entries. Getters perform an atomic lookup, subscribe only if they must wait, and then recheck after the subscription acknowledgement; this closes the notification race with a concurrent write or close.

Stream records carry a storage kind. inline embeds the encoded chunk, redis references the separate blob hash, and tombstone preserves metadata after clear_data. The reserved s3 kind makes the format forward-compatible with an S3-compatible blob backend; reading it currently returns an UNIMPLEMENTED status.

General-purpose client¶

Commands are binary safe: Python accepts strings or bytes-like arguments and returns an owned RedisReply. Avoid blocking commands such as XREAD BLOCK and BLPOP on the multiplexed command connection. For wake-ups, use a subscription and ordinary non-blocking state queries, as RedisChunkStore does. A command's effective deadline is the earlier of its explicit absolute deadline and command_timeout.

a11.redis.client.RedisClient ¶

RedisClient(options: RedisClientOptions | dict[str, Any] | None = None)

An asynchronous, binary-safe hiredis/libuv client using A11 futures.

Begin connecting with validated native options.

A plain mapping is validated into the same bound options object used by C++, rather than creating a second Python configuration model.

options `property` ¶

options: RedisClientOptions

A copy of the client's connection options.

create `staticmethod` ¶

create(options: RedisClientOptions | dict[str, Any] | None = None) -> RedisClient

Create a client and begin connecting without blocking.

close ¶

close() -> None

Begin an idempotent asynchronous disconnect.

command `async` ¶

command(parts: Sequence[str | bytes | bytearray | memoryview], deadline: Time | None = None) -> RedisReply

Execute one binary-safe Redis command.

eval `async` ¶

eval(script: str | bytes, keys: Sequence[str | bytes], arguments: Sequence[str | bytes | bytearray | memoryview] = tuple(), deadline: Time | None = None) -> RedisReply

Execute Lua with every cluster-sensitive key declared.

ready `async` ¶

ready() -> None

Wait until command and Pub/Sub connections are initialized.

subscribe `async` ¶

subscribe(channel: str, deadline: Time | None = None) -> RedisSubscription

Subscribe and wait for Redis's acknowledgement.

a11.redis.client.RedisClientOptions ¶

RedisClientOptions(host: str = '127.0.0.1', port: SupportsInt = 6379, username: str = '', password: str = '', database: SupportsInt = 0, client_name: str = 'a11', connect_timeout: Any | None = None, command_timeout: Any | None = None)

Connection and timeout policy for Redis.

Construct validated Redis connection options.

client_name `property` `writable` ¶

client_name: str

Name reported through CLIENT SETNAME.

command_timeout `property` `writable` ¶

command_timeout: Duration

Default maximum time allowed for one command.

connect_timeout `property` `writable` ¶

connect_timeout: Duration

Maximum time allowed for establishing a connection.

database `property` `writable` ¶

database: int

Logical Redis database selected after connection.

host `property` `writable` ¶

host: str

Redis host name or IP address.

password `property` `writable` ¶

password: str

ACL password, if authentication is enabled.

port `property` `writable` ¶

port: int

Redis TCP port.

username `property` `writable` ¶

username: str

ACL username, if authentication is enabled.

from_environment `staticmethod` ¶

from_environment() -> RedisClientOptions

Read A11_REDIS_URL or the individual A11_REDIS_* variables.

from_url `staticmethod` ¶

from_url(url: str) -> RedisClientOptions

Parse a redis://[user:password@]host[:port][/database] URL.

validate ¶

validate() -> None

Raise if these connection options are invalid.

a11.redis.client.RedisReply ¶

An owned, binary-safe RESP value returned by RedisClient.

is_null `property` ¶

is_null: bool

Whether this is a null RESP value.

type `property` ¶

type: RedisReplyType

The RESP kind of this value.

as_boolean ¶

as_boolean() -> bool

Return the boolean payload, raising for another RESP kind.

as_bytes ¶

as_bytes() -> bytes

Return string payload bytes, raising for another RESP kind.

as_elements ¶

as_elements() -> list[RedisReply]

Return aggregate children; maps alternate key and value entries.

as_float ¶

as_float() -> float

Return the double payload, raising for another RESP kind.

as_integer ¶

as_integer() -> int

Return the integer payload, raising for another RESP kind.

debug_string ¶

debug_string() -> str

Return a diagnostic representation of the RESP value.

a11.redis.client.RedisReplyType ¶

RedisReplyType(value: SupportsInt)

The RESP value kind held by RedisReply.

Members:

NULL

STRING

INTEGER

DOUBLE

BOOLEAN

ARRAY

MAP

SET

a11.redis.client.RedisSubscription ¶

A non-buffering broadcast subscription for invalidation events.

channel `property` ¶

channel: str

The subscribed Redis channel.

generation `property` ¶

generation: int

The current broadcast generation.

wait `async` ¶

wait(after: int, deadline: Time | None = None) -> int

Wait until a message advances the generation beyond after.

a11.redis.client.default_client ¶

default_client() -> RedisClient

Return the process-global environment-configured Redis client.

a11.redis.client.set_default_client ¶

set_default_client(client: RedisClient) -> None

Replace the process-global client used by new Redis chunk stores.

a11.redis.client.reset_default_client ¶

reset_default_client() -> None

Clear the process-global client so the environment is read again.

Redis¶

Shared node identity¶

Configuration¶

Chunk-store layout and atomicity¶

General-purpose client¶

a11.redis.client.RedisClient ¶

options property ¶

create staticmethod ¶

close ¶

command async ¶

eval async ¶

ready async ¶

subscribe async ¶

a11.redis.client.RedisClientOptions ¶

client_name property writable ¶

command_timeout property writable ¶

connect_timeout property writable ¶

database property writable ¶

host property writable ¶

password property writable ¶

port property writable ¶

username property writable ¶

from_environment staticmethod ¶

from_url staticmethod ¶