hms-mediaengine/packages/protocol/hms_protocol/connection.py

"""IPC-Verbindung zwischen Control Core und Renderer (PLAN.md §6.2).

Server (Renderer-seitig) und Client (Control-Core-seitig) auf 127.0.0.1.
Verbindungsablauf:

1. Client verbindet, sendet hello mit Protokollversion + Capabilities
2. Server prüft Version, antwortet welcome mit eigenen Capabilities
3. Server sendet vollständigen state snapshot
4. danach inkrementelle Deltas mit monotoner Revision
5. Heartbeat mindestens alle 500 ms in beide Richtungen
6. Ack für jedes zustandsändernde Command (idempotent über message_id)
7. Nach Reconnect: Re-Sync (neuer Snapshot), Deltas erst danach akzeptiert
"""

from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass

from hms_protocol.envelope import PROTOCOL_VERSION, Envelope, MessageType
from hms_protocol.framing import encode_frame, read_frame_async

HEARTBEAT_INTERVAL_S = 0.5  # §6.2: mindestens alle 500 ms


@dataclass
class ProtocolError(Exception):
    """Protokollverstoß; Verbindung wird getrennt."""

    code: str
    message: str

    def __str__(self) -> str:
        return f"{self.code}: {self.message}"


@dataclass
class HandshakeInfo:
    """Ergebnis des Handshakes mit Capabilities der Gegenseite."""

    peer_name: str
    peer_capabilities: dict
    protocol_version: int = PROTOCOL_VERSION


class IpcServer:
    """Renderer-seitiger IPC-Server. Lauscht ausschließlich auf 127.0.0.1.

    Liefert dem Renderer:
    - accept(handler) → wartet auf Client, führt Handshake durch
    - receive() → nächste Nachricht (command/event/heartbeat)
    - send(envelope) → Nachricht an Control Core
    """

    def __init__(self, port: int = 0, host: str = "127.0.0.1") -> None:
        if host not in ("127.0.0.1", "localhost", "::1"):
            raise ValueError("IPC-Server darf nur auf Loopback lauschen (§6.2)")
        self._host = host
        self._port = port
        self._server: asyncio.Server | None = None
        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None
        self._handler_task: asyncio.Task | None = None
        self._heartbeat_task: asyncio.Task | None = None
        self._last_peer_heartbeat_ns: int = 0
        self.capabilities: dict = {}
        self.peer_name: str = ""
        self.peer_capabilities: dict = {}
        self.name: str = "hms-renderer"

    @property
    def port(self) -> int:
        if self._server is None:
            return self._port
        return self._server.sockets[0].getsockname()[1] if self._server.sockets else self._port

    async def start(self) -> int:
        """Startet den Listener; gibt den tatsächlichen Port zurück."""
        self._server = await asyncio.start_server(self._on_client, self._host, self._port)
        return self.port

    async def stop(self) -> None:
        if self._heartbeat_task:
            self._heartbeat_task.cancel()
        if self._handler_task:
            self._handler_task.cancel()
        if self._writer:
            self._writer.close()
        if self._server:
            self._server.close()
            await self._server.wait_closed()

    async def _on_client(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
    ) -> None:
        """Nimmt genau einen Client an (V1: eine Verbindung)."""
        self._reader = reader
        self._writer = writer
        # auf hello warten
        hello_raw = await read_frame_async(reader)
        action = hello_raw.get("payload", {}).get("action")
        if hello_raw.get("type") != "command" or action != "hello":
            got = hello_raw.get("type")
            raise ProtocolError("EXPECTED_HELLO", f"got {got}")
        peer_caps = hello_raw.get("payload", {}).get("capabilities", {})
        peer_name = hello_raw.get("payload", {}).get("name", "unknown")
        self.peer_name = peer_name
        self.peer_capabilities = peer_caps
        if hello_raw.get("protocol_version") != PROTOCOL_VERSION:
            err = Envelope(
                type=MessageType.ERROR,
                payload={"code": "VERSION_MISMATCH", "expected": PROTOCOL_VERSION},
            )
            writer.write(_encode_envelope(err))
            await writer.drain()
            writer.close()
            raise ProtocolError("VERSION_MISMATCH", "client protocol mismatch")
        # welcome senden
        welcome = Envelope(
            type=MessageType.EVENT,
            payload={"action": "welcome", "name": self.name, "capabilities": self.capabilities},
        )
        writer.write(_encode_envelope(welcome))
        await writer.drain()
        self._last_peer_heartbeat_ns = time.monotonic_ns()
        self._heartbeat_task = asyncio.get_running_loop().create_task(self._send_heartbeats())

    async def _send_heartbeats(self) -> None:
        while True:
            await asyncio.sleep(HEARTBEAT_INTERVAL_S)
            if self._writer is None:
                return
            hb = Envelope(type=MessageType.HEARTBEAT, payload={"source": self.name})
            self._writer.write(_encode_envelope(hb))
            await self._writer.drain()

    async def receive(self) -> Envelope | None:
        """Liest die nächste Nachricht; None bei Verbindungsabbruch."""
        if self._reader is None:
            return None
        try:
            raw = await read_frame_async(self._reader)
        except (asyncio.IncompleteReadError, ConnectionError):
            return None
        if raw.get("type") == "heartbeat":
            self._last_peer_heartbeat_ns = time.monotonic_ns()
        return Envelope.model_validate(raw)

    async def send(self, envelope: Envelope) -> None:
        if self._writer is None:
            raise ConnectionError("IPC client not connected")
        self._writer.write(_encode_envelope(envelope))
        await self._writer.drain()

    @property
    def peer_alive(self) -> bool:
        """True, wenn letzter Peer-Heartbeat < 2× Intervall zurückliegt."""
        if self._last_peer_heartbeat_ns == 0:
            return False
        return (time.monotonic_ns() - self._last_peer_heartbeat_ns) < 2 * HEARTBEAT_INTERVAL_S * 1e9


class IpcClient:
    """Control-Core-seitiger IPC-Client. Verbindet sich mit 127.0.0.1:port.

    - connect() → Handshake, gibt HandshakeInfo zurück
    - receive() → nächste Nachricht (snapshot/event/ack/heartbeat)
    - send(envelope) → Nachricht an Renderer
    - Nach Reconnect: connect() erneut → neuer Snapshot (Re-Sync)
    """

    def __init__(self, port: int, host: str = "127.0.0.1", name: str = "control-core") -> None:
        if host not in ("127.0.0.1", "localhost", "::1"):
            raise ValueError("IPC-Client darf nur Loopback verbinden (§6.2)")
        self._host = host
        self._port = port
        self._name = name
        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None
        self._heartbeat_task: asyncio.Task | None = None
        self._last_peer_heartbeat_ns: int = 0
        self.capabilities: dict = {}

    async def connect(self) -> HandshakeInfo:
        """Baut Verbindung auf, führt Handshake durch."""
        self._reader, self._writer = await asyncio.open_connection(self._host, self._port)
        hello = Envelope(
            type=MessageType.COMMAND,
            payload={"action": "hello", "name": self._name, "capabilities": self.capabilities},
        )
        self._writer.write(_encode_envelope(hello))
        await self._writer.drain()
        raw = await read_frame_async(self._reader)
        if raw.get("type") == "error":
            raise ProtocolError(
                raw.get("payload", {}).get("code", "REMOTE_ERROR"),
                str(raw.get("payload", {})),
            )
        if raw.get("type") != "event" or raw.get("payload", {}).get("action") != "welcome":
            raise ProtocolError("EXPECTED_WELCOME", f"got {raw.get('type')}")
        if raw.get("protocol_version") != PROTOCOL_VERSION:
            sent = raw.get("protocol_version")
            raise ProtocolError("VERSION_MISMATCH", f"server sent version {sent}")
        self._last_peer_heartbeat_ns = time.monotonic_ns()
        self._heartbeat_task = asyncio.get_running_loop().create_task(self._send_heartbeats())
        return HandshakeInfo(
            peer_name=raw.get("payload", {}).get("name", "unknown"),
            peer_capabilities=raw.get("payload", {}).get("capabilities", {}),
        )

    async def disconnect(self) -> None:
        if self._heartbeat_task:
            self._heartbeat_task.cancel()
            self._heartbeat_task = None
        if self._writer:
            self._writer.close()
            try:
                await self._writer.wait_closed()
            except (ConnectionError, asyncio.CancelledError):
                pass
            self._writer = None
            self._reader = None

    async def receive(self) -> Envelope | None:
        if self._reader is None:
            return None
        try:
            raw = await read_frame_async(self._reader)
        except (asyncio.IncompleteReadError, ConnectionError):
            return None
        if raw.get("type") == "heartbeat":
            self._last_peer_heartbeat_ns = time.monotonic_ns()
        return Envelope.model_validate(raw)

    async def send(self, envelope: Envelope) -> None:
        if self._writer is None:
            raise ConnectionError("IPC not connected")
        self._writer.write(_encode_envelope(envelope))
        await self._writer.drain()

    async def _send_heartbeats(self) -> None:
        while True:
            await asyncio.sleep(HEARTBEAT_INTERVAL_S)
            if self._writer is None:
                return
            hb = Envelope(type=MessageType.HEARTBEAT, payload={"source": self._name})
            self._writer.write(_encode_envelope(hb))
            await self._writer.drain()

    @property
    def peer_alive(self) -> bool:
        if self._last_peer_heartbeat_ns == 0:
            return False
        return (time.monotonic_ns() - self._last_peer_heartbeat_ns) < 2 * HEARTBEAT_INTERVAL_S * 1e9


def _encode_envelope(envelope: Envelope) -> bytes:
    return encode_frame(envelope.model_dump(mode="json"))