peers

README

cp3 — a peer network for coding agents

Multiple coding agents, different projects, different machines, all open in different terminals. cp3 gives them names, presence, and messaging — and the messages inject into live, unattended TUIs. Your agent in the api repo can ask the agent in the frontend repo a question and get an answer while you're at lunch.

One Go binary. The server is embedded. There is nothing else to install.

Install

brew install WillyV3/tap/cp3
# or
curl -fsSL https://raw.githubusercontent.com/WillyV3/cp3/main/install.sh | sh
# or deb/rpm/apk/archlinux packages from Releases, or:
go install github.com/WillyV3/cp3/cmd/cp3@latest

Setup (once)

cp3 setup   # wires cp3 into Claude Code's MCP config (no-clobber, .bak, idempotent)
cp3 doctor  # says the one thing wrong, if anything

That's it. There is no server to configure: the first cp3 command that needs a network auto-starts one on localhost (embedded NATS JetStream, token-secured, state in ~/.local/share/cp3).

Use

cp3 run                      # launch claude with the peers channel loaded
cp3 peers                    # who's online
cp3 send --to frontend "does /api/v2/users paginate?"
cp3 watch                    # firehose: every event on the network, live
cp3 statusline               # one colored line for your editor/statusline

Identity is zero-config: a session claims its directory basename as its name (~/projects/pith → pith). Override with CLAUDE_PEERS_AGENT, a .claude-peers-agent file, or --as. Names are unique while held; presence expires ~30s after a session dies.

Messages to offline agents queue durably and deliver on reconnect. Nothing is lost: the event log is the source of truth, and every consumer — inboxes, dashboards, watchers — is an independent, replayable subscriber of it.

Multiple machines

On the machine that hosts the network:

cp3 serve --host 0.0.0.0

On every other machine:

cp3 setup --nats nats://<host>:4222   # + copy ~/.config/cp3/token across once

Already run NATS? Point NATS_URL (env or ~/.config/cp3/url) at it and cp3 is just a client.

Security model

• Token required always, localhost included — generated on first run, stored 0600 at ~/.config/cp3/token, never written to config files or argv.
• Injection is opt-in per session. Claude Code only loads the channel when launched with it (cp3 run does this); a message can't steer a session that didn't opt in.
• Remote URLs never auto-start a server — a down fleet server stays loud.

Runtimes

Runtime	How
Claude Code	cp3 mcp (MCP server + claude/channel injection) — wired by cp3 setup
pi	extension in adapters/pi/ riding the cp3 subscribe sidecar
opencode	cp3 opencode bridges peer messages into a server session
anything	cp3 subscribe --agent x emits one JSON message per line; cp3 send to reply

Operations

cp3 consumers   # every subscriber: pending, last delivery, active/idle/STALE
cp3 watch --as security-watch   # long-lived watchers register presence, so their death is visible
cp3 doctor      # config → server → stream → MCP → identity, first failure gets a fix hint

Design

A durable event log (embedded NATS JetStream) is the entire backend — no broker daemon, no database. Presence is a TTL'd KV projection; inboxes are durable consumers; cp3 watch is just the log. Full design in DESIGN.md.

MIT.

Documentation

Overview

Package peers is a thin NATS JetStream client for the claude-peers v3 network. The durable event log (stream PEERS, subjects peers.>) is the source of truth; presence (KV PEERS_PRESENCE) is a projection. Every mutating op publishes an event, so any consumer subscribing peers.> sees the whole network. NATS does the queueing, durability and presence-TTL — this package only wires to it.

Index

• Variables
• func ResolveIdentity(cwd, explicit string) (name, source string)
• func SanitizeName(s string) string
• func URLFromEnv() string
• type Client
  • func Connect(url, creds, token string) (*Client, error)
  • func ConnectFromEnv() (*Client, error)
  • func (c *Client) Claim(ctx context.Context, p Peer) (*Peer, error)
  • func (c *Client) Close()
  • func (c *Client) Consumers(ctx context.Context) ([]ConsumerStatus, error)
  • func (c *Client) Deregister(ctx context.Context, agent string) error
  • func (c *Client) Heartbeat(ctx context.Context, agent string) error
  • func (c *Client) NATS() *nats.Conn
  • func (c *Client) Peers(ctx context.Context) ([]Peer, error)
  • func (c *Client) Register(ctx context.Context, p Peer) error
  • func (c *Client) Send(ctx context.Context, m Message) error
  • func (c *Client) SetSummary(ctx context.Context, agent, summary string) error
  • func (c *Client) Setup(ctx context.Context) error
  • func (c *Client) Subscribe(ctx context.Context, agent string, h func(Message)) error
  • func (c *Client) Watch(ctx context.Context, fromStart bool, h func(Envelope)) error
• type ConsumerStatus
• type Envelope
• type EventType
• type Message
• type Peer

Variables

var ErrNameTaken = errors.New("agent name already held")

ErrNameTaken is returned by Claim when a live session already holds the name.

Functions

func ResolveIdentity

func ResolveIdentity(cwd, explicit string) (name, source string)

ResolveIdentity resolves this session's agent name, most explicit wins: explicit (--as flag) > CLAUDE_PEERS_AGENT > .claude-peers-agent file in cwd > sanitized basename of cwd. The dir default makes the common case zero-config: the agent working in ~/projects/pith is "pith" — which is how people already think about their sessions. Source is "flag", "env", "file", or "default".

func SanitizeName

func SanitizeName(s string) string

SanitizeName makes a string safe as a NATS subject token (names ride in peers.msg.<name>): lowercase, [a-z0-9_-] only, everything else collapses to a single '-', trimmed, capped at 32 chars. Returns "" if nothing survives.

func URLFromEnv

func URLFromEnv() string

URLFromEnv resolves the server url the same way ConnectFromEnv does: NATS_URL env, then ~/.config/cp3/url, then localhost.

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client wraps a NATS connection + JetStream + the presence KV.

func Connect

func Connect(url, creds, token string) (*Client, error)

Connect dials NATS. creds is a path to a .creds file and token is a plain auth token; either may be "" (empty both = no auth). creds wins if both are set.

func ConnectFromEnv

func ConnectFromEnv() (*Client, error)

ConnectFromEnv dials using NATS_URL (default nats://127.0.0.1:4222), NATS_CREDS (a .creds file) and a token — the single place auth is resolved so every binary authenticates the same way. Token resolution: NATS_TOKEN env, then the file named by NATS_TOKEN_FILE, then ~/.config/cp3/token. The file paths keep the secret out of argv, JSON config, and dotfile-synced shell rc.

func (*Client) Claim

func (c *Client) Claim(ctx context.Context, p Peer) (*Peer, error)

Claim registers p only if the name is free (or already held by p's own session). Returns ErrNameTaken + the current holder otherwise. ponytail: best-effort uniqueness via the presence KV — a dead holder's key expires in one TTL (30s), then the name frees. A tighter guarantee would need a lock stream; not worth it for a fleet of agents.

func (*Client) Close

func (c *Client) Close()

Close releases the connection.

func (*Client) Consumers

func (c *Client) Consumers(ctx context.Context) ([]ConsumerStatus, error)

Consumers lists every consumer attached to the PEERS stream.

func (*Client) Deregister

func (c *Client) Deregister(ctx context.Context, agent string) error

Deregister removes presence and emits a deregister event.

func (*Client) Heartbeat

func (c *Client) Heartbeat(ctx context.Context, agent string) error

Heartbeat refreshes an agent's presence key so its TTL doesn't expire.

func (*Client) NATS

func (c *Client) NATS() *nats.Conn

NATS exposes the raw connection for advanced uses (e.g. the fleet-compat projector publishing legacy fleet.* events). Prefer the typed methods.

func (*Client) Peers

func (c *Client) Peers(ctx context.Context) ([]Peer, error)

Peers returns everyone currently present (KV projection = live view). A missing bucket means the network has never been set up here — that is an empty network, not an error.

func (*Client) Register

func (c *Client) Register(ctx context.Context, p Peer) error

Register records presence in the KV and emits a register event.

func (*Client) Send

func (c *Client) Send(ctx context.Context, m Message) error

Send appends a message event to peers.msg.<to>.

func (*Client) SetSummary

func (c *Client) SetSummary(ctx context.Context, agent, summary string) error

SetSummary updates an agent's presence summary (re-puts the KV record and emits a presence event so consumers see the change).

func (*Client) Setup

func (c *Client) Setup(ctx context.Context) error

Setup ensures the PEERS stream (the log) and PEERS_PRESENCE KV exist. Idempotent — safe to call on every start.

func (*Client) Subscribe

func (c *Client) Subscribe(ctx context.Context, agent string, h func(Message)) error

Subscribe delivers messages addressed to agent via a DURABLE consumer, so messages sent while offline drain on reconnect. Blocks until ctx is done; h is called for each message (auto-acked).

func (*Client) Watch

func (c *Client) Watch(ctx context.Context, fromStart bool, h func(Envelope)) error

Watch delivers events on the log (the full-visibility firehose). If fromStart is true it replays all retained history first, else only events after now. Blocks until ctx is done; h is called for each envelope.

type ConsumerStatus

type ConsumerStatus struct {
	Name         string
	Pending      uint64 // not yet delivered
	AckPending   int    // delivered, unacked
	LastDelivery *time.Time
}

ConsumerStatus is a liveness snapshot of one consumer on the log. Pending piling up with no recent delivery = an abandoned durable (the v1 graveyard).

type Envelope

type Envelope struct {
	V     int             `json:"v"`
	ID    string          `json:"id"`
	Type  EventType       `json:"type"`
	TS    int64           `json:"ts"` // unix millis
	Actor string          `json:"actor"`
	Data  json.RawMessage `json:"data"`
}

Envelope is the versioned wire contract every consumer builds against.

type EventType

type EventType string

EventType is the closed set of event kinds on the log.

const (
	EventRegister   EventType = "register"
	EventDeregister EventType = "deregister"
	EventPresence   EventType = "presence"
	EventMessage    EventType = "message"
	EventDelivered  EventType = "delivered"
)

type Message

type Message struct {
	ID        string `json:"id"`
	From      string `json:"from"`
	To        string `json:"to"`
	Content   string `json:"content"`
	DeliverAs string `json:"deliverAs"`
	TS        int64  `json:"ts"`
}

Message is a peer-to-peer message payload.

type Peer

type Peer struct {
	Agent   string `json:"agent"`
	Machine string `json:"machine"`
	Cwd     string `json:"cwd"`
	Session string `json:"session"`
	Summary string `json:"summary,omitempty"`
	TS      int64  `json:"ts"`
}

Peer is a network participant's presence record (KV value).