Step1/pr3 ipc runtime

shared/protocol/COMMANDS.md

@@ -0,0 +1,15 @@
+# VibeMouse Agent Commands
+
+Semantic commands that can be sent to the agent via IPC or resolved from input events via bindings.
+
+| Command | Description |
+|---------|-------------|
+| `noop` | No operation; event is ignored |
+| `toggle_recording` | Start or stop voice recording |
+| `trigger_secondary_action` | In idle: send Enter. In recording: stop and send transcript to OpenClaw |
+| `submit_recording` | Stop recording and send transcript to OpenClaw |
+| `send_enter` | Send Enter key to focused input |
+| `workspace_left` | Switch workspace left (e.g. Hyprland) |
+| `workspace_right` | Switch workspace right |
+| `reload_config` | Reload config.json |
+| `shutdown` | Gracefully shut down the agent |

shared/protocol/EVENTS.md

@@ -0,0 +1,30 @@
+# VibeMouse Normalized Input Events
+
+Device-agnostic event names produced by the listener and consumed by the binding resolver.
+
+## Mouse events
+
+| Event | Description |
+|-------|-------------|
+| `mouse.side_front.press` | Front side button press |
+| `mouse.side_rear.press` | Rear side button press |
+
+## Keyboard events
+
+| Event | Description |
+|-------|-------------|
+| `hotkey.record_toggle` | Recording toggle hotkey (e.g. Ctrl+Alt+Space) |
+| `hotkey.recording_submit` | Recording submit hotkey (optional) |
+
+## Gesture events
+
+| Event | Description |
+|-------|-------------|
+| `gesture.up` | Upward gesture |
+| `gesture.down` | Downward gesture |
+| `gesture.left` | Leftward gesture |
+| `gesture.right` | Rightward gesture |
+
+## Event naming convention
+
+Events use dot-separated segments: `category.subcategory.action`. All lowercase, alphanumeric and underscore only.

shared/schema/ipc.schema.json

@@ -0,0 +1,47 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://vibemouse.local/schema/ipc.schema.json",
+  "title": "VibeMouse IPC Messages",
+  "description": "LPJSON message types for agent-listener communication over stdio",
+  "oneOf": [
+    { "$ref": "#/$defs/EventMessage" },
+    { "$ref": "#/$defs/CommandMessage" }
+  ],
+  "$defs": {
+    "EventMessage": {
+      "type": "object",
+      "required": ["type", "event"],
+      "additionalProperties": false,
+      "properties": {
+        "type": { "const": "event" },
+        "event": {
+          "type": "string",
+          "description": "Normalized input event name, e.g. mouse.side_front.press",
+          "pattern": "^[a-z][a-z0-9_]*(\\.[a-z][a-z0-9_]*)*$"
+        }
+      }
+    },
+    "CommandMessage": {
+      "type": "object",
+      "required": ["type", "command"],
+      "additionalProperties": false,
+      "properties": {
+        "type": { "const": "command" },
+        "command": {
+          "type": "string",
+          "enum": [
+            "noop",
+            "reload_config",
+            "send_enter",
+            "shutdown",
+            "submit_recording",
+            "toggle_recording",
+            "trigger_secondary_action",
+            "workspace_left",
+            "workspace_right"
+          ]
+        }
+      }
+    }
+  }
+}

tests/cli/test_main.py

@@ -33,6 +33,7 @@ def test_default_invocation_runs_app(self) -> None:
         cfg = SimpleNamespace(log_level="INFO")
         with (
             patch("vibemouse.main.load_config", return_value=cfg) as load_config,
+            patch("vibemouse.main.resolve_config_path", return_value="/tmp/config.json"),
             patch(
                 "vibemouse.main.VoiceMouseApp", return_value=app_instance
             ) as app_ctor,
@@ -42,20 +43,51 @@ def test_default_invocation_runs_app(self) -> None:
         self.assertEqual(rc, 0)
         self.assertEqual(load_config.call_count, 1)
         self.assertEqual(app_ctor.call_count, 1)
+        self.assertEqual(
+            app_ctor.call_args.kwargs,
+            {
+                "listener_mode": "inline",
+                "config_path": "/tmp/config.json",
+            },
+        )
         self.assertEqual(app_instance.run.call_count, 1)
 
     def test_explicit_run_subcommand_runs_app(self) -> None:
         app_instance = MagicMock()
         cfg = SimpleNamespace(log_level="INFO")
         with (
             patch("vibemouse.main.load_config", return_value=cfg),
+            patch("vibemouse.main.resolve_config_path", return_value="/tmp/config.json"),
             patch("vibemouse.main.VoiceMouseApp", return_value=app_instance),
         ):
             rc = main(["run"])
 
         self.assertEqual(rc, 0)
         self.assertEqual(app_instance.run.call_count, 1)
 
+    def test_agent_run_off_mode_is_forwarded_to_app(self) -> None:
+        app_instance = MagicMock()
+        cfg = SimpleNamespace(log_level="INFO")
+        with (
+            patch("vibemouse.main.load_config", return_value=cfg),
+            patch(
+                "vibemouse.main.resolve_config_path",
+                return_value="/tmp/agent-config.json",
+            ),
+            patch("vibemouse.main.VoiceMouseApp", return_value=app_instance) as app_ctor,
+        ):
+            rc = main(["agent", "run", "--listener", "off"])
+
+        self.assertEqual(rc, 0)
+        self.assertEqual(
+            app_ctor.call_args.kwargs,
+            {
+                "listener_mode": "off",
+                "config_path": "/tmp/agent-config.json",
+            },
+        )
+        self.assertEqual(app_instance.run.call_count, 1)
+
     def test_deploy_subcommand_dispatches_to_deploy(self) -> None:
         with (
             patch("vibemouse.main.run_deploy", return_value=5) as run_deploy,

tests/core/test_app.py

@@ -11,7 +11,12 @@
 from typing import cast
 from unittest.mock import patch
 
-from vibemouse.core.commands import COMMAND_SEND_ENTER, EVENT_MOUSE_SIDE_FRONT_PRESS
+from vibemouse.core.commands import (
+    COMMAND_RELOAD_CONFIG,
+    COMMAND_SEND_ENTER,
+    COMMAND_SHUTDOWN,
+    EVENT_MOUSE_SIDE_FRONT_PRESS,
+)
 from vibemouse.app import VoiceMouseApp
 
 
@@ -80,7 +85,10 @@ def test_set_recording_status_writes_recording_payload(self) -> None:
                 dict[str, object],
                 json.loads(status_file.read_text(encoding="utf-8")),
             )
-            self.assertEqual(payload, {"recording": True, "state": "recording"})
+            self.assertEqual(
+                payload,
+                {"recording": True, "state": "recording", "listener_mode": "inline"},
+            )
 
     def test_set_recording_status_writes_idle_payload(self) -> None:
         subject = self._make_subject()
@@ -98,7 +106,37 @@ def test_set_recording_status_writes_idle_payload(self) -> None:
                 dict[str, object],
                 json.loads(status_file.read_text(encoding="utf-8")),
             )
-            self.assertEqual(payload, {"recording": False, "state": "idle"})
+            self.assertEqual(
+                payload,
+                {"recording": False, "state": "idle", "listener_mode": "inline"},
+            )
+
+    def test_set_recording_status_includes_ipc_port_when_command_server_is_running(self) -> None:
+        subject = self._make_subject()
+        with tempfile.TemporaryDirectory(prefix="vibemouse-status-") as tmp:
+            status_file = Path(tmp) / "status.json"
+            setattr(subject, "_config", SimpleNamespace(status_file=status_file))
+            setattr(subject, "_command_server", SimpleNamespace(port=43125))
+
+            set_status = cast(
+                Callable[[bool], None],
+                getattr(subject, "_set_recording_status"),
+            )
+            set_status(False)
+
+            payload = cast(
+                dict[str, object],
+                json.loads(status_file.read_text(encoding="utf-8")),
+            )
+            self.assertEqual(
+                payload,
+                {
+                    "ipc_port": 43125,
+                    "listener_mode": "inline",
+                    "recording": False,
+                    "state": "idle",
+                },
+            )
 
 
 class VoiceMouseAppButtonBehaviorTests(unittest.TestCase):
@@ -304,6 +342,34 @@ def test_handle_input_event_routes_through_binding_resolver(self) -> None:
 
         self.assertEqual(send_enter_calls, ["enter"])
 
+    def test_execute_command_reload_config_dispatches_to_reload_handler(self) -> None:
+        subject = self._make_subject()
+        reload_calls: list[bool] = []
+        setattr(subject, "_command_lock", threading.RLock())
+        setattr(subject, "_reload_config", lambda: reload_calls.append(True))
+
+        execute_command = cast(
+            Callable[[str], None],
+            getattr(subject, "_execute_command"),
+        )
+        execute_command(COMMAND_RELOAD_CONFIG)
+
+        self.assertEqual(reload_calls, [True])
+
+    def test_execute_command_shutdown_sets_stop_event(self) -> None:
+        subject = self._make_subject()
+        stop_event = threading.Event()
+        setattr(subject, "_command_lock", threading.RLock())
+        setattr(subject, "_stop_event", stop_event)
+
+        execute_command = cast(
+            Callable[[str], None],
+            getattr(subject, "_execute_command"),
+        )
+        execute_command(COMMAND_SHUTDOWN)
+
+        self.assertTrue(stop_event.is_set())
+
     def test_transcribe_and_output_openclaw_uses_openclaw_sender(self) -> None:
         subject = self._make_subject()
         recording = SimpleNamespace(duration_s=1.0, path=Path("/tmp/transcribe.wav"))

tests/ipc/__init__.py

@@ -0,0 +1 @@
+# IPC tests

tests/ipc/test_client_server.py

@@ -0,0 +1,99 @@
+"""Tests for IPC client-server round-trip over pipes."""
+
+from __future__ import annotations
+
+import io
+import json
+import socket
+import struct
+import threading
+import time
+
+from vibemouse.ipc.client import IPCClient
+from vibemouse.ipc.messages import make_command_message, write_lpjson_frame
+from vibemouse.ipc.server import AgentCommandServer, IPCServer
+
+
+def test_ipc_client_send_event() -> None:
+    """IPCClient.send_event writes valid LPJSON to stdout (simulating pipe)."""
+    stdout_buf = io.BytesIO()
+    stdin_buf = io.BytesIO()
+    client = IPCClient(stdin=stdin_buf, stdout=stdout_buf)
+    client.send_event("hotkey.record_toggle")
+    data = stdout_buf.getvalue()
+    assert len(data) >= 4
+    length, = struct.unpack("<I", data[:4])
+    assert length == len(data) - 4
+    payload = json.loads(data[4:].decode("utf-8"))
+    assert payload == {"type": "event", "event": "hotkey.record_toggle"}
+
+
+def test_ipc_server_receives_event() -> None:
+    """IPCServer receives event from reader and calls on_event."""
+    payload = {"type": "event", "event": "mouse.side_rear.press"}
+    body = json.dumps(payload).encode("utf-8")
+    frame = struct.pack("<I", len(body)) + body
+    reader = io.BytesIO(frame)
+    writer = io.BytesIO()
+    received: list[str] = []
+    server = IPCServer(
+        reader=reader,
+        writer=writer,
+        on_event=lambda e: received.append(e),
+    )
+    server.start()
+    time.sleep(0.1)
+    server.stop()
+    assert received == ["mouse.side_rear.press"]
+
+
+def test_ipc_server_send_command() -> None:
+    """IPCServer.send_command writes valid LPJSON to writer."""
+    reader = io.BytesIO()  # empty, no events
+    writer = io.BytesIO()
+    server = IPCServer(
+        reader=reader,
+        writer=writer,
+        on_event=lambda e: None,
+    )
+    server.send_command("shutdown")
+    data = writer.getvalue()
+    assert len(data) >= 4
+    length, = struct.unpack("<I", data[:4])
+    assert length == len(data) - 4
+    payload = json.loads(data[4:].decode("utf-8"))
+    assert payload == {"type": "command", "command": "shutdown"}
+
+
+def test_ipc_server_receives_command_when_handler_is_configured() -> None:
+    payload = {"type": "command", "command": "reload_config"}
+    body = json.dumps(payload).encode("utf-8")
+    frame = struct.pack("<I", len(body)) + body
+    reader = io.BytesIO(frame)
+    received: list[str] = []
+    server = IPCServer(reader=reader, on_command=lambda cmd: received.append(cmd))
+    server.start()
+    time.sleep(0.1)
+    server.stop()
+    assert received == ["reload_config"]
+
+
+def test_agent_command_server_accepts_loopback_command() -> None:
+    received: list[str] = []
+    ready = threading.Event()
+
+    def on_command(command_name: str) -> None:
+        received.append(command_name)
+        ready.set()
+
+    server = AgentCommandServer(on_command=on_command)
+    server.start()
+    try:
+        with socket.create_connection(("127.0.0.1", server.port), timeout=2) as conn:
+            stream = conn.makefile("rwb")
+            write_lpjson_frame(stream, make_command_message("shutdown"))
+            stream.close()
+        assert ready.wait(timeout=2)
+        assert received == ["shutdown"]
+    finally:
+        server.stop()