feat(host-agent): Backend-Anbindung — RVS + Bridge-Registry + Brain-Tools (Phase B-D)

Damit kann ARIA Host-Agenten tatsaechlich nutzen (Ende-zu-Ende):

- RVS: host_hello/host_ping/host_command/host_result in ALLOWED_TYPES.
- Bridge: _hosts-Registry (host_hello/host_ping/host_result), _host_list,
  _host_request (requestId->Future wie Satelliten), HTTP-Endpoints
  /internal/host-list und /internal/host (op via action).
- Brain: Tools host_list/host_exec/host_read/host_write/host_info/
  host_screenshot + _dispatch_host (ruft /internal/host*). host_screenshot
  speichert das PNG unter /shared/host-screenshots. host_exec kann sudo=true.

Naechste (optionale) Phase E: Host-Agenten in der Diagnostic-Flotte anzeigen.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

aria-brain/agent.py

@@ -20,6 +20,7 @@ import json
 import logging
 import os
 import re
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -1262,6 +1263,104 @@ META_TOOLS = [
             },
         },
     },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_list",
+            "description": (
+                "Zeigt die ARIA-Host-Agenten (Rechner, auf denen ein Agent DIREKT "
+                "laeuft und sich per RVS meldet) die ONLINE sind + was sie koennen "
+                "(exec/read/write/info/screenshot). Ein Host-Agent gibt Dir vollen "
+                "Zugriff auf GENAU DIESEN Rechner — auch wenn er hinter NAT/Firewall "
+                "sitzt. Nutze das, wenn Stefan etwas 'auf meinem Laptop/PC/Server X' "
+                "machen will, das kein Geraet im Netz ist."
+            ),
+            "parameters": {"type": "object", "properties": {}},
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_exec",
+            "description": (
+                "Fuehrt EIN Shell-Kommando auf einem Host-Agenten aus (bash -lc). "
+                "Gibt exit_code + stdout (gefenstert: contains/offset/max_chars) + "
+                "stderr. Fuer Root-Rechte sudo=true setzen (Agent nutzt root/"
+                "SUDO_PASSWORD/NOPASSWD automatisch). Sei vorsichtig — das ist "
+                "vollwertiger Rechnerzugriff."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "host": {"type": "string", "description": "Host-ID/Name aus host_list."},
+                    "cmd": {"type": "string", "description": "Shell-Kommando."},
+                    "sudo": {"type": "boolean", "description": "Mit Root-Rechten ausfuehren."},
+                    "timeout": {"type": "number", "description": "max. Laufzeit in Sekunden (Default 60)."},
+                },
+                "required": ["host", "cmd"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_read",
+            "description": "Liest eine Datei von einem Host-Agenten (Text). Fuer Binaerdateien liefert der Agent Base64.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "host": {"type": "string", "description": "Host-ID/Name."},
+                    "path": {"type": "string", "description": "Absoluter Pfad auf dem Rechner."},
+                },
+                "required": ["host", "path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_write",
+            "description": "Schreibt Text in eine Datei auf einem Host-Agenten (ueberschreibt; append=true haengt an).",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "host": {"type": "string", "description": "Host-ID/Name."},
+                    "path": {"type": "string", "description": "Zielpfad."},
+                    "text": {"type": "string", "description": "Inhalt."},
+                    "append": {"type": "boolean", "description": "Anhaengen statt ueberschreiben."},
+                },
+                "required": ["host", "path", "text"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_info",
+            "description": "System-Info eines Host-Agenten: OS, CPU/RAM/Disk-Auslastung, Uptime, IP.",
+            "parameters": {
+                "type": "object",
+                "properties": {"host": {"type": "string", "description": "Host-ID/Name."}},
+                "required": ["host"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "host_screenshot",
+            "description": (
+                "Macht ein Bildschirmfoto auf einem Host-Agenten (nur wenn dort eine "
+                "grafische Session laeuft). Gibt an, dass ein Bild erstellt wurde; das "
+                "Bild wird der Konversation angehaengt."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {"host": {"type": "string", "description": "Host-ID/Name."}},
+                "required": ["host"],
+            },
+        },
+    },
     {
         "type": "function",
         "function": {
@@ -2428,6 +2527,114 @@ class Agent:
         except Exception as exc:
             return f"FEHLER: Satellit/Bridge nicht erreichbar: {exc}"
 
+    def _dispatch_host(self, name: str, arguments: dict) -> str:
+        """host_list / host_exec / host_read / host_write / host_info /
+        host_screenshot — via Bridge (/internal/host*) → RVS → Host-Agent."""
+        import base64 as _b64
+
+        def _post(path: str, body: dict, timeout: float) -> dict:
+            data = json.dumps(body).encode("utf-8")
+            req = urllib.request.Request(f"{BRIDGE_URL}{path}", data=data, method="POST",
+                                         headers={"Content-Type": "application/json"})
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return json.loads(resp.read().decode("utf-8", "ignore"))
+        try:
+            if name == "host_list":
+                result = _post("/internal/host-list", {}, 10)
+                hosts = result.get("hosts") or []
+                if not hosts:
+                    return ("Gerade ist kein Host-Agent online. (Ein Host-Agent laeuft "
+                            "direkt auf einem Rechner und erlaubt Dir, ihn zu steuern — "
+                            "auch hinter NAT/Firewall.)")
+                lines = []
+                for h in hosts:
+                    status = "online" if h.get("online") else "offline"
+                    caps = ", ".join(h.get("caps") or [])
+                    lines.append(f"- {h.get('name')} (id={h.get('hostId')}, {status}, "
+                                 f"{h.get('os', '')}; kann: {caps})")
+                return "Host-Agenten (Rechner):\n" + "\n".join(lines)
+
+            host = (arguments.get("host") or "").strip()
+            if not host:
+                return "FEHLER: host ist Pflicht (siehe host_list)."
+
+            if name == "host_exec":
+                cmd = (arguments.get("cmd") or "").strip()
+                if not cmd:
+                    return "FEHLER: cmd ist Pflicht."
+                params = {"cmd": cmd}
+                if arguments.get("sudo"):
+                    params["sudo"] = True
+                timeout = float(arguments.get("timeout") or 60)
+                result = _post("/internal/host", {"host": host, "action": "exec",
+                                                  "params": params, "timeout": timeout + 10},
+                               timeout + 20)
+                if not result.get("ok"):
+                    return f"FEHLER: {result.get('error')}"
+                r = result.get("result") or {}
+                note = f"exit={r.get('exit_code')}"
+                if r.get("truncated"):
+                    note += f" (stdout gekuerzt, {r.get('total_chars')} Zeichen gesamt)"
+                out = f"[{note}]\n{r.get('body', '')}"
+                stderr = (r.get("stderr") or "").strip()
+                if stderr:
+                    out += f"\n--- stderr ---\n{stderr[:1500]}"
+                return out
+
+            if name == "host_read":
+                path = (arguments.get("path") or "").strip()
+                result = _post("/internal/host", {"host": host, "action": "read",
+                                                  "params": {"path": path}}, 30)
+                if not result.get("ok"):
+                    return f"FEHLER: {result.get('error')}"
+                r = result.get("result") or {}
+                try:
+                    text = _b64.b64decode(r.get("base64", "")).decode("utf-8", "replace")
+                except Exception:
+                    text = "(Binaerdatei — nicht als Text darstellbar)"
+                trunc = " (gekuerzt)" if r.get("truncated") else ""
+                return f"{r.get('path')} ({r.get('size')} Bytes){trunc}:\n{text[:8000]}"
+
+            if name == "host_write":
+                text = arguments.get("text") or ""
+                params = {"path": (arguments.get("path") or "").strip(),
+                          "base64": _b64.b64encode(text.encode("utf-8")).decode("ascii"),
+                          "append": bool(arguments.get("append"))}
+                result = _post("/internal/host", {"host": host, "action": "write",
+                                                  "params": params}, 30)
+                if not result.get("ok"):
+                    return f"FEHLER: {result.get('error')}"
+                r = result.get("result") or {}
+                return f"OK — {r.get('bytes')} Bytes nach {r.get('path')} geschrieben."
+
+            if name == "host_info":
+                result = _post("/internal/host", {"host": host, "action": "info",
+                                                  "params": {}}, 20)
+                if not result.get("ok"):
+                    return f"FEHLER: {result.get('error')}"
+                return f"System-Info {host}:\n" + json.dumps(result.get("result") or {},
+                                                             ensure_ascii=False, indent=2)
+
+            if name == "host_screenshot":
+                result = _post("/internal/host", {"host": host, "action": "screenshot",
+                                                  "params": {}}, 30)
+                if not result.get("ok"):
+                    return f"FEHLER: {result.get('error')}"
+                r = result.get("result") or {}
+                try:
+                    d = "/shared/host-screenshots"
+                    os.makedirs(d, exist_ok=True)
+                    fp = os.path.join(d, f"{host}-{int(time.time())}.png")
+                    with open(fp, "wb") as f:
+                        f.write(_b64.b64decode(r.get("base64", "")))
+                    return f"Screenshot von {host} gespeichert: {fp} ({r.get('bytes')} Bytes)."
+                except Exception as exc:
+                    return f"Screenshot erstellt ({r.get('bytes')} Bytes), Speichern fehlgeschlagen: {exc}"
+
+            return f"FEHLER: unbekanntes Host-Tool {name}"
+        except Exception as exc:
+            return f"FEHLER: Host/Bridge nicht erreichbar: {exc}"
+
     def _dispatch_tool(self, name: str, arguments: dict, project_id: str = "") -> str:
         """Fuehrt einen Tool-Call aus und gibt ein kurzes Text-Resultat zurueck.
         Niemals werfen — Fehler werden als Text-Resultat reportet damit Claude
@@ -3220,6 +3427,9 @@ class Agent:
                 return f"OK — Projekt '{updated['name']}' ist wieder ein normaler Chat."
             if name in ("satellite_list", "satellite_devices", "satellite_command"):
                 return self._dispatch_satellite(name, arguments)
+            if name in ("host_list", "host_exec", "host_read", "host_write",
+                        "host_info", "host_screenshot"):
+                return self._dispatch_host(name, arguments)
             if name == "vm_register":
                 pid = (project_id or "").strip()
                 if not pid:

bridge/aria_bridge.py

@@ -766,6 +766,11 @@ class ARIABridge:
         # requestId → Future (sat_devices / sat_result), analog _pending_flux.
         self._satellites: dict[str, dict] = {}
         self._pending_sat: dict[str, asyncio.Future] = {}
+        # Host-Agenten (Direktzugriff auf einen Rechner). hostId → {name, os,
+        # caps, control, last_seen}. Registrierung via host_hello/host_ping.
+        # _pending_host: requestId → Future (host_result), analog _pending_sat.
+        self._hosts: dict[str, dict] = {}
+        self._pending_host: dict[str, asyncio.Future] = {}
         # Compute-Fleet: GPU-Worker (voxtral/whisper/f5tts/llm) melden sich per
         # worker_hello, halten sich per worker_ping frisch. instanceId →
         # {service, node, gpus, model, busy, last_seen}. Genutzt fuer die
@@ -3550,6 +3555,40 @@ class ARIABridge:
                             self._satellites[sid]["caps"], self._satellites[sid]["control"])
             return
 
+        elif msg_type == "host_hello":
+            hid = (payload.get("hostId") or "").strip()
+            if hid:
+                new = hid not in self._hosts
+                self._hosts[hid] = {
+                    "hostId": hid,
+                    "name": payload.get("name") or hid,
+                    "os": payload.get("os") or "",
+                    "caps": payload.get("caps") or [],
+                    "control": bool(payload.get("control")),
+                    "last_seen": time.time(),
+                }
+                if new:
+                    logger.info("[host] Agent online: %s (%s) caps=%s control=%s",
+                                hid, self._hosts[hid]["name"],
+                                self._hosts[hid]["caps"], self._hosts[hid]["control"])
+            return
+
+        elif msg_type == "host_ping":
+            hid = (payload.get("hostId") or "").strip()
+            if hid and hid in self._hosts:
+                self._hosts[hid]["last_seen"] = time.time()
+            return
+
+        elif msg_type == "host_result":
+            req_id = payload.get("requestId", "")
+            future = self._pending_host.get(req_id)
+            if future is not None and not future.done():
+                future.set_result(payload)
+            hid = (payload.get("hostId") or "").strip()
+            if hid and hid in self._hosts:
+                self._hosts[hid]["last_seen"] = time.time()
+            return
+
         elif msg_type == "worker_hello":
             iid = (payload.get("instanceId") or "").strip()
             if iid:
@@ -4523,6 +4562,24 @@ class ARIABridge:
                 elif method in ("GET", "POST") and path == "/internal/worker-list":
                     # Diagnostic/Brain fragt: welche Compute-Worker sind online (Flotte).
                     await _send_response(writer, 200, {"ok": True, "workers": self._worker_list()})
+                elif method == "POST" and path == "/internal/host-list":
+                    # Brain fragt: welche Host-Agenten (Rechner) sind online + Capabilities.
+                    await _send_response(writer, 200, {"ok": True, "hosts": self._host_list()})
+                elif method == "POST" and path == "/internal/host":
+                    # Brain-Tool: Kommando an einen Host-Agenten.
+                    # body: {host, action, params?, timeout?}
+                    try:
+                        data = json.loads(body.decode("utf-8", "ignore"))
+                    except Exception as exc:
+                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
+                        return
+                    result = await self._host_request(
+                        host=str(data.get("host") or ""),
+                        action=str(data.get("action") or ""),
+                        params=data.get("params") if isinstance(data.get("params"), dict) else {},
+                        timeout=float(data.get("timeout") or 60.0),
+                    )
+                    await _send_response(writer, 200, result)
                 elif method == "POST" and path == "/internal/satellite":
                     # Brain-Tool: Discovery oder Command an einen Satelliten.
                     # body: {op:'discover'|'command', satellite, device?, action?, params?}
@@ -4862,6 +4919,44 @@ class ARIABridge:
         finally:
             self._pending_sat.pop(request_id, None)
 
+    def _host_list(self) -> list[dict]:
+        """Bekannte Host-Agenten (frisch = in den letzten 5 Min gesehen)."""
+        now = time.time()
+        return [{
+            "hostId": h["hostId"], "name": h.get("name") or h["hostId"],
+            "os": h.get("os") or "", "caps": h.get("caps") or [],
+            "control": bool(h.get("control")),
+            "online": (now - h.get("last_seen", 0)) < 300,
+        } for h in self._hosts.values()]
+
+    async def _host_request(self, host: str = "", action: str = "",
+                            params: Optional[dict] = None,
+                            timeout: float = 60.0) -> dict:
+        """Schickt host_command an einen Host-Agenten (via RVS) und wartet auf
+        host_result. Muster identisch zu _satellite_request."""
+        if self.ws_rvs is None:
+            return {"ok": False, "error": "RVS-Verbindung nicht aktiv"}
+        request_id = str(uuid.uuid4())
+        loop = asyncio.get_event_loop()
+        future: asyncio.Future = loop.create_future()
+        self._pending_host[request_id] = future
+        try:
+            msg = {"type": "host_command",
+                   "payload": {"requestId": request_id, "host": host,
+                               "action": action, "params": params or {}},
+                   "timestamp": int(time.time() * 1000)}
+            ok = await self._send_to_rvs(msg)
+            if not ok:
+                return {"ok": False, "error": "Host-Request konnte nicht gesendet werden"}
+            result = await asyncio.wait_for(future, timeout=timeout)
+            return result if isinstance(result, dict) else {"ok": False, "error": "ungueltige Antwort"}
+        except asyncio.TimeoutError:
+            return {"ok": False, "error": f"Host '{host or 'all'}' antwortet nicht (Timeout)."}
+        except Exception as exc:
+            return {"ok": False, "error": str(exc)}
+        finally:
+            self._pending_host.pop(request_id, None)
+
     async def _delete_chat_message(self, ts: int) -> dict:
         """Entfernt eine Bubble: aus chat_backup.jsonl + Brain conversation,
         broadcastet chat_message_deleted via RVS.

rvs/server.js

@@ -84,6 +84,10 @@ const ALLOWED_TYPES = new Set([
   // (SNMP/HTTP/FritzBox), der Satellit speichert sie verschluesselt.
   "sat_creds_set", "sat_creds_delete", "sat_creds_list",
   "sat_creds_result", "sat_creds_list_result",
+  // Host-Agenten: ein Agent laeuft direkt auf einem Rechner, meldet sich mit
+  // host_hello/host_ping und fuehrt host_command aus (exec/read/write/info/
+  // screenshot) -> host_result. ARIA steuert so Rechner auch hinter NAT.
+  "host_hello", "host_ping", "host_command", "host_result",
   // Compute-Flotte (AI-Boxen): Worker (f5tts/whisper/voxtral/llm-adapter) melden
   // sich per worker_hello an und pingen per worker_ping; der Diagnostic-Server
   // aggregiert das und broadcastet worker_update/worker_list an die Browser-UI.