From d376e529089f61163af68c4d37d93e180fbf8871 Mon Sep 17 00:00:00 2001 From: Mortdecai Date: Sun, 29 Mar 2026 19:05:15 -0400 Subject: [PATCH] feat: JSON-lines protocol with command and event dataclasses Co-Authored-By: Claude Sonnet 4.6 --- src/kitty_workbench/protocol.py | 135 ++++++++++++++++++++++++++++++++ tests/conftest.py | 13 +++ tests/test_protocol.py | 48 ++++++++++++ 3 files changed, 196 insertions(+) create mode 100644 src/kitty_workbench/protocol.py create mode 100644 tests/conftest.py create mode 100644 tests/test_protocol.py diff --git a/src/kitty_workbench/protocol.py b/src/kitty_workbench/protocol.py new file mode 100644 index 0000000..01bea76 --- /dev/null +++ b/src/kitty_workbench/protocol.py @@ -0,0 +1,135 @@ +"""JSON-lines protocol for server ↔ TUI communication.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field, asdict +from typing import Optional + + +# --- Server → TUI Commands --- + +@dataclass +class InitCmd: + project: str + title: str + image_protocol: str + description: str = "" + cmd: str = field(default="init", init=False) + +@dataclass +class DisplayCmd: + widget: str + pane: str = "main" + clear: bool = False + content: Optional[str] = None + items: Optional[list] = None + id: Optional[str] = None + label: Optional[str] = None + placeholder: Optional[str] = None + cmd: str = field(default="display", init=False) + +@dataclass +class ImageCmd: + path: str + pane: str = "main" + clear: bool = True + cmd: str = field(default="image", init=False) + +@dataclass +class LogCmd: + entry: str + level: str = "info" + cmd: str = field(default="log", init=False) + +@dataclass +class ClearCmd: + pane: str = "main" + cmd: str = field(default="clear", init=False) + +@dataclass +class LayoutCmd: + panes: dict = field(default_factory=dict) + cmd: str = field(default="layout", init=False) + +@dataclass +class NotifyCmd: + message: str + level: str = "info" + cmd: str = field(default="notify", init=False) + +@dataclass +class ShutdownCmd: + cmd: str = field(default="shutdown", init=False) + + +# --- TUI → Server Events --- + +@dataclass +class ReadyEvent: + event: str = field(default="ready", init=False) + +@dataclass +class ChecklistToggleEvent: + pane: str + index: int + label: str + checked: bool + event: str = field(default="checklist_toggle", init=False) + +@dataclass +class ButtonClickEvent: + pane: str + id: str + event: str = field(default="button_click", init=False) + +@dataclass +class InputSubmitEvent: + pane: str + id: str + value: str + event: str = field(default="input_submit", init=False) + + +# --- Registry for decoding --- + +_CMD_TYPES = { + "init": InitCmd, + "display": DisplayCmd, + "image": ImageCmd, + "log": LogCmd, + "clear": ClearCmd, + "layout": LayoutCmd, + "notify": NotifyCmd, + "shutdown": ShutdownCmd, +} + +_EVENT_TYPES = { + "ready": ReadyEvent, + "checklist_toggle": ChecklistToggleEvent, + "button_click": ButtonClickEvent, + "input_submit": InputSubmitEvent, +} + + +def encode_message(msg) -> str: + """Serialize a command or event dataclass to a JSON line (no trailing newline).""" + return json.dumps(asdict(msg), separators=(",", ":")) + + +def decode_message(line: str): + """Deserialize a JSON line to a command or event dataclass. Returns None if unknown.""" + data = json.loads(line) + if "cmd" in data: + cls = _CMD_TYPES.get(data["cmd"]) + if cls is None: + return None + kwargs = {k: v for k, v in data.items() if k != "cmd"} + return cls(**kwargs) + elif "event" in data: + cls = _EVENT_TYPES.get(data["event"]) + if cls is None: + return None + kwargs = {k: v for k, v in data.items() if k != "event"} + return cls(**kwargs) + return None diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..471a9e5 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,13 @@ +import pytest + +@pytest.fixture +def tmp_workbench(tmp_path): + """Provide a temporary ~/Kitty-Workbench directory.""" + wb = tmp_path / "Kitty-Workbench" + wb.mkdir() + return wb + +@pytest.fixture +def socket_path(tmp_path): + """Provide a temporary socket path.""" + return str(tmp_path / "test.sock") diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..564f843 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,48 @@ +import json +from kitty_workbench.protocol import ( + DisplayCmd, ImageCmd, LogCmd, ClearCmd, LayoutCmd, InitCmd, ShutdownCmd, + ReadyEvent, ChecklistToggleEvent, ButtonClickEvent, InputSubmitEvent, + encode_message, decode_message, +) + + +def test_display_cmd_round_trip(): + cmd = DisplayCmd(widget="markdown", content="# Hello", pane="main", clear=False) + line = encode_message(cmd) + decoded = decode_message(line) + assert decoded == cmd + + +def test_init_cmd_round_trip(): + cmd = InitCmd(project="io102", title="Test Project", image_protocol="sixel") + line = encode_message(cmd) + decoded = decode_message(line) + assert decoded == cmd + assert decoded.project == "io102" + + +def test_checklist_event_round_trip(): + evt = ChecklistToggleEvent(pane="sidebar", index=2, label="Check R412", checked=True) + line = encode_message(evt) + decoded = decode_message(line) + assert decoded == evt + assert decoded.checked is True + + +def test_ready_event_round_trip(): + evt = ReadyEvent() + line = encode_message(evt) + decoded = decode_message(line) + assert isinstance(decoded, ReadyEvent) + + +def test_encode_produces_single_line(): + cmd = LogCmd(entry="test entry", level="info") + line = encode_message(cmd) + assert "\n" not in line + assert json.loads(line)["cmd"] == "log" + + +def test_decode_unknown_message_returns_none(): + result = decode_message('{"cmd": "unknown_thing", "data": 123}') + assert result is None