diff --git a/.env.example b/.env.example
index b91625d..a73985a 100644
--- a/.env.example
+++ b/.env.example
@@ -88,6 +88,9 @@
 # and the registry being down never blocks anything.
 #HIVE_URL=http://192.168.1.168:8087
 ANNOUNCE_INTERVAL_SEC=120
+# Which profile answers incoming peer asks, and how long one may run.
+PEER_ASK_PROFILE=server_admin
+PEER_ASK_TIMEOUT_SEC=120
 
 # ── gnexus-auth OAuth ────────────────────────────────────────────────────────────
 GNAUTH_BASE_URL=https://auth.your-domain.com
diff --git a/deploy/env.template b/deploy/env.template
index 0363f52..fca288f 100644
--- a/deploy/env.template
+++ b/deploy/env.template
@@ -51,3 +51,5 @@
 # To join a swarm: copy .swarm-key from the hive host, uncomment, restart.
 #HIVE_URL=http://192.168.1.168:8087
 ANNOUNCE_INTERVAL_SEC=120
+PEER_ASK_PROFILE=server_admin
+PEER_ASK_TIMEOUT_SEC=120
diff --git a/docs/config.md b/docs/config.md
index 77ba49d..10b7786 100644
--- a/docs/config.md
+++ b/docs/config.md
@@ -118,6 +118,8 @@
 | `ANNOUNCE_INTERVAL_SEC` | int | `120` | How often this navi re-announces itself to the hive. |
 | `SWARM_KEY_FILE` | str | `.swarm-key` | Shared PSK file (gitignored, 0600, generated by `deploy/install.sh`). |
 | `INSTANCE_FILE` | str | `instance.json` | This navi's identity for the swarm: generated name (adjective-animal) + instance id. Rename = edit the file, restart. |
+| `PEER_ASK_PROFILE` | str | `server_admin` | Which profile answers incoming `/peer/ask` questions (one-shot agent run). |
+| `PEER_ASK_TIMEOUT_SEC` | int | `120` | Hard ceiling for a single peer ask — the LLM turn inside `/peer/ask`. |
 
 See [`deploy/README.md`](../deploy/README.md) for the one-command server deployment (`bash deploy/install.sh`): dockerized PostgreSQL + systemd + `navi-code` in PATH, web UI and auth off by default.
 
diff --git a/docs/tools.md b/docs/tools.md
index 681f5bd..eccb3cc 100644
--- a/docs/tools.md
+++ b/docs/tools.md
@@ -52,6 +52,7 @@
 | `CreateMcpServerTool` | `create_mcp_server` | Scaffold a new MCP server directory with boilerplate |
 | `TestMcpToolTool` | `test_mcp_tool` | Execute a single MCP tool call in isolation for diagnostics |
 | `ReflectTool` | `reflect` | Self-reflection and analysis |
+| `PeerTool` | `peer` | Swarm communication: `list` machines, `status` of one peer, `ask` a peer a question (answered by a real agent run on the other side) |
 | `PlanTool` | `plan` | Agent-invoked planning: fresh plan or re-plan with a `reason` (and optional `updated_goal`). The tool result instructs the agent to wait for user confirmation when the task is complex, and to proceed immediately otherwise |
 | `ScheduleRecallTool` | `schedule_recall` | Schedule a headless callback for the current session (once/recurring/immediate) |
 | `ManageRecallTool` | `manage_recall` | Cancel, skip, or list scheduled recalls for the current session |
diff --git a/hive/README.md b/hive/README.md
index a3f01c1..e6d1453 100644
--- a/hive/README.md
+++ b/hive/README.md
@@ -58,4 +58,20 @@
 `.env` (install.sh generates one; copy the same file to every machine and the
 hive host to form a swarm). No `HIVE_URL` — the navi is standalone and keeps
 working; the registry being down is also non-blocking: the agent just gets a
-context note that the book is unreachable.
\ No newline at end of file
+context note that the book is unreachable.
+## Peer channel (stage 2)
+
+Navi instances talk to each other directly — the hive is only the address
+book, never on the message path. Each navi exposes on its API port (8099):
+
+| Endpoint | Auth | Purpose |
+|---|---|---|
+| `GET /peer/hello` | none | name + uuid prefix + version (discovery ping) |
+| `GET /peer/status` | PSK | identity, uptime, machine facts, its hive view |
+| `POST /peer/ask` | PSK | ask a peer a question; a one-shot agent run under `PEER_ASK_PROFILE` answers |
+
+Agents use the built-in `peer` tool (`list` / `status` / `ask`). Loop
+safety is deterministic: the answering agent is created with `peer`
+excluded, so asks cannot recurse; an ask claiming to come from our own
+uuid is refused (409). Every ask is audited on both sides via structlog
+(`peer.ask_sent` / `peer.ask_received` / `peer.ask_answered`).
diff --git a/navi/api/routes/peer.py b/navi/api/routes/peer.py
new file mode 100644
index 0000000..58e5611
--- /dev/null
+++ b/navi/api/routes/peer.py
@@ -0,0 +1,181 @@
+"""Peer-to-peer swarm endpoints — other navi instances talk to this one here.
+
+The peer channel lives on the main API port (8099). Three endpoints:
+
+  GET  /peer/hello   OPEN  — name, uuid prefix, version, port. A caller that
+                            stumbled onto this port learns only that a navi
+                            lives here; nothing sensitive is exposed.
+  GET  /peer/status  PSK   — identity + machine facts + hive reachability.
+  POST /peer/ask     PSK   — one-shot question answered by a real agent run
+                            under settings.peer_ask_profile.
+
+Loop safety: the answering agent run is created with ``peer`` excluded from
+its tools, so a remote question can never recursively spawn another peer
+ask — the recursion is cut at the tool level, deterministically. As a second
+belt, an ask whose ``from_instance_id`` matches our own uuid is refused.
+
+Audit: every ask is logged on both the request shape (peer name/uuid,
+question length, duration) and the outcome — structlog events
+``peer.ask_received`` / ``peer.ask_answered`` / ``peer.ask_failed``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import platform
+import time
+import uuid
+
+import structlog
+from fastapi import APIRouter, Header, HTTPException
+from pydantic import BaseModel, Field
+
+from navi import __version__
+from navi.config import settings
+from navi.identity import load_identity
+
+router = APIRouter(prefix="/peer", tags=["peer"])
+
+log = structlog.get_logger()
+
+_PROCESS_START = time.time()
+
+# Peer asks run real LLM turns — a burst of them would stack GPU work. One
+# concurrent peer answer at a time; extra callers wait for the semaphore.
+_ASK_SEMAPHORE = asyncio.Semaphore(1)
+
+_MAX_QUESTION_CHARS = 4000
+_MAX_ANSWER_CHARS = 16_000
+
+
+def _identity() -> dict:
+    try:
+        ident = load_identity(settings.instance_file)
+        return {"name": ident.name, "instance_id": ident.instance_id}
+    except Exception:
+        # A navi whose identity file vanished still answers hello/status;
+        # ask refuses below (nobody to pin the answer to).
+        return {"name": "unnamed", "instance_id": ""}
+
+
+def _require_swarm_key(x_swarm_key: str | None) -> None:
+    from navi.swarm import verify_swarm_key
+
+    if not x_swarm_key:
+        raise HTTPException(status_code=401, detail="X-Swarm-Key header required")
+    if not verify_swarm_key(x_swarm_key):
+        raise HTTPException(status_code=403, detail="invalid swarm key")
+
+
+@router.get("/hello")
+async def peer_hello() -> dict:
+    ident = _identity()
+    return {
+        "navi": True,
+        "name": ident["name"],
+        "instance_id": ident["instance_id"][:8] if ident["instance_id"] else "",
+        "version": __version__,
+        "port": settings.navi_port,
+    }
+
+
+@router.get("/status")
+async def peer_status(x_swarm_key: str | None = Header(None)) -> dict:
+    _require_swarm_key(x_swarm_key)
+    ident = _identity()
+    status: dict = {
+        "name": ident["name"],
+        "instance_id": ident["instance_id"],
+        "version": __version__,
+        "port": settings.navi_port,
+        "uptime_sec": round(time.time() - _PROCESS_START),
+    }
+    from navi.swarm import get_announcer
+
+    announcer = get_announcer()
+    status["hive"] = announcer.get_status() if announcer else {"configured": False}
+    try:
+        status["machine"] = {
+            "hostname": os.uname().nodename,
+            "os": f"{platform.system()} {platform.release()}",
+            "cpu_cores": os.cpu_count(),
+        }
+    except Exception:  # machine facts are best-effort
+        pass
+    return status
+
+
+class PeerAskPayload(BaseModel):
+    from_name: str = Field(min_length=1, max_length=64)
+    from_instance_id: str = Field(min_length=1, max_length=64)
+    question: str = Field(min_length=1, max_length=_MAX_QUESTION_CHARS)
+
+
+@router.post("/ask")
+async def peer_ask(
+    payload: PeerAskPayload,
+    x_swarm_key: str | None = Header(None),
+) -> dict:
+    _require_swarm_key(x_swarm_key)
+    ident = _identity()
+    if not ident["instance_id"]:
+        raise HTTPException(status_code=503, detail="instance identity missing")
+    # Loop guard (belt): a question claiming to come FROM us has travelled
+    # a full circle — refuse it instead of answering ourselves recursively.
+    if payload.from_instance_id == ident["instance_id"]:
+        log.warning("peer.ask_loop_detected", from_name=payload.from_name)
+        raise HTTPException(status_code=409, detail="ask loop detected - this question originated here")
+
+    request_id = uuid.uuid4().hex[:12]
+    log.info(
+        "peer.ask_received",
+        request_id=request_id,
+        from_name=payload.from_name,
+        from_instance_id=payload.from_instance_id,
+        question_chars=len(payload.question),
+    )
+
+    briefing = (
+        f"Another navi instance '{payload.from_name}' "
+        f"(instance {payload.from_instance_id[:8]}) is asking you a question "
+        "over the swarm peer channel. Answer for a fellow autonomous agent: "
+        "be direct and factual, include concrete values/paths/commands where "
+        "relevant. You cannot ask other peers from here — if the answer needs "
+        "something you cannot check, say so explicitly."
+    )
+    started = time.time()
+    async with _ASK_SEMAPHORE:
+        try:
+            from navi.api.deps import get_agent
+
+            agent = get_agent()
+            answer, completed = await asyncio.wait_for(
+                agent.run_ephemeral(
+                    user_message=payload.question,
+                    profile_id=settings.peer_ask_profile,
+                    exclude_tools=["peer"],  # deterministic loop cut
+                    briefing=briefing,
+                    timeout_seconds=settings.peer_ask_timeout_sec,
+                ),
+                timeout=settings.peer_ask_timeout_sec + 30,  # LLM ceiling + slack
+            )
+        except asyncio.TimeoutError:
+            log.warning("peer.ask_timeout", request_id=request_id, peer=payload.from_name,
+                        timeout=settings.peer_ask_timeout_sec)
+            raise HTTPException(status_code=504, detail="peer ask timed out")
+        except Exception as e:
+            log.error("peer.ask_failed", request_id=request_id, peer=payload.from_name,
+                      error=f"{type(e).__name__}: {e}")
+            raise HTTPException(status_code=500, detail="peer ask failed")
+
+    answer = (answer or "")[:_MAX_ANSWER_CHARS]
+    log.info(
+        "peer.ask_answered",
+        request_id=request_id,
+        peer=payload.from_name,
+        completed=completed,
+        duration_sec=round(time.time() - started, 1),
+        answer_chars=len(answer),
+    )
+    return {"answer": answer, "completed": completed, "request_id": request_id}
\ No newline at end of file
diff --git a/navi/config.py b/navi/config.py
index b34b8b2..4c01394 100644
--- a/navi/config.py
+++ b/navi/config.py
@@ -135,6 +135,13 @@
     hive_url: str = ""
     announce_interval_sec: int = 120
 
+    # Peer-to-peer swarm channel (stage 2). Incoming /peer/ask requests are
+    # answered by a one-shot agent run under this profile — pick a profile
+    # with broad but non-destructive tools (default: server_admin).
+    peer_ask_profile: str = "server_admin"
+    # Hard ceiling for a single peer ask (the LLM run inside /peer/ask).
+    peer_ask_timeout_sec: int = 120
+
     # Auth session cookie encryption (Fernet key, 32-byte base64)
     navi_auth_encryption_key: str = ""
     navi_auth_cookie_name: str = "navi_auth_session"
diff --git a/navi/core/registry.py b/navi/core/registry.py
index c54edce..865de4e 100644
--- a/navi/core/registry.py
+++ b/navi/core/registry.py
@@ -16,6 +16,7 @@
     ListProfilesTool,
     ManageRecallTool,
     MemoryTool,
+    PeerTool,
     ReflectTool,
     PlanTool,
     ScheduleRecallTool,
@@ -241,6 +242,7 @@
                 TodoTool(kv_store=kv_store), ScratchpadTool(kv_store=kv_store),
                 ReflectTool(ai_helper=ai_helper),
                 PlanTool(),
+                PeerTool(),
                 reload_tool, list_tool, manual_tool,
                 mcp_status_tool, create_mcp_server_tool, test_mcp_tool_tool,
                 schedule_recall_tool, manage_recall_tool,
diff --git a/navi/main.py b/navi/main.py
index 13b321d..374d55b 100644
--- a/navi/main.py
+++ b/navi/main.py
@@ -13,7 +13,7 @@
 from fastapi.staticfiles import StaticFiles
 
 from navi.api.deps import require_admin
-from navi.api.routes import agents, api_tokens, auth, health, messages, sessions, webhooks
+from navi.api.routes import agents, api_tokens, auth, health, messages, peer, sessions, webhooks
 from navi.api.routes.admin import router as admin_router
 from navi.api.websocket import router as ws_router
 from navi.config import settings
@@ -244,6 +244,7 @@
 )
 
 app.include_router(health.router)
+app.include_router(peer.router)
 app.include_router(auth.router)
 app.include_router(api_tokens.router)
 app.include_router(agents.router)
diff --git a/navi/profiles/developer/config.json b/navi/profiles/developer/config.json
index ecd9b69..3391163 100644
--- a/navi/profiles/developer/config.json
+++ b/navi/profiles/developer/config.json
@@ -55,7 +55,8 @@
         "content_publish",
         "gmail",
         "schedule_recall",
-        "manage_recall"
+        "manage_recall",
+        "peer"
       ],
       "mcp": {
         "navi-web": [
diff --git a/navi/profiles/navi_code/config.json b/navi/profiles/navi_code/config.json
index 432b35e..cda9cde 100644
--- a/navi/profiles/navi_code/config.json
+++ b/navi/profiles/navi_code/config.json
@@ -56,7 +56,8 @@
         "ssh_exec",
         "spawn_agent",
         "schedule_recall",
-        "manage_recall"
+        "manage_recall",
+        "peer"
       ],
       "mcp": {
         "navi-web": [
diff --git a/navi/profiles/server_admin/config.json b/navi/profiles/server_admin/config.json
index efcb4b9..3e56301 100644
--- a/navi/profiles/server_admin/config.json
+++ b/navi/profiles/server_admin/config.json
@@ -54,7 +54,8 @@
         "content_publish",
         "gmail",
         "schedule_recall",
-        "manage_recall"
+        "manage_recall",
+        "peer"
       ],
       "mcp": {
         "gnexus-book": [
diff --git a/navi/swarm.py b/navi/swarm.py
index 891a4f2..9751728 100644
--- a/navi/swarm.py
+++ b/navi/swarm.py
@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import asyncio
+import hmac
 import os
 import platform
 import socket
@@ -45,6 +46,23 @@
     return value or None
 
 
+def verify_swarm_key(provided: str | None) -> bool:
+    """Receiving-side check of an X-Swarm-Key header.
+
+    Accepts the current key AND the ``.previous`` one (rotation window — the
+    sender may still be on the old key while it propagates). Constant-time.
+    """
+    from navi.config import settings
+
+    if not provided:
+        return False
+    current = read_swarm_key(settings.swarm_key_file)
+    previous = read_swarm_key(Path(settings.swarm_key_file).with_name(
+        Path(settings.swarm_key_file).name + ".previous"
+    ))
+    return any(k and hmac.compare_digest(k, provided) for k in (current, previous))
+
+
 def _machine_meta() -> dict:
     """Basic machine facts for the address book (the mini knowledge base)."""
     meta: dict = {
diff --git a/navi/tools/__init__.py b/navi/tools/__init__.py
index 0251975..fe8ea1c 100644
--- a/navi/tools/__init__.py
+++ b/navi/tools/__init__.py
@@ -16,6 +16,7 @@
 from .list_profiles import ListProfilesTool
 from .reflect import ReflectTool
 from .plan import PlanRunner, PlanTool
+from .peer import PeerTool
 
 __all__ = [
     "Tool",
@@ -38,4 +39,5 @@
     "ReflectTool",
     "PlanRunner",
     "PlanTool",
+    "PeerTool",
 ]
diff --git a/navi/tools/peer.py b/navi/tools/peer.py
new file mode 100644
index 0000000..d934c87
--- /dev/null
+++ b/navi/tools/peer.py
@@ -0,0 +1,268 @@
+"""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. 'quiet-otter'). "
+        "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.")
\ No newline at end of file
diff --git a/tests/unit/test_peer_routes.py b/tests/unit/test_peer_routes.py
new file mode 100644
index 0000000..08d75ea
--- /dev/null
+++ b/tests/unit/test_peer_routes.py
@@ -0,0 +1,128 @@
+"""Peer endpoints: hello is open, status/ask need the PSK, loop guard works."""
+
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from navi import swarm
+from navi.api.routes import peer as peer_routes
+from navi.identity import ensure_identity
+
+KEY = "secret-key-123"
+
+
+@pytest.fixture()
+def peer_env(tmp_path, monkeypatch):
+    """Real instance.json + .swarm-key in tmp; settings point there."""
+    import navi.api.deps  # noqa: F401 - resolve the container BEFORE settings are swapped
+
+    ensure_identity(tmp_path / "instance.json")
+    (tmp_path / ".swarm-key").write_text(KEY)
+    fake = SimpleNamespace(
+        instance_file=str(tmp_path / "instance.json"),
+        swarm_key_file=str(tmp_path / ".swarm-key"),
+        peer_ask_profile="server_admin",
+        peer_ask_timeout_sec=5,
+        navi_port=8099,
+    )
+    monkeypatch.setattr(peer_routes, "settings", fake)
+    monkeypatch.setattr("navi.config.settings", fake)
+    from navi.identity import load_identity
+
+    return load_identity(tmp_path / "instance.json")
+
+
+@pytest.fixture()
+def client(peer_env):
+    app = FastAPI()
+    app.include_router(peer_routes.router)
+    return TestClient(app)
+
+
+def _headers(key: str = KEY) -> dict:
+    return {"X-Swarm-Key": key}
+
+
+def test_hello_is_open(client, peer_env):
+    resp = client.get("/peer/hello")  # no key
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["navi"] is True
+    assert body["name"] == peer_env.name
+    assert body["instance_id"] == peer_env.instance_id[:8]  # prefix only, not full uuid
+    assert len(body["instance_id"]) == 8
+    assert body["port"] == 8099
+
+
+def test_status_requires_key(client):
+    assert client.get("/peer/status").status_code == 401
+    assert client.get("/peer/status", headers=_headers("wrong")).status_code == 403
+    resp = client.get("/peer/status", headers=_headers())
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["name"]
+    assert "machine" in body
+    assert body["hive"] == {"configured": False}  # no announcer in tests
+
+
+def test_ask_requires_key(client, peer_env):
+    payload = {"from_name": "amber-falcon", "from_instance_id": "x" * 32, "question": "hi"}
+    assert client.post("/peer/ask", json=payload).status_code == 401
+    assert client.post("/peer/ask", json=payload, headers=_headers("bad")).status_code == 403
+
+
+def test_ask_loop_from_own_uuid_refused(client, peer_env, monkeypatch):
+    payload = {"from_name": peer_env.name, "from_instance_id": peer_env.instance_id, "question": "echo?"}
+    resp = client.post("/peer/ask", json=payload, headers=_headers())
+    assert resp.status_code == 409
+
+
+def test_ask_runs_agent_without_peer_tool(client, peer_env, monkeypatch):
+    captured = {}
+
+    class FakeAgent:
+        async def run_ephemeral(self, user_message, profile_id, exclude_tools=None,
+                                briefing=None, timeout_seconds=None, **kw):
+            captured.update(
+                message=user_message, profile=profile_id, exclude=exclude_tools,
+                briefing=briefing, timeout=timeout_seconds,
+            )
+            return "the answer from the other side", True
+
+    monkeypatch.setattr("navi.api.deps.get_agent", lambda: FakeAgent())
+    payload = {"from_name": "amber-falcon", "from_instance_id": "y" * 32, "question": "what is your load?"}
+    resp = client.post("/peer/ask", json=payload, headers=_headers())
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["answer"] == "the answer from the other side"
+    assert body["completed"] is True
+    assert body["request_id"]
+    # loop guard: answering agent has no peer tool, and the right profile
+    assert captured["exclude"] == ["peer"]
+    assert captured["profile"] == "server_admin"
+    assert captured["message"] == "what is your load?"
+    assert "amber-falcon" in captured["briefing"]
+
+
+def test_ask_question_length_capped(client, peer_env):
+    payload = {"from_name": "amber-falcon", "from_instance_id": "y" * 32, "question": "q" * 5000}
+    assert client.post("/peer/ask", json=payload, headers=_headers()).status_code == 422
+
+
+def test_previous_swarm_key_accepted(client, peer_env, tmp_path, monkeypatch):
+    # rotation window: .swarm-key.previous still authenticates
+    (tmp_path / ".swarm-key.previous").write_text("old-key")
+    resp = client.get("/peer/status", headers={"X-Swarm-Key": "old-key"})
+    assert resp.status_code == 200
+
+
+def test_verify_swarm_key_rejects_garbage(tmp_path, monkeypatch):
+    (tmp_path / ".swarm-key").write_text("real")
+    fake = SimpleNamespace(swarm_key_file=str(tmp_path / ".swarm-key"))
+    monkeypatch.setattr("navi.config.settings", fake)
+    assert swarm.verify_swarm_key("real") is True
+    assert swarm.verify_swarm_key("wrong") is False
+    assert swarm.verify_swarm_key("") is False
+    assert swarm.verify_swarm_key(None) is False
\ No newline at end of file
diff --git a/tests/unit/tools/test_peer.py b/tests/unit/tools/test_peer.py
new file mode 100644
index 0000000..135a8eb
--- /dev/null
+++ b/tests/unit/tools/test_peer.py
@@ -0,0 +1,184 @@
+"""peer tool: hive book with stale cache, ask payload shape, self-ask guard."""
+
+import json
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from navi.identity import ensure_identity
+from navi.tools import peer as peer_tool
+from navi.tools.peer import PeerTool
+
+KEY = "secret-key-123"
+
+ME = {"name": "quiet-otter", "instance_id": "11111111-2222-3333-4444-555555555555"}
+FALCON = {
+    "name": "amber-falcon",
+    "instance_id": "99999999-2222-3333-4444-555555555555",
+    "host": "192.168.1.50",
+    "port": 8099,
+    "address": "192.168.1.50:8099",
+    "online": True,
+    "meta": {"os": "Linux 6.1", "cpu_cores": 16, "ram_mb": 65536},
+    "version": "0.1.0",
+}
+
+
+@pytest.fixture()
+def tool(tmp_path, monkeypatch):
+    """Tool wired to a tmp identity/key and injectable http transport."""
+    ensure_identity(tmp_path / "instance.json")
+    (tmp_path / ".swarm-key").write_text(KEY)
+    fake = SimpleNamespace(
+        hive_url="http://hive.test",
+        swarm_key_file=str(tmp_path / ".swarm-key"),
+        instance_file=str(tmp_path / "instance.json"),
+        peer_ask_timeout_sec=30,
+    )
+    monkeypatch.setattr(peer_tool, "settings", fake)
+
+    real_client = httpx.AsyncClient
+
+    def make_client(transport):
+        # The tool builds its own httpx.AsyncClient; swap in the mock transport.
+        monkeypatch.setattr(
+            httpx, "AsyncClient",
+            lambda timeout=None: real_client(transport=transport, timeout=timeout),
+        )
+
+    yield make_client
+    peer_tool._peers_cache = None
+
+
+def _hive_transport(state: dict):
+    """Transport serving the hive /peers endpoint (and peer endpoints)."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/peers":
+            if state.get("hive_down"):
+                raise httpx.ConnectError("registry unreachable")
+            return httpx.Response(200, json={"peers": state.get("peers", []), "ttl_sec": 300})
+        if request.url.path == "/peer/status" and state.get("status_ok", True):
+            return httpx.Response(200, json={
+                "name": FALCON["name"], "instance_id": FALCON["instance_id"],
+                "version": "0.1.0", "port": 8099, "uptime_sec": 1234,
+                "machine": {"os": "Linux 6.1", "cpu_cores": 16},
+                "hive": {"reachable": True},
+            })
+        if request.url.path == "/peer/ask" and state.get("ask_ok", True):
+            body = json.loads(request.content)
+            state["last_ask"] = {"body": body, "key": request.headers.get("X-Swarm-Key")}
+            return httpx.Response(200, json={"answer": "load is 0.42", "completed": True, "request_id": "abc"})
+        raise httpx.ConnectError("no route")
+
+    return httpx.MockTransport(handler)
+
+
+@pytest.mark.asyncio
+async def test_list_success_excludes_self(tool):
+    tool(_hive_transport({"peers": [FALCON]}))
+    result = await PeerTool().execute({"action": "list"})
+    assert result.success
+    assert "amber-falcon" in result.output
+    assert "192.168.1.50:8099" in result.output
+    assert "online" in result.output
+
+
+@pytest.mark.asyncio
+async def test_list_falls_back_to_stale_cache(tool):
+    state = {"peers": [FALCON]}
+    tool(_hive_transport(state))
+    assert (await PeerTool().execute({"action": "list"})).success  # warm the cache
+
+    # hive goes down -> cached copy still answers, marked stale
+    state["hive_down"] = True
+    peer_tool._peers_cache["peers"][0]["online"] = False  # mutate what we cached
+    result = await PeerTool().execute({"action": "list"})
+    assert result.success
+    assert "STALE" in result.output
+    assert "amber-falcon" in result.output
+    assert "offline" in result.output  # last-known state
+
+
+@pytest.mark.asyncio
+async def test_list_fails_when_no_cache_and_hive_down(tool):
+    tool(_hive_transport({"peers": [FALCON], "hive_down": True}))
+    result = await PeerTool().execute({"action": "list"})
+    assert not result.success
+    assert "Cannot reach the hive" in result.output
+
+
+@pytest.mark.asyncio
+async def test_ask_sends_identity_and_key(tool):
+    from navi.identity import load_identity
+
+    me = load_identity(peer_tool.settings.instance_file)
+
+    state = {"peers": [FALCON]}
+    tool(_hive_transport(state))
+    result = await PeerTool().execute({"action": "ask", "peer": "amber-falcon", "question": "what is your load?"})
+    assert result.success
+    assert "load is 0.42" in result.output
+    sent = state["last_ask"]
+    assert sent["key"] == KEY
+    assert sent["body"]["question"] == "what is your load?"
+    assert sent["body"]["from_name"] == me.name
+    assert sent["body"]["from_instance_id"] == me.instance_id  # our real generated identity
+    assert result.output.startswith("amber-falcon answered:")
+
+
+@pytest.mark.asyncio
+async def test_ask_unknown_peer_lists_known(tool):
+    tool(_hive_transport({"peers": [FALCON]}))
+    result = await PeerTool().execute({"action": "ask", "peer": "ghost", "question": "hi"})
+    assert not result.success
+    assert "Unknown peer 'ghost'" in result.output
+    assert "amber-falcon" in result.output  # helpful: names the known peers
+
+
+@pytest.mark.asyncio
+async def test_ask_refuses_self(tool):
+    from navi.identity import load_identity
+
+    me = load_identity(peer_tool.settings.instance_file)
+    tool(_hive_transport({"peers": [FALCON]}))
+    result = await PeerTool().execute({"action": "ask", "peer": me.name, "question": "hi"})
+    assert not result.success
+    assert "this very navi" in result.output
+
+
+@pytest.mark.asyncio
+async def test_ask_reports_peer_refusal(tool):
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/peers":
+            return httpx.Response(200, json={"peers": [FALCON], "ttl_sec": 300})
+        return httpx.Response(409, json={"detail": "ask loop detected - this question originated here"})
+
+    tool(httpx.MockTransport(handler))
+    result = await PeerTool().execute({"action": "ask", "peer": "amber-falcon", "question": "hi"})
+    assert not result.success
+    assert "refused" in result.output
+    assert "ask loop detected" in result.output
+
+
+@pytest.mark.asyncio
+async def test_status_formats_peer_state(tool):
+    tool(_hive_transport({"peers": [FALCON]}))
+    result = await PeerTool().execute({"action": "status", "peer": "amber-falcon"})
+    assert result.success
+    assert "uptime: 1234s" in result.output
+    assert "Linux 6.1" in result.output
+
+
+@pytest.mark.asyncio
+async def test_incomplete_answer_is_flagged(tool):
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/peers":
+            return httpx.Response(200, json={"peers": [FALCON], "ttl_sec": 300})
+        return httpx.Response(200, json={"answer": "partial", "completed": False, "request_id": "x"})
+
+    tool(httpx.MockTransport(handler))
+    result = await PeerTool().execute({"action": "ask", "peer": "amber-falcon", "question": "hi"})
+    assert result.success
+    assert "turn limit" in result.output
\ No newline at end of file