fix: HTTP transport + RPC endpoint to prevent bridge zombie accumulation

apps/memos-local-plugin/adapters/hermes/memos_provider/__init__.py

@@ -71,7 +71,6 @@
     ensure_viewer_daemon,
     kill_zombie_bridges,
     probe_viewer_status,
-    startup_lock_active,
 )
 
 
@@ -428,48 +427,54 @@ def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[ov
             else:
                 viewer_status = "free"  # force stdio fallback below
         elif viewer_status == "free":
-            # Re-probe after a short wait only when another process may be
-            # mid-startup (startup lock is held). On a cold first-launch the
-            # lock doesn't exist, so we skip the delay entirely.
-            if startup_lock_active():
-                time.sleep(1.0)
-                viewer_status = probe_viewer_status()
-            if viewer_status == "running_memos" and self._connect_http_bridge(session_id):
-                logger.info(
-                    "MemOS: bridge ready (HTTP, late probe) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
+            # Brief wait: another process may be about to bind the port.
+            time.sleep(0.5)
+            viewer_status = probe_viewer_status()
+            if viewer_status == "running_memos":
+                if self._connect_http_bridge(session_id):
+                    logger.info(
+                        "MemOS: bridge ready (HTTP, late probe) session=%s platform=%s (episode deferred)",
+                        self._session_id,
+                        self._platform,
+                    )
 
         if self._bridge is None:
             try:
                 ensure_viewer_daemon()
             except Exception as err:
                 logger.warning("MemOS: viewer daemon check failed — %s", err)
-            new_bridge: MemosBridgeClient | None = None
-            try:
-                new_bridge = MemosBridgeClient()
-                # Register the fallback LLM handler BEFORE we open the
-                # session so it is available the very first time the
-                # plugin's facade asks for help (e.g. on the first
-                # `turn.start` retrieval call).
-                new_bridge.register_host_handler(
-                    "host.llm.complete",
-                    self._handle_host_llm_complete,
-                )
-                self._bridge = new_bridge
-                self._open_session(session_id, timeout=60.0)
-                logger.info(
-                    "MemOS: bridge ready (stdio) session=%s platform=%s (episode deferred)",
-                    self._session_id,
-                    self._platform,
-                )
-            except Exception as err:
-                logger.warning("MemOS: bridge init failed — %s", err)
-                if new_bridge is not None:
-                    with contextlib.suppress(Exception):
-                        new_bridge.close()
-                self._bridge = None
+
+            # Re-probe: daemon may have just started — try HTTP before stdio.
+            if self._bridge is None:
+                viewer_status = probe_viewer_status()
+                if viewer_status == "running_memos":
+                    self._connect_http_bridge(session_id)
+
+            if self._bridge is None:
+                new_bridge: MemosBridgeClient | None = None
+                try:
+                    new_bridge = MemosBridgeClient()
+                    # Register the fallback LLM handler BEFORE we open the
+                    # session so it is available the very first time the
+                    # plugin's facade asks for help (e.g. on the first
+                    # `turn.start` retrieval call).
+                    new_bridge.register_host_handler(
+                        "host.llm.complete",
+                        self._handle_host_llm_complete,
+                    )
+                    self._bridge = new_bridge
+                    self._open_session(session_id, timeout=60.0)
+                    logger.info(
+                        "MemOS: bridge ready (stdio) session=%s platform=%s (episode deferred)",
+                        self._session_id,
+                        self._platform,
+                    )
+                except Exception as err:
+                    logger.warning("MemOS: bridge init failed — %s", err)
+                    if new_bridge is not None:
+                        with contextlib.suppress(Exception):
+                            new_bridge.close()
+                    self._bridge = None
         # Register a Hermes plugin hook to capture tool calls as they
         # happen. The `post_tool_call` hook fires after every tool
         # dispatch (write_file, terminal, search_files, etc.) with the
@@ -1970,6 +1975,14 @@ def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> N
                 ensure_viewer_daemon()
             except Exception as err:
                 logger.warning("MemOS: viewer daemon check failed during reconnect — %s", err)
+
+            # Re-probe: daemon may have just started — try HTTP before stdio.
+            viewer_status = probe_viewer_status()
+            if viewer_status == "running_memos":
+                if self._connect_http_bridge(session_id, timeout=timeout):
+                    logger.info("MemOS: reconnected via HTTP (post-daemon start)")
+                    return
+
             new_bridge: MemosBridgeClient | None = None
             try:
                 new_bridge = MemosBridgeClient()

apps/memos-local-plugin/adapters/hermes/memos_provider/bridge_client.py

@@ -19,6 +19,9 @@
 import shutil
 import subprocess
 import threading
+import time
+import urllib.error
+import urllib.request
 
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -531,4 +534,123 @@ def _resolve(self, msg: dict[str, Any]) -> None:
             entry["error"] = msg["error"]
         else:
             entry["result"] = msg.get("result")
-        entry["event"].set()
+
+
+class MemosHttpClient:
+    """JSON-RPC 2.0 client that talks to the daemon bridge over HTTP.
+
+    Drop-in replacement for ``MemosBridgeClient`` when a daemon is already
+    running on the viewer port. Instead of spawning a new subprocess, this
+    client POSTs JSON-RPC envelopes to ``/api/v1/rpc`` on the daemon's HTTP
+    server. This eliminates zombie bridge accumulation.
+
+    Limitations vs. stdio client:
+    - No reverse-direction RPC (``host.llm.complete``). The daemon's own
+      stdio bridge handles host LLM fallback internally.
+    - No ``notify()`` (notifications have no response; use ``request()``
+      for all calls — the daemon handles both).
+    """
+
+    def __init__(
+        self,
+        *,
+        port: int = 18800,
+        host: str = "127.0.0.1",
+        api_key: str | None = None,
+    ) -> None:
+        self._base_url = f"http://{host}:{port}/api/v1/rpc"
+        self._lock = threading.Lock()
+        self._next_id = 1
+        self._closed = False
+        self._api_key = api_key
+        self._host_handlers: dict[str, Callable[[dict[str, Any]], Any]] = {}
+
+    @property
+    def pid(self) -> int:
+        """Return 0 — HTTP client has no subprocess."""
+        return 0
+
+    # ─── Public API (matches MemosBridgeClient) ──
+
+    def request(
+        self,
+        method: str,
+        params: Any = None,
+        *,
+        timeout: float = 30.0,
+    ) -> dict[str, Any]:
+        if self._closed:
+            raise BridgeError("transport_closed", "HTTP client is closed")
+        with self._lock:
+            rpc_id = self._next_id
+            self._next_id += 1
+
+        envelope = {
+            "jsonrpc": "2.0",
+            "id": rpc_id,
+            "method": method,
+            "params": params,
+        }
+        payload = json.dumps(envelope, ensure_ascii=False).encode("utf-8")
+
+        req = urllib.request.Request(
+            self._base_url,
+            data=payload,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        if self._api_key:
+            req.add_header("Authorization", f"Bearer {self._api_key}")
+
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                body = resp.read(4_194_304)  # 4 MiB safety cap
+        except urllib.error.HTTPError as exc:
+            # Try to read the JSON-RPC error body
+            try:
+                err_body = exc.read(4_194_304)
+                err_json = json.loads(err_body.decode("utf-8", errors="replace"))
+                err_obj = err_json.get("error") or {}
+                raise BridgeError(
+                    err_obj.get("data", {}).get("code") or str(err_obj.get("code", "internal")),
+                    err_obj.get("message", f"HTTP {exc.code}"),
+                    err_obj.get("data"),
+                ) from exc
+            except (json.JSONDecodeError, UnicodeDecodeError):
+                raise BridgeError("internal", f"HTTP {exc.code}: {exc.reason}") from exc
+        except urllib.error.URLError as exc:
+            raise BridgeError("transport_closed", str(exc)) from exc
+
+        result = json.loads(body.decode("utf-8", errors="replace"))
+        if "error" in result:
+            e = result["error"]
+            raise BridgeError(
+                (e.get("data") or {}).get("code") or str(e.get("code", "internal")),
+                e.get("message", "unknown error"),
+                e.get("data"),
+            )
+        return result.get("result") or {}
+
+    def notify(self, method: str, params: Any = None) -> None:
+        """No-op for HTTP — use request() for all calls."""
+        try:
+            self.request(method, params, timeout=5.0)
+        except Exception:
+            pass
+
+    def on_event(self, cb: Callable[[dict[str, Any]], None]) -> None:
+        """No-op — SSE events not supported over HTTP transport."""
+
+    def on_log(self, cb: Callable[[dict[str, Any]], None]) -> None:
+        """No-op — SSE logs not supported over HTTP transport."""
+
+    def register_host_handler(
+        self,
+        method: str,
+        handler: Callable[[dict[str, Any]], Any],
+    ) -> None:
+        """Store the handler but it won't be called (daemon handles host LLM internally)."""
+        self._host_handlers[method] = handler
+
+    def close(self) -> None:
+        self._closed = True

apps/memos-local-plugin/adapters/hermes/memos_provider/daemon_manager.py

@@ -479,6 +479,108 @@ def kill_zombie_bridges() -> int:
     return killed
 
 
+def probe_viewer_status() -> str:
+    """Return the current viewer daemon status without side effects.
+
+    Returns one of: ``"running_memos"``, ``"free"``, ``"blocked"``.
+    This is a cheap, lock-free probe suitable for deciding whether to
+    spawn a new stdio bridge or connect to the existing daemon over HTTP.
+    """
+    return _probe_viewer()
+
+
+def startup_lock_active() -> bool:
+    """Return True when the viewer-start.lock directory exists.
+
+    Used by the provider to skip the cold-start sleep when there is no
+    concurrent daemon launch in progress.
+    """
+    return (_plugin_root() / "daemon" / "viewer-start.lock").exists()
+
+
+def kill_zombie_bridges() -> int:
+    """Kill all bridge.cjs processes that are NOT the daemon on port 18800.
+
+    Returns the number of zombies killed. The daemon (the process that
+    owns port 18800) is left alone. This should be called early in the
+    provider's lifecycle to clean up leftovers from crashed sessions.
+    """
+    # Find the PID that owns port 18800 (the real daemon).
+    # ss(8) is Linux-only; fall back to lsof on macOS.
+    daemon_pid: int | None = None
+    try:
+        ss_out = subprocess.check_output(
+            ["ss", "-tlnp"],
+            timeout=2.0,
+            text=True,
+        )
+        for line in ss_out.splitlines():
+            if ":18800" in line:
+                # ss output: users:(("node",pid=21246,fd=24))
+                m = re.search(r"pid=(\d+)", line)
+                if m:
+                    daemon_pid = int(m.group(1))
+                    break
+    except Exception:
+        pass
+
+    if daemon_pid is None:
+        # macOS fallback — lsof is available on both Linux and macOS.
+        try:
+            lsof_out = subprocess.check_output(
+                ["lsof", "-iTCP:18800", "-sTCP:LISTEN", "-n", "-P"],
+                timeout=2.0,
+                text=True,
+            )
+            for line in lsof_out.splitlines()[1:]:  # skip header
+                parts = line.split()
+                if len(parts) >= 2:
+                    try:
+                        daemon_pid = int(parts[1])
+                        break
+                    except ValueError:
+                        continue
+        except Exception:
+            pass
+
+    # If we still can't identify the daemon PID, skip killing entirely to
+    # avoid terminating the daemon itself.
+    if daemon_pid is None:
+        logger.debug("MemOS: zombie scan skipped — could not identify daemon PID on port 18800")
+        return 0
+
+    # Find all bridge.cjs processes
+    killed = 0
+    try:
+        ps_out = subprocess.check_output(
+            ["ps", "aux"],
+            timeout=2.0,
+            text=True,
+        )
+        for line in ps_out.splitlines():
+            if "bridge.cjs" not in line or "grep" in line:
+                continue
+            parts = line.split()
+            if len(parts) < 2:
+                continue
+            try:
+                pid = int(parts[1])
+            except ValueError:
+                continue
+            if pid == daemon_pid:
+                continue
+            try:
+                os.kill(pid, signal.SIGTERM)
+                killed += 1
+                logger.info("MemOS: killed zombie bridge pid=%d", pid)
+            except (OSError, ProcessLookupError):
+                pass
+    except Exception as err:
+        logger.debug("MemOS: zombie scan failed — %s", err)
+
+    return killed
+
+
 def wait_for_process_exit(pid: int, timeout: float = 5.0) -> bool:
     """Wait for a process to exit.
 

apps/memos-local-plugin/bridge.cts

@@ -468,6 +468,9 @@ async function main(): Promise<void> {
       }
     }
 
+    bridgeStatus?.markConnected();
+    bridgeHeartbeat = bridgeStatus?.startHeartbeat();
+
     const shutdownDaemon = async (sig: string) => {
       process.stderr.write(`bridge: daemon received ${sig}, shutting down\n`);
       removeOwnedPidFile();