fix(gateway): control-socket hardening from #92447 post-merge review

- bind under umask 0o177 so the socket is never world-connectable, even
  pre-chmod (review pt 3)
- verb handlers run in an executor: state-file reads stay off the
  adapter event loop (pt 2)
- inventory dedupes one multiplex gateway answering identify for several
  homes — one runtime record per pid, with regression test (pt 1)
- v1 wire contract (one request per connection) documented in the module
  docstring (pt 5); /tmp-unwritable skip in the short-home test

Live-verified: perms 600 at bind, identify 4.3ms via executor path, 20
rapid queries healthy.

gateway/control_socket.py

@@ -31,6 +31,11 @@ Transport:
 Never a TCP port. Filesystem/pipe ACLs are the auth boundary — the same
 trust model as ``gateway_state.json`` today.
 
+v1 wire contract: ONE request per connection — a single JSON line in, a
+single JSON line out, then the server closes. Clients must not rely on
+keep-alive or pipelining. Verb handlers may touch disk (they run in an
+executor server-side) but must stay fast; the client budget is small.
+
 Consumers (``hermes update --plan`` inventory, the post-update fleet version
 matrix) PREFER the socket when it answers and fall back to the existing
 state-file/scan layer when it doesn't — old gateways mid-upgrade and crashed
@@ -271,9 +276,16 @@ class GatewayControlServer:
         with contextlib.suppress(OSError):
             if bind_path.exists():
                 bind_path.unlink()
-        self._server = await asyncio.start_unix_server(
-            self._handle_connection, path=str(bind_path)
-        )
+        # Bind under a restrictive umask so the socket is never
+        # world-connectable, even for the instant before an explicit chmod
+        # could run. Restore the process umask immediately after.
+        old_umask = os.umask(0o177)
+        try:
+            self._server = await asyncio.start_unix_server(
+                self._handle_connection, path=str(bind_path)
+            )
+        finally:
+            os.umask(old_umask)
         with contextlib.suppress(OSError):
             os.chmod(bind_path, 0o600)
         self._bind_path = bind_path
@@ -379,7 +391,14 @@ class GatewayControlServer:
             )
             if not raw or len(raw) > _MAX_REQUEST_BYTES:
                 return
-            writer.write(self.handle_request_line(raw.rstrip(b"\n")))
+            # Handlers read state files from disk; keep that off the
+            # gateway's event loop (the same loop drives every platform
+            # adapter), so a fast-polling consumer can't stall heartbeats.
+            loop = asyncio.get_running_loop()
+            response = await loop.run_in_executor(
+                None, self.handle_request_line, raw.rstrip(b"\n")
+            )
+            writer.write(response)
             await writer.drain()
         except (asyncio.TimeoutError, ConnectionError, OSError):
             pass

hermes_cli/update_inventory.py

@@ -199,6 +199,11 @@ def collect_runtime_inventory() -> UpdatePlan:
                 except (TypeError, ValueError):
                     sock_pid = None
                 if sock_pid is not None:
+                    if sock_pid in seen_pids:
+                        # One multiplex gateway can answer identify for
+                        # several profile homes — one runtime record per
+                        # process, not per home.
+                        continue
                     seen_pids.add(sock_pid)
                     declared = identity.get("supervisor")
                     supervisor = (

tests/gateway/test_control_socket.py

@@ -50,7 +50,10 @@ def test_short_home_binds_in_home(tmp_path: Path):
     # system temp root directly.
     import tempfile
 
-    short_root = Path(tempfile.mkdtemp(prefix="hgw-", dir="/tmp"))
+    try:
+        short_root = Path(tempfile.mkdtemp(prefix="hgw-", dir="/tmp"))
+    except OSError:
+        pytest.skip("/tmp not writable on this host")
     try:
         short_home = short_root / ".hermes"
         short_home.mkdir()
@@ -337,6 +340,39 @@ def test_collect_fleet_versions_falls_back_to_state_file(tmp_path: Path, monkeyp
     assert "source" not in fleet[0]
 
 
+def test_runtime_inventory_dedupes_same_pid_across_homes(tmp_path: Path, monkeypatch):
+    """One multiplex gateway answering identify for two profile homes must
+    yield exactly ONE runtime record (reviewer point on #92447)."""
+    import hermes_cli.update_inventory as ui
+
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    profiles_root = tmp_path / "profiles"
+    (profiles_root / "coder").mkdir(parents=True)
+
+    monkeypatch.setattr(
+        "hermes_cli.profiles._get_default_hermes_home", lambda: home
+    )
+    monkeypatch.setattr(
+        "hermes_cli.profiles._get_profiles_root", lambda: profiles_root
+    )
+    monkeypatch.setattr(
+        "hermes_cli.gateway._get_service_pids", lambda all_profiles=False: set()
+    )
+    monkeypatch.setattr(
+        "hermes_cli.gateway.find_profile_gateway_processes", lambda: []
+    )
+    monkeypatch.setattr(
+        "gateway.control_socket.identify_gateway",
+        lambda h, **kw: _fake_identity(777, "SHA777"),
+    )
+
+    plan = ui.collect_runtime_inventory()
+    gws = [r for r in plan.runtimes if r.kind == "gateway"]
+    assert len(gws) == 1, [r.__dict__ for r in gws]
+    assert gws[0].pid == 777
+
+
 def test_runtime_inventory_prefers_socket_supervisor(tmp_path: Path, monkeypatch):
     import hermes_cli.update_inventory as ui