"""peer — talk to other navi instances in the swarm.
The address book lives in the hive registry (hive_url); asks themselves go
DIRECTLY peer-to-peer over the 8099 API port — the hive is only for
discovery and is never on the message path.
Subcommands (``action`` param):
list — every machine in the swarm (from the hive; cached copy if the
hive is down, marked stale)
status — one peer's live state (uptime, version, machine, hive view)
ask — ask a peer a question; a real agent on the other side answers
Loop safety on the asking side: the answering agent has no ``peer`` tool
(see navi/api/routes/peer.py), so asks cannot recurse. Asking a peer whose
question would loop back to ourselves is refused by the remote side.
"""
from __future__ import annotations
import structlog
from navi.config import settings
from navi.tools._internal.base import Tool, ToolContext, ToolResult
log = structlog.get_logger()
# Last successful hive /peers response — the stale fallback when the hive
# is unreachable. {fetched_at: <iso>, peers: [...]}
_peers_cache: dict | None = None
def _swarm_key() -> str | None:
from navi.swarm import read_swarm_key
return read_swarm_key(settings.swarm_key_file)
def _self_identity():
from navi.identity import load_identity
return load_identity(settings.instance_file)
class _PeerError(Exception):
"""Control-flow error with a machine-readable code for ToolResult."""
def __init__(self, message: str, error: str):
super().__init__(message)
self.error = error
class PeerTool(Tool):
name = "peer"
description = (
"Communicate with other navi instances in the swarm (the local network of "
"navi agents this machine belongs to).\n\n"
"Actions:\n"
"· list — all machines in the swarm with name, address, online status\n"
"· status <peer> — one peer's live state: uptime, version, machine facts\n"
"· ask <peer> <question> — ask a peer a question; its agent investigates "
"on ITS machine and returns a text answer. Use for anything this peer "
"can check better than you: its services, files, hardware, local state.\n\n"
"Address the peer by its swarm name (e.g. 'yuki-grace'). "
"Answers can take up to a couple of minutes - the peer runs a real agent "
"turn. Do not ask peers about this machine's state; check it locally."
)
parameters = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["list", "ask", "status"],
"description": "list = swarm overview, ask = question to a peer, status = peer health",
},
"peer": {
"type": "string",
"description": "Peer's swarm name (for ask/status), e.g. 'amber-falcon'",
},
"question": {
"type": "string",
"description": "The question for 'ask' - concrete, self-contained, answerable by the peer on its machine",
},
},
"required": ["action"],
}
async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
action = params.get("action", "")
try:
if action == "list":
return await self._list()
if action == "ask":
return await self._ask(params)
if action == "status":
return await self._status(params)
return ToolResult(success=False, output=f"Unknown action: {action!r} (use list/ask/status)",
error="bad_params")
except _PeerError as e:
return ToolResult(success=False, output=str(e), error=e.error)
except Exception as e:
return ToolResult(success=False, output=f"{type(e).__name__}: {e}", error="peer_tool_error")
# ── hive book ────────────────────────────────────────────────────────
async def _peers_from_hive(self) -> list | None:
"""Fresh peer list from the hive, or None when it is unreachable.
Updates the module cache on success so a later offline call can fall
back to the last known book.
"""
import httpx
key = _swarm_key()
if not settings.hive_url:
raise _PeerError("Swarm is not configured: HIVE_URL is empty on this machine.",
"swarm_unconfigured")
if not key:
raise _PeerError("Swarm key (.swarm-key) is missing on this machine.",
"swarm_unconfigured")
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(f"{settings.hive_url.rstrip('/')}/peers",
headers={"X-Swarm-Key": key})
resp.raise_for_status()
except Exception:
return None
global _peers_cache
peers = resp.json().get("peers", [])
_peers_cache = {"peers": peers}
return peers
# ── list ────────────────────────────────────────────────────────────
async def _list(self) -> ToolResult:
peers = await self._peers_from_hive()
stale = False
if peers is None:
if _peers_cache is None:
raise _PeerError(f"Cannot reach the hive registry ({settings.hive_url}) and no cached "
"peer list exists yet.", "hive_unreachable")
peers = _peers_cache["peers"]
stale = True
try:
myself = _self_identity()
except Exception:
myself = None
rows = []
for p in peers:
if myself and p.get("instance_id") == myself.instance_id:
continue
state = "online" if p.get("online") else "offline"
meta = p.get("meta", {})
rows.append(
f"- {p.get('name')} — {p.get('address')} [{state}] "
f"{meta.get('os', '')}, {meta.get('cpu_cores', '?')} cores"
)
header = "Swarm peers" + (" (STALE — hive unreachable, last known list)" if stale else "")
return ToolResult(success=True, output=f"{header}:\n" + "\n".join(rows) if rows
else f"{header}: no other machines announced yet")
# ── resolve ─────────────────────────────────────────────────────────
async def _resolve(self, name: str) -> dict:
"""Find a peer record by swarm name (hive first, stale cache as fallback)."""
peers = await self._peers_from_hive()
if peers is None:
peers = _peers_cache["peers"] if _peers_cache else []
for p in peers:
if p.get("name") == name:
return p
known = ", ".join(p.get("name", "?") for p in peers) or "none known"
raise _PeerError(f"Unknown peer '{name}'. Known peers: {known}", "unknown_peer")
# ── status ──────────────────────────────────────────────────────────
async def _status(self, params: dict) -> ToolResult:
import httpx
name = params.get("peer") or ""
if not name:
return ToolResult(success=False, output="'status' needs a peer name.", error="bad_params")
peer = await self._resolve(name)
key = _swarm_key()
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.get(f"http://{peer['address']}/peer/status",
headers={"X-Swarm-Key": key})
resp.raise_for_status()
except Exception as e:
return ToolResult(success=False,
output=f"Peer '{name}' at {peer['address']} is unreachable: "
f"{type(e).__name__}: {e}", error="peer_unreachable")
s = resp.json()
machine = s.get("machine", {})
lines = [
f"{s.get('name')} (instance {str(s.get('instance_id'))[:8]})",
f"address: {peer['address']}, port {s.get('port')}, version {s.get('version')}",
f"uptime: {s.get('uptime_sec')}s",
f"machine: {machine.get('os', '?')}, {machine.get('cpu_cores', '?')} cores",
f"its hive view: {s.get('hive', {}).get('reachable', '?')}",
]
return ToolResult(success=True, output="\n".join(lines))
# ── ask ─────────────────────────────────────────────────────────────
async def _ask(self, params: dict) -> ToolResult:
import httpx
name = params.get("peer") or ""
question = params.get("question") or ""
if not name or not question:
return ToolResult(success=False,
output="'ask' needs both a peer name and a question.",
error="bad_params")
key = _swarm_key()
if not key:
return ToolResult(success=False, output="Swarm key (.swarm-key) is missing on this machine.",
error="swarm_unconfigured")
try:
me = _self_identity()
except Exception:
return ToolResult(success=False,
output="Instance identity (instance.json) is missing - cannot identify "
"myself to peers.", error="identity_missing")
if me.name == name:
return ToolResult(success=False,
output=f"'{name}' is this very navi. Ask yourself that directly or check "
"locally - no need to go through the peer channel.",
error="self_ask")
peer = await self._resolve(name)
payload = {
"from_name": me.name,
"from_instance_id": me.instance_id,
"question": question[:4000],
}
timeout = settings.peer_ask_timeout_sec + 30
log.info("peer.ask_sent", peer=name, address=peer["address"],
question_chars=len(payload["question"]), timeout=timeout)
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(f"http://{peer['address']}/peer/ask",
json=payload, headers={"X-Swarm-Key": key})
resp.raise_for_status()
except httpx.HTTPStatusError as e:
detail = ""
try:
detail = e.response.json().get("detail", "")
except Exception:
pass
return ToolResult(success=False,
output=f"Peer '{name}' refused the ask "
f"(HTTP {e.response.status_code}): {detail}",
error="peer_refused")
except Exception as e:
return ToolResult(success=False,
output=f"Peer '{name}' at {peer['address']} is unreachable: "
f"{type(e).__name__}: {e}", error="peer_unreachable")
body = resp.json()
answer = body.get("answer", "")
log.info("peer.ask_answer", peer=name, answer_chars=len(answer),
completed=body.get("completed"))
if not body.get("completed"):
answer += "\n\n[note: the peer hit its turn limit - the answer may be incomplete]"
return ToolResult(success=True,
output=f"{name} answered:\n\n{answer}" if answer else
f"{name} returned an empty answer.")
