""" CONTROL_PLANE — shared memory region for commands, target symbols, venue lifecycle management, and system management. This is a SEPARATE region from the data plane (book/account/fulfilment). External systems communicate with MALKHUT via this region. All writes are atomic (single UVZINC01 frame). All reads are lock-free. """ from __future__ import annotations import json import struct import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Mapping, Optional, Sequence from malkhut.ipc.zinc_plane import ( SharedRegionReader, SharedRegionWriter, _decode_payload, _encode_payload, DEFAULT_REGION_SIZE, _HDR_SIZE, ) class ControlCommand(str, Enum): START = "START" STOP = "STOP" PAUSE = "PAUSE" RESUME = "RESUME" SET_SYMBOLS = "SET_SYMBOLS" CONNECT_VENUE = "CONNECT_VENUE" DISCONNECT_VENUE = "DISCONNECT_VENUE" SET_MODE = "SET_MODE" EMERGENCY_STOP = "EMERGENCY_STOP" HOT_RELOAD_POLICY = "HOT_RELOAD_POLICY" STATUS_REQUEST = "STATUS_REQUEST" class VenueLifecycle(str, Enum): DISCONNECTED = "DISCONNECTED" CONNECTING = "CONNECTING" CONNECTED = "CONNECTED" RECONNECTING = "RECONNECTING" ERROR = "ERROR" @dataclass(frozen=True, slots=True) class ControlPlaneFrame: """Single atomic control plane frame.""" command: str ts_ns: int target_symbols: tuple[str, ...] = () venue: str = "" venue_lifecycle: str = "" mode: str = "" policy_version: str = "" params: Mapping[str, Any] = field(default_factory=dict) source: str = "" ack_required: bool = False class MalkhutControlPlane: """ CONTROL_PLANE shared memory region. External systems (other agents, TUI, supervisor, venue adapters) write commands here. MALKHUT reads and processes them. Region name: malkhut_control (Zinc object: /dev/shm/zinc_malkhut_control) Protocol: 1. External system writes a ControlPlaneFrame to the region. 2. MALKHUT reads the frame, processes the command. 3. If ack_required, MALKHUT writes an ACK frame back. """ REGION_NAME = "malkhut_control" def __init__(self, capacity: int = DEFAULT_REGION_SIZE) -> None: self._writer = SharedRegionWriter(self.REGION_NAME, capacity) self._reader = SharedRegionReader(self.REGION_NAME) self._seq = 0 def publish_command(self, frame: ControlPlaneFrame) -> None: """Write a control command to the region.""" self._seq += 1 data = { "command": frame.command, "ts_ns": frame.ts_ns, "target_symbols": list(frame.target_symbols), "venue": frame.venue, "venue_lifecycle": frame.venue_lifecycle, "mode": frame.mode, "policy_version": frame.policy_version, "params": dict(frame.params), "source": frame.source, "ack_required": frame.ack_required, "seq": self._seq, } self._writer.write(data) def read_command(self, timeout_ms: int = 100) -> Optional[ControlPlaneFrame]: """Read latest control command. Returns None on timeout/error.""" try: data, seq = self._reader.read(timeout_ms) return ControlPlaneFrame( command=data.get("command", ""), ts_ns=data.get("ts_ns", 0), target_symbols=tuple(data.get("target_symbols", [])), venue=data.get("venue", ""), venue_lifecycle=data.get("venue_lifecycle", ""), mode=data.get("mode", ""), policy_version=data.get("policy_version", ""), params=data.get("params", {}), source=data.get("source", ""), ack_required=data.get("ack_required", False), ) except (ValueError, KeyError, TypeError): return None def publish_ack(self, command: str, ts_ns: int, status: str, details: str = "") -> None: """Write an ACK frame back to the control plane.""" self._seq += 1 data = { "command": f"ACK_{command}", "ts_ns": ts_ns, "status": status, "details": details, "seq": self._seq, } self._writer.write(data) def close(self) -> None: self._writer.close() self._reader.close()