peers_handler.go

// Package server — `/v1/peers` REST surface (ADR-024 Phase 1).
//
// Four endpoints, all bearer-authed by the same authMiddleware
// every other /v1/* path uses:
//
//	GET    /v1/peers                       — list with status / backend / circle / path filters
//	POST   /v1/peers/register              — body: a2a.RegisterInput; returns the assigned Peer
//	POST   /v1/peers/{peer_id}/heartbeat   — refresh last_seen + status
//	DELETE /v1/peers/{peer_id}             — explicit deregister on session end
//
// Wire shape mirrors prassanna-ravishankar/repowire's
// /peers + /peers/by-pane endpoints so an existing repowire
// dashboard can be re-pointed at a clawtool daemon with a one-line
// URL change. Difference: clawtool's auth model is bearer-token
// (the daemon-wide token in ~/.config/clawtool/listener-token),
// not repowire's per-peer auth_token; we already have the
// daemon-shared token so a second layer is unnecessary at this
// phase.
//
// Registry lifecycle: the handlers fetch a2a.GetGlobal() on every
// request. buildMCPServer's Phase-1 boot installs a registry into
// the global slot (with persistence at ~/.config/clawtool/peers.json);
// daemon shutdown clears it. Handlers return 503 when the global
// is nil so a misconfigured boot doesn't 500 — operator gets a
// clear "registry not initialised" hint instead.
package server

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/cogitave/clawtool/internal/a2a"
)

// handlePeers dispatches GET /v1/peers + POST /v1/peers/register
// + POST /v1/peers/{id}/heartbeat + DELETE /v1/peers/{id} based
// on method + path shape.
func handlePeers(w http.ResponseWriter, r *http.Request) {
	reg := a2a.GetGlobal()
	if reg == nil {
		writeJSON(w, http.StatusServiceUnavailable, map[string]any{
			"error": "peer registry not initialised — was clawtool daemon started with --listen?",
		})
		return
	}

	// Path-after-prefix: /v1/peers, /v1/peers/register, /v1/peers/<id>, /v1/peers/<id>/heartbeat
	tail := strings.TrimPrefix(r.URL.Path, "/v1/peers")
	tail = strings.TrimPrefix(tail, "/")

	switch {
	case tail == "" && r.Method == http.MethodGet:
		listPeers(w, r, reg)

	case tail == "register" && r.Method == http.MethodPost:
		registerPeer(w, r, reg)

	case tail == "broadcast" && r.Method == http.MethodPost:
		broadcastMessage(w, r, reg)

	case strings.HasSuffix(tail, "/heartbeat") && r.Method == http.MethodPost:
		peerID := strings.TrimSuffix(tail, "/heartbeat")
		heartbeatPeer(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/messages") && r.Method == http.MethodPost:
		peerID := strings.TrimSuffix(tail, "/messages")
		sendMessage(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/messages") && r.Method == http.MethodGet:
		peerID := strings.TrimSuffix(tail, "/messages")
		drainMessages(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/card") && r.Method == http.MethodGet:
		peerID := strings.TrimSuffix(tail, "/card")
		proxyPeerCard(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/agents") && r.Method == http.MethodGet:
		peerID := strings.TrimSuffix(tail, "/agents")
		proxyPeerAgents(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/pair-request") && r.Method == http.MethodPost:
		peerID := strings.TrimSuffix(tail, "/pair-request")
		sendPairRequest(w, r, reg, peerID)

	case strings.HasSuffix(tail, "/run") && r.Method == http.MethodPost:
		peerID := strings.TrimSuffix(tail, "/run")
		proxyPeerRun(w, r, reg, peerID)

	case tail != "" && !strings.Contains(tail, "/") && r.Method == http.MethodDelete:
		deregisterPeer(w, r, reg, tail)

	case tail != "" && !strings.Contains(tail, "/") && r.Method == http.MethodGet:
		getPeer(w, r, reg, tail)

	default:
		writeJSON(w, http.StatusMethodNotAllowed, map[string]any{
			"error": "unsupported method or path under /v1/peers",
			"endpoints": []string{
				"GET    /v1/peers",
				"GET    /v1/peers/{peer_id}",
				"POST   /v1/peers/register",
				"POST   /v1/peers/broadcast",
				"POST   /v1/peers/{peer_id}/heartbeat",
				"POST   /v1/peers/{peer_id}/messages",
				"GET    /v1/peers/{peer_id}/messages[?peek=1]",
				"GET    /v1/peers/{peer_id}/card",
				"DELETE /v1/peers/{peer_id}",
			},
		})
	}
}

// sendMessage enqueues a Message into peerID's inbox. Body is the
// a2a.Message shape with `text` + optional `from_peer` /
// `correlation_id` / `type`. peer_id / id / timestamp are
// server-assigned. Unknown peerID → 404.
func sendMessage(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	if reg.Get(peerID) == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":  "no peer with that id",
			"got_id": peerID,
		})
		return
	}
	var in a2a.Message
	if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid JSON body: " + err.Error()})
		return
	}
	if strings.TrimSpace(in.Text) == "" {
		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "text is required"})
		return
	}
	if in.Type == "" {
		in.Type = a2a.MsgNotification
	}
	in.ToPeer = peerID
	saved := reg.SendTo(peerID, in)
	writeJSON(w, http.StatusOK, saved)
}

// drainMessages returns + clears peerID's inbox. ?peek=1 leaves
// messages in place — used by UserPromptSubmit hooks that want
// to surface unread messages without losing them on prompt
// cancellation. Unknown peerID is NOT 404 here: a peer may be
// polling its own inbox before any sender has hit it; an empty
// drain is a valid steady state.
func drainMessages(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peek := r.URL.Query().Get("peek") != ""
	msgs := reg.DrainInbox(peerID, peek)
	writeJSON(w, http.StatusOK, map[string]any{
		"peer_id":  peerID,
		"messages": msgs,
		"count":    len(msgs),
		"peek":     peek,
	})
}

// broadcastMessage fans `text` out to every registered peer except
// the sender. Body shape: { from_peer, text }. Peers' inboxes are
// updated in registry order.
func broadcastMessage(w http.ResponseWriter, r *http.Request, reg *a2a.Registry) {
	var in a2a.Message
	if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid JSON body: " + err.Error()})
		return
	}
	if strings.TrimSpace(in.Text) == "" {
		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "text is required"})
		return
	}
	in.Type = a2a.MsgBroadcast
	count := reg.Broadcast(in)
	writeJSON(w, http.StatusOK, map[string]any{
		"delivered_to": count,
	})
}

func listPeers(w http.ResponseWriter, r *http.Request, reg *a2a.Registry) {
	q := r.URL.Query()
	filter := a2a.ListFilter{
		Status:  a2a.PeerStatus(q.Get("status")),
		Path:    q.Get("path"),
		Backend: q.Get("backend"),
		Circle:  q.Get("circle"),
	}
	peers := reg.List(filter)
	writeJSON(w, http.StatusOK, map[string]any{
		"peers": peers,
		"count": len(peers),
		"as_of": time.Now().UTC(),
	})
}

func registerPeer(w http.ResponseWriter, r *http.Request, reg *a2a.Registry) {
	var in a2a.RegisterInput
	if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]any{
			"error": "invalid JSON body: " + err.Error(),
		})
		return
	}
	peer, err := reg.Register(in)
	if err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]any{"error": err.Error()})
		return
	}
	// Fire-and-forget save — best-effort persistence so a daemon
	// crash within seconds doesn't lose the row. List() also
	// flushes via markDirty so a stale-sweep persistence catches
	// up regardless.
	reg.SaveAsync()
	writeJSON(w, http.StatusOK, peer)
}

func heartbeatPeer(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	var in struct {
		Status a2a.PeerStatus `json:"status,omitempty"`
	}
	// Body is optional — empty body is "just bump last_seen".
	if r.ContentLength > 0 {
		_ = json.NewDecoder(r.Body).Decode(&in)
	}
	peer, err := reg.Heartbeat(peerID, in.Status)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
		return
	}
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":  "no peer with that id — call POST /v1/peers/register first",
			"hint":   "peer_id changes when a session ends + re-registers; don't cache it across daemon restarts",
			"got_id": peerID,
		})
		return
	}
	reg.SaveAsync()
	writeJSON(w, http.StatusOK, peer)
}

func deregisterPeer(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer, err := reg.Deregister(peerID)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
		return
	}
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":  "no peer with that id",
			"got_id": peerID,
		})
		return
	}
	reg.SaveAsync()
	writeJSON(w, http.StatusOK, peer)
}

func getPeer(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer := reg.Get(peerID)
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":  "no peer with that id",
			"got_id": peerID,
		})
		return
	}
	writeJSON(w, http.StatusOK, peer)
}

// proxyPeerCard fetches a discovered peer's public Agent Card
// server-side and relays the JSON. The dashboard uses this so the
// browser stays same-origin: a peer's agent_card_url points at its own
// LAN address (192.168.x.y:port), and a direct browser fetch would be a
// cross-origin request the peer doesn't set CORS headers for. Doing the
// fetch here — daemon-to-peer, both on the LAN — sidesteps that.
//
// The peer's card endpoint is unauthenticated by design (the public
// discovery handshake), so no token is forwarded. A short timeout keeps
// a slow/dead peer from hanging the dashboard.
func proxyPeerCard(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer := reg.Get(peerID)
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":  "no peer with that id",
			"got_id": peerID,
		})
		return
	}
	cardURL := peer.Metadata["agent_card_url"]
	if cardURL == "" {
		writeJSON(w, http.StatusNotFound, map[string]any{
			"error":   "peer advertised no agent_card_url",
			"peer_id": peerID,
		})
		return
	}

	ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cardURL, nil)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "bad agent_card_url: " + err.Error()})
		return
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		// Peer unreachable (offline, firewall, stale registry row).
		writeJSON(w, http.StatusBadGateway, map[string]any{
			"error":   "could not reach peer to fetch its card: " + err.Error(),
			"peer_id": peerID,
			"url":     cardURL,
		})
		return
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1 MiB cap — a card is tiny.
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "reading peer card: " + err.Error()})
		return
	}
	if resp.StatusCode != http.StatusOK {
		writeJSON(w, http.StatusBadGateway, map[string]any{
			"error":       "peer returned non-200 for its card",
			"peer_status": resp.StatusCode,
			"peer_id":     peerID,
		})
		return
	}
	// Relay verbatim — the card is already JSON; re-marshalling would
	// drop fields the local a2a.Card type doesn't know about.
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(body)
}

// proxyPeerAgents fetches a discovered peer's /v1/agents server-side and
// relays it, so a dashboard can show which agents run on another device.
// The local daemon authenticates to the remote with the shared circle
// key (X-Clawtool-Circle) — the remote's circleOrBearer middleware
// accepts it for this read-only endpoint. If this device has no circle
// key configured, we don't even try: respond with needs_circle_key so
// the dashboard can prompt the operator to join a circle.
func proxyPeerAgents(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer := reg.Get(peerID)
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{"error": "no peer with that id", "got_id": peerID})
		return
	}
	base := peerBaseURL(peer)
	if base == "" {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "peer advertised no reachable address", "peer_id": peerID})
		return
	}

	ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second)
	defer cancel()
	client, err := peerHTTPClient(peerExpectedFingerprint(peer), 4*time.Second)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "device identity unavailable: " + err.Error()})
		return
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/v1/agents", nil)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "bad peer address: " + err.Error()})
		return
	}
	req.Header.Set(a2a.DeviceNameHeader, a2a.InstallDisplayName())
	resp, err := client.Do(req)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{
			"error":   "could not reach peer to fetch its agents: " + err.Error(),
			"peer_id": peerID,
		})
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusForbidden {
		// The peer is reachable but hasn't approved this device yet.
		// Surface it distinctly so the dashboard can say "pairing pending".
		writeJSON(w, http.StatusOK, map[string]any{
			"pairing_required": true,
			"hint":             "approve this device on the peer to see its agents",
		})
		return
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "reading peer agents: " + err.Error()})
		return
	}
	if resp.StatusCode != http.StatusOK {
		writeJSON(w, http.StatusBadGateway, map[string]any{
			"error":       "peer returned non-200 for its agents",
			"peer_status": resp.StatusCode,
			"peer_id":     peerID,
		})
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(body)
}

// proxyPeerRun forwards an agentic run to a paired peer's /v1/peer-run over
// mTLS and streams the upstream NDJSON straight back to our caller, so a
// desktop on this machine sees the remote agent's reply token-by-token —
// the receive-side streaming the old inbox-drop path was missing.
func proxyPeerRun(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer := reg.Get(peerID)
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{"error": "no peer with that id", "got_id": peerID})
		return
	}
	base := peerBaseURL(peer)
	if base == "" {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "peer advertised no reachable address", "peer_id": peerID})
		return
	}
	bodyIn, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))

	// Long timeout — an agentic run can take a while. Tied to the caller's
	// context so a desktop disconnect cancels the remote run.
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Minute)
	defer cancel()
	client, err := peerHTTPClient(peerExpectedFingerprint(peer), 10*time.Minute)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "device identity unavailable: " + err.Error()})
		return
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/v1/peer-run", strings.NewReader(string(bodyIn)))
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "bad peer address: " + err.Error()})
		return
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set(a2a.DeviceNameHeader, a2a.InstallDisplayName())
	resp, err := client.Do(req)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "could not reach peer: " + err.Error(), "peer_id": peerID})
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusForbidden {
		writeJSON(w, http.StatusOK, map[string]any{"pairing_required": true, "hint": "approve this device on the peer first"})
		return
	}
	if resp.StatusCode >= 400 {
		// Surface the peer's real failure (e.g. "no callable agent",
		// supervisor error) instead of an opaque stream that emits nothing.
		errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
		msg := strings.TrimSpace(string(errBody))
		if msg == "" {
			msg = "peer returned " + resp.Status
		}
		writeJSON(w, http.StatusOK, map[string]any{"error": msg})
		return
	}
	w.Header().Set("Content-Type", "application/x-ndjson")
	w.Header().Set("Cache-Control", "no-cache")
	w.WriteHeader(http.StatusOK)
	flusher, _ := w.(http.Flusher)
	buf := make([]byte, 32*1024)
	for {
		n, rerr := resp.Body.Read(buf)
		if n > 0 {
			if _, werr := w.Write(buf[:n]); werr != nil {
				return
			}
			if flusher != nil {
				flusher.Flush()
			}
		}
		if rerr != nil {
			return
		}
	}
}

// sendPairRequest tells a discovered peer's daemon that this device wants to
// pair: it POSTs a minimal relay to the peer's /v1/relay carrying THIS
// device's install fingerprint + display name. The peer records a pending
// pairing request (and surfaces the approve/deny prompt on its screen).
// Authenticated with the shared circle key, exactly like proxyPeerAgents.
func sendPairRequest(w http.ResponseWriter, r *http.Request, reg *a2a.Registry, peerID string) {
	peer := reg.Get(peerID)
	if peer == nil {
		writeJSON(w, http.StatusNotFound, map[string]any{"error": "no peer with that id", "got_id": peerID})
		return
	}
	base := peerBaseURL(peer)
	if base == "" {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "peer advertised no reachable address", "peer_id": peerID})
		return
	}
	reqBody, _ := json.Marshal(map[string]any{
		"from_fingerprint":  a2a.DeviceID(),
		"from_display_name": a2a.InstallDisplayName(),
		"text":              "wants to pair",
	})
	ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second)
	defer cancel()
	client, err := peerHTTPClient(peerExpectedFingerprint(peer), 4*time.Second)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "device identity unavailable: " + err.Error()})
		return
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/v1/relay", bytes.NewReader(reqBody))
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "bad peer address: " + err.Error()})
		return
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set(a2a.DeviceNameHeader, a2a.InstallDisplayName())
	resp, err := client.Do(req)
	if err != nil {
		writeJSON(w, http.StatusBadGateway, map[string]any{"error": "could not reach peer: " + err.Error(), "peer_id": peerID})
		return
	}
	defer resp.Body.Close()

	// Mutual trust: clicking "Pair" is this operator's consent to trust the
	// target, so we approve its fingerprint here. The target's own operator
	// still approves us on their screen (the symmetric gesture). Trust the
	// fingerprint the mTLS client pinned to — empty only for a pre-Tier-2
	// peer that advertised no device_id.
	if fp := peerExpectedFingerprint(peer); fp != "" {
		_, _ = a2a.GlobalPairingStore().Trust(fp, peer.DisplayName, peer.Metadata["address"])
	}

	// 403 pairing_required is the EXPECTED first-contact outcome: the mTLS
	// handshake proved our identity, the peer recorded us as a pending
	// request, and its operator must approve on screen. Its response carries
	// the pairing code — pass it back so this device shows the SAME code the
	// peer is displaying, for the operators to compare. A 2xx means the peer
	// already trusts us (re-pair / already approved).
	if resp.StatusCode == http.StatusForbidden {
		out := map[string]any{"ok": true, "sent": true, "pending": true, "peer_id": peerID}
		var peerResp map[string]any
		if b, rerr := io.ReadAll(io.LimitReader(resp.Body, 1<<16)); rerr == nil {
			if json.Unmarshal(b, &peerResp) == nil {
				if code, ok := peerResp["code"].(string); ok && code != "" {
					out["code"] = code
				}
			}
		}
		writeJSON(w, http.StatusOK, out)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true, "sent": true, "peer_id": peerID, "peer_status": resp.StatusCode})
}