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>
This commit is contained in:
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user