orchestrata_ultimate.postman.sync_router

Reference › orchestrata_ultimate › postman › sync_router

The synchronous routing path of the broker, built as a mixin for Postman. The kernel's PipelineRunner runs in plain synchronous code, so it cannot use the async route; SyncRouterMixin.route_sync is what it calls for every hop, including the first packet of an intent. It serves SPEC-001 story US-002 and SPEC-002 (REQ-002-US-009-001, REQ-002-FR1-013, REQ-002-FR4-006). The mixin expects its host to provide the Redis client, handler table, auditor, message log and in-memory fallback.

route_sync does the following, in order:

1. Stamp the packet with a deterministic message id built from sender, recipient, subject, thread id and timestamp (cxp.deterministic_ids).
2. Drop duplicates: _is_duplicate_sync adds the id to a Redis set and treats a failed add as a duplicate, falling back to an in-memory set when Redis is down or errors; a duplicate returns None.
3. Append to the message log and to the thread's Redis list (_persist_to_thread_sync), the audit trail that get_thread_packets reads back.
4. Push the packet onto the recipient's queue (_enqueue_sync, a Redis list postman:q:<volk>, with the in-memory fallback), and write it to the CXP auditor.
5. _route_sync_deliver looks up the handler. With none registered, or when it returns nothing, the packet is marked queued and the method returns None. Otherwise it removes the queue entry and returns the handler's reply after _stamp_response gives the reply its own deterministic id, logs it, persists it and records the request-reply link in the auditor.

_dispatch_sync calls the handler directly; for an async handler, _run_async_handler uses asyncio.run, going through a one-thread pool when an event loop is already running. get_thread_packets returns a thread's packets from Redis or memory, and get_log returns string forms of everything routed.

At a glance

Class diagram

Dependency diagram

API

Classes

SyncRouterMixin

Decorators: traces(spec='001', story='US-002-001', reqs=['REQ-001-US-002-001']) · Line 36

Synchronous routing methods for PipelineRunner compatibility.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/sync_router.py

# mypy: ignore-errors
"""
Synchronous routing mixin for Postman.

Extracts all sync routing methods used by PipelineRunner
into a composable mixin class. Postman inherits from
SyncRouterMixin to gain these capabilities.

Requires the host class to initialise:
    - self._redis / self._redis_ok
    - self._handlers
    - self._auditor  (CXPAuditor)
    - self.message_log  (List[CXPPacket])
    - self._fallback_ids  (set)
    - self._in_memory_fallback_enqueue(dest, data)
"""

# Spec: 002 - Postman Message Broker

import asyncio
import concurrent.futures
import inspect
import json
import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.core.packet import DeliveryStatus as CXPDeliveryStatus
from orchestrata_ultimate.cxp.deterministic_ids import generate_deterministic_id

logger = logging.getLogger(__name__)


@traces(spec="001", story="US-002-001", reqs=["REQ-001-US-002-001"])
class SyncRouterMixin:
    """Synchronous routing methods for PipelineRunner compatibility.

    This mixin provides the sync routing pipeline used by the
    kernel's PipelineRunner.  It relies on attributes initialised
    by the host Postman class (see module docstring).
    """

    THREAD_PREFIX: str
    DEDUP_KEY: str
    _redis: Any
    _redis_ok: bool
    _handlers: Any
    _auditor: Any
    message_log: list[CXPPacket]
    _fallback_ids: set
    _in_memory_fallback_enqueue: Any

    @traces(spec="002", story="US-009", reqs=["REQ-002-US-009-001"])
    def route_sync(self, packet: CXPPacket) -> CXPPacket | None:
        """Synchronous message routing for PipelineRunner.

        Enqueue, persist, deliver, and return response.
        Uses Redis for persistence with in-memory fallback.

        Args:
            packet: The CXP packet to route.

        Returns:
            Response CXPPacket from the handler, or None if
            no handler registered or delivery queued.
        """
        packet.message_id = generate_deterministic_id(
            packet.from_agent,
            packet.to_agent,
            packet.subject,
            packet.thread_id,
            packet.timestamp,
        )
        if self._is_duplicate_sync(packet.message_id):
            return None
        self.message_log.append(packet)
        self._persist_to_thread_sync(packet)
        # Route by instance_id when available (swarm), fall back to alias
        queue_id = getattr(packet, "to_instance", "") or packet.to_agent
        self._enqueue_sync(queue_id, packet)
        self._auditor.log_packet(packet)
        return self._route_sync_deliver(packet)

    @traces(spec="001", story="US-002-001", reqs=["REQ-001-US-002-001"])
    def _route_sync_deliver(self, packet: CXPPacket) -> CXPPacket | None:
        """Handle delivery and response stamping for route_sync.

        Args:
            packet: The routed CXP packet.

        Returns:
            Stamped response packet or None.
        """
        # Handlers are registered by alias, not instance_id
        destination = packet.to_agent
        queue_id = getattr(packet, "to_instance", "") or destination
        if destination not in self._handlers:
            packet.delivery_status = CXPDeliveryStatus.QUEUED
            return None
        response = self._dispatch_sync(destination, packet)
        if not response:
            packet.delivery_status = CXPDeliveryStatus.QUEUED
            return None
        self._dequeue_sync(queue_id)
        return self._stamp_response(packet, response)

    def _stamp_response(self, packet: CXPPacket, response: CXPPacket) -> CXPPacket:
        """Stamp response with deterministic ID and persist.

        Args:
            packet: The original routed packet.
            response: The handler's response packet.

        Returns:
            The stamped response packet.
        """
        response.message_id = generate_deterministic_id(
            response.from_agent,
            response.to_agent,
            response.subject,
            response.thread_id,
            response.timestamp,
        )
        self.message_log.append(response)
        self._persist_to_thread_sync(response)
        self._auditor.log_packet(response)
        self._auditor.log_routing(
            packet.message_id,
            response.message_id,
            packet.thread_id,
            packet.subject,
        )
        return response

    @traces(
        spec="002",
        story="FR-001",
        reqs=["REQ-002-FR1-013", "REQ-002-US-003-008"],
    )
    def _is_duplicate_sync(self, message_id: str) -> bool:
        """Check if a message ID has already been seen.

        Uses Redis SADD for deduplication with in-memory fallback.

        Args:
            message_id: The deterministic message ID to check.

        Returns:
            True if the message is a duplicate, False otherwise.
        """
        if not self._redis_ok:
            if message_id in self._fallback_ids:
                return True
            self._fallback_ids.add(message_id)
            return False
        try:
            return not self._redis.sadd(self.DEDUP_KEY, message_id)
        except Exception as exc:
            logger.warning("Redis dedup check failed: %s", exc)
            return False

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-017"])
    def _persist_to_thread_sync(self, packet: CXPPacket) -> None:
        """Persist a packet to the Redis thread audit trail.

        Args:
            packet: The CXP packet to persist.
        """
        if not self._redis_ok:
            return
        try:
            data = json.dumps(packet.to_dict(), default=str)
            key = f"{self.THREAD_PREFIX}{packet.thread_id}"
            self._redis.rpush(key, data)
        except Exception as exc:
            logger.warning("Thread persist failed: %s", exc)

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-006"])
    def _enqueue_sync(self, volk_id: str, packet: CXPPacket) -> None:
        """Enqueue a packet to the sync delivery queue.

        Uses Redis LPUSH on ``postman:q:{volk_id}`` with
        in-memory fallback on transient errors.

        Args:
            volk_id: Destination volk identifier.
            packet: The CXP packet to enqueue.
        """
        fallback_data = {
            "packet_id": packet.message_id,
            "priority": 0,
        }
        if not self._redis_ok:
            self._in_memory_fallback_enqueue(volk_id, fallback_data)
            return
        try:
            data = json.dumps(packet.to_dict(), default=str)
            self._redis.lpush(f"postman:q:{volk_id}", data)
        except Exception as exc:
            logger.warning("Redis enqueue failed: %s", exc)
            self._in_memory_fallback_enqueue(volk_id, fallback_data)

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-007"])
    def _dequeue_sync(self, volk_id: str) -> str | None:
        """Dequeue a packet from the sync delivery queue.

        Args:
            volk_id: Volk identifier to dequeue from.

        Returns:
            Raw dequeued data string, or None.
        """
        if not self._redis_ok:
            return None
        try:
            return self._redis.rpop(f"postman:q:{volk_id}")
        except Exception as exc:
            logger.debug("Redis dequeue failed: %s", exc)
            return None

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-011"])
    def _dispatch_sync(self, destination: str, packet: CXPPacket) -> CXPPacket | None:
        """Dispatch a packet to a registered handler synchronously.

        Bridges async handlers using ThreadPoolExecutor when an
        event loop is already running.

        Args:
            destination: The target volk identifier.
            packet: The CXP packet to deliver.

        Returns:
            Response CXPPacket from the handler, or None.
        """
        handler = self._handlers[destination]
        if inspect.iscoroutinefunction(handler):
            return self._run_async_handler(handler, packet)
        return handler(packet)

    def _run_async_handler(self, handler: Any, packet: CXPPacket) -> CXPPacket | None:
        """Run an async handler from a sync context.

        Args:
            handler: The async handler callable.
            packet: The CXP packet to pass to the handler.

        Returns:
            Response CXPPacket from the handler, or None.
        """
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            with concurrent.futures.ThreadPoolExecutor() as pool:
                return pool.submit(asyncio.run, handler(packet)).result()
        return asyncio.run(handler(packet))

    @traces(
        spec="004",
        story="US-008",
        reqs=["REQ-004-US-008-005", "REQ-004-US-008-006"],
    )
    def get_thread_packets(self, thread_id: str) -> list[CXPPacket]:
        """Get all packets for a thread from Redis or in-memory.

        Tries Redis first, falls back to filtering the in-memory
        message log.

        Args:
            thread_id: The thread identifier to retrieve packets for.

        Returns:
            List of CXPPacket objects for the given thread.
        """
        if self._redis_ok:
            try:
                key = f"{self.THREAD_PREFIX}{thread_id}"
                raw_packets = self._redis.lrange(key, 0, -1)
                return [CXPPacket.from_dict(json.loads(r)) for r in raw_packets]
            except Exception as exc:
                logger.debug("Redis thread fetch failed: %s", exc)
        return [p for p in self.message_log if p.thread_id == thread_id]

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-008-001"])
    def get_log(self) -> list[str]:
        """Get string representations of all routed messages.

        Returns:
            List of repr strings for each logged packet.
        """
        return [repr(p) for p in self.message_log]