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.