Agent Codex: AG1_AetherBus Library

# Agent Codex: AG1_AetherBus Library

**Version:** (e.g., 0.6.0 - Increment after cleanup)

**Agent Type:** Core Infrastructure Library / Message Bus SDK

**Primary Goal:**
To provide a reliable, asynchronous, Redis Streams-based message bus for inter-agent communication within the AG1 ecosystem. It defines core message structures (Envelope), publishing/subscribing mechanisms, and RPC-like patterns.

**Key Components & Modules:**

1.  **`envelope.py` - The `Envelope` Class:**
    *   **Purpose**: Defines the standardized message wrapper for all AetherBus communications.
    *   **Core Fields**: `role`, `content`, `agent_name`, `user_id`, `session_code`, `reply_to`, `correlation_id`, `envelope_type`, `meta`, `timestamp`, `envelope_id`.
    *   **Functionality**: Serialization (`to_dict`), deserialization (`from_dict`), unique ID generation.

2.  **`keys.py` - `StreamKeyBuilder` Class:**
    *   **Purpose**: Provides a centralized and consistent way to generate Redis stream names and keys for various AetherBus entities (agent inboxes, edge responses, temporary RPC replies, registration channels, etc.).
    *   **Functionality**: Methods like `agent_inbox(name)`, `edge_response(edge_name, user_id)`, `edge_register(edge_name)`, `temp_response(unique_id_part)`.

3.  **`bus.py` - Core Bus Operations:**
    *   **`build_redis_url(...)`**: Constructs the Redis connection URL from configuration (environment variables or direct parameters).
    *   **`publish_envelope(redis_client, stream_name, envelope)`**: Serializes an `Envelope` and publishes it to a specified Redis Stream using `XADD`.
    *   **`subscribe_simple(redis_client, stream_name, callback, group_name, consumer_name, mkstream, block_ms, is_temp_stream)`**: A simplified, robust function for an agent to subscribe to a single stream using a consumer group, process messages via a callback, and handle acknowledgments.
    *   **`subscribe(redis_client, stream_patterns_or_names, callback, group, consumer, mkstream, batch_size, block_ms)`**: (If this is a more general or older version) A function for subscribing to one or more streams/patterns. *Focus for cleanup might be to ensure `subscribe_simple` is the primary, well-tested method for single stream listening by agents, or to make `subscribe` equally robust.*

4.  **`rpc.py` - Request-Response Pattern Implementation:**
    *   **`bus_rpc_call(redis_client, target_stream, request_envelope, timeout)`**:
        *   Sends `request_envelope` to `target_stream`.
        *   **Crucially, it manages a temporary, unique reply stream name which it sets in `request_envelope.reply_to` before sending.**
        *   Listens on this temporary reply stream for a single response envelope (as a JSON string).
        *   Handles timeouts.
        *   Returns the JSON string of the response envelope or `None`.
    *   **`bus_rpc_envelope(redis_client, target_inbox, request_envelope, timeout)`**:
        *   A wrapper around `bus_rpc_call`.
        *   Ensures `request_envelope.reply_to` is appropriately set (using a fallback if necessary) before calling `bus_rpc_call`.
        *   Takes the JSON string response from `bus_rpc_call` and deserializes it into an `Envelope` object (or an error dictionary).

5.  **`bus_adapterV2.py` - `BusAdapterV2` Class:**
    *   **Purpose**: An abstraction layer for agents to easily manage subscriptions to their inboxes and other streams.
    *   **Functionality**:
        *   Initializes with an `agent_id`, `core_handler` (like `agent.handle_bus_envelope`), `redis_client`, and `patterns` to listen on.
        *   Uses `subscribe_simple` (or `subscribe`) internally to manage the Redis Stream subscriptions and consumer groups.
        *   Provides `start()` and `stop()` methods for managing the lifecycle of the listeners.
        *   Handles dispatching received messages to the `core_handler`.
        *   May include methods like `add_subscription()` for dynamic subscriptions.

**Dependencies:**
*   `redis` (asyncio version: `redis.asyncio`)
*   Python 3.x (specify version, e.g., 3.9+)

**Key Design Principles for Robustness:**
*   Clear error handling and logging in all bus operations.
*   Resilience to Redis connection issues (e.g., retry mechanisms where appropriate).
*   Proper message acknowledgment (`XACK`) in subscription loops to prevent reprocessing of faulty messages.
*   Consistent envelope serialization and deserialization.
*   Well-defined and unique stream naming via `StreamKeyBuilder`.
*   Non-blocking operations suitable for an `asyncio` environment.

**Interaction with Other Agents:**
This library is the fundamental communication layer. All AetherBus agents (Muse2, HeartBeat, Echo, EthAgent, Gatekeeper, uFetch_ASIOne_Client_Agent, ufetch_asi_edge_handler, Relay, etc.) will use these components to send and receive messages.