| """Newline-delimited JSON protocol for the local Unix socket.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| from collections.abc import Callable |
| from dataclasses import dataclass |
| from typing import Any |
|
|
| from .errors import RelayError, ValidationError |
| from .service import RelayService |
|
|
| MAX_FRAME_BYTES = 128 * 1024 |
|
|
|
|
| @dataclass |
| class RelayProtocol: |
| service: RelayService |
|
|
| def handle(self, frame: dict[str, Any]) -> dict[str, Any]: |
| request_id = frame.get("id") |
| try: |
| if set(frame) - {"id", "caller", "method", "params"}: |
| raise ValidationError("unknown protocol fields") |
| caller = str(frame.get("caller", "")) |
| method = str(frame.get("method", "")) |
| params = frame.get("params", {}) |
| if not isinstance(params, dict): |
| raise ValidationError("params must be an object") |
| result = self._dispatch(caller, method, params) |
| return {"id": request_id, "ok": True, "result": result} |
| except RelayError as exc: |
| return { |
| "id": request_id, |
| "ok": False, |
| "error": {"code": exc.code, "message": str(exc)}, |
| } |
| except Exception: |
| return { |
| "id": request_id, |
| "ok": False, |
| "error": {"code": "internal_error", "message": "relay operation failed"}, |
| } |
|
|
| def _dispatch(self, caller: str, method: str, params: dict[str, Any]) -> dict[str, Any]: |
| handlers: dict[str, Callable[[], dict[str, Any]]] = { |
| "relay_health": lambda: self.service.health(caller), |
| "relay_list_agents": lambda: self.service.list_agents(caller), |
| "relay_send_message": lambda: self.service.send(caller, params), |
| "relay_read_inbox": lambda: self.service.read_inbox( |
| caller, after=params.get("after"), limit=int(params.get("limit", 50)) |
| ), |
| "relay_get_thread": lambda: self.service.get_thread( |
| caller, str(params.get("thread_id", "")) |
| ), |
| "relay_ack_message": lambda: self.service.acknowledge( |
| caller, str(params.get("message_id", "")) |
| ), |
| "relay_cancel_task": lambda: self.service.cancel_task( |
| caller, str(params.get("message_id", "")) |
| ), |
| } |
| try: |
| handler = handlers[method] |
| except KeyError as exc: |
| raise ValidationError(f"unknown relay method: {method}") from exc |
| return handler() |
|
|
| def handle_line(self, raw: bytes) -> bytes: |
| if len(raw) > MAX_FRAME_BYTES: |
| response: dict[str, Any] = { |
| "id": None, |
| "ok": False, |
| "error": {"code": "validation_error", "message": "frame too large"}, |
| } |
| else: |
| try: |
| parsed = json.loads(raw) |
| if not isinstance(parsed, dict): |
| raise ValueError |
| response = self.handle(parsed) |
| except (json.JSONDecodeError, ValueError): |
| response = { |
| "id": None, |
| "ok": False, |
| "error": {"code": "validation_error", "message": "invalid JSON object"}, |
| } |
| return (json.dumps(response, ensure_ascii=False) + "\n").encode() |
|
|
