rpc: unix-socket puppeteering server (raw keys + introspection + screen)

Run the TUI with --rpc <sock> and it renders normally while an asyncio
listener on the same loop lets another process drive it. Handlers touch
the UI directly (same loop, no thread hop), so every injected key has the
identical on-screen effect a keyboard would — the point for livestreaming.

Newline-delimited JSON. v1 methods: keys/text (raw injection, via the same
_press_keys the pilot uses; text can interleave wait:<ms> for a typed-out
look), state/view/screen/functions (introspection; screen() is a full
plain-text render of exactly what the viewer sees). No auth by design —
the socket is 0600, local only.

tests/rpc_smoke.py drives it end-to-end over a real socket (7 green).

1 files changed, 20 insertions, 1 deletions

diff --git a/idatui/app.py b/idatui/app.py
index 11d54aa..ebf928c 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -16,6 +16,7 @@ Design notes:
 
 from __future__ import annotations
 
+import asyncio
 import os
 import re
 import subprocess
@@ -1731,12 +1732,14 @@ class IdaTui(App):
     ]
 
     def __init__(self, url: str, db: str | None, keepalive: bool = True,
-                 open_path: str | None = None) -> None:
+                 open_path: str | None = None, rpc_path: str | None = None) -> None:
         super().__init__()
         self._url = url
         self._db = db
         self._open_path = open_path
         self._do_keepalive = keepalive
+        self._rpc_path = rpc_path
+        self._rpc = None
         self.client: IDAClient | None = None
         self.program: Program | None = None
         self._ka = None
@@ -1803,6 +1806,22 @@ class IdaTui(App):
         # (pseudocode) so app bindings work before anything is opened.
         self.query_one(DecompView).focus()
         self._connect()
+        if self._rpc_path:
+            self._start_rpc()
+
+    def _start_rpc(self) -> None:
+        from .rpc import RpcServer
+        self._rpc = RpcServer(self, self._rpc_path)
+
+        async def _serve() -> None:
+            await self._rpc.start()
+            self._status(f"rpc: listening on {self._rpc_path}")
+
+        asyncio.get_running_loop().create_task(_serve())
+
+    async def on_unmount(self) -> None:
+        if self._rpc is not None:
+            await self._rpc.stop()
 
     # -- status helper ----------------------------------------------------- #
     def _status(self, text: str) -> None: