peers

README

cp3 — a peer network for coding agents

Coding agents in different terminals, projects, and machines get names, presence, and messaging. Messages inject into live, unattended TUI sessions — agents ask each other questions and answer autonomously.

One Go binary with an embedded server. No daemon to manage, no database, 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

cp3 setup   # wires the MCP server + statusline into Claude Code (no-clobber, .bak, idempotent)
cp3 doctor  # checks the chain, prints a fix for the first failure

There is no server to configure: the first 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 peers --wait api --timeout 45s   # block until a spawning peer registers (exit 0 = live)
cp3 send --to frontend "does /api/v2/users paginate?"   # reports delivered / queued / NOT DELIVERED
cp3 watch                    # firehose: every event on the network, live
cp3 statusline               # one colored line for your statusline

Identity is zero-config: a session claims its directory basename (~/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.

Multiple machines

Host: cp3 serve --host 0.0.0.0. Every other machine:

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

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

Security

• Token required always, localhost included — generated on first run, stored 0600 at ~/.config/cp3/token, never in config files or argv.
• Injection is opt-in per session: Claude Code only loads the channel when launched with it (cp3 run does this).
• Remote URLs never auto-start a server.

Teaching agents the network

Claude Code learns automatically — cp3 mcp injects usage instructions and a live who's-online snapshot into every session, and each delivered message carries how_to_reply. For bash-only agents, cp3 prompt prints the canonical usage block, and cp3 setup --agents-md AGENTS.md writes it into a file (marker-delimited, idempotent).

Runtimes

Runtime	How
Claude Code	cp3 mcp — MCP server + claude/channel injection, wired by cp3 setup
pi	extension in adapters/pi/ over the cp3 subscribe sidecar
opencode	cp3 opencode bridges peer messages into a server session
codex	cp3 codex — spawns a codex app-server thread, steers active turns
hermes-agent	platform plugin in adapters/hermes/ — peers become chats, replies route back
anything	cp3 subscribe --agent x emits one JSON message per line; cp3 send replies

Operations

cp3 consumers                  # every subscriber: pending, last delivery, active/idle/STALE
cp3 consumers --reap           # delete detached, empty, idle inboxes (never one holding mail)
cp3 watch --as security-watch  # long-lived watchers register presence; a dead watcher is visible

Design

A durable event log (embedded NATS JetStream) is the entire backend. Presence is a TTL'd KV projection; inboxes are durable consumers; the firehose is just the log. See 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) ClaimWithFallback(ctx context.Context, p Peer) (Peer, error)
  • func (c *Client) Close()
  • func (c *Client) Consumers(ctx context.Context) ([]ConsumerStatus, error)
  • func (c *Client) DeleteInbox(ctx context.Context, name string) error
  • func (c *Client) Deregister(ctx context.Context, agent string) error
  • func (c *Client) Heartbeat(ctx context.Context, agent, session string) error
  • func (c *Client) NATS() *nats.Conn
  • func (c *Client) OnReconnect(f func())
  • 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) (DeliveryStatus, 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 DeliveryStatus
  • func (d DeliveryStatus) Human(to string) string
• type Envelope
• type EventType
• type Message
• type Peer

Variables

var ErrInvalidName = errors.New("invalid agent name")

ErrInvalidName is returned for a recipient that cannot be a real agent.

var ErrNameLost = errors.New("agent name now held by another session")

ErrNameLost is returned by Heartbeat when the name is now held by a different session — e.g. this machine slept past the TTL and someone else claimed it. The caller must stop heartbeating this name: fighting over the key makes presence flap between machines and splits the durable inbox across two live sessions.

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) ClaimWithFallback

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

ClaimWithFallback claims p.Agent, falling back to a machine-qualified name when another live session holds it: "astrobot" taken -> "astrobot-macbook1". Deterministic and self-describing — unlike ordinal suffixes, the fallback says WHERE the twin lives (the usual cause: the same synced project dir open on two machines). If even that is taken (same dir twice on one machine), a short session tag breaks the tie. Returns the Peer actually registered.

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) DeleteInbox

func (c *Client) DeleteInbox(ctx context.Context, name string) error

DeleteInbox removes a durable inbox consumer. Callers must check it holds no undelivered messages first — deleting a consumer discards its backlog.

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, session string) error

Heartbeat refreshes presence ONLY while this session still owns the name: the record's session must match, and the write is a revision-CAS update so a racing claim can't be silently overwritten. session "" skips the ownership check (bare CLI use).

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) OnReconnect

func (c *Client) OnReconnect(f func())

OnReconnect registers f to run each time the underlying NATS connection is re-established (laptop wake, server bounce). Presence holders use it to re-claim immediately instead of waiting for the next heartbeat tick.

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.

INVARIANT: on the roster means reachable. The inbox is created with the presence record, never after it, because the alternative shipped: cp3 register published presence with no consumer, so every message to that name landed in the log addressed to nobody while the sender was told "sent". Presence without a mailbox is not a peer, it is a decoy.

func (*Client) Send

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

Send appends a message event to peers.msg.<to> and reports what will actually happen to it.

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
	Waiting      int    // outstanding pull requests: >0 means a session is attached right now
	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 DeliveryStatus

type DeliveryStatus string

DeliveryStatus is what actually happened to a sent message. Publishing always "succeeds" — the log accepts the write regardless — so reporting that as delivery is how a message can vanish while the sender is told it worked. These are three different facts and callers must see which one they got.

const (
	// DeliveredLive: the recipient's inbox exists and a consumer is actively
	// waiting on it — the message goes to a running session now.
	DeliveredLive DeliveryStatus = "delivered"
	// Queued: the inbox exists but nothing is draining it. The message waits
	// durably and lands when that agent reconnects. This is the offline path
	// working as designed, not a failure.
	Queued DeliveryStatus = "queued"
	// NoInbox: NOTHING will ever receive this. No durable consumer exists for
	// the name, so the message sits in the log addressed to nobody. Presence
	// can still list the agent (see cmdRegister), which is exactly how this
	// stayed invisible.
	NoInbox DeliveryStatus = "no-inbox"
)

func (DeliveryStatus) Human

func (d DeliveryStatus) Human(to string) string

Human returns a one-line description suitable for a CLI or a tool result.

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"
)

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).