feat: JSON-lines protocol with command and event dataclasses

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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

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")

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