+i<dZddlmZddlmZddlmZddlmZddlm Z GddeZ eGd d Z dd l Z dd l Z dd lmZmZe jeZd ZdZGddZGddZy )u core/bridge/openclaw_bridge.py OpenClaw<>Genesis Bridge — Inbound Message Schema (Story 8.01) AIVA RLM Nexus PRD v2, Module 8 Defines the standard message format for bidirectional communication between AIVA (running on Mac Mini via OpenClaw gateway) and the Genesis swarm running on the E: drive WSL2 environment. All messages conform to ``OpenClawMessage`` regardless of direction. Priority routing and expiry enforcement are the responsibility of the routing layer (BridgeWriter / BridgeReader — Stories 8.02 / 8.03), not the schema. Usage:: from core.bridge.openclaw_bridge import MessageDirection, OpenClawMessage from datetime import datetime, timezone import uuid msg = OpenClawMessage( message_id=str(uuid.uuid4()), session_id="session-abc123", direction=MessageDirection.AIVA_TO_GENESIS, payload={"intent": "task_request", "body": "Book George's Cairns call"}, priority=2, created_at=datetime.now(timezone.utc), ) ) annotations) dataclass)datetime)Enum)OptionalceZdZdZdZdZy)MessageDirectionaDirection of travel for an OpenClaw bridge message. Values ------ AIVA_TO_GENESIS Message originates on the Mac Mini (AIVA / OpenClaw gateway) and travels to the Genesis WSL2 environment for swarm execution. GENESIS_TO_AIVA Message originates in the Genesis swarm and travels to AIVA on the Mac Mini for injection into her active session or cron jobs. aiva_to_genesisgenesis_to_aivaN)__name__ __module__ __qualname____doc__AIVA_TO_GENESISGENESIS_TO_AIVA4/mnt/e/genesis-system/core/bridge/openclaw_bridge.pyr r ,s (O'Orr c\eZdZUdZded<ded<ded<ded<d ed <d ed <d Zded<y )OpenClawMessageu*Standard envelope for all OpenClaw<>Genesis bridge traffic. Fields ------ message_id : str Unique identifier for this message (UUID v4 recommended). session_id : str Identifier for the AIVA or Genesis session that produced the message. Used for grouping related messages and injecting context. direction : MessageDirection Whether this message is AIVA→Genesis or Genesis→AIVA. payload : dict Arbitrary structured data. Typical keys: * ``intent`` — classified intent signal from AIVA (AIVA_TO_GENESIS) * ``task_id`` — swarm task ID for result correlation * ``body`` — free-form text or structured result * ``injection`` — context to inject into AIVA's next session priority : int Routing priority for the message: * ``1`` — critical (SLA: route immediately, no batching) * ``2`` — high (SLA: route within 500 ms) * ``3`` — normal (SLA: route within 3 s) Validation of out-of-range values is intentionally left to the routing layer (BridgeWriter) so this schema remains a pure value object with no side effects. created_at : datetime UTC wall-clock time when the message was constructed. MUST be timezone-aware (``tzinfo`` set). expires_at : Optional[datetime] Optional absolute expiry time. A ``None`` value means the message does not expire. When set, BridgeReader must discard messages where ``expires_at < datetime.now(timezone.utc)``. str message_id session_idr directiondictpayloadintpriorityr created_atNzOptional[datetime] expires_at)r r rr__annotations__r rrrrrBs5#JOO MM%)J")rrN)rtimezonezbridge:queue:aiva_to_genesiszbridge:queue:genesis_to_aivaceZdZdZdZddZy) BridgeWriteruWrites OpenClawMessages to Redis bridge queues. Serialises each message to JSON and RPUSH-es it to the appropriate Redis list, based on the message's direction. Parameters ---------- redis_client: Redis client exposing an ``rpush(key, value)`` method. Pass a :class:`unittest.mock.MagicMock` in tests — no live Redis connection is required. c||_yNredisself redis_clients r__init__zBridgeWriter.__init__ ! rcK|jtjtj}|j}|j |j tj}||kr0tjd|j|jy|jjtjjk(rt}nt }|j|j"|jj|j$|j&|j(j|j|jjndd} |j*j-|t/j0|tjd|j|y#t2$r$tj5d|j|YywxYww) uSerialise *message* to JSON and RPUSH to the appropriate bridge queue. Parameters ---------- message: The :class:`OpenClawMessage` to enqueue. Returns ------- bool ``True`` on success. ``False`` when: * the message has expired (``expires_at`` is in the past), or * the underlying Redis call raises an exception. Notes ----- * Uses ``RPUSH`` (not ``LPUSH``) to preserve FIFO order. * Expired check normalises naive datetimes to UTC before comparison. * Serialises via ``json.dumps`` — never pickle. Ntzinfou5BridgeWriter: message %s expired at %s — discardingFrrrrrrr z%BridgeWriter: pushed message %s to %sTz-BridgeWriter: failed to push message %s to %s)r rnowr"utcr0replaceloggerdebugr isoformatrvaluer rBRIDGE_QUEUE_AIVA_TO_GENESISBRIDGE_QUEUE_GENESIS_TO_AIVArrrrr(rpushjsondumps Exception exception)r*messager2expires queue_keyrs rsendzBridgeWriter.sends4    ),,x||,C((G~~%!///>} K&&%%'     " "&6&F&F&L&L L4I4I ",,!,, **00((!,,668292D2D2P"",,.VZ    JJ  Y 7(; < LL7""      ?""    s+EG#"AF32G#3*G G#G  G#N)r@rreturnbool)r r rrr,rCrrrr$r$s "Rrr$c,eZdZdZdZ d ddZy) BridgeReaderuReads OpenClawMessages from Redis bridge queues using blocking pop. Calls ``BLPOP`` with a configurable timeout so callers can process the next available message without spinning in a tight loop. Parameters ---------- redis_client: Redis client exposing a ``blpop(keys, timeout)`` method that mirrors the redis-py ``blpop`` signature: * ``blpop(keys, timeout=0)`` — blocks up to *timeout* seconds. * Returns ``(key_bytes, value_bytes)`` on success. * Returns ``None`` on timeout (no message available). Pass a :class:`unittest.mock.MagicMock` in tests — no live Redis connection is required. c||_yr&r'r)s rr,zBridgeReader.__init__r-rc >K|jtjjk(rt}nt}|j j ||}|y|\}} tj|} t|d} |d} tj| } | j | jt j"} d} |j%dDtj|d} | j | jt j"} t'|d |d | |d |d | | S#tjtf$r}td|d||d}~wwxYw#t(ttf$r}td|d||d}~wwxYww)u-Pop the next message from the bridge queue for *direction*. Uses ``BLPOP`` (blocking left-pop) so the call parks the coroutine for up to *timeout_s* seconds rather than spinning. Parameters ---------- direction : MessageDirection Which queue to read from. Mapped as: * ``AIVA_TO_GENESIS`` → ``bridge:queue:aiva_to_genesis`` * ``GENESIS_TO_AIVA`` → ``bridge:queue:genesis_to_aiva`` timeout_s : int, optional Maximum seconds to block waiting for a message. Defaults to 1. Pass ``0`` for an immediate non-blocking pop. Returns ------- OpenClawMessage The next message if one was available within *timeout_s*. None If the queue was empty and the timeout elapsed (no exception). Raises ------ ValueError If the raw bytes retrieved from Redis cannot be decoded as JSON or are missing required ``OpenClawMessage`` fields. Notes ----- * Uses ``BLPOP``, never ``LPOP`` in a loop — a single blocking call. * ``timeout_s`` is passed through to Redis verbatim (not hardcoded). * Direction enum is resolved via ``.value`` to stay robust against module-reload edge cases in tests. * Datetime fields are parsed from ISO 8601 strings back to :class:`~datetime.datetime` objects with UTC timezone. * Direction string is parsed back to the :class:`MessageDirection` enum member. )timeoutNz(BridgeReader: malformed JSON from queue z: rrr/r rrrrr1z1BridgeReader: failed to deserialise message from )r8r rr9r:r(blpopr<loadsJSONDecodeError TypeError ValueErrorr fromisoformatr0r4r"r3getrKeyError) r*r timeout_srBresult_keyrawdataexc msg_directioncreated_at_rawrr s rpollzBridgeReader.polls` ??.>>DD D4I4I !!)Y!? >  c  ::c?D ,T+->?M!,/N!//?J  ('//x||/D -1Jxx %1%33D4FG $$,!+!3!38<rcs<#!(t(, ,*,* ,*z '   8 $==ccjBBr