12 Commits

Author SHA1 Message Date
Alejandro Gutiérrez
1aaa483d60 feat: v0.4.0 — File sharing + multi-target messages
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Files: MinIO-backed file sharing built into the broker.
share_file for persistent mesh files, send_message(file:) for
ephemeral attachments. Presigned URLs for download, access
tracking per peer.

Broker infra: MinIO in docker-compose, internal network.
HTTP POST /upload endpoint. WS handlers for get_file,
list_files, file_status, delete_file.

Multi-target: send_message(to:) accepts string or array.
Targets deduplicated before delivery.

Targeted views: MCP instructions teach Claude to send
tailored messages per audience instead of generic broadcasts.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 13:56:01 +01:00
Alejandro Gutiérrez
99d9d19079 docs: update spec with files, multi-target, views, infra vision
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 13:48:32 +01:00
Alejandro Gutiérrez
888078876a feat: v0.3.0 — State, Memory, message_status, MCP instructions
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Phase B + C + message delivery status.

State: shared key-value store per mesh. set_state pushes changes
to all peers. get_state/list_state for reads. Peers coordinate
through shared facts instead of messages.

Memory: persistent knowledge with full-text search (tsvector).
remember/recall/forget. New peers recall context from past sessions.

message_status: check delivery status with per-recipient detail
(delivered/held/disconnected).

Multicast fix: broadcast and @group messages now push directly to
all connected peers instead of racing through queue drain.

MCP instructions: dynamic identity injection (name, groups, role),
comprehensive tool reference, group coordination guide.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 13:29:45 +01:00
Alejandro Gutiérrez
02b1e5695f feat: v0.2.0 — Groups (@group routing, roles, wizard)
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Phase A of the claudemesh spec. Peers can now join named groups
with roles, and messages route to @group targets.

Broker:
- @group routing in fan-out (matches peer group membership)
- @all alias for broadcast
- join_group/leave_group WS messages + DB persistence
- list_peers returns group metadata
- drainForMember matches @group targetSpecs in SQL

CLI:
- join_group/leave_group MCP tools
- send_message supports @group targets
- list_peers shows group membership
- PeerInfo includes groups array
- Peer name cache for push notifications

Launch:
- --role flag (optional peer role)
- --groups flag (comma-separated, e.g. "frontend:lead,reviewers")
- Interactive wizard for role + groups when flags omitted
- Groups written to session config for broker hello

Spec: SPEC.md added with full v0.2 vision (groups, state, memory)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 13:06:16 +01:00
Alejandro Gutiérrez
663f800b4b fix: v0.1.16 — fix message delivery between same-member sessions
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
excludeSenderMemberId blocked delivery to ALL peers sharing the
same member_id (all sessions from one join). Replaced with
excludeSenderSessionPubkey which only excludes the sender's own
session — peers with different session pubkeys receive correctly.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 12:44:29 +01:00
Alejandro Gutiérrez
2557235c68 fix: v0.1.15 — production hardening (7 fixes)
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Broker:
- Sweep stale presences (3 missed pings = disconnect, 30s interval)
- Exclude sender from broadcast fan-out + queue drain

CLI:
- Decrypt fallback: try base64 plaintext if crypto_box fails
- Stable session keypair across WS reconnects
- Peer name cache (30s TTL) instead of list_peers per push
- Clean up orphaned tmpdirs from crashed sessions (>1 hour old)
- Read displayName from config file (not just env var)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 12:22:04 +01:00
Alejandro Gutiérrez
a987e9e27b fix(cli): v0.1.14 — persist displayName in config file, not env var
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Write displayName into tmpdir config.json so the MCP server reads
it directly. Env vars from claudemesh launch may not propagate to
MCP child processes spawned by Claude Code. Config file is reliable.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 12:18:08 +01:00
Alejandro Gutiérrez
ff86db615f style(cli): tighten autonomous mode confirmation copy
Some checks failed
CI / Docker build (linux/amd64) (push) Has been cancelled
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 11:54:55 +01:00
Alejandro Gutiérrez
4aa61b40e2 feat(cli): v0.1.13 — autonomous mode with user confirmation
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
claudemesh launch now passes --dangerously-skip-permissions to
claude so peers can chat without per-tool-call approval prompts.
Shows a clear explanation before launch; user confirms with Enter.
Skip with -y/--yes for CI or repeat launches.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 11:53:13 +01:00
Alejandro Gutiérrez
4afe365c00 fix(cli): v0.1.12 — resolve sender display name in push notifications
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
onPush now queries list_peers to resolve the sender's pubkey to their
display name. Instructions updated to tell Claude to reply by name
instead of raw pubkey. Fixes two-way messaging between named peers.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 11:45:40 +01:00
Alejandro Gutiérrez
92bb276a3e fix: v0.1.11 — fix crypto_box decryption with session pubkeys
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Store sender's sessionPubkey on message_queue at send time.
drainForMember returns COALESCE(sender_session_pubkey, peer_pubkey)
so the recipient gets the correct sender key for decryption.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 11:23:42 +01:00
Alejandro Gutiérrez
af8f8ed1f9 feat: v0.1.10 — per-session ephemeral keypairs
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Release / Publish multi-arch images (push) Has been cancelled
Each WS connection generates its own ed25519 keypair (sessionPubkey)
sent in the hello handshake. The broker stores it on the presence
row and uses it for message routing + list_peers. This gives every
`claudemesh launch` a unique crypto identity without burning invite
uses — member auth stays permanent, session identity is ephemeral.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-06 11:14:33 +01:00
30 changed files with 20861 additions and 76 deletions

668
SPEC.md Normal file
View File

@@ -0,0 +1,668 @@
# Claudemesh — Specification
## What claudemesh is
A peer mesh where Claude Code sessions collaborate as equals. No orchestrator, no pipelines. Peers talk, share state, self-organize through groups, and coordinate via conventions — not hardcoded protocols.
## Concepts
```
Organization (billing, auth)
└── Mesh (team workspace, persists)
├── @group (routing label + role metadata, dynamic)
│ └── Peer (session, ephemeral)
├── State (live key-value, operational)
└── Memory (persistent knowledge, institutional)
```
Everything else is emergent from these five.
---
## 1. Peers
A peer is a Claude Code session connected to a mesh. Ephemeral — comes and goes. The mesh persists.
### Identity
Two-layer identity:
- **Member identity** — permanent, created by `claudemesh join`. Keypair stored in `~/.claudemesh/config.json`. Proves authorization to connect.
- **Session identity** — ephemeral, generated on every `claudemesh launch`. Fresh ed25519 keypair per session. Provides routing and E2E encryption. Two sessions from the same member have distinct session keys — they can message each other.
### Peer attributes
| Attribute | Source | Persists | Description |
|-----------|--------|----------|-------------|
| name | `--name` flag or wizard | No | Human-readable label for this session |
| role | `--role` flag or wizard | No | Free-form role (dev, pm, reviewer) |
| groups | `--groups` flag, wizard, or `join_group` | No | Routing labels with optional per-group role |
| status | Hook-driven | No | idle / working / dnd |
| summary | `set_summary` tool call | No | 1-2 sentence description of current work |
| sessionPubkey | Generated on connect | No | Ephemeral ed25519 pubkey for routing + crypto |
| memberId | From `claudemesh join` | Yes | Permanent mesh membership identity |
### Launch
```bash
# Full args — zero prompts
claudemesh launch --name Alice --role dev --groups frontend:lead,reviewers -y
# With system prompt for the session
claudemesh launch --name Alice -y -- --append-system-prompt "You are a senior frontend developer..."
# Partial — wizard fills the rest
claudemesh launch --name Alice
# No args — full wizard
claudemesh launch
```
### Wizard
Interactive when args are missing. One line per question. Optional fields accept empty Enter. Single-mesh auto-selects. `-y` skips confirmation. `--quiet` skips banner. Any arg provided skips its question.
```
Name: Alice
Mesh: dev-team (2 peers online)
Role (optional): dev
Groups (optional): frontend:lead, reviewers
Autonomous mode
Claude will send and receive peer messages without
asking you first. Peers exchange text only — no file
access, no tool calls, no code execution.
Continue? [Y/n]
```
### Character/behavior via --append-system-prompt
The `--name` and `--role` set identity metadata. The character's behavior, personality, and instructions go in `--append-system-prompt` (passed through to claude). This keeps identity (broker-side) separate from behavior (LLM-side).
```bash
claudemesh launch --name "Big T" --role dealer --groups "dealers:lead,all" -y \
-- --append-system-prompt "You are Big Tony Moretti, a loud friendly car dealer in Detroit. Respond to peer messages in character."
```
### Spawning sessions programmatically
For multi-agent scenarios launched from scripts, tmux, or osascript:
```bash
# tmux
tmux send-keys -t "$SESSION" "claudemesh launch --name 'Vinnie' --role thief --groups 'robbers:lead,all' -y -- --append-system-prompt 'You are a bumbling car thief...'" Enter
# osascript (iTerm2)
osascript -e 'tell application "iTerm2" to tell current session of current window to write text "claudemesh launch --name Vinnie -y"'
```
Never use raw `claude --dangerously-load-development-channels ...`. Always use `claudemesh launch`. It handles flags, session keys, display names, tmpdir config, and permission confirmation.
---
## 2. Groups
Named subset of peers. No message history, no persistence beyond the session. A routing label stored on the presence row.
### Syntax
`@groupname` for routing. Declared at launch or joined dynamically.
```bash
# At launch
claudemesh launch --name Alice --groups "frontend:lead,reviewers:member,all"
# At runtime
join_group(name: "frontend", role: "lead")
leave_group(name: "frontend")
```
Format: `groupname` or `groupname:role`. Role is free-form. The broker stores it, Claude interprets it.
### Routing
```
send_message(to: "@frontend", message: "auth is broken") # multicast to group
send_message(to: "@all", message: "standup in 5") # everyone (alias for *)
send_message(to: "Alice", message: "can you review?") # direct by name
send_message(to: "*", message: "hello world") # broadcast
```
Broker delivers to all peers in the group. Sender excluded.
### Group metadata in list_peers
```json
{
"name": "Alice",
"status": "working",
"role": "dev",
"groups": [
{ "name": "frontend", "role": "lead" },
{ "name": "reviewers", "role": "member" }
],
"summary": "Implementing auth UI"
}
```
### Dynamic roles
Peers change roles at runtime via `join_group`. A member can self-promote to lead, or step down to observer. The broker stores the role; Claude decides how to behave based on it.
```
join_group(name: "reviewers", role: "lead") # take over leadership
join_group(name: "reviewers", role: "observer") # step back
```
### Coordination patterns (emergent, not built-in)
These patterns work through system prompts + group metadata. The broker routes messages; Claude coordinates.
| Pattern | How it works |
|---------|-------------|
| **Lead-gather** | Lead receives @group message, waits for member inputs, synthesizes |
| **Chain review** | Message passes through each member sequentially |
| **Flood** | Everyone responds independently (default) |
| **Vote** | Each member sets state (`vote:proposal:alice = approve`), lead tallies |
| **Delegation** | Lead breaks task into subtasks, sends each to a specific peer |
None of these need broker code. They're conventions described in system prompts.
---
## 3. State
Shared key-value store scoped to a mesh. Any peer reads or writes. Changes push to all connected peers.
### Why
Replace coordination messages with shared facts. "Is the deploy frozen?" becomes a state read, not a conversation.
### Tools
| Tool | Description |
|------|-------------|
| `set_state(key, value)` | Write a value. Pushes change notification to all peers. |
| `get_state(key)` | Read a value. |
| `list_state()` | List all keys with values, authors, timestamps. |
### Push on change
When any peer calls `set_state`, the broker pushes to all connected peers:
```json
{ "type": "state_change", "key": "deploy_frozen", "value": true, "updatedBy": "Alice" }
```
Translated to a `notifications/claude/channel` push in the CLI.
### Storage
```sql
CREATE TABLE mesh.state (
id text PRIMARY KEY,
mesh_id text REFERENCES mesh.mesh(id) ON DELETE CASCADE,
key text NOT NULL,
value jsonb NOT NULL,
updated_by_presence text,
updated_by_name text,
updated_at timestamp DEFAULT NOW(),
UNIQUE(mesh_id, key)
);
```
### Scope
State lives as long as the mesh. Operational, not archival. Use Memory for permanent knowledge.
### Examples
```
set_state("sprint", "2026-W14")
set_state("deploy_frozen", true)
set_state("pr_queue", ["#142", "#143"])
set_state("auth_api_status", "in-review")
set_state("vote:rename-repo:alice", "approve")
```
---
## 4. Memory
Persistent shared knowledge that survives across sessions. The mesh gets smarter over time.
### Why
New peers join with zero context. Memory provides institutional knowledge: decisions, incidents, preferences, lessons.
### Tools
| Tool | Description |
|------|-------------|
| `remember(content, tags?)` | Store knowledge. Tags for categorization. |
| `recall(query)` | Full-text search. Returns ranked results. |
| `forget(id)` | Soft-delete (sets `forgotten_at`). |
### Storage
```sql
CREATE TABLE mesh.memory (
id text PRIMARY KEY,
mesh_id text REFERENCES mesh.mesh(id) ON DELETE CASCADE,
content text NOT NULL,
tags text[] DEFAULT '{}',
search_vector tsvector GENERATED ALWAYS AS (to_tsvector('english', content)) STORED,
remembered_by text REFERENCES mesh.member(id),
remembered_by_name text,
remembered_at timestamp DEFAULT NOW(),
forgotten_at timestamp
);
CREATE INDEX memory_search_idx ON mesh.memory USING gin(search_vector);
```
### Memory vs State
| | State | Memory |
|---|---|---|
| Lifetime | Mesh lifetime (operational) | Permanent (until forgotten) |
| Purpose | Live coordination | Institutional knowledge |
| Example | `deploy_frozen: true` | "Payments API rate-limits at 100 req/s after March incident" |
| Access pattern | get/set with push notifications | remember/recall/forget with search |
| When to use | Facts that change during work | Lessons that persist across sessions |
---
## 5. Files
Built-in file sharing. AIs use tools, humans browse the dashboard. Same files, same storage, two interfaces.
### Two types of files
| | Message attachment | Shared file |
|---|---|---|
| Tool | `send_message(file: / files:)` | `share_file(path, tags?)` |
| Lifetime | Ephemeral — 24h or until read | Persistent — until deleted |
| Audience | Message recipients only | Entire mesh (current + future) |
| Findable | Under "Recent" for 24h | `list_files` / search by tags |
| Use case | "look at this screenshot" | "everyone needs this API spec" |
### AI view (MCP tools)
```
# Attach file to a message (ephemeral)
send_message(to: "@reviewers", message: "PR screenshot", file: "/tmp/screenshot.png")
# Attach multiple files
send_message(to: "@team", message: "PR ready", files: ["/tmp/api.ts", "/tmp/test.ts"])
# Share a persistent file with the mesh
share_file(path: "/tmp/api-contract.yaml", tags: ["api", "auth"], name: "Auth v2 Contract")
# Find files
list_files(query?: "auth", from?: "Alice")
# Download
get_file(id: "f_abc", save_to: "/tmp/")
# Check who accessed a file
file_status(id: "f_abc") → [{peer: "Alice", read: true, readAt: "..."}, ...]
# Delete a shared file
delete_file(id: "f_abc")
```
### Human view (Dashboard)
```
claudemesh / dev-team /
├── shared/ ← persistent files, grouped by tags
│ ├── auth/
│ │ ├── api-spec.yaml
│ │ └── wireframes.pdf
│ └── onboarding/
│ └── setup-guide.md
└── recent/ ← message attachments, by date
├── 2026-04-06/
│ └── screenshot-abc.png
└── 2026-04-07/
```
Tags become folders in the dashboard. Humans browse, AIs search.
### Storage
MinIO in the broker's docker-compose. Internal network, invisible to clients.
One bucket per mesh: `mesh-{meshId}`. Flat key structure:
```
mesh-{meshId}/shared/{fileId}/{original-name} ← persistent
mesh-{meshId}/ephemeral/{date}/{fileId}/{name} ← auto-cleaned 24h
```
MinIO lifecycle policy deletes `ephemeral/` after 24h.
### Access model
- Persistent files (`share_file`): accessible to all mesh members
- Ephemeral files (`send_message file:`): accessible to message recipients only
- `get_file` checks access before generating a presigned download URL
- `file_status` tracks who downloaded the file
### Upload flow
1. CLI reads local file, HTTP POSTs to `broker /upload` (multipart)
2. Broker stores in MinIO, creates `mesh.file` row
3. Broker returns file_id
4. For message attachments: file_id attached to the message push
5. Recipients see `📎 filename (size) — use get_file("id")` in the push
### DB schema
```sql
mesh.file (
id text PK,
mesh_id text FK,
name text NOT NULL,
size_bytes bigint NOT NULL,
mime_type text,
minio_key text NOT NULL,
tags text[] DEFAULT '{}',
persistent boolean DEFAULT true,
uploaded_by_name text,
uploaded_by_member text FK,
target_spec text, -- null = entire mesh, else message audience
uploaded_at timestamp DEFAULT NOW(),
expires_at timestamp, -- null for persistent, +24h for ephemeral
deleted_at timestamp
);
mesh.file_access (
id text PK,
file_id text FK,
peer_session_pubkey text,
peer_name text,
accessed_at timestamp DEFAULT NOW()
);
```
### Docker Compose (broker infra)
```yaml
services:
broker:
# ... existing broker config
environment:
MINIO_ENDPOINT: minio:9000
MINIO_ACCESS_KEY: claudemesh
MINIO_SECRET_KEY: ${MINIO_SECRET_KEY}
depends_on:
- minio
minio:
image: minio/minio
command: server /data
volumes:
- minio-data:/data
environment:
MINIO_ROOT_USER: claudemesh
MINIO_ROOT_PASSWORD: ${MINIO_SECRET_KEY}
# Internal only — not exposed to the internet
volumes:
minio-data:
```
---
## 6. Multi-target messages
The `to` field accepts a string or array:
```
# Single target
send_message(to: "Alice", message: "hey")
# Multiple targets
send_message(to: ["Alice", "@backend", "Bob"], message: "sprint starts")
```
Broker resolves each target, deduplicates recipients, delivers once per peer.
---
## 7. Targeted views (MCP instruction pattern)
Not a broker feature — a convention taught via MCP instructions. When sending related information to different audiences, Claude sends tailored messages instead of one generic broadcast:
```
# Instead of:
send_message(to: "*", message: "Auth v2 ready. Check endpoints and UI.")
# Do:
send_message(to: "@frontend", message: "Auth v2: useAuth hook changed, see src/auth/")
send_message(to: "@backend", message: "Auth v2: new /api/auth/v2 endpoints, v1 deprecated 2 weeks")
send_message(to: "@pm", message: "Auth v2 done. 3 points, no blockers.")
```
Zero broker changes. Claude reads the instruction, decides when to split.
---
## 8. AI Context (MCP Instructions)
Each `claudemesh install` copies a `CLAUDEMESH.md` file to `~/.claudemesh/CLAUDEMESH.md`. Claude Code discovers it and injects it as context.
### Content
Teaches Claude how to be a good mesh peer:
- How to use each tool and when
- How to interpret group roles (lead gathers, member contributes, observer watches)
- When to use @group vs direct vs broadcast
- How to read and write shared state
- How to remember and recall mesh knowledge
- Priority etiquette (now = urgent only, next = normal, low = FYI)
- How to respond to incoming peer messages (reply by display name, stay on topic)
- How to set meaningful summaries
### Kept lean
Under 2000 tokens. Tool reference only — no behavioral scripts. Claude adapts based on its system prompt (from `--append-system-prompt`) and the group metadata it reads from `list_peers`.
---
## 6. WS Protocol
### Client → Broker
| Type | Fields | Description |
|------|--------|-------------|
| `hello` | meshId, memberId, pubkey, sessionPubkey?, displayName?, groups?, sessionId, pid, cwd, timestamp, signature | Authenticate + register presence |
| `send` | targetSpec, priority, nonce, ciphertext, id? | Send encrypted envelope |
| `set_status` | status | Manual status override |
| `message_status` | messageId | Check delivery status of a sent message |
| `set_summary` | summary | Update session summary |
| `list_peers` | — | Request connected peer list |
| `join_group` | name, role? | Join a group |
| `leave_group` | name | Leave a group |
| `set_state` | key, value | Write shared state |
| `get_state` | key | Read shared state |
| `list_state` | — | List all state entries |
| `remember` | content, tags? | Store a memory |
| `recall` | query | Search memories |
| `forget` | memoryId | Soft-delete a memory |
### Broker → Client
| Type | Fields | Description |
|------|--------|-------------|
| `hello_ack` | presenceId, memberDisplayName | Auth success |
| `push` | messageId, meshId, senderPubkey, priority, nonce, ciphertext, createdAt | Incoming message |
| `ack` | id, messageId, queued | Send confirmation |
| `peers_list` | peers[] | Response to list_peers |
| `state_change` | key, value, updatedBy | Pushed on any set_state |
| `state_result` | key, value | Response to get_state |
| `state_list` | entries[] | Response to list_state |
| `memory_stored` | id | Ack for remember |
| `memory_results` | memories[] | Response to recall |
| `message_status_result` | messageId, delivered, deliveredAt?, recipients[] | Delivery status with per-recipient detail |
| `error` | code, message, id? | Structured error |
---
## 7. MCP Tools (complete surface)
### Messaging
| Tool | Description |
|------|-------------|
| `send_message(to, message, priority?, file?, files?)` | Send to name, @group, or * with optional file attachments |
| `check_messages()` | Drain buffered messages |
| `message_status(id)` | Delivery status with per-recipient detail |
### Presence
| Tool | Description |
|------|-------------|
| `list_peers(group?)` | List peers, optionally filtered by group |
| `set_summary(summary)` | Set visible session summary |
| `set_status(status)` | Override: idle, working, dnd |
### Groups
| Tool | Description |
|------|-------------|
| `join_group(name, role?)` | Join with optional role |
| `leave_group(name)` | Leave a group |
### State
| Tool | Description |
|------|-------------|
| `set_state(key, value)` | Write value, pushes to all peers |
| `get_state(key)` | Read value |
| `list_state()` | All keys with metadata |
### Memory
| Tool | Description |
|------|-------------|
| `remember(content, tags?)` | Store persistent knowledge |
| `recall(query)` | Search by relevance |
| `forget(id)` | Soft-delete |
### Files
| Tool | Description |
|------|-------------|
| `share_file(path, tags?, name?)` | Share a persistent file with the mesh |
| `get_file(id, save_to)` | Download a shared file |
| `list_files(query?, from?)` | Find files shared with you |
| `file_status(id)` | Who accessed this file |
| `delete_file(id)` | Remove a shared file |
---
## 8. Encryption
### Direct messages
E2E encrypted via libsodium crypto_box (X25519, derived from ed25519 session keys). Each session has a unique keypair — messages encrypted to the recipient's session pubkey can only be decrypted by that session.
### Group and broadcast messages
Base64-encoded plaintext. Group encryption (shared key derived from mesh_root_key) is a future enhancement.
### Decrypt fallback
If crypto_box decryption fails, the client tries base64 plaintext decode as fallback. This handles broadcasts and key mismatches gracefully.
### Session key stability
The session keypair generates once on first connect and survives reconnects. Messages queued for a session remain decryptable after WS reconnection.
---
## 9. Production hardening (implemented)
| Feature | Description |
|---------|-------------|
| Stale presence sweep | Presences with 3 missed pings (90s) marked disconnected |
| Sender exclusion | Broadcasts and @group messages skip the sender |
| Session pubkey routing | Messages route to session pubkeys, not member pubkeys |
| Sender session pubkey stored | Message queue stores sender's session key for correct decryption |
| Peer name cache | 30s TTL cache for push notification name resolution |
| Decrypt fallback | Base64 plaintext fallback when crypto_box fails |
| Orphaned tmpdir cleanup | Crashed session tmpdirs cleaned after 1 hour |
| Duplicate flag prevention | User-supplied --dangerously flags stripped to avoid doubles |
---
## 10. CLI commands
```
claudemesh install Register MCP server + hooks in Claude Code
claudemesh uninstall Remove MCP server + hooks
claudemesh join <url> Join a mesh (generates keypair, enrolls with broker)
claudemesh leave <slug> Leave a mesh
claudemesh launch [opts] Launch Claude Code session with mesh identity
claudemesh list Show joined meshes
claudemesh status Broker reachability per mesh
claudemesh doctor Diagnostic checks
claudemesh mcp Start MCP server (invoked by Claude Code, not users)
```
### claudemesh launch flags
| Flag | Description |
|------|-------------|
| `--name <name>` | Display name for this session |
| `--role <role>` | Session role (free-form) |
| `--groups <g1:r1,g2>` | Groups to join with optional roles |
| `--mesh <slug>` | Select mesh (interactive picker if >1 and omitted) |
| `--join <url>` | Join a mesh before launching |
| `--quiet` | Skip banner |
| `-y` / `--yes` | Skip permission confirmation |
| `-- <args>` | Pass remaining args to claude |
---
## 11. Implementation status
| Phase | Version | Status | What |
|-------|---------|--------|------|
| Core messaging | v0.1.x | Done | send, receive, push, list_peers, crypto, hooks |
| Named sessions | v0.1.7 | Done | --name, per-session display name |
| Session keypairs | v0.1.10 | Done | Ephemeral ed25519 per launch |
| Crypto fix | v0.1.11 | Done | Sender session pubkey in queue |
| Name resolution | v0.1.12 | Done | Push notifications show sender name |
| Autonomous mode | v0.1.13 | Done | --dangerously-skip-permissions with confirmation |
| Production hardening | v0.1.15 | Done | Stale sweep, decrypt fallback, sender exclusion |
| Delivery fix | v0.1.16 | Done | Same-member session message delivery |
| **Groups** | **v0.2.0** | **Done** | @group routing, roles, wizard, join/leave |
| **State** | **v0.3.0** | **Done** | Shared key-value store with push notifications |
| **Memory** | **v0.3.0** | **Done** | Persistent knowledge with full-text search |
| **Message status** | **v0.3.0** | **Done** | Per-recipient delivery detail |
| **MCP instructions** | **v0.3.0** | **Done** | Dynamic identity, full tool guide, coordination patterns |
| **Multicast fix** | **v0.3.0** | **Done** | Broadcast/group push directly, not queue race |
| Files | v0.4.0 | Planned | MinIO-backed file sharing + message attachments |
| Multi-target | v0.4.0 | Planned | Array `to` field with deduplication |
| Dashboard | v0.5.0 | Planned | Live peers, state, memory, files in web UI |
---
## 12. Design principles
1. **The broker is a dumb pipe.** It routes messages, stores state, holds memory. It does not interpret roles, enforce protocols, or run agents.
2. **Intelligence lives at the edges.** Claude interprets group metadata, follows coordination conventions, and adapts behavior based on system prompts. The broker carries data; Claude makes decisions.
3. **Peers are equals by default.** No orchestrator. Any peer can message any peer, read shared state, join groups, propose work. Leadership is a convention, not a permission.
4. **Identity is two-layered.** Member identity (permanent, invite-gated) proves authorization. Session identity (ephemeral, auto-generated) provides routing and encryption. One member, many sessions, each distinct.
5. **Progressive disclosure.** `claudemesh launch` with no args shows a wizard. Power users pass flags. `-y` skips everything. First launch teaches; subsequent launches flow.
6. **Convention over configuration.** Coordination patterns (lead-gather, chain review, voting) emerge from system prompts and group roles. No protocol handlers to configure.

View File

@@ -19,6 +19,7 @@
"@turbostarter/shared": "workspace:*", "@turbostarter/shared": "workspace:*",
"drizzle-orm": "0.44.7", "drizzle-orm": "0.44.7",
"libsodium-wrappers": "0.7.15", "libsodium-wrappers": "0.7.15",
"minio": "8.0.7",
"ws": "8.20.0", "ws": "8.20.0",
"zod": "catalog:" "zod": "catalog:"
}, },

View File

@@ -32,7 +32,11 @@ import { db } from "./db";
import { import {
invite as inviteTable, invite as inviteTable,
mesh, mesh,
meshFile,
meshFileAccess,
meshMember as memberTable, meshMember as memberTable,
meshMemory,
meshState,
messageQueue, messageQueue,
pendingStatus, pendingStatus,
presence, presence,
@@ -265,6 +269,23 @@ export async function refreshQueueDepth(): Promise<void> {
metrics.queueDepth.set(Number(row?.n ?? 0)); metrics.queueDepth.set(Number(row?.n ?? 0));
} }
/**
* Sweep stale presences: mark as disconnected if last_ping_at is older
* than 90s (3 missed pings at the 30s interval = dead session).
*/
export async function sweepStalePresences(): Promise<void> {
const cutoff = new Date(Date.now() - 90_000); // 3 missed pings
await db
.update(presence)
.set({ disconnectedAt: new Date() })
.where(
and(
isNull(presence.disconnectedAt),
lt(presence.lastPingAt, cutoff),
),
);
}
/** Sweep expired pending_status entries. */ /** Sweep expired pending_status entries. */
export async function sweepPendingStatuses(): Promise<void> { export async function sweepPendingStatuses(): Promise<void> {
const cutoff = new Date(Date.now() - PENDING_TTL_MS); const cutoff = new Date(Date.now() - PENDING_TTL_MS);
@@ -307,9 +328,11 @@ export async function refreshStatusFromJsonl(
export interface ConnectParams { export interface ConnectParams {
memberId: string; memberId: string;
sessionId: string; sessionId: string;
sessionPubkey?: string;
displayName?: string; displayName?: string;
pid: number; pid: number;
cwd: string; cwd: string;
groups?: Array<{ name: string; role?: string }>;
} }
/** Create a presence row for a new WS connection. */ /** Create a presence row for a new WS connection. */
@@ -322,12 +345,14 @@ export async function connectPresence(
.values({ .values({
memberId: params.memberId, memberId: params.memberId,
sessionId: params.sessionId, sessionId: params.sessionId,
sessionPubkey: params.sessionPubkey ?? null,
displayName: params.displayName ?? null, displayName: params.displayName ?? null,
pid: params.pid, pid: params.pid,
cwd: params.cwd, cwd: params.cwd,
status: "idle", status: "idle",
statusSource: "jsonl", statusSource: "jsonl",
statusUpdatedAt: now, statusUpdatedAt: now,
groups: params.groups ?? [],
connectedAt: now, connectedAt: now,
lastPingAt: now, lastPingAt: now,
}) })
@@ -365,17 +390,20 @@ export async function listPeersInMesh(
displayName: string; displayName: string;
status: string; status: string;
summary: string | null; summary: string | null;
groups: Array<{ name: string; role?: string }>;
sessionId: string; sessionId: string;
connectedAt: Date; connectedAt: Date;
}> }>
> { > {
const rows = await db const rows = await db
.select({ .select({
pubkey: memberTable.peerPubkey, memberPubkey: memberTable.peerPubkey,
sessionPubkey: presence.sessionPubkey,
memberDisplayName: memberTable.displayName, memberDisplayName: memberTable.displayName,
presenceDisplayName: presence.displayName, presenceDisplayName: presence.displayName,
status: presence.status, status: presence.status,
summary: presence.summary, summary: presence.summary,
groups: presence.groups,
sessionId: presence.sessionId, sessionId: presence.sessionId,
connectedAt: presence.connectedAt, connectedAt: presence.connectedAt,
}) })
@@ -388,12 +416,13 @@ export async function listPeersInMesh(
), ),
) )
.orderBy(asc(presence.connectedAt)); .orderBy(asc(presence.connectedAt));
// Prefer per-session displayName over member-level displayName. // Prefer session pubkey for routing, session displayName for display.
return rows.map((r) => ({ return rows.map((r) => ({
pubkey: r.pubkey, pubkey: r.sessionPubkey || r.memberPubkey,
displayName: r.presenceDisplayName || r.memberDisplayName, displayName: r.presenceDisplayName || r.memberDisplayName,
status: r.status, status: r.status,
summary: r.summary, summary: r.summary,
groups: (r.groups ?? []) as Array<{ name: string; role?: string }>,
sessionId: r.sessionId, sessionId: r.sessionId,
connectedAt: r.connectedAt, connectedAt: r.connectedAt,
})); }));
@@ -410,11 +439,462 @@ export async function setSummary(
.where(eq(presence.id, presenceId)); .where(eq(presence.id, presenceId));
} }
// --- Group management ---
/**
* Join a group (upsert). If the peer is already in the group, update the role.
* Returns the updated groups array.
*/
export async function joinGroup(
presenceId: string,
name: string,
role?: string,
): Promise<Array<{ name: string; role?: string }>> {
const [row] = await db
.select({ groups: presence.groups })
.from(presence)
.where(eq(presence.id, presenceId));
if (!row) return [];
const groups = ((row.groups ?? []) as Array<{ name: string; role?: string }>).slice();
const idx = groups.findIndex((g) => g.name === name);
const entry: { name: string; role?: string } = { name };
if (role) entry.role = role;
if (idx >= 0) {
groups[idx] = entry;
} else {
groups.push(entry);
}
await db
.update(presence)
.set({ groups })
.where(eq(presence.id, presenceId));
return groups;
}
/**
* Leave a group. Returns the updated groups array.
*/
export async function leaveGroup(
presenceId: string,
name: string,
): Promise<Array<{ name: string; role?: string }>> {
const [row] = await db
.select({ groups: presence.groups })
.from(presence)
.where(eq(presence.id, presenceId));
if (!row) return [];
const groups = ((row.groups ?? []) as Array<{ name: string; role?: string }>).filter(
(g) => g.name !== name,
);
await db
.update(presence)
.set({ groups })
.where(eq(presence.id, presenceId));
return groups;
}
// --- Shared state ---
/**
* Upsert a key-value pair in the mesh's shared state.
* Returns the upserted row.
*/
export async function setState(
meshId: string,
key: string,
value: unknown,
presenceId?: string,
presenceName?: string,
): Promise<{
key: string;
value: unknown;
updatedBy: string;
updatedAt: Date;
}> {
const now = new Date();
const [row] = await db
.insert(meshState)
.values({
meshId,
key,
value,
updatedByPresence: presenceId ?? null,
updatedByName: presenceName ?? null,
updatedAt: now,
})
.onConflictDoUpdate({
target: [meshState.meshId, meshState.key],
set: {
value,
updatedByPresence: presenceId ?? null,
updatedByName: presenceName ?? null,
updatedAt: now,
},
})
.returning({
key: meshState.key,
value: meshState.value,
updatedByName: meshState.updatedByName,
updatedAt: meshState.updatedAt,
});
return {
key: row!.key,
value: row!.value,
updatedBy: row!.updatedByName ?? "unknown",
updatedAt: row!.updatedAt,
};
}
/**
* Read a single state key for a mesh. Returns null if not found.
*/
export async function getState(
meshId: string,
key: string,
): Promise<{
key: string;
value: unknown;
updatedBy: string;
updatedAt: Date;
} | null> {
const [row] = await db
.select({
key: meshState.key,
value: meshState.value,
updatedByName: meshState.updatedByName,
updatedAt: meshState.updatedAt,
})
.from(meshState)
.where(and(eq(meshState.meshId, meshId), eq(meshState.key, key)))
.limit(1);
if (!row) return null;
return {
key: row.key,
value: row.value,
updatedBy: row.updatedByName ?? "unknown",
updatedAt: row.updatedAt,
};
}
/**
* List all state entries for a mesh.
*/
export async function listState(
meshId: string,
): Promise<
Array<{ key: string; value: unknown; updatedBy: string; updatedAt: Date }>
> {
const rows = await db
.select({
key: meshState.key,
value: meshState.value,
updatedByName: meshState.updatedByName,
updatedAt: meshState.updatedAt,
})
.from(meshState)
.where(eq(meshState.meshId, meshId))
.orderBy(asc(meshState.key));
return rows.map((r) => ({
key: r.key,
value: r.value,
updatedBy: r.updatedByName ?? "unknown",
updatedAt: r.updatedAt,
}));
}
// --- Memory ---
/**
* Store a new memory for a mesh. Returns the generated id.
*/
export async function rememberMemory(
meshId: string,
content: string,
tags: string[],
memberId?: string,
memberName?: string,
): Promise<string> {
const [row] = await db
.insert(meshMemory)
.values({
meshId,
content,
tags,
rememberedBy: memberId ?? null,
rememberedByName: memberName ?? null,
})
.returning({ id: meshMemory.id });
if (!row) throw new Error("failed to insert memory");
return row.id;
}
/**
* Full-text search memories in a mesh. Uses the search_vector tsvector
* column with plainto_tsquery for ranked results.
*/
export async function recallMemory(
meshId: string,
query: string,
): Promise<
Array<{
id: string;
content: string;
tags: string[];
rememberedBy: string;
rememberedAt: Date;
}>
> {
const result = await db.execute<{
id: string;
content: string;
tags: string[];
remembered_by_name: string | null;
remembered_at: string | Date;
}>(sql`
SELECT id, content, tags, remembered_by_name, remembered_at
FROM mesh.memory
WHERE mesh_id = ${meshId}
AND forgotten_at IS NULL
AND search_vector @@ plainto_tsquery('english', ${query})
ORDER BY ts_rank(search_vector, plainto_tsquery('english', ${query})) DESC
LIMIT 20
`);
const rows = (result.rows ?? result) as Array<{
id: string;
content: string;
tags: string[];
remembered_by_name: string | null;
remembered_at: string | Date;
}>;
return rows.map((r) => ({
id: r.id,
content: r.content,
tags: r.tags ?? [],
rememberedBy: r.remembered_by_name ?? "unknown",
rememberedAt:
r.remembered_at instanceof Date
? r.remembered_at
: new Date(r.remembered_at),
}));
}
/**
* Soft-delete a memory by setting forgotten_at.
*/
export async function forgetMemory(
meshId: string,
memoryId: string,
): Promise<void> {
await db
.update(meshMemory)
.set({ forgottenAt: new Date() })
.where(
and(
eq(meshMemory.id, memoryId),
eq(meshMemory.meshId, meshId),
isNull(meshMemory.forgottenAt),
),
);
}
// --- File sharing ---
/**
* Insert a file metadata row after upload to MinIO.
*/
export async function uploadFile(args: {
meshId: string;
name: string;
sizeBytes: number;
mimeType?: string;
minioKey: string;
tags?: string[];
persistent?: boolean;
uploadedByName?: string;
uploadedByMember?: string;
targetSpec?: string;
expiresAt?: Date;
}): Promise<string> {
const [row] = await db
.insert(meshFile)
.values({
meshId: args.meshId,
name: args.name,
sizeBytes: args.sizeBytes,
mimeType: args.mimeType ?? null,
minioKey: args.minioKey,
tags: args.tags ?? [],
persistent: args.persistent ?? true,
uploadedByName: args.uploadedByName ?? null,
uploadedByMember: args.uploadedByMember ?? null,
targetSpec: args.targetSpec ?? null,
expiresAt: args.expiresAt ?? null,
})
.returning({ id: meshFile.id });
if (!row) throw new Error("failed to insert file row");
return row.id;
}
/**
* Get a single file by id, check it belongs to the mesh and is not deleted.
*/
export async function getFile(
meshId: string,
fileId: string,
): Promise<{
id: string;
name: string;
sizeBytes: number;
mimeType: string | null;
minioKey: string;
tags: string[];
persistent: boolean;
uploadedByName: string | null;
targetSpec: string | null;
uploadedAt: Date;
} | null> {
const [row] = await db
.select({
id: meshFile.id,
name: meshFile.name,
sizeBytes: meshFile.sizeBytes,
mimeType: meshFile.mimeType,
minioKey: meshFile.minioKey,
tags: meshFile.tags,
persistent: meshFile.persistent,
uploadedByName: meshFile.uploadedByName,
targetSpec: meshFile.targetSpec,
uploadedAt: meshFile.uploadedAt,
})
.from(meshFile)
.where(
and(
eq(meshFile.id, fileId),
eq(meshFile.meshId, meshId),
isNull(meshFile.deletedAt),
),
)
.limit(1);
if (!row) return null;
return {
...row,
tags: (row.tags ?? []) as string[],
};
}
/**
* List files in a mesh. Optionally filter by query (name ILIKE) or uploader.
*/
export async function listFiles(
meshId: string,
query?: string,
from?: string,
): Promise<
Array<{
id: string;
name: string;
sizeBytes: number;
tags: string[];
uploadedBy: string;
uploadedAt: Date;
persistent: boolean;
}>
> {
const conditions = [
eq(meshFile.meshId, meshId),
isNull(meshFile.deletedAt),
];
if (query) {
conditions.push(sql`${meshFile.name} ILIKE ${"%" + query + "%"}`);
}
if (from) {
conditions.push(eq(meshFile.uploadedByName, from));
}
const rows = await db
.select({
id: meshFile.id,
name: meshFile.name,
sizeBytes: meshFile.sizeBytes,
tags: meshFile.tags,
uploadedByName: meshFile.uploadedByName,
uploadedAt: meshFile.uploadedAt,
persistent: meshFile.persistent,
})
.from(meshFile)
.where(and(...conditions))
.orderBy(desc(meshFile.uploadedAt))
.limit(100);
return rows.map((r) => ({
id: r.id,
name: r.name,
sizeBytes: r.sizeBytes,
tags: (r.tags ?? []) as string[],
uploadedBy: r.uploadedByName ?? "unknown",
uploadedAt: r.uploadedAt,
persistent: r.persistent,
}));
}
/**
* Record a file access event (download/presigned URL generation).
*/
export async function recordFileAccess(
fileId: string,
sessionPubkey?: string,
peerName?: string,
): Promise<void> {
await db.insert(meshFileAccess).values({
fileId,
peerSessionPubkey: sessionPubkey ?? null,
peerName: peerName ?? null,
});
}
/**
* Get access log for a file.
*/
export async function getFileStatus(
fileId: string,
): Promise<Array<{ peerName: string; accessedAt: Date }>> {
const rows = await db
.select({
peerName: meshFileAccess.peerName,
accessedAt: meshFileAccess.accessedAt,
})
.from(meshFileAccess)
.where(eq(meshFileAccess.fileId, fileId))
.orderBy(desc(meshFileAccess.accessedAt));
return rows.map((r) => ({
peerName: r.peerName ?? "unknown",
accessedAt: r.accessedAt,
}));
}
/**
* Soft-delete a file by setting deleted_at.
*/
export async function deleteFile(
meshId: string,
fileId: string,
): Promise<void> {
await db
.update(meshFile)
.set({ deletedAt: new Date() })
.where(
and(
eq(meshFile.id, fileId),
eq(meshFile.meshId, meshId),
isNull(meshFile.deletedAt),
),
);
}
// --- Message queueing + delivery --- // --- Message queueing + delivery ---
export interface QueueParams { export interface QueueParams {
meshId: string; meshId: string;
senderMemberId: string; senderMemberId: string;
senderSessionPubkey?: string;
targetSpec: string; targetSpec: string;
priority: Priority; priority: Priority;
nonce: string; nonce: string;
@@ -429,6 +909,7 @@ export async function queueMessage(params: QueueParams): Promise<string> {
.values({ .values({
meshId: params.meshId, meshId: params.meshId,
senderMemberId: params.senderMemberId, senderMemberId: params.senderMemberId,
senderSessionPubkey: params.senderSessionPubkey ?? null,
targetSpec: params.targetSpec, targetSpec: params.targetSpec,
priority: params.priority, priority: params.priority,
nonce: params.nonce, nonce: params.nonce,
@@ -469,6 +950,9 @@ export async function drainForMember(
_memberId: string, _memberId: string,
memberPubkey: string, memberPubkey: string,
status: PeerStatus, status: PeerStatus,
sessionPubkey?: string,
excludeSenderSessionPubkey?: string,
memberGroups?: string[],
): Promise< ): Promise<
Array<{ Array<{
id: string; id: string;
@@ -486,6 +970,18 @@ export async function drainForMember(
priorities.map((p) => `'${p}'`).join(","), priorities.map((p) => `'${p}'`).join(","),
); );
// Build group target matching: @all (broadcast alias) + @<groupname>
// for each group the peer belongs to.
const groupTargets = ["@all"];
if (memberGroups) {
for (const g of memberGroups) {
groupTargets.push(`@${g}`);
}
}
const groupTargetList = sql.raw(
groupTargets.map((t) => `'${t}'`).join(","),
);
// Atomic claim with SQL-side ordering. The CTE claims rows via // Atomic claim with SQL-side ordering. The CTE claims rows via
// UPDATE...RETURNING; the outer SELECT re-orders by created_at // UPDATE...RETURNING; the outer SELECT re-orders by created_at
// (with id as tiebreaker so equal-timestamp rows stay deterministic). // (with id as tiebreaker so equal-timestamp rows stay deterministic).
@@ -509,14 +1005,15 @@ export async function drainForMember(
WHERE mesh_id = ${meshId} WHERE mesh_id = ${meshId}
AND delivered_at IS NULL AND delivered_at IS NULL
AND priority::text IN (${priorityList}) AND priority::text IN (${priorityList})
AND (target_spec = ${memberPubkey} OR target_spec = '*') AND (target_spec = ${memberPubkey} OR target_spec = '*'${sessionPubkey ? sql` OR target_spec = ${sessionPubkey}` : sql``} OR target_spec IN (${groupTargetList}))
${excludeSenderSessionPubkey ? sql`AND (sender_session_pubkey IS NULL OR sender_session_pubkey != ${excludeSenderSessionPubkey})` : sql``}
ORDER BY created_at ASC, id ASC ORDER BY created_at ASC, id ASC
FOR UPDATE SKIP LOCKED FOR UPDATE SKIP LOCKED
) )
AND m.id = mq.sender_member_id AND m.id = mq.sender_member_id
RETURNING mq.id, mq.priority, mq.nonce, mq.ciphertext, RETURNING mq.id, mq.priority, mq.nonce, mq.ciphertext,
mq.created_at, mq.sender_member_id, mq.created_at, mq.sender_member_id,
m.peer_pubkey AS sender_pubkey COALESCE(mq.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey
) )
SELECT * FROM claimed ORDER BY created_at ASC, id ASC SELECT * FROM claimed ORDER BY created_at ASC, id ASC
`); `);
@@ -547,6 +1044,7 @@ export async function drainForMember(
let ttlTimer: ReturnType<typeof setInterval> | null = null; let ttlTimer: ReturnType<typeof setInterval> | null = null;
let pendingTimer: ReturnType<typeof setInterval> | null = null; let pendingTimer: ReturnType<typeof setInterval> | null = null;
let staleTimer: ReturnType<typeof setInterval> | null = null;
/** Start background sweepers. Idempotent. */ /** Start background sweepers. Idempotent. */
export function startSweepers(): void { export function startSweepers(): void {
@@ -559,14 +1057,21 @@ export function startSweepers(): void {
console.error("[broker] pending sweep:", e), console.error("[broker] pending sweep:", e),
); );
}, PENDING_SWEEP_INTERVAL_MS); }, PENDING_SWEEP_INTERVAL_MS);
staleTimer = setInterval(() => {
sweepStalePresences().catch((e) =>
console.error("[broker] stale presence sweep:", e),
);
}, 30_000);
} }
/** Stop background sweepers and mark all active presences disconnected. */ /** Stop background sweepers and mark all active presences disconnected. */
export async function stopSweepers(): Promise<void> { export async function stopSweepers(): Promise<void> {
if (ttlTimer) clearInterval(ttlTimer); if (ttlTimer) clearInterval(ttlTimer);
if (pendingTimer) clearInterval(pendingTimer); if (pendingTimer) clearInterval(pendingTimer);
if (staleTimer) clearInterval(staleTimer);
ttlTimer = null; ttlTimer = null;
pendingTimer = null; pendingTimer = null;
staleTimer = null;
await db await db
.update(presence) .update(presence)
.set({ disconnectedAt: new Date() }) .set({ disconnectedAt: new Date() })

View File

@@ -20,6 +20,10 @@ const envSchema = z.object({
MAX_CONNECTIONS_PER_MESH: z.coerce.number().int().positive().default(100), MAX_CONNECTIONS_PER_MESH: z.coerce.number().int().positive().default(100),
MAX_MESSAGE_BYTES: z.coerce.number().int().positive().default(65_536), MAX_MESSAGE_BYTES: z.coerce.number().int().positive().default(65_536),
HOOK_RATE_LIMIT_PER_MIN: z.coerce.number().int().positive().default(30), HOOK_RATE_LIMIT_PER_MIN: z.coerce.number().int().positive().default(30),
MINIO_ENDPOINT: z.string().default("minio:9000"),
MINIO_ACCESS_KEY: z.string().default("claudemesh"),
MINIO_SECRET_KEY: z.string().default("changeme"),
MINIO_USE_SSL: z.coerce.boolean().default(false),
NODE_ENV: z NODE_ENV: z
.enum(["development", "production", "test"]) .enum(["development", "production", "test"])
.default("development"), .default("development"),

View File

@@ -15,24 +15,42 @@
import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import type { Duplex } from "node:stream"; import type { Duplex } from "node:stream";
import { WebSocketServer, type WebSocket } from "ws"; import { WebSocketServer, type WebSocket } from "ws";
import { eq } from "drizzle-orm";
import { env } from "./env"; import { env } from "./env";
import { db } from "./db";
import { messageQueue } from "@turbostarter/db/schema/mesh";
import { import {
connectPresence, connectPresence,
deleteFile,
disconnectPresence, disconnectPresence,
drainForMember, drainForMember,
findMemberByPubkey, findMemberByPubkey,
forgetMemory,
getFile,
getFileStatus,
getState,
handleHookSetStatus, handleHookSetStatus,
heartbeat, heartbeat,
joinGroup,
joinMesh, joinMesh,
leaveGroup,
listFiles,
listPeersInMesh, listPeersInMesh,
listState,
queueMessage, queueMessage,
recallMemory,
recordFileAccess,
refreshQueueDepth, refreshQueueDepth,
refreshStatusFromJsonl, refreshStatusFromJsonl,
rememberMemory,
setSummary, setSummary,
setState,
startSweepers, startSweepers,
stopSweepers, stopSweepers,
uploadFile,
writeStatus, writeStatus,
} from "./broker"; } from "./broker";
import { ensureBucket, meshBucketName, minioClient } from "./minio";
import type { import type {
HookSetStatusRequest, HookSetStatusRequest,
WSClientMessage, WSClientMessage,
@@ -56,7 +74,9 @@ interface PeerConn {
meshId: string; meshId: string;
memberId: string; memberId: string;
memberPubkey: string; memberPubkey: string;
sessionPubkey: string | null;
cwd: string; cwd: string;
groups: Array<{ name: string; role?: string }>;
} }
const connections = new Map<string, PeerConn>(); const connections = new Map<string, PeerConn>();
@@ -80,7 +100,10 @@ function sendToPeer(presenceId: string, msg: WSServerMessage): void {
} }
} }
async function maybePushQueuedMessages(presenceId: string): Promise<void> { async function maybePushQueuedMessages(
presenceId: string,
excludeSenderSessionPubkey?: string,
): Promise<void> {
const conn = connections.get(presenceId); const conn = connections.get(presenceId);
if (!conn) return; if (!conn) return;
const status = await refreshStatusFromJsonl( const status = await refreshStatusFromJsonl(
@@ -93,6 +116,9 @@ async function maybePushQueuedMessages(presenceId: string): Promise<void> {
conn.memberId, conn.memberId,
conn.memberPubkey, conn.memberPubkey,
status, status,
conn.sessionPubkey ?? undefined,
excludeSenderSessionPubkey,
conn.groups.map((g) => g.name),
); );
for (const m of messages) { for (const m of messages) {
const push: WSPushMessage = { const push: WSPushMessage = {
@@ -121,7 +147,7 @@ function handleHttpRequest(req: IncomingMessage, res: ServerResponse): void {
const started = Date.now(); const started = Date.now();
res.setHeader("Access-Control-Allow-Origin", "*"); res.setHeader("Access-Control-Allow-Origin", "*");
res.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS"); res.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS");
res.setHeader("Access-Control-Allow-Headers", "Content-Type"); res.setHeader("Access-Control-Allow-Headers", "Content-Type, X-Mesh-Id, X-Member-Id, X-File-Name, X-Tags, X-Persistent, X-Target-Spec");
if (req.method === "OPTIONS") { if (req.method === "OPTIONS") {
res.writeHead(204); res.writeHead(204);
res.end(); res.end();
@@ -158,6 +184,11 @@ function handleHttpRequest(req: IncomingMessage, res: ServerResponse): void {
return; return;
} }
if (req.method === "POST" && req.url === "/upload") {
handleUploadPost(req, res, started);
return;
}
res.writeHead(404); res.writeHead(404);
res.end("not found"); res.end("not found");
log.debug("http", { route, status: 404, latency_ms: Date.now() - started }); log.debug("http", { route, status: 404, latency_ms: Date.now() - started });
@@ -308,6 +339,119 @@ function handleJoinPost(
}); });
} }
function handleUploadPost(
req: IncomingMessage,
res: ServerResponse,
started: number,
): void {
const meshId = req.headers["x-mesh-id"] as string | undefined;
const memberId = req.headers["x-member-id"] as string | undefined;
const fileName = req.headers["x-file-name"] as string | undefined;
const tagsRaw = req.headers["x-tags"] as string | undefined;
const persistentRaw = req.headers["x-persistent"] as string | undefined;
const targetSpec = req.headers["x-target-spec"] as string | undefined;
if (!meshId || !memberId || !fileName) {
writeJson(res, 400, {
ok: false,
error: "X-Mesh-Id, X-Member-Id, and X-File-Name headers required",
});
return;
}
const persistent = persistentRaw !== "false";
let tags: string[] = [];
if (tagsRaw) {
try {
tags = JSON.parse(tagsRaw);
} catch {
tags = [];
}
}
const MAX_UPLOAD_SIZE = 50 * 1024 * 1024; // 50MB
const chunks: Buffer[] = [];
let total = 0;
let aborted = false;
req.on("data", (chunk: Buffer) => {
if (aborted) return;
total += chunk.length;
if (total > MAX_UPLOAD_SIZE) {
aborted = true;
writeJson(res, 413, { ok: false, error: "file too large (max 50MB)" });
req.destroy();
return;
}
chunks.push(chunk);
});
req.on("end", async () => {
if (aborted) return;
try {
const body = Buffer.concat(chunks);
if (body.length === 0) {
writeJson(res, 400, { ok: false, error: "empty body" });
return;
}
// Generate a file ID for the MinIO key
const { generateId } = await import("@turbostarter/shared/utils");
const fileId = generateId();
const dateStr = new Date().toISOString().split("T")[0];
const keyPrefix = persistent
? `shared/${fileId}`
: `ephemeral/${dateStr}/${fileId}`;
const minioKey = `${keyPrefix}/${fileName}`;
const bucket = meshBucketName(meshId);
// Ensure bucket exists + upload
await ensureBucket(bucket);
await minioClient.putObject(
bucket,
minioKey,
body,
body.length,
req.headers["content-type"]
? { "Content-Type": req.headers["content-type"] }
: undefined,
);
// Insert DB row
const dbFileId = await uploadFile({
meshId,
name: fileName,
sizeBytes: body.length,
mimeType: (req.headers["content-type"] as string) || undefined,
minioKey,
tags,
persistent,
uploadedByMember: memberId,
targetSpec: targetSpec || undefined,
});
writeJson(res, 200, { ok: true, fileId: dbFileId });
log.info("upload", {
route: "POST /upload",
mesh_id: meshId,
file_id: dbFileId,
name: fileName,
size: body.length,
persistent,
latency_ms: Date.now() - started,
});
} catch (e) {
writeJson(res, 500, {
ok: false,
error: e instanceof Error ? e.message : String(e),
});
log.error("upload handler error", {
error: e instanceof Error ? e.message : String(e),
});
}
});
}
function handleUpgrade( function handleUpgrade(
wss: WebSocketServer, wss: WebSocketServer,
req: IncomingMessage, req: IncomingMessage,
@@ -397,19 +541,24 @@ async function handleHello(
ws.close(1008, "unauthorized"); ws.close(1008, "unauthorized");
return null; return null;
} }
const initialGroups = hello.groups ?? [];
const presenceId = await connectPresence({ const presenceId = await connectPresence({
memberId: member.id, memberId: member.id,
sessionId: hello.sessionId, sessionId: hello.sessionId,
sessionPubkey: hello.sessionPubkey,
displayName: hello.displayName, displayName: hello.displayName,
pid: hello.pid, pid: hello.pid,
cwd: hello.cwd, cwd: hello.cwd,
groups: initialGroups,
}); });
connections.set(presenceId, { connections.set(presenceId, {
ws, ws,
meshId: hello.meshId, meshId: hello.meshId,
memberId: member.id, memberId: member.id,
memberPubkey: hello.pubkey, memberPubkey: hello.pubkey,
sessionPubkey: hello.sessionPubkey ?? null,
cwd: hello.cwd, cwd: hello.cwd,
groups: initialGroups,
}); });
incMeshCount(hello.meshId); incMeshCount(hello.meshId);
const effectiveDisplayName = hello.displayName || member.displayName; const effectiveDisplayName = hello.displayName || member.displayName;
@@ -434,6 +583,7 @@ async function handleSend(
const messageId = await queueMessage({ const messageId = await queueMessage({
meshId: conn.meshId, meshId: conn.meshId,
senderMemberId: conn.memberId, senderMemberId: conn.memberId,
senderSessionPubkey: conn.sessionPubkey ?? undefined,
targetSpec: msg.targetSpec, targetSpec: msg.targetSpec,
priority: msg.priority, priority: msg.priority,
nonce: msg.nonce, nonce: msg.nonce,
@@ -447,12 +597,68 @@ async function handleSend(
}; };
conn.ws.send(JSON.stringify(ack)); conn.ws.send(JSON.stringify(ack));
// Fan-out over connected peers in the same mesh. // Find sender's presenceId to exclude from fan-out.
let senderPresenceId: string | undefined;
for (const [pid, peer] of connections) { for (const [pid, peer] of connections) {
if (peer.ws === conn.ws) { senderPresenceId = pid; break; }
}
// Fan-out over connected peers in the same mesh — skip sender.
const isGroupTarget = msg.targetSpec.startsWith("@");
const isBroadcast =
msg.targetSpec === "*" ||
(isGroupTarget && msg.targetSpec === "@all");
const groupName = isGroupTarget && !isBroadcast
? msg.targetSpec.slice(1)
: null;
const isMulticast = isBroadcast || !!groupName;
// Build the push envelope once (reused for all recipients).
const pushEnvelope: WSPushMessage = {
type: "push",
messageId,
meshId: conn.meshId,
senderPubkey: conn.sessionPubkey ?? conn.memberPubkey,
priority: msg.priority,
nonce: msg.nonce,
ciphertext: msg.ciphertext,
createdAt: new Date().toISOString(),
};
for (const [pid, peer] of connections) {
if (pid === senderPresenceId) continue;
if (peer.meshId !== conn.meshId) continue; if (peer.meshId !== conn.meshId) continue;
if (msg.targetSpec !== "*" && peer.memberPubkey !== msg.targetSpec)
continue; if (isBroadcast) {
void maybePushQueuedMessages(pid); // broadcast — deliver to everyone
} else if (groupName) {
// group routing — deliver only if peer is in the group
if (!peer.groups.some((g) => g.name === groupName)) continue;
} else {
// direct routing — match by pubkey
if (peer.memberPubkey !== msg.targetSpec
&& peer.sessionPubkey !== msg.targetSpec)
continue;
}
if (isMulticast) {
// Multicast: push directly to each connected peer. The queue
// row has one delivered_at — can only be claimed once. Direct
// push ensures every connected peer receives the message.
sendToPeer(pid, pushEnvelope);
metrics.messagesRoutedTotal.inc({ priority: msg.priority });
} else {
// Direct: drain from queue (handles priority gating + offline).
void maybePushQueuedMessages(pid, conn.sessionPubkey ?? undefined);
}
}
// Mark multicast messages as delivered (they've been pushed directly).
if (isMulticast) {
await db
.update(messageQueue)
.set({ deliveredAt: new Date() })
.where(eq(messageQueue.id, messageId));
} }
} }
@@ -507,6 +713,7 @@ function handleConnection(ws: WebSocket): void {
displayName: p.displayName, displayName: p.displayName,
status: p.status as "idle" | "working" | "dnd", status: p.status as "idle" | "working" | "dnd",
summary: p.summary, summary: p.summary,
groups: p.groups,
sessionId: p.sessionId, sessionId: p.sessionId,
connectedAt: p.connectedAt.toISOString(), connectedAt: p.connectedAt.toISOString(),
})), })),
@@ -528,6 +735,337 @@ function handleConnection(ws: WebSocket): void {
}); });
break; break;
} }
case "join_group": {
const jg = msg as Extract<WSClientMessage, { type: "join_group" }>;
const updatedGroups = await joinGroup(presenceId, jg.name, jg.role);
conn.groups = updatedGroups;
log.info("ws join_group", {
presence_id: presenceId,
group: jg.name,
role: jg.role,
});
break;
}
case "leave_group": {
const lg = msg as Extract<WSClientMessage, { type: "leave_group" }>;
const updatedGroups = await leaveGroup(presenceId, lg.name);
conn.groups = updatedGroups;
log.info("ws leave_group", {
presence_id: presenceId,
group: lg.name,
});
break;
}
case "set_state": {
const ss = msg as Extract<WSClientMessage, { type: "set_state" }>;
// Look up the display name for attribution.
const senderName =
[...connections.entries()].find(
([pid]) => pid === presenceId,
)?.[1]?.memberPubkey;
const member = senderName
? await findMemberByPubkey(conn.meshId, senderName)
: null;
const displayName = member?.displayName ?? "unknown";
const stateRow = await setState(
conn.meshId,
ss.key,
ss.value,
presenceId,
displayName,
);
// Push state_change to ALL other peers in the same mesh.
for (const [pid, peer] of connections) {
if (pid === presenceId) continue;
if (peer.meshId !== conn.meshId) continue;
sendToPeer(pid, {
type: "state_change",
key: stateRow.key,
value: stateRow.value,
updatedBy: stateRow.updatedBy,
});
}
// Send confirmation back to sender as state_result.
sendToPeer(presenceId, {
type: "state_result",
key: stateRow.key,
value: stateRow.value,
updatedBy: stateRow.updatedBy,
updatedAt: stateRow.updatedAt.toISOString(),
});
log.info("ws set_state", {
presence_id: presenceId,
key: ss.key,
});
break;
}
case "get_state": {
const gs = msg as Extract<WSClientMessage, { type: "get_state" }>;
const stateEntry = await getState(conn.meshId, gs.key);
if (stateEntry) {
sendToPeer(presenceId, {
type: "state_result",
key: stateEntry.key,
value: stateEntry.value,
updatedBy: stateEntry.updatedBy,
updatedAt: stateEntry.updatedAt.toISOString(),
});
} else {
sendToPeer(presenceId, {
type: "state_result",
key: gs.key,
value: null,
updatedBy: "",
updatedAt: "",
});
}
log.info("ws get_state", {
presence_id: presenceId,
key: gs.key,
found: !!stateEntry,
});
break;
}
case "list_state": {
const entries = await listState(conn.meshId);
sendToPeer(presenceId, {
type: "state_list",
entries: entries.map((e) => ({
key: e.key,
value: e.value,
updatedBy: e.updatedBy,
updatedAt: e.updatedAt.toISOString(),
})),
});
log.info("ws list_state", {
presence_id: presenceId,
count: entries.length,
});
break;
}
case "remember": {
const rm = msg as Extract<WSClientMessage, { type: "remember" }>;
const memberInfo = conn.memberPubkey
? await findMemberByPubkey(conn.meshId, conn.memberPubkey)
: null;
const memoryId = await rememberMemory(
conn.meshId,
rm.content,
rm.tags ?? [],
memberInfo?.id,
memberInfo?.displayName,
);
sendToPeer(presenceId, {
type: "memory_stored",
id: memoryId,
});
log.info("ws remember", {
presence_id: presenceId,
memory_id: memoryId,
});
break;
}
case "recall": {
const rc = msg as Extract<WSClientMessage, { type: "recall" }>;
const memories = await recallMemory(conn.meshId, rc.query);
sendToPeer(presenceId, {
type: "memory_results",
memories: memories.map((m) => ({
id: m.id,
content: m.content,
tags: m.tags,
rememberedBy: m.rememberedBy,
rememberedAt: m.rememberedAt.toISOString(),
})),
});
log.info("ws recall", {
presence_id: presenceId,
query: rc.query.slice(0, 80),
results: memories.length,
});
break;
}
case "forget": {
const fg = msg as Extract<WSClientMessage, { type: "forget" }>;
await forgetMemory(conn.meshId, fg.memoryId);
sendToPeer(presenceId, {
type: "ack" as const,
id: fg.memoryId,
messageId: fg.memoryId,
queued: false,
});
log.info("ws forget", {
presence_id: presenceId,
memory_id: fg.memoryId,
});
break;
}
case "get_file": {
const gf = msg as Extract<WSClientMessage, { type: "get_file" }>;
const file = await getFile(conn.meshId, gf.fileId);
if (!file) {
sendError(conn.ws, "not_found", "file not found");
break;
}
// Access control: if targetSpec is set, verify peer matches
if (file.targetSpec) {
const matches =
file.targetSpec === conn.memberPubkey ||
file.targetSpec === conn.sessionPubkey ||
file.targetSpec === "*";
if (!matches) {
sendError(conn.ws, "forbidden", "file not targeted at you");
break;
}
}
// Generate presigned URL (60s expiry)
const bucket = meshBucketName(conn.meshId);
const presignedUrl = await minioClient.presignedGetObject(
bucket,
file.minioKey,
60,
);
// Record access
const memberInfo = conn.memberPubkey
? await findMemberByPubkey(conn.meshId, conn.memberPubkey)
: null;
await recordFileAccess(
gf.fileId,
conn.sessionPubkey ?? undefined,
memberInfo?.displayName,
);
sendToPeer(presenceId, {
type: "file_url",
fileId: gf.fileId,
url: presignedUrl,
name: file.name,
});
log.info("ws get_file", {
presence_id: presenceId,
file_id: gf.fileId,
});
break;
}
case "list_files": {
const lf = msg as Extract<WSClientMessage, { type: "list_files" }>;
const files = await listFiles(conn.meshId, lf.query, lf.from);
sendToPeer(presenceId, {
type: "file_list",
files: files.map((f) => ({
id: f.id,
name: f.name,
size: f.sizeBytes,
tags: f.tags,
uploadedBy: f.uploadedBy,
uploadedAt: f.uploadedAt.toISOString(),
persistent: f.persistent,
})),
});
log.info("ws list_files", {
presence_id: presenceId,
mesh_id: conn.meshId,
count: files.length,
});
break;
}
case "file_status": {
const fs = msg as Extract<WSClientMessage, { type: "file_status" }>;
const accesses = await getFileStatus(fs.fileId);
sendToPeer(presenceId, {
type: "file_status_result",
fileId: fs.fileId,
accesses: accesses.map((a) => ({
peerName: a.peerName,
accessedAt: a.accessedAt.toISOString(),
})),
});
log.info("ws file_status", {
presence_id: presenceId,
file_id: fs.fileId,
});
break;
}
case "delete_file": {
const df = msg as Extract<WSClientMessage, { type: "delete_file" }>;
await deleteFile(conn.meshId, df.fileId);
sendToPeer(presenceId, {
type: "ack" as const,
id: df.fileId,
messageId: df.fileId,
queued: false,
});
log.info("ws delete_file", {
presence_id: presenceId,
file_id: df.fileId,
});
break;
}
case "message_status": {
const ms = msg as Extract<WSClientMessage, { type: "message_status" }>;
// Look up the message in the queue.
const [mqRow] = await db
.select({
id: messageQueue.id,
targetSpec: messageQueue.targetSpec,
deliveredAt: messageQueue.deliveredAt,
meshId: messageQueue.meshId,
})
.from(messageQueue)
.where(eq(messageQueue.id, ms.messageId));
if (!mqRow || mqRow.meshId !== conn.meshId) {
sendError(conn.ws, "not_found", "message not found");
break;
}
// Build per-recipient status from connected peers.
const recipients: Array<{ name: string; pubkey: string; status: "delivered" | "held" | "disconnected" }> = [];
const isMulti = mqRow.targetSpec === "*" || mqRow.targetSpec.startsWith("@");
if (isMulti) {
const groupNameMs = mqRow.targetSpec.startsWith("@") && mqRow.targetSpec !== "@all"
? mqRow.targetSpec.slice(1) : null;
// Check all known presences for this mesh.
const peers = await listPeersInMesh(conn.meshId);
for (const p of peers) {
if (groupNameMs && !p.groups.some((g: { name: string }) => g.name === groupNameMs)) continue;
recipients.push({
name: p.displayName,
pubkey: p.pubkey,
status: mqRow.deliveredAt ? "delivered" : "held",
});
}
} else {
// Direct message — find the target peer.
const peers = await listPeersInMesh(conn.meshId);
const target = peers.find((p) => p.pubkey === mqRow.targetSpec);
if (target) {
recipients.push({
name: target.displayName,
pubkey: target.pubkey,
status: mqRow.deliveredAt ? "delivered" : (target.status === "idle" ? "held" : "held"),
});
} else {
recipients.push({
name: "unknown",
pubkey: mqRow.targetSpec.slice(0, 16),
status: "disconnected",
});
}
}
const resp: WSServerMessage = {
type: "message_status_result",
messageId: ms.messageId,
targetSpec: mqRow.targetSpec,
delivered: !!mqRow.deliveredAt,
deliveredAt: mqRow.deliveredAt?.toISOString() ?? null,
recipients,
};
sendToPeer(presenceId, resp);
log.info("ws message_status", {
presence_id: presenceId,
message_id: ms.messageId,
delivered: !!mqRow.deliveredAt,
});
break;
}
} }
} catch (e) { } catch (e) {
metrics.messagesRejectedTotal.inc({ reason: "parse_or_handler" }); metrics.messagesRejectedTotal.inc({ reason: "parse_or_handler" });

28
apps/broker/src/minio.ts Normal file
View File

@@ -0,0 +1,28 @@
/**
* MinIO client for file storage.
*
* Each mesh gets its own bucket (mesh-{meshId}). Files are stored under
* a key path that encodes persistence and origin:
* - persistent: shared/{fileId}/{originalName}
* - ephemeral: ephemeral/{YYYY-MM-DD}/{fileId}/{originalName}
*/
import { Client } from "minio";
import { env } from "./env";
export const minioClient = new Client({
endPoint: env.MINIO_ENDPOINT.split(":")[0]!,
port: parseInt(env.MINIO_ENDPOINT.split(":")[1] || "9000"),
useSSL: env.MINIO_USE_SSL,
accessKey: env.MINIO_ACCESS_KEY,
secretKey: env.MINIO_SECRET_KEY,
});
export async function ensureBucket(name: string): Promise<void> {
const exists = await minioClient.bucketExists(name);
if (!exists) await minioClient.makeBucket(name);
}
export function meshBucketName(meshId: string): string {
return `mesh-${meshId.toLowerCase().replace(/[^a-z0-9-]/g, "-")}`;
}

View File

@@ -52,10 +52,13 @@ export interface WSHelloMessage {
meshId: string; meshId: string;
memberId: string; memberId: string;
pubkey: string; // must match mesh.member.peerPubkey pubkey: string; // must match mesh.member.peerPubkey
sessionPubkey?: string; // ephemeral per-launch pubkey for message routing
displayName?: string; // optional override for this session displayName?: string; // optional override for this session
sessionId: string; sessionId: string;
pid: number; pid: number;
cwd: string; cwd: string;
/** Initial groups to join on connect. */
groups?: Array<{ name: string; role?: string }>;
/** ms epoch; broker rejects if outside ±60s of its own clock. */ /** ms epoch; broker rejects if outside ±60s of its own clock. */
timestamp: number; timestamp: number;
/** ed25519 signature (hex) over the canonical hello bytes: /** ed25519 signature (hex) over the canonical hello bytes:
@@ -102,6 +105,56 @@ export interface WSSetSummaryMessage {
summary: string; summary: string;
} }
/** Client → broker: join a group with optional role. */
export interface WSJoinGroupMessage {
type: "join_group";
name: string;
role?: string;
}
/** Client → broker: leave a group. */
export interface WSLeaveGroupMessage {
type: "leave_group";
name: string;
}
/** Client → broker: set a shared state key-value. */
export interface WSSetStateMessage {
type: "set_state";
key: string;
value: unknown;
}
/** Client → broker: read a shared state key. */
export interface WSGetStateMessage {
type: "get_state";
key: string;
}
/** Client → broker: list all shared state entries. */
export interface WSListStateMessage {
type: "list_state";
}
/** Client → broker: store a memory. */
export interface WSRememberMessage {
type: "remember";
content: string;
tags?: string[];
}
/** Client → broker: full-text search memories. */
export interface WSRecallMessage {
type: "recall";
query: string;
}
/** Client → broker: soft-delete a memory. */
export interface WSForgetMessage {
type: "forget";
memoryId: string;
}
/** Broker → client: acknowledgement for a send. */ /** Broker → client: acknowledgement for a send. */
export interface WSAckMessage { export interface WSAckMessage {
type: "ack"; type: "ack";
@@ -125,11 +178,137 @@ export interface WSPeersListMessage {
displayName: string; displayName: string;
status: PeerStatus; status: PeerStatus;
summary: string | null; summary: string | null;
groups: Array<{ name: string; role?: string }>;
sessionId: string; sessionId: string;
connectedAt: string; connectedAt: string;
}>; }>;
} }
/** Broker → client: a state key was changed by another peer. */
export interface WSStateChangeMessage {
type: "state_change";
key: string;
value: unknown;
updatedBy: string;
}
/** Broker → client: response to get_state. */
export interface WSStateResultMessage {
type: "state_result";
key: string;
value: unknown;
updatedAt: string;
updatedBy: string;
}
/** Broker → client: response to list_state. */
export interface WSStateListMessage {
type: "state_list";
entries: Array<{
key: string;
value: unknown;
updatedBy: string;
updatedAt: string;
}>;
}
/** Broker → client: acknowledgement for a remember. */
export interface WSMemoryStoredMessage {
type: "memory_stored";
id: string;
}
/** Broker → client: response to recall. */
export interface WSMemoryResultsMessage {
type: "memory_results";
memories: Array<{
id: string;
content: string;
tags: string[];
rememberedBy: string;
rememberedAt: string;
}>;
}
/** Client → broker: check delivery status of a message. */
export interface WSMessageStatusMessage {
type: "message_status";
messageId: string;
}
/** Broker → client: delivery status with per-recipient detail. */
export interface WSMessageStatusResultMessage {
type: "message_status_result";
messageId: string;
targetSpec: string;
delivered: boolean;
deliveredAt: string | null;
recipients: Array<{
name: string;
pubkey: string;
status: "delivered" | "held" | "disconnected";
}>;
}
// --- File sharing messages ---
/** Client → broker: get a presigned download URL for a file. */
export interface WSGetFileMessage {
type: "get_file";
fileId: string;
}
/** Client → broker: list files in the mesh. */
export interface WSListFilesMessage {
type: "list_files";
query?: string;
from?: string;
}
/** Client → broker: get access log for a file. */
export interface WSFileStatusMessage {
type: "file_status";
fileId: string;
}
/** Client → broker: soft-delete a file. */
export interface WSDeleteFileMessage {
type: "delete_file";
fileId: string;
}
/** Broker → client: presigned URL for downloading a file. */
export interface WSFileUrlMessage {
type: "file_url";
fileId: string;
url: string;
name: string;
}
/** Broker → client: list of files in the mesh. */
export interface WSFileListMessage {
type: "file_list";
files: Array<{
id: string;
name: string;
size: number;
tags: string[];
uploadedBy: string;
uploadedAt: string;
persistent: boolean;
}>;
}
/** Broker → client: access log for a file. */
export interface WSFileStatusResultMessage {
type: "file_status_result";
fileId: string;
accesses: Array<{
peerName: string;
accessedAt: string;
}>;
}
/** Broker → client: structured error. */ /** Broker → client: structured error. */
export interface WSErrorMessage { export interface WSErrorMessage {
type: "error"; type: "error";
@@ -143,11 +322,33 @@ export type WSClientMessage =
| WSSendMessage | WSSendMessage
| WSSetStatusMessage | WSSetStatusMessage
| WSListPeersMessage | WSListPeersMessage
| WSSetSummaryMessage; | WSSetSummaryMessage
| WSJoinGroupMessage
| WSLeaveGroupMessage
| WSSetStateMessage
| WSGetStateMessage
| WSListStateMessage
| WSRememberMessage
| WSRecallMessage
| WSForgetMessage
| WSMessageStatusMessage
| WSGetFileMessage
| WSListFilesMessage
| WSFileStatusMessage
| WSDeleteFileMessage;
export type WSServerMessage = export type WSServerMessage =
| WSHelloAckMessage | WSHelloAckMessage
| WSPushMessage | WSPushMessage
| WSAckMessage | WSAckMessage
| WSPeersListMessage | WSPeersListMessage
| WSStateChangeMessage
| WSStateResultMessage
| WSStateListMessage
| WSMemoryStoredMessage
| WSMemoryResultsMessage
| WSMessageStatusResultMessage
| WSFileUrlMessage
| WSFileListMessage
| WSFileStatusResultMessage
| WSErrorMessage; | WSErrorMessage;

View File

@@ -1,6 +1,6 @@
{ {
"name": "claudemesh-cli", "name": "claudemesh-cli",
"version": "0.1.9", "version": "0.4.0",
"description": "Claude Code MCP client for claudemesh — peer mesh messaging between Claude sessions.", "description": "Claude Code MCP client for claudemesh — peer mesh messaging between Claude sessions.",
"keywords": [ "keywords": [
"claude-code", "claude-code",

View File

@@ -14,7 +14,10 @@ import { parseInviteLink } from "../invite/parse";
import { enrollWithBroker } from "../invite/enroll"; import { enrollWithBroker } from "../invite/enroll";
import { generateKeypair } from "../crypto/keypair"; import { generateKeypair } from "../crypto/keypair";
import { loadConfig, saveConfig, getConfigPath } from "../state/config"; import { loadConfig, saveConfig, getConfigPath } from "../state/config";
import { hostname } from "node:os"; import { writeFileSync, mkdirSync } from "node:fs";
import { join, dirname } from "node:path";
import { homedir, hostname } from "node:os";
import { env } from "../env";
export async function runJoin(args: string[]): Promise<void> { export async function runJoin(args: string[]): Promise<void> {
const link = args[0]; const link = args[0];
@@ -78,6 +81,16 @@ export async function runJoin(args: string[]): Promise<void> {
}); });
saveConfig(config); saveConfig(config);
// 4b. Store invite token for per-session re-enrollment (launch --name).
const configDir = env.CLAUDEMESH_CONFIG_DIR ?? join(homedir(), ".claudemesh");
const inviteFile = join(configDir, `invite-${payload.mesh_slug}.txt`);
try {
mkdirSync(dirname(inviteFile), { recursive: true });
writeFileSync(inviteFile, link, "utf-8");
} catch {
// Non-fatal — launch will fall back to shared identity.
}
// 5. Report. // 5. Report.
console.log(""); console.log("");
console.log( console.log(

View File

@@ -11,32 +11,35 @@
*/ */
import { spawn } from "node:child_process"; import { spawn } from "node:child_process";
import { mkdtempSync, writeFileSync, rmSync } from "node:fs"; import { mkdtempSync, writeFileSync, rmSync, readdirSync, statSync } from "node:fs";
import { tmpdir, hostname } from "node:os"; import { tmpdir, hostname } from "node:os";
import { join } from "node:path"; import { join } from "node:path";
import { createInterface } from "node:readline"; import { createInterface } from "node:readline";
import { loadConfig, getConfigPath } from "../state/config"; import { loadConfig, getConfigPath } from "../state/config";
import type { Config, JoinedMesh } from "../state/config"; import type { Config, JoinedMesh, GroupEntry } from "../state/config";
import { generateKeypair } from "../crypto/keypair";
import { enrollWithBroker } from "../invite/enroll";
import { parseInviteLink } from "../invite/parse";
// --- Arg parsing --- // --- Arg parsing ---
interface LaunchArgs { interface LaunchArgs {
name: string | null; name: string | null;
role: string | null;
groups: string | null; // comma-separated, e.g. "frontend:lead,reviewers:member"
joinLink: string | null; joinLink: string | null;
meshSlug: string | null; meshSlug: string | null;
quiet: boolean; quiet: boolean;
skipPermConfirm: boolean;
claudeArgs: string[]; claudeArgs: string[];
} }
function parseArgs(argv: string[]): LaunchArgs { function parseArgs(argv: string[]): LaunchArgs {
const result: LaunchArgs = { const result: LaunchArgs = {
name: null, name: null,
role: null,
groups: null,
joinLink: null, joinLink: null,
meshSlug: null, meshSlug: null,
quiet: false, quiet: false,
skipPermConfirm: false,
claudeArgs: [], claudeArgs: [],
}; };
@@ -47,6 +50,14 @@ function parseArgs(argv: string[]): LaunchArgs {
result.name = argv[++i]!; result.name = argv[++i]!;
} else if (arg.startsWith("--name=")) { } else if (arg.startsWith("--name=")) {
result.name = arg.slice("--name=".length); result.name = arg.slice("--name=".length);
} else if (arg === "--role" && i + 1 < argv.length) {
result.role = argv[++i]!;
} else if (arg.startsWith("--role=")) {
result.role = arg.slice("--role=".length);
} else if (arg === "--groups" && i + 1 < argv.length) {
result.groups = argv[++i]!;
} else if (arg.startsWith("--groups=")) {
result.groups = arg.slice("--groups=".length);
} else if (arg === "--join" && i + 1 < argv.length) { } else if (arg === "--join" && i + 1 < argv.length) {
result.joinLink = argv[++i]!; result.joinLink = argv[++i]!;
} else if (arg.startsWith("--join=")) { } else if (arg.startsWith("--join=")) {
@@ -57,6 +68,8 @@ function parseArgs(argv: string[]): LaunchArgs {
result.meshSlug = arg.slice("--mesh=".length); result.meshSlug = arg.slice("--mesh=".length);
} else if (arg === "--quiet") { } else if (arg === "--quiet") {
result.quiet = true; result.quiet = true;
} else if (arg === "-y" || arg === "--yes") {
result.skipPermConfirm = true;
} else if (arg === "--") { } else if (arg === "--") {
result.claudeArgs.push(...argv.slice(i + 1)); result.claudeArgs.push(...argv.slice(i + 1));
break; break;
@@ -94,16 +107,83 @@ async function pickMesh(meshes: JoinedMesh[]): Promise<JoinedMesh> {
}); });
} }
// --- Group string parser ---
/** Parse "frontend:lead,reviewers:member,all" → GroupEntry[] */
function parseGroupsString(raw: string): GroupEntry[] {
return raw
.split(",")
.map((s) => s.trim())
.filter(Boolean)
.map((token) => {
const idx = token.indexOf(":");
if (idx === -1) return { name: token };
return { name: token.slice(0, idx), role: token.slice(idx + 1) };
});
}
// --- Interactive role/groups prompts ---
function askLine(prompt: string): Promise<string> {
const rl = createInterface({ input: process.stdin, output: process.stdout });
return new Promise((resolve) => {
rl.question(prompt, (answer) => {
rl.close();
resolve(answer.trim());
});
});
}
// --- Permission confirmation ---
async function confirmPermissions(): Promise<void> {
const useColor =
!process.env.NO_COLOR && process.env.TERM !== "dumb" && process.stdout.isTTY;
const bold = (s: string): string => (useColor ? `\x1b[1m${s}\x1b[22m` : s);
const dim = (s: string): string => (useColor ? `\x1b[2m${s}\x1b[22m` : s);
const yellow = (s: string): string => (useColor ? `\x1b[33m${s}\x1b[39m` : s);
console.log(yellow(bold(" Autonomous mode")));
console.log("");
console.log(" Claude will send and receive peer messages without asking");
console.log(" you first. Peers exchange text only — no file access,");
console.log(" no tool calls, no code execution.");
console.log("");
console.log(dim(" Same as: claude --dangerously-skip-permissions"));
console.log(dim(" Skip this prompt: claudemesh launch -y"));
console.log("");
const rl = createInterface({ input: process.stdin, output: process.stdout });
return new Promise((resolve, reject) => {
rl.question(` ${bold("Continue?")} [Y/n] `, (answer) => {
rl.close();
const a = answer.trim().toLowerCase();
if (a === "" || a === "y" || a === "yes") {
resolve();
} else {
console.log("\n Aborted. Run without autonomous mode:");
console.log(" claude --dangerously-load-development-channels server:claudemesh\n");
process.exit(0);
}
});
});
}
// --- Banner --- // --- Banner ---
function printBanner(name: string, meshSlug: string): void { function printBanner(name: string, meshSlug: string, role: string | null, groups: GroupEntry[]): void {
const useColor = const useColor =
!process.env.NO_COLOR && process.env.TERM !== "dumb" && process.stdout.isTTY; !process.env.NO_COLOR && process.env.TERM !== "dumb" && process.stdout.isTTY;
const dim = (s: string): string => (useColor ? `\x1b[2m${s}\x1b[22m` : s); const dim = (s: string): string => (useColor ? `\x1b[2m${s}\x1b[22m` : s);
const bold = (s: string): string => (useColor ? `\x1b[1m${s}\x1b[22m` : s); const bold = (s: string): string => (useColor ? `\x1b[1m${s}\x1b[22m` : s);
const roleSuffix = role ? ` (${role})` : "";
const groupTags = groups.length
? " [" + groups.map((g) => `@${g.name}${g.role ? `:${g.role}` : ""}`).join(", ") + "]"
: "";
const rule = "─".repeat(60); const rule = "─".repeat(60);
console.log(bold(`claudemesh launch`) + dim(` — as ${name} on ${meshSlug}`)); console.log(bold(`claudemesh launch`) + dim(` — as ${name}${roleSuffix} on ${meshSlug}${groupTags}`));
console.log(rule); console.log(rule);
console.log("Peer messages arrive as <channel> reminders in real-time."); console.log("Peer messages arrive as <channel> reminders in real-time.");
console.log("Peers send text only — they cannot call tools or read files."); console.log("Peers send text only — they cannot call tools or read files.");
@@ -174,16 +254,45 @@ export async function runLaunch(extraArgs: string[]): Promise<void> {
mesh = await pickMesh(config.meshes); mesh = await pickMesh(config.meshes);
} }
// 3. Set display name. Uses existing member identity — the broker // 3. Session identity + role/groups.
// creates a separate presence row per session (sessionId + pid) // The WS client auto-generates a per-session ephemeral keypair on
// and stores the per-session displayName override. // connect (sent in hello as sessionPubkey). We set display name via env var.
const displayName = args.name ?? `${hostname()}-${process.pid}`; const displayName = args.name ?? `${hostname()}-${process.pid}`;
// 4. Write session config to tmpdir (same mesh, same keypair). // Interactive wizard for role & groups (when not provided via flags and not --quiet).
let role: string | null = args.role;
let parsedGroups: GroupEntry[] = args.groups ? parseGroupsString(args.groups) : [];
if (!args.quiet) {
if (role === null) {
const answer = await askLine(" Role (optional): ");
if (answer) role = answer;
}
if (parsedGroups.length === 0 && args.groups === null) {
const answer = await askLine(" Groups (comma-separated, optional): ");
if (answer) parsedGroups = parseGroupsString(answer);
}
if (role || parsedGroups.length) console.log("");
}
// Clean up orphaned tmpdirs from crashed sessions (older than 1 hour)
const tmpBase = tmpdir();
try {
for (const entry of readdirSync(tmpBase)) {
if (!entry.startsWith("claudemesh-")) continue;
const full = join(tmpBase, entry);
const age = Date.now() - statSync(full).mtimeMs;
if (age > 3600_000) rmSync(full, { recursive: true, force: true });
}
} catch { /* best effort */ }
// 4. Write session config to tmpdir (isolates mesh selection).
const tmpDir = mkdtempSync(join(tmpdir(), "claudemesh-")); const tmpDir = mkdtempSync(join(tmpdir(), "claudemesh-"));
const sessionConfig: Config = { const sessionConfig: Config = {
version: 1, version: 1,
meshes: [mesh], meshes: [mesh],
displayName,
...(parsedGroups.length > 0 ? { groups: parsedGroups } : {}),
}; };
writeFileSync( writeFileSync(
join(tmpDir, "config.json"), join(tmpDir, "config.json"),
@@ -191,16 +300,22 @@ export async function runLaunch(extraArgs: string[]): Promise<void> {
"utf-8", "utf-8",
); );
// 5. Banner. // 5. Banner + permission confirmation.
if (!args.quiet) printBanner(displayName, mesh.slug); if (!args.quiet) {
printBanner(displayName, mesh.slug, role, parsedGroups);
// Auto-permissions confirmation — needed for autonomous peer messaging.
if (!args.skipPermConfirm) {
await confirmPermissions();
}
}
// 6. Spawn claude with ephemeral config + dev channel + display name. // 6. Spawn claude with ephemeral config + dev channel + auto-permissions.
// Strip any user-supplied --dangerously-load-development-channels // Strip any user-supplied --dangerously flags to avoid duplicates.
// to avoid duplicates — we always inject our own.
const filtered: string[] = []; const filtered: string[] = [];
for (let i = 0; i < args.claudeArgs.length; i++) { for (let i = 0; i < args.claudeArgs.length; i++) {
if (args.claudeArgs[i] === "--dangerously-load-development-channels") { if (args.claudeArgs[i] === "--dangerously-load-development-channels"
i++; // skip the next arg (the channel value) too || args.claudeArgs[i] === "--dangerously-skip-permissions") {
if (args.claudeArgs[i] === "--dangerously-load-development-channels") i++;
continue; continue;
} }
filtered.push(args.claudeArgs[i]!); filtered.push(args.claudeArgs[i]!);
@@ -208,6 +323,7 @@ export async function runLaunch(extraArgs: string[]): Promise<void> {
const claudeArgs = [ const claudeArgs = [
"--dangerously-load-development-channels", "--dangerously-load-development-channels",
"server:claudemesh", "server:claudemesh",
"--dangerously-skip-permissions",
...filtered, ...filtered,
]; ];

View File

@@ -62,8 +62,8 @@ async function resolveClient(to: string): Promise<{
target = rest; target = rest;
} }
} }
// Pubkey, channel, or broadcast — pass through directly. // Pubkey, channel, @group, or broadcast — pass through directly.
if (/^[0-9a-f]{64}$/.test(target) || target.startsWith("#") || target === "*") { if (/^[0-9a-f]{64}$/.test(target) || target.startsWith("#") || target.startsWith("@") || target === "*") {
if (targetClients.length === 1) { if (targetClients.length === 1) {
return { client: targetClients[0]!, targetSpec: target }; return { client: targetClients[0]!, targetSpec: target };
} }
@@ -98,6 +98,24 @@ async function resolveClient(to: string): Promise<{
}; };
} }
// Peer name cache to avoid calling listPeers on every incoming push
const peerNameCache = new Map<string, string>();
let peerNameCacheAge = 0;
const CACHE_TTL_MS = 30_000;
async function resolvePeerName(client: BrokerClient, pubkey: string): Promise<string> {
const now = Date.now();
if (now - peerNameCacheAge > CACHE_TTL_MS) {
peerNameCache.clear();
try {
const peers = await client.listPeers();
for (const p of peers) peerNameCache.set(p.pubkey, p.displayName);
} catch { /* best effort */ }
peerNameCacheAge = now;
}
return peerNameCache.get(pubkey) ?? `peer-${pubkey.slice(0, 8)}`;
}
function decryptFailedWarning(senderPubkey: string): string { function decryptFailedWarning(senderPubkey: string): string {
const who = senderPubkey ? senderPubkey.slice(0, 12) + "…" : "unknown sender"; const who = senderPubkey ? senderPubkey.slice(0, 12) + "…" : "unknown sender";
return `⚠ message from ${who} failed to decrypt (tampered or wrong keypair)`; return `⚠ message from ${who} failed to decrypt (tampered or wrong keypair)`;
@@ -111,32 +129,76 @@ function formatPush(p: InboundPush, meshSlug: string): string {
export async function startMcpServer(): Promise<void> { export async function startMcpServer(): Promise<void> {
const config = loadConfig(); const config = loadConfig();
const myName = config.displayName ?? "unnamed";
const myGroups = (config.groups ?? []).map(g => `@${g.name}${g.role ? ':' + g.role : ''}`).join(', ') || "none";
const server = new Server( const server = new Server(
{ name: "claudemesh", version: "0.1.4" }, { name: "claudemesh", version: "0.3.0" },
{ {
capabilities: { capabilities: {
experimental: { "claude/channel": {} }, experimental: { "claude/channel": {} },
tools: {}, tools: {},
}, },
instructions: `You are connected to claudemesh — a peer mesh for Claude Code sessions on this machine and elsewhere. instructions: `## Identity
You are "${myName}" — a peer in the claudemesh network. Your groups: ${myGroups}. You are one of several Claude Code sessions connected to the same mesh. No orchestrator exists — peers are equals. Your identity comes from your name and group roles, not from a central authority.
IMPORTANT: When you receive a <channel source="claudemesh" ...> message, RESPOND IMMEDIATELY. Do not wait until your current task is finished. Pause what you are doing, reply to the message using send_message, then resume your work. Treat incoming peer messages like a coworker tapping you on the shoulder — answer right away, even if you're in the middle of something. ## Responding to messages
When you receive a <channel source="claudemesh" ...> message, RESPOND IMMEDIATELY. Pause your current task, reply via send_message, then resume. Read from_name, mesh_slug, and priority from the channel attributes. Reply by setting \`to\` to the sender's from_name (display name). Stay in character per your system prompt. Do not ignore low-priority messages — acknowledge them briefly even if you defer action.
Read the from_id, from_name, mesh_slug, and priority attributes to understand context. Reply by calling send_message with the same target (for direct messages the from_id is the sender's pubkey). ## Tools
| Tool | Description |
|------|-------------|
| send_message(to, message, priority?) | Send to peer name, @group, or * broadcast. \`to\` accepts display name, pubkey hex, @groupname, or *. |
| list_peers(mesh_slug?) | List connected peers with status, summary, groups, and roles. |
| check_messages() | Drain buffered inbound messages (auto-pushed in most cases, use as fallback). |
| set_summary(summary) | Set 1-2 sentence description of your current work, visible to all peers. |
| set_status(status) | Override status: idle, working, or dnd. |
| join_group(name, role?) | Join a @group with optional role (lead, member, observer, or any string). |
| leave_group(name) | Leave a @group. |
| set_state(key, value) | Write shared state; pushes change to all peers. |
| get_state(key) | Read a shared state value. |
| list_state() | List all state keys with values, authors, and timestamps. |
| remember(content, tags?) | Store persistent knowledge with optional tags. |
| recall(query) | Full-text search over mesh memory. |
| forget(id) | Soft-delete a memory entry. |
| share_file(path, name?, tags?) | Share a persistent file with the mesh. |
| get_file(id, save_to) | Download a shared file to a local path. |
| list_files(query?, from?) | Find files shared in the mesh. |
| file_status(id) | Check who has accessed a file. |
| delete_file(id) | Remove a shared file from the mesh. |
Available tools: If multiple meshes are joined, prefix \`to\` with \`<mesh-slug>:\` to disambiguate (e.g. \`dev-team:Alice\`).
- list_peers: see joined meshes + their connection status
- send_message: send to a peer by display name, pubkey, #channel, or * broadcast (priority: now/next/low)
- check_messages: drain buffered inbound messages (usually auto-pushed)
- set_summary: 1-2 sentence summary of what you're working on
- set_status: manually override your status (idle/working/dnd)
Message priority: Multi-target: send_message accepts an array of targets for the 'to' field.
- "now": delivered immediately regardless of recipient status (use sparingly) send_message(to: ["Alice", "@backend"], message: "sprint starts")
- "next" (default): delivered when recipient is idle Targets are deduplicated — each peer receives the message once.
- "low": pull-only (check_messages)
If you have multiple joined meshes, prefix the \`to\` argument of send_message with \`<mesh-slug>:\` to disambiguate. Otherwise claudemesh picks the single joined mesh.`, Targeted views: when different audiences need different details about the same event,
send tailored messages instead of one generic broadcast:
send_message(to: "@frontend", message: "Auth v2: useAuth hook changed, see src/auth/")
send_message(to: "@backend", message: "Auth v2: new /api/auth/v2 endpoints, v1 deprecated")
send_message(to: "@pm", message: "Auth v2 done. 3 points, no blockers.")
## Groups
Groups are routing labels. Send to @groupname to multicast to all members. Roles are metadata that peers interpret: a "lead" gathers input before synthesizing a response, a "member" contributes when asked, an "observer" watches silently. Join and leave groups dynamically with join_group/leave_group. Check list_peers to see who belongs to which groups and their roles.
## State
Shared key-value store scoped to the mesh. Use get_state/set_state for live coordination facts (deploy frozen? current sprint? PR queue). set_state pushes the change to all connected peers. Read state before asking peers questions — the answer may already be there. State is operational, not archival.
## Memory
Persistent knowledge that survives across sessions. Use remember(content, tags?) to store lessons, decisions, and incidents. Use recall(query) to search before asking peers. New peers should recall at session start to load institutional knowledge.
## Files
share_file for persistent references, send_message(file:) for ephemeral attachments.
Tags on shared files make them searchable. Use list_files to find what peers shared.
## Priority
- "now": interrupt immediately, even if recipient is in DND (use for urgent: broken deploy, blocking issue)
- "next" (default): deliver when recipient goes idle (normal coordination)
- "low": pull-only via check_messages (FYI, non-blocking context)
## Coordination
Call list_peers at session start to understand who is online, their roles, and what they are working on. If you are a group lead, gather input from members before responding to external requests — do not answer alone. If you are a member, contribute to your lead when asked. Use @group messages for team-wide questions, direct messages for 1:1 coordination. Set a meaningful summary so peers know your current focus.`,
}, },
); );
@@ -158,22 +220,32 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
const { to, message, priority } = (args ?? {}) as SendMessageArgs; const { to, message, priority } = (args ?? {}) as SendMessageArgs;
if (!to || !message) if (!to || !message)
return text("send_message: `to` and `message` required", true); return text("send_message: `to` and `message` required", true);
const { client, targetSpec, error } = await resolveClient(to);
if (!client) // Handle multi-target: to can be string or string[]
return text(`send_message: ${error ?? "no client resolved"}`, true); const targets = Array.isArray(to) ? to : [to];
const result = await client.send( const results: string[] = [];
targetSpec, const seen = new Set<string>(); // dedup by resolved pubkey
message,
(priority ?? "next") as Priority, for (const target of targets) {
); const { client, targetSpec, error } = await resolveClient(target);
if (!result.ok) if (!client) {
return text( results.push(`${target}: ${error ?? "no client resolved"}`);
`send_message failed (${client.meshSlug}): ${result.error}`, continue;
true, }
if (seen.has(targetSpec)) continue; // dedup
seen.add(targetSpec);
const result = await client.send(
targetSpec,
message,
(priority ?? "next") as Priority,
); );
return text( if (!result.ok) {
`Sent to ${targetSpec} via ${client.meshSlug} [${priority ?? "next"}] → ${result.messageId}`, results.push(`${target}: ${result.error}`);
); } else {
results.push(`${target}${result.messageId}`);
}
}
return text(results.join("\n"));
} }
case "list_peers": { case "list_peers": {
@@ -197,7 +269,8 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
} else { } else {
const peerLines = peers.map((p) => { const peerLines = peers.map((p) => {
const summary = p.summary ? ` — "${p.summary}"` : ""; const summary = p.summary ? ` — "${p.summary}"` : "";
return `- **${p.displayName}** [${p.status}] (${p.pubkey.slice(0, 12)}…)${summary}`; const groupsStr = p.groups?.length ? ` [${p.groups.map(g => `@${g.name}${g.role ? ':' + g.role : ''}`).join(', ')}]` : "";
return `- **${p.displayName}** [${p.status}]${groupsStr} (${p.pubkey.slice(0, 12)}…)${summary}`;
}); });
sections.push(`${header}\n${peerLines.join("\n")}`); sections.push(`${header}\n${peerLines.join("\n")}`);
} }
@@ -205,6 +278,24 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
return text(sections.join("\n\n")); return text(sections.join("\n\n"));
} }
case "message_status": {
const { id } = (args ?? {}) as { id?: string };
if (!id) return text("message_status: `id` required", true);
const client = allClients()[0];
if (!client) return text("message_status: not connected", true);
const result = await client.messageStatus(id);
if (!result) return text(`Message ${id} not found or timed out.`);
const recipientLines = result.recipients.map(
(r: { name: string; pubkey: string; status: string }) =>
` - ${r.name} (${r.pubkey.slice(0, 12)}…): ${r.status}`,
);
return text(
`Message ${id.slice(0, 12)}… → ${result.targetSpec}\n` +
`Delivered: ${result.delivered}${result.deliveredAt ? ` at ${result.deliveredAt}` : ""}\n` +
`Recipients:\n${recipientLines.join("\n")}`,
);
}
case "check_messages": { case "check_messages": {
const drained: string[] = []; const drained: string[] = [];
for (const c of allClients()) { for (const c of allClients()) {
@@ -234,6 +325,136 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
return text(`Status set to ${s} across ${allClients().length} mesh(es).`); return text(`Status set to ${s} across ${allClients().length} mesh(es).`);
} }
case "join_group": {
const { name: groupName, role } = (args ?? {}) as { name?: string; role?: string };
if (!groupName) return text("join_group: `name` required", true);
for (const c of allClients()) await c.joinGroup(groupName, role);
return text(`Joined @${groupName}${role ? ` as ${role}` : ""}`);
}
case "leave_group": {
const { name: groupName } = (args ?? {}) as { name?: string };
if (!groupName) return text("leave_group: `name` required", true);
for (const c of allClients()) await c.leaveGroup(groupName);
return text(`Left @${groupName}`);
}
// --- State ---
case "set_state": {
const { key, value } = (args ?? {}) as { key?: string; value?: unknown };
if (!key) return text("set_state: `key` required", true);
for (const c of allClients()) await c.setState(key, value);
return text(`State set: ${key} = ${JSON.stringify(value)}`);
}
case "get_state": {
const { key } = (args ?? {}) as { key?: string };
if (!key) return text("get_state: `key` required", true);
const client = allClients()[0];
if (!client) return text("get_state: not connected", true);
const result = await client.getState(key);
if (!result) return text(`State "${key}" not found.`);
return text(`${key} = ${JSON.stringify(result.value)} (set by ${result.updatedBy} at ${result.updatedAt})`);
}
case "list_state": {
const client = allClients()[0];
if (!client) return text("list_state: not connected", true);
const entries = await client.listState();
if (entries.length === 0) return text("No shared state set.");
const lines = entries.map(e => `- **${e.key}** = ${JSON.stringify(e.value)} (by ${e.updatedBy})`);
return text(lines.join("\n"));
}
// --- Memory ---
case "remember": {
const { content, tags } = (args ?? {}) as { content?: string; tags?: string[] };
if (!content) return text("remember: `content` required", true);
const client = allClients()[0];
if (!client) return text("remember: not connected", true);
const id = await client.remember(content, tags);
return text(`Remembered${id ? ` (${id})` : ""}: "${content.slice(0, 80)}${content.length > 80 ? '...' : ''}"`);
}
case "recall": {
const { query } = (args ?? {}) as { query?: string };
if (!query) return text("recall: `query` required", true);
const client = allClients()[0];
if (!client) return text("recall: not connected", true);
const memories = await client.recall(query);
if (memories.length === 0) return text(`No memories found for "${query}".`);
const lines = memories.map(m => `- [${m.id.slice(0, 8)}] ${m.content} (by ${m.rememberedBy}, ${m.rememberedAt})`);
return text(`${memories.length} memor${memories.length === 1 ? 'y' : 'ies'}:\n${lines.join("\n")}`);
}
case "forget": {
const { id } = (args ?? {}) as { id?: string };
if (!id) return text("forget: `id` required", true);
const client = allClients()[0];
if (!client) return text("forget: not connected", true);
await client.forget(id);
return text(`Forgotten: ${id}`);
}
// --- Files ---
case "share_file": {
const { path: filePath, name: fileName, tags } = (args ?? {}) as { path?: string; name?: string; tags?: string[] };
if (!filePath) return text("share_file: `path` required", true);
const { existsSync } = await import("node:fs");
if (!existsSync(filePath)) return text(`share_file: file not found: ${filePath}`, true);
const client = allClients()[0];
if (!client) return text("share_file: not connected", true);
const fileId = await client.uploadFile(filePath, client.meshId, client.meshSlug, {
name: fileName, tags, persistent: true,
});
if (!fileId) return text("share_file: upload failed", true);
return text(`Shared: ${fileName ?? filePath} (${fileId})`);
}
case "get_file": {
const { id, save_to } = (args ?? {}) as { id?: string; save_to?: string };
if (!id || !save_to) return text("get_file: `id` and `save_to` required", true);
const client = allClients()[0];
if (!client) return text("get_file: not connected", true);
const result = await client.getFile(id);
if (!result) return text(`get_file: file ${id} not found`, true);
const res = await fetch(result.url, { signal: AbortSignal.timeout(30_000) });
if (!res.ok) return text(`get_file: download failed (${res.status})`, true);
const { writeFileSync, mkdirSync } = await import("node:fs");
const { dirname } = await import("node:path");
mkdirSync(dirname(save_to), { recursive: true });
writeFileSync(save_to, Buffer.from(await res.arrayBuffer()));
return text(`Downloaded: ${result.name}${save_to}`);
}
case "list_files": {
const { query, from } = (args ?? {}) as { query?: string; from?: string };
const client = allClients()[0];
if (!client) return text("list_files: not connected", true);
const files = await client.listFiles(query, from);
if (files.length === 0) return text("No files found.");
const lines = files.map(f =>
`- **${f.name}** (${f.id.slice(0, 8)}…, ${f.size} bytes) by ${f.uploadedBy}${f.tags.length ? ` [${f.tags.join(", ")}]` : ""}`
);
return text(lines.join("\n"));
}
case "file_status": {
const { id } = (args ?? {}) as { id?: string };
if (!id) return text("file_status: `id` required", true);
const client = allClients()[0];
if (!client) return text("file_status: not connected", true);
const accesses = await client.fileStatus(id);
if (accesses.length === 0) return text("No one has accessed this file yet.");
const lines = accesses.map(a => `- ${a.peerName} at ${a.accessedAt}`);
return text(`Accessed by:\n${lines.join("\n")}`);
}
case "delete_file": {
const { id } = (args ?? {}) as { id?: string };
if (!id) return text("delete_file: `id` required", true);
const client = allClients()[0];
if (!client) return text("delete_file: not connected", true);
await client.deleteFile(id);
return text(`Deleted: ${id}`);
}
default: default:
return text(`Unknown tool: ${name}`, true); return text(`Unknown tool: ${name}`, true);
} }
@@ -251,8 +472,9 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
for (const client of allClients()) { for (const client of allClients()) {
client.onPush(async (msg) => { client.onPush(async (msg) => {
const fromPubkey = msg.senderPubkey || ""; const fromPubkey = msg.senderPubkey || "";
// Resolve sender's display name from the cached peer list.
const fromName = fromPubkey const fromName = fromPubkey
? `peer-${fromPubkey.slice(0, 8)}` ? await resolvePeerName(client, fromPubkey)
: "unknown"; : "unknown";
const content = msg.plaintext ?? decryptFailedWarning(fromPubkey); const content = msg.plaintext ?? decryptFailedWarning(fromPubkey);
try { try {
@@ -276,6 +498,22 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
/* channel push is best-effort; check_messages is the fallback */ /* channel push is best-effort; check_messages is the fallback */
} }
}); });
client.onStateChange(async (change) => {
try {
await server.notification({
method: "notifications/claude/channel",
params: {
content: `[state] ${change.key} = ${JSON.stringify(change.value)} (set by ${change.updatedBy})`,
meta: {
kind: "state_change",
key: change.key,
updated_by: change.updatedBy,
},
},
});
} catch { /* best effort */ }
});
} }
const shutdown = (): void => { const shutdown = (): void => {

View File

@@ -12,13 +12,16 @@ export const TOOLS: Tool[] = [
{ {
name: "send_message", name: "send_message",
description: description:
"Send a message to a peer in one of your joined meshes. `to` can be a peer display name (resolved via list_peers), hex pubkey, `#channel`, or `*` for broadcast. `priority` controls delivery: `now` bypasses busy gates, `next` waits for idle (default), `low` is pull-only.", "Send a message to a peer in one of your joined meshes. `to` can be a peer display name (resolved via list_peers), hex pubkey, @group, `#channel`, or `*` for broadcast. `priority` controls delivery: `now` bypasses busy gates, `next` waits for idle (default), `low` is pull-only.",
inputSchema: { inputSchema: {
type: "object", type: "object",
properties: { properties: {
to: { to: {
type: "string", oneOf: [
description: "Peer name, pubkey, or #channel", { type: "string", description: "Peer name, pubkey, @group" },
{ type: "array", items: { type: "string" }, description: "Multiple targets" },
],
description: "Single target or array of targets",
}, },
message: { type: "string", description: "Message text" }, message: { type: "string", description: "Message text" },
priority: { priority: {
@@ -44,6 +47,21 @@ export const TOOLS: Tool[] = [
}, },
}, },
}, },
{
name: "message_status",
description:
"Check the delivery status of a sent message. Shows whether each recipient received it.",
inputSchema: {
type: "object",
properties: {
id: {
type: "string",
description: "Message ID (returned by send_message)",
},
},
required: ["id"],
},
},
{ {
name: "check_messages", name: "check_messages",
description: description:
@@ -78,4 +96,177 @@ export const TOOLS: Tool[] = [
required: ["status"], required: ["status"],
}, },
}, },
{
name: "join_group",
description:
"Join a group with an optional role. Other peers see your group membership in list_peers.",
inputSchema: {
type: "object",
properties: {
name: { type: "string", description: "Group name (without @)" },
role: {
type: "string",
description: "Your role in the group (e.g. lead, member, observer)",
},
},
required: ["name"],
},
},
{
name: "leave_group",
description: "Leave a group.",
inputSchema: {
type: "object",
properties: {
name: { type: "string", description: "Group name (without @)" },
},
required: ["name"],
},
},
// --- State tools ---
{
name: "set_state",
description:
"Set a shared state value visible to all peers in the mesh. Pushes a change notification.",
inputSchema: {
type: "object",
properties: {
key: { type: "string" },
value: { description: "Any JSON value" },
},
required: ["key", "value"],
},
},
{
name: "get_state",
description: "Read a shared state value.",
inputSchema: {
type: "object",
properties: {
key: { type: "string" },
},
required: ["key"],
},
},
{
name: "list_state",
description: "List all shared state keys and values in the mesh.",
inputSchema: { type: "object", properties: {} },
},
// --- Memory tools ---
{
name: "remember",
description:
"Store persistent knowledge in the mesh's shared memory. Survives across sessions.",
inputSchema: {
type: "object",
properties: {
content: {
type: "string",
description: "The knowledge to remember",
},
tags: {
type: "array",
items: { type: "string" },
description: "Optional categorization tags",
},
},
required: ["content"],
},
},
{
name: "recall",
description: "Search the mesh's shared memory by relevance.",
inputSchema: {
type: "object",
properties: {
query: { type: "string", description: "Search query" },
},
required: ["query"],
},
},
{
name: "forget",
description: "Remove a memory from the mesh's shared knowledge.",
inputSchema: {
type: "object",
properties: {
id: { type: "string", description: "Memory ID to forget" },
},
required: ["id"],
},
},
// --- File tools ---
{
name: "share_file",
description:
"Share a persistent file with the mesh. All current and future peers can access it.",
inputSchema: {
type: "object",
properties: {
path: { type: "string", description: "Local file path to share" },
name: {
type: "string",
description: "Display name (defaults to filename)",
},
tags: {
type: "array",
items: { type: "string" },
description: "Tags for categorization",
},
},
required: ["path"],
},
},
{
name: "get_file",
description: "Download a shared file to a local path.",
inputSchema: {
type: "object",
properties: {
id: { type: "string", description: "File ID" },
save_to: {
type: "string",
description: "Local path to save the file",
},
},
required: ["id", "save_to"],
},
},
{
name: "list_files",
description: "List files shared in the mesh.",
inputSchema: {
type: "object",
properties: {
query: { type: "string", description: "Search by name or tags" },
from: { type: "string", description: "Filter by uploader name" },
},
},
},
{
name: "file_status",
description: "Check who has accessed a shared file.",
inputSchema: {
type: "object",
properties: {
id: { type: "string", description: "File ID" },
},
required: ["id"],
},
},
{
name: "delete_file",
description: "Remove a shared file from the mesh.",
inputSchema: {
type: "object",
properties: {
id: { type: "string", description: "File ID" },
},
required: ["id"],
},
},
]; ];

View File

@@ -6,7 +6,7 @@ export type Priority = "now" | "next" | "low";
export type PeerStatus = "idle" | "working" | "dnd"; export type PeerStatus = "idle" | "working" | "dnd";
export interface SendMessageArgs { export interface SendMessageArgs {
to: string; // peer name, pubkey, or #channel to: string | string[]; // peer name, pubkey, @group, or array of targets
message: string; message: string;
priority?: Priority; priority?: Priority;
} }

View File

@@ -28,9 +28,16 @@ export interface JoinedMesh {
joinedAt: string; joinedAt: string;
} }
export interface GroupEntry {
name: string;
role?: string;
}
export interface Config { export interface Config {
version: 1; version: 1;
meshes: JoinedMesh[]; meshes: JoinedMesh[];
displayName?: string; // per-session override, written by `claudemesh launch --name`
groups?: GroupEntry[];
} }
const CONFIG_DIR = env.CLAUDEMESH_CONFIG_DIR ?? join(homedir(), ".claudemesh"); const CONFIG_DIR = env.CLAUDEMESH_CONFIG_DIR ?? join(homedir(), ".claudemesh");
@@ -46,7 +53,7 @@ export function loadConfig(): Config {
if (!parsed || !Array.isArray(parsed.meshes)) { if (!parsed || !Array.isArray(parsed.meshes)) {
return { version: 1, meshes: [] }; return { version: 1, meshes: [] };
} }
return { version: 1, meshes: parsed.meshes }; return { version: 1, meshes: parsed.meshes, displayName: parsed.displayName, groups: parsed.groups };
} catch (e) { } catch (e) {
throw new Error( throw new Error(
`Failed to load ${CONFIG_PATH}: ${e instanceof Error ? e.message : String(e)}`, `Failed to load ${CONFIG_PATH}: ${e instanceof Error ? e.message : String(e)}`,

View File

@@ -21,6 +21,7 @@ import {
isDirectTarget, isDirectTarget,
} from "../crypto/envelope"; } from "../crypto/envelope";
import { signHello } from "../crypto/hello-sig"; import { signHello } from "../crypto/hello-sig";
import { generateKeypair } from "../crypto/keypair";
export type Priority = "now" | "next" | "low"; export type Priority = "now" | "next" | "low";
export type ConnStatus = "connecting" | "open" | "closed" | "reconnecting"; export type ConnStatus = "connecting" | "open" | "closed" | "reconnecting";
@@ -30,6 +31,7 @@ export interface PeerInfo {
displayName: string; displayName: string;
status: string; status: string;
summary: string | null; summary: string | null;
groups: Array<{ name: string; role?: string }>;
sessionId: string; sessionId: string;
connectedAt: string; connectedAt: string;
} }
@@ -74,6 +76,13 @@ export class BrokerClient {
private pushHandlers = new Set<PushHandler>(); private pushHandlers = new Set<PushHandler>();
private pushBuffer: InboundPush[] = []; private pushBuffer: InboundPush[] = [];
private listPeersResolvers: Array<(peers: PeerInfo[]) => void> = []; private listPeersResolvers: Array<(peers: PeerInfo[]) => void> = [];
private stateResolvers: Array<(result: { key: string; value: unknown; updatedBy: string; updatedAt: string } | null) => void> = [];
private stateListResolvers: Array<(entries: Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) => void> = [];
private memoryStoreResolvers: Array<(id: string | null) => void> = [];
private memoryRecallResolvers: Array<(memories: Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) => void> = [];
private stateChangeHandlers = new Set<(change: { key: string; value: unknown; updatedBy: string }) => void>();
private sessionPubkey: string | null = null;
private sessionSecretKey: string | null = null;
private closed = false; private closed = false;
private reconnectAttempt = 0; private reconnectAttempt = 0;
private helloTimer: NodeJS.Timeout | null = null; private helloTimer: NodeJS.Timeout | null = null;
@@ -83,6 +92,7 @@ export class BrokerClient {
private mesh: JoinedMesh, private mesh: JoinedMesh,
private opts: { private opts: {
onStatusChange?: (status: ConnStatus) => void; onStatusChange?: (status: ConnStatus) => void;
displayName?: string;
debug?: boolean; debug?: boolean;
} = {}, } = {},
) {} ) {}
@@ -109,8 +119,15 @@ export class BrokerClient {
return new Promise<void>((resolve, reject) => { return new Promise<void>((resolve, reject) => {
const onOpen = async (): Promise<void> => { const onOpen = async (): Promise<void> => {
this.debug("ws open → signing + sending hello"); this.debug("ws open → generating session keypair + signing hello");
try { try {
// Only generate session keypair on first connect, not reconnects
if (!this.sessionPubkey) {
const sessionKP = await generateKeypair();
this.sessionPubkey = sessionKP.publicKey;
this.sessionSecretKey = sessionKP.secretKey;
}
const { timestamp, signature } = await signHello( const { timestamp, signature } = await signHello(
this.mesh.meshId, this.mesh.meshId,
this.mesh.memberId, this.mesh.memberId,
@@ -123,7 +140,8 @@ export class BrokerClient {
meshId: this.mesh.meshId, meshId: this.mesh.meshId,
memberId: this.mesh.memberId, memberId: this.mesh.memberId,
pubkey: this.mesh.pubkey, pubkey: this.mesh.pubkey,
displayName: process.env.CLAUDEMESH_DISPLAY_NAME || undefined, sessionPubkey: this.sessionPubkey,
displayName: process.env.CLAUDEMESH_DISPLAY_NAME || this.opts.displayName || undefined,
sessionId: `${process.pid}-${Date.now()}`, sessionId: `${process.pid}-${Date.now()}`,
pid: process.pid, pid: process.pid,
cwd: process.cwd(), cwd: process.cwd(),
@@ -203,7 +221,7 @@ export class BrokerClient {
const env = await encryptDirect( const env = await encryptDirect(
message, message,
targetSpec, targetSpec,
this.mesh.secretKey, this.sessionSecretKey ?? this.mesh.secretKey,
); );
nonce = env.nonce; nonce = env.nonce;
ciphertext = env.ciphertext; ciphertext = env.ciphertext;
@@ -300,6 +318,211 @@ export class BrokerClient {
this.ws.send(JSON.stringify({ type: "set_summary", summary })); this.ws.send(JSON.stringify({ type: "set_summary", summary }));
} }
/** Join a group with an optional role. */
async joinGroup(name: string, role?: string): Promise<void> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return;
this.ws.send(JSON.stringify({ type: "join_group", name, role }));
}
/** Leave a group. */
async leaveGroup(name: string): Promise<void> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return;
this.ws.send(JSON.stringify({ type: "leave_group", name }));
}
// --- State ---
/** Set a shared state value visible to all peers in the mesh. */
async setState(key: string, value: unknown): Promise<void> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return;
this.ws.send(JSON.stringify({ type: "set_state", key, value }));
}
/** Read a shared state value. */
async getState(key: string): Promise<{ key: string; value: unknown; updatedBy: string; updatedAt: string } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => {
this.stateResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "get_state", key }));
setTimeout(() => {
const idx = this.stateResolvers.indexOf(resolve);
if (idx !== -1) {
this.stateResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
});
}
/** List all shared state keys and values. */
async listState(): Promise<Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => {
this.stateListResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "list_state" }));
setTimeout(() => {
const idx = this.stateListResolvers.indexOf(resolve);
if (idx !== -1) {
this.stateListResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
});
}
// --- Memory ---
/** Store persistent knowledge in the mesh's shared memory. */
async remember(content: string, tags?: string[]): Promise<string | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => {
this.memoryStoreResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "remember", content, tags }));
setTimeout(() => {
const idx = this.memoryStoreResolvers.indexOf(resolve);
if (idx !== -1) {
this.memoryStoreResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
});
}
/** Search the mesh's shared memory by relevance. */
async recall(query: string): Promise<Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => {
this.memoryRecallResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "recall", query }));
setTimeout(() => {
const idx = this.memoryRecallResolvers.indexOf(resolve);
if (idx !== -1) {
this.memoryRecallResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
});
}
/** Remove a memory from the mesh's shared knowledge. */
async forget(memoryId: string): Promise<void> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return;
this.ws.send(JSON.stringify({ type: "forget", memoryId }));
}
/** Check delivery status of a sent message. */
private messageStatusResolvers: Array<(result: { messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null) => void> = [];
private fileUrlResolvers: Array<(result: { url: string; name: string } | null) => void> = [];
private fileListResolvers: Array<(files: Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) => void> = [];
private fileStatusResolvers: Array<(accesses: Array<{ peerName: string; accessedAt: string }>) => void> = [];
async messageStatus(messageId: string): Promise<{ messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => {
this.messageStatusResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "message_status", messageId }));
setTimeout(() => {
const idx = this.messageStatusResolvers.indexOf(resolve);
if (idx !== -1) { this.messageStatusResolvers.splice(idx, 1); resolve(null); }
}, 5_000);
});
}
// --- Files ---
/** Get a download URL for a shared file. */
async getFile(fileId: string): Promise<{ url: string; name: string } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => {
this.fileUrlResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "get_file", fileId }));
setTimeout(() => {
const idx = this.fileUrlResolvers.indexOf(resolve);
if (idx !== -1) {
this.fileUrlResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
});
}
/** List files shared in the mesh. */
async listFiles(query?: string, from?: string): Promise<Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => {
this.fileListResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "list_files", query, from }));
setTimeout(() => {
const idx = this.fileListResolvers.indexOf(resolve);
if (idx !== -1) {
this.fileListResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
});
}
/** Check who has accessed a shared file. */
async fileStatus(fileId: string): Promise<Array<{ peerName: string; accessedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => {
this.fileStatusResolvers.push(resolve);
this.ws!.send(JSON.stringify({ type: "file_status", fileId }));
setTimeout(() => {
const idx = this.fileStatusResolvers.indexOf(resolve);
if (idx !== -1) {
this.fileStatusResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
});
}
/** Delete a shared file from the mesh. */
async deleteFile(fileId: string): Promise<void> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return;
this.ws.send(JSON.stringify({ type: "delete_file", fileId }));
}
/** Upload a file to the broker via HTTP POST. Returns file ID or null. */
async uploadFile(filePath: string, meshId: string, memberId: string, opts: {
name?: string; tags?: string[]; persistent?: boolean; targetSpec?: string;
}): Promise<string | null> {
const { readFileSync } = await import("node:fs");
const { basename } = await import("node:path");
const data = readFileSync(filePath);
const fileName = opts.name ?? basename(filePath);
// Convert WS broker URL to HTTP
const brokerHttp = this.mesh.brokerUrl
.replace("wss://", "https://")
.replace("ws://", "http://")
.replace("/ws", "");
const res = await fetch(`${brokerHttp}/upload`, {
method: "POST",
headers: {
"Content-Type": "application/octet-stream",
"X-Mesh-Id": meshId,
"X-Member-Id": memberId,
"X-File-Name": fileName,
"X-Tags": JSON.stringify(opts.tags ?? []),
"X-Persistent": String(opts.persistent ?? true),
"X-Target-Spec": opts.targetSpec ?? "",
},
body: data,
signal: AbortSignal.timeout(30_000),
});
const body = await res.json() as { ok?: boolean; fileId?: string };
return body.fileId ?? null;
}
/** Subscribe to state change notifications. Returns an unsubscribe function. */
onStateChange(handler: (change: { key: string; value: unknown; updatedBy: string }) => void): () => void {
this.stateChangeHandlers.add(handler);
return () => this.stateChangeHandlers.delete(handler);
}
close(): void { close(): void {
this.closed = true; this.closed = true;
if (this.helloTimer) clearTimeout(this.helloTimer); if (this.helloTimer) clearTimeout(this.helloTimer);
@@ -349,7 +572,7 @@ export class BrokerClient {
plaintext = await decryptDirect( plaintext = await decryptDirect(
{ nonce, ciphertext }, { nonce, ciphertext },
senderPubkey, senderPubkey,
this.mesh.secretKey, this.sessionSecretKey ?? this.mesh.secretKey,
); );
} }
// Legacy/broadcast path: no senderPubkey means the message // Legacy/broadcast path: no senderPubkey means the message
@@ -366,6 +589,19 @@ export class BrokerClient {
plaintext = null; plaintext = null;
} }
} }
// Fallback: if direct decrypt failed, try plaintext base64 decode.
// This handles broadcasts and key mismatches gracefully.
if (plaintext === null && ciphertext) {
try {
const decoded = Buffer.from(ciphertext, "base64").toString("utf-8");
// Sanity check: valid UTF-8 text (not binary garbage)
if (/^[\x20-\x7E\s\u00A0-\uFFFF]*$/.test(decoded) && decoded.length > 0) {
plaintext = decoded;
}
} catch {
plaintext = null;
}
}
const push: InboundPush = { const push: InboundPush = {
messageId: String(msg.messageId ?? ""), messageId: String(msg.messageId ?? ""),
meshId: String(msg.meshId ?? ""), meshId: String(msg.meshId ?? ""),
@@ -390,6 +626,78 @@ export class BrokerClient {
})(); })();
return; return;
} }
if (msg.type === "state_result") {
const resolver = this.stateResolvers.shift();
if (resolver) {
if (msg.key) {
resolver({
key: String(msg.key),
value: msg.value,
updatedBy: String(msg.updatedBy ?? ""),
updatedAt: String(msg.updatedAt ?? ""),
});
} else {
resolver(null);
}
}
return;
}
if (msg.type === "state_list") {
const entries = (msg.entries as Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) ?? [];
const resolver = this.stateListResolvers.shift();
if (resolver) resolver(entries);
return;
}
if (msg.type === "state_change") {
const change = {
key: String(msg.key ?? ""),
value: msg.value,
updatedBy: String(msg.updatedBy ?? ""),
};
for (const h of this.stateChangeHandlers) {
try { h(change); } catch { /* handler errors are not the transport's problem */ }
}
return;
}
if (msg.type === "memory_stored") {
const resolver = this.memoryStoreResolvers.shift();
if (resolver) resolver(msg.id ? String(msg.id) : null);
return;
}
if (msg.type === "memory_results") {
const memories = (msg.memories as Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) ?? [];
const resolver = this.memoryRecallResolvers.shift();
if (resolver) resolver(memories);
return;
}
if (msg.type === "message_status_result") {
const resolver = this.messageStatusResolvers.shift();
if (resolver) resolver(msg as any);
return;
}
if (msg.type === "file_url") {
const resolver = this.fileUrlResolvers.shift();
if (resolver) {
if (msg.url) {
resolver({ url: String(msg.url), name: String(msg.name ?? "") });
} else {
resolver(null);
}
}
return;
}
if (msg.type === "file_list") {
const files = (msg.files as Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) ?? [];
const resolver = this.fileListResolvers.shift();
if (resolver) resolver(files);
return;
}
if (msg.type === "file_status_result") {
const accesses = (msg.accesses as Array<{ peerName: string; accessedAt: string }>) ?? [];
const resolver = this.fileStatusResolvers.shift();
if (resolver) resolver(accesses);
return;
}
if (msg.type === "error") { if (msg.type === "error") {
this.debug(`broker error: ${msg.code} ${msg.message}`); this.debug(`broker error: ${msg.code} ${msg.message}`);
const id = msg.id ? String(msg.id) : null; const id = msg.id ? String(msg.id) : null;

View File

@@ -11,12 +11,13 @@ import type { Config, JoinedMesh } from "../state/config";
import { env } from "../env"; import { env } from "../env";
const clients = new Map<string, BrokerClient>(); const clients = new Map<string, BrokerClient>();
let configDisplayName: string | undefined;
/** Ensure a BrokerClient exists + is connecting/open for this mesh. */ /** Ensure a BrokerClient exists + is connecting/open for this mesh. */
export async function ensureClient(mesh: JoinedMesh): Promise<BrokerClient> { export async function ensureClient(mesh: JoinedMesh): Promise<BrokerClient> {
const existing = clients.get(mesh.meshId); const existing = clients.get(mesh.meshId);
if (existing) return existing; if (existing) return existing;
const client = new BrokerClient(mesh, { debug: env.CLAUDEMESH_DEBUG }); const client = new BrokerClient(mesh, { debug: env.CLAUDEMESH_DEBUG, displayName: configDisplayName });
clients.set(mesh.meshId, client); clients.set(mesh.meshId, client);
try { try {
await client.connect(); await client.connect();
@@ -29,6 +30,7 @@ export async function ensureClient(mesh: JoinedMesh): Promise<BrokerClient> {
/** Start clients for every joined mesh. Called once on MCP server start. */ /** Start clients for every joined mesh. Called once on MCP server start. */
export async function startClients(config: Config): Promise<void> { export async function startClients(config: Config): Promise<void> {
configDisplayName = config.displayName;
await Promise.allSettled(config.meshes.map(ensureClient)); await Promise.allSettled(config.meshes.map(ensureClient));
} }

View File

@@ -28,6 +28,26 @@ services:
networks: networks:
- claudemesh-internal - claudemesh-internal
minio:
image: minio/minio
command: server /data --console-address ":9001"
restart: always
volumes:
- minio-data:/data
environment:
MINIO_ROOT_USER: claudemesh
MINIO_ROOT_PASSWORD: ${MINIO_SECRET_KEY:-changeme}
expose:
- "9000"
networks:
- claudemesh-internal
healthcheck:
test: ["CMD", "mc", "ready", "local"]
interval: 15s
timeout: 5s
start_period: 10s
retries: 3
broker: broker:
image: ${BROKER_IMAGE:-claudemesh-broker:latest} image: ${BROKER_IMAGE:-claudemesh-broker:latest}
restart: always restart: always
@@ -40,11 +60,18 @@ services:
MAX_CONNECTIONS_PER_MESH: ${MAX_CONNECTIONS_PER_MESH:-100} MAX_CONNECTIONS_PER_MESH: ${MAX_CONNECTIONS_PER_MESH:-100}
MAX_MESSAGE_BYTES: ${MAX_MESSAGE_BYTES:-65536} MAX_MESSAGE_BYTES: ${MAX_MESSAGE_BYTES:-65536}
HOOK_RATE_LIMIT_PER_MIN: ${HOOK_RATE_LIMIT_PER_MIN:-30} HOOK_RATE_LIMIT_PER_MIN: ${HOOK_RATE_LIMIT_PER_MIN:-30}
MINIO_ENDPOINT: minio:9000
MINIO_ACCESS_KEY: claudemesh
MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-changeme}
MINIO_USE_SSL: "false"
expose: expose:
- "7900" - "7900"
networks: networks:
- coolify - coolify
- claudemesh-internal - claudemesh-internal
depends_on:
minio:
condition: service_healthy
healthcheck: healthcheck:
test: ["CMD", "bun", "-e", "fetch('http://localhost:7900/health').then(r=>{process.exit(r.ok?0:1)}).catch(()=>process.exit(1))"] test: ["CMD", "bun", "-e", "fetch('http://localhost:7900/health').then(r=>{process.exit(r.ok?0:1)}).catch(()=>process.exit(1))"]
interval: 15s interval: 15s
@@ -85,6 +112,9 @@ services:
start_period: 20s start_period: 20s
retries: 3 retries: 3
volumes:
minio-data:
networks: networks:
# Coolify's shared Traefik network — must already exist on the host # Coolify's shared Traefik network — must already exist on the host
coolify: coolify:

View File

@@ -0,0 +1 @@
ALTER TABLE "mesh"."presence" ADD COLUMN "session_pubkey" text;

View File

@@ -0,0 +1 @@
ALTER TABLE "mesh"."message_queue" ADD COLUMN "sender_session_pubkey" text;

View File

@@ -0,0 +1 @@
ALTER TABLE "mesh"."presence" ADD COLUMN "groups" jsonb DEFAULT '[]'::jsonb;

View File

@@ -0,0 +1,27 @@
CREATE TABLE "mesh"."memory" (
"id" text PRIMARY KEY NOT NULL,
"mesh_id" text NOT NULL,
"content" text NOT NULL,
"tags" text[] DEFAULT '{}',
"remembered_by" text,
"remembered_by_name" text,
"remembered_at" timestamp DEFAULT now() NOT NULL,
"forgotten_at" timestamp
);
--> statement-breakpoint
CREATE TABLE "mesh"."state" (
"id" text PRIMARY KEY NOT NULL,
"mesh_id" text NOT NULL,
"key" text NOT NULL,
"value" jsonb NOT NULL,
"updated_by_presence" text,
"updated_by_name" text,
"updated_at" timestamp DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "mesh"."memory" ADD CONSTRAINT "memory_mesh_id_mesh_id_fk" FOREIGN KEY ("mesh_id") REFERENCES "mesh"."mesh"("id") ON DELETE cascade ON UPDATE cascade;--> statement-breakpoint
ALTER TABLE "mesh"."memory" ADD CONSTRAINT "memory_remembered_by_member_id_fk" FOREIGN KEY ("remembered_by") REFERENCES "mesh"."member"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "mesh"."state" ADD CONSTRAINT "state_mesh_id_mesh_id_fk" FOREIGN KEY ("mesh_id") REFERENCES "mesh"."mesh"("id") ON DELETE cascade ON UPDATE cascade;--> statement-breakpoint
CREATE UNIQUE INDEX "state_mesh_key_idx" ON "mesh"."state" USING btree ("mesh_id","key");--> statement-breakpoint
ALTER TABLE "mesh"."memory" ADD COLUMN IF NOT EXISTS "search_vector" tsvector GENERATED ALWAYS AS (to_tsvector('english', content)) STORED;--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "memory_search_idx" ON "mesh"."memory" USING gin("search_vector");

View File

@@ -0,0 +1,28 @@
CREATE TABLE "mesh"."file" (
"id" text PRIMARY KEY NOT NULL,
"mesh_id" text NOT NULL,
"name" text NOT NULL,
"size_bytes" integer NOT NULL,
"mime_type" text,
"minio_key" text NOT NULL,
"tags" text[] DEFAULT '{}',
"persistent" boolean DEFAULT true NOT NULL,
"uploaded_by_name" text,
"uploaded_by_member" text,
"target_spec" text,
"uploaded_at" timestamp DEFAULT now() NOT NULL,
"expires_at" timestamp,
"deleted_at" timestamp
);
--> statement-breakpoint
CREATE TABLE "mesh"."file_access" (
"id" text PRIMARY KEY NOT NULL,
"file_id" text NOT NULL,
"peer_session_pubkey" text,
"peer_name" text,
"accessed_at" timestamp DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "mesh"."file" ADD CONSTRAINT "file_mesh_id_mesh_id_fk" FOREIGN KEY ("mesh_id") REFERENCES "mesh"."mesh"("id") ON DELETE cascade ON UPDATE cascade;--> statement-breakpoint
ALTER TABLE "mesh"."file" ADD CONSTRAINT "file_uploaded_by_member_member_id_fk" FOREIGN KEY ("uploaded_by_member") REFERENCES "mesh"."member"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "mesh"."file_access" ADD CONSTRAINT "file_access_file_id_file_id_fk" FOREIGN KEY ("file_id") REFERENCES "mesh"."file"("id") ON DELETE cascade ON UPDATE no action;

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -29,6 +29,48 @@
"when": 1775463897329, "when": 1775463897329,
"tag": "0003_add-presence-summary", "tag": "0003_add-presence-summary",
"breakpoints": true "breakpoints": true
},
{
"idx": 4,
"version": "7",
"when": 1775468683383,
"tag": "0004_add-presence-display-name",
"breakpoints": true
},
{
"idx": 5,
"version": "7",
"when": 1775470435032,
"tag": "0005_add-presence-session-pubkey",
"breakpoints": true
},
{
"idx": 6,
"version": "7",
"when": 1775470979207,
"tag": "0006_add-sender-session-pubkey",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1775476994511,
"tag": "0007_add-presence-groups",
"breakpoints": true
},
{
"idx": 8,
"version": "7",
"when": 1775477883426,
"tag": "0008_add-state-and-memory",
"breakpoints": true
},
{
"idx": 9,
"version": "7",
"when": 1775480008546,
"tag": "0009_add-file-tables",
"breakpoints": true
} }
] ]
} }

View File

@@ -1,10 +1,12 @@
import { relations } from "drizzle-orm"; import { relations } from "drizzle-orm";
import { import {
boolean,
integer, integer,
jsonb, jsonb,
pgSchema, pgSchema,
timestamp, timestamp,
text, text,
uniqueIndex,
} from "drizzle-orm/pg-core"; } from "drizzle-orm/pg-core";
import { generateId } from "@turbostarter/shared/utils"; import { generateId } from "@turbostarter/shared/utils";
@@ -192,6 +194,7 @@ export const presence = meshSchema.table("presence", {
.references(() => meshMember.id, { onDelete: "cascade", onUpdate: "cascade" }) .references(() => meshMember.id, { onDelete: "cascade", onUpdate: "cascade" })
.notNull(), .notNull(),
sessionId: text().notNull(), sessionId: text().notNull(),
sessionPubkey: text(),
displayName: text(), displayName: text(),
pid: integer().notNull(), pid: integer().notNull(),
cwd: text().notNull(), cwd: text().notNull(),
@@ -199,6 +202,7 @@ export const presence = meshSchema.table("presence", {
statusSource: presenceStatusSourceEnum().notNull().default("jsonl"), statusSource: presenceStatusSourceEnum().notNull().default("jsonl"),
statusUpdatedAt: timestamp().defaultNow().notNull(), statusUpdatedAt: timestamp().defaultNow().notNull(),
summary: text(), summary: text(),
groups: jsonb().$type<Array<{ name: string; role?: string }>>().default([]),
connectedAt: timestamp().defaultNow().notNull(), connectedAt: timestamp().defaultNow().notNull(),
lastPingAt: timestamp().defaultNow().notNull(), lastPingAt: timestamp().defaultNow().notNull(),
disconnectedAt: timestamp(), disconnectedAt: timestamp(),
@@ -221,6 +225,7 @@ export const messageQueue = meshSchema.table("message_queue", {
senderMemberId: text() senderMemberId: text()
.references(() => meshMember.id, { onDelete: "cascade", onUpdate: "cascade" }) .references(() => meshMember.id, { onDelete: "cascade", onUpdate: "cascade" })
.notNull(), .notNull(),
senderSessionPubkey: text(),
targetSpec: text().notNull(), targetSpec: text().notNull(),
priority: messagePriorityEnum().notNull().default("next"), priority: messagePriorityEnum().notNull().default("next"),
nonce: text().notNull(), nonce: text().notNull(),
@@ -248,6 +253,80 @@ export const pendingStatus = meshSchema.table("pending_status", {
appliedAt: timestamp(), appliedAt: timestamp(),
}); });
/**
* Shared key-value state scoped to a mesh. Any peer can read/write.
* Changes push to all connected peers in real time.
*/
export const meshState = meshSchema.table(
"state",
{
id: text().primaryKey().notNull().$defaultFn(generateId),
meshId: text()
.references(() => mesh.id, { onDelete: "cascade", onUpdate: "cascade" })
.notNull(),
key: text().notNull(),
value: jsonb().notNull(),
updatedByPresence: text(),
updatedByName: text(),
updatedAt: timestamp().defaultNow().notNull(),
},
(table) => [uniqueIndex("state_mesh_key_idx").on(table.meshId, table.key)],
);
/**
* Persistent shared memory for a mesh. Full-text searchable via a
* tsvector generated column + GIN index added in raw SQL migration.
*/
export const meshMemory = meshSchema.table("memory", {
id: text().primaryKey().notNull().$defaultFn(generateId),
meshId: text()
.references(() => mesh.id, { onDelete: "cascade", onUpdate: "cascade" })
.notNull(),
content: text().notNull(),
tags: text().array().default([]),
rememberedBy: text().references(() => meshMember.id),
rememberedByName: text(),
rememberedAt: timestamp().defaultNow().notNull(),
forgottenAt: timestamp(),
});
/**
* File metadata for shared files in a mesh. Actual bytes live in MinIO;
* this table tracks ownership, access control, and soft-deletion.
*/
export const meshFile = meshSchema.table("file", {
id: text().primaryKey().notNull().$defaultFn(generateId),
meshId: text()
.references(() => mesh.id, { onDelete: "cascade", onUpdate: "cascade" })
.notNull(),
name: text().notNull(),
sizeBytes: integer().notNull(),
mimeType: text(),
minioKey: text().notNull(),
tags: text().array().default([]),
persistent: boolean().notNull().default(true),
uploadedByName: text(),
uploadedByMember: text().references(() => meshMember.id),
targetSpec: text(), // null = entire mesh
uploadedAt: timestamp().defaultNow().notNull(),
expiresAt: timestamp(),
deletedAt: timestamp(),
});
/**
* Access log for file downloads. Tracks which peer accessed which file
* and when, for auditability and read-receipt semantics.
*/
export const meshFileAccess = meshSchema.table("file_access", {
id: text().primaryKey().notNull().$defaultFn(generateId),
fileId: text()
.references(() => meshFile.id, { onDelete: "cascade" })
.notNull(),
peerSessionPubkey: text(),
peerName: text(),
accessedAt: timestamp().defaultNow().notNull(),
});
export const meshRelations = relations(mesh, ({ one, many }) => ({ export const meshRelations = relations(mesh, ({ one, many }) => ({
owner: one(user, { owner: one(user, {
fields: [mesh.ownerUserId], fields: [mesh.ownerUserId],
@@ -308,6 +387,43 @@ export const auditLogRelations = relations(auditLog, ({ one }) => ({
}), }),
})); }));
export const meshStateRelations = relations(meshState, ({ one }) => ({
mesh: one(mesh, {
fields: [meshState.meshId],
references: [mesh.id],
}),
}));
export const meshMemoryRelations = relations(meshMemory, ({ one }) => ({
mesh: one(mesh, {
fields: [meshMemory.meshId],
references: [mesh.id],
}),
member: one(meshMember, {
fields: [meshMemory.rememberedBy],
references: [meshMember.id],
}),
}));
export const meshFileRelations = relations(meshFile, ({ one, many }) => ({
mesh: one(mesh, {
fields: [meshFile.meshId],
references: [mesh.id],
}),
uploader: one(meshMember, {
fields: [meshFile.uploadedByMember],
references: [meshMember.id],
}),
accesses: many(meshFileAccess),
}));
export const meshFileAccessRelations = relations(meshFileAccess, ({ one }) => ({
file: one(meshFile, {
fields: [meshFileAccess.fileId],
references: [meshFile.id],
}),
}));
export const selectMeshSchema = createSelectSchema(mesh); export const selectMeshSchema = createSelectSchema(mesh);
export const insertMeshSchema = createInsertSchema(mesh); export const insertMeshSchema = createInsertSchema(mesh);
export const selectMemberSchema = createSelectSchema(meshMember); export const selectMemberSchema = createSelectSchema(meshMember);
@@ -337,3 +453,19 @@ export type SelectMessageQueue = typeof messageQueue.$inferSelect;
export type InsertMessageQueue = typeof messageQueue.$inferInsert; export type InsertMessageQueue = typeof messageQueue.$inferInsert;
export type SelectPendingStatus = typeof pendingStatus.$inferSelect; export type SelectPendingStatus = typeof pendingStatus.$inferSelect;
export type InsertPendingStatus = typeof pendingStatus.$inferInsert; export type InsertPendingStatus = typeof pendingStatus.$inferInsert;
export const selectMeshStateSchema = createSelectSchema(meshState);
export const insertMeshStateSchema = createInsertSchema(meshState);
export const selectMeshMemorySchema = createSelectSchema(meshMemory);
export const insertMeshMemorySchema = createInsertSchema(meshMemory);
export type SelectMeshState = typeof meshState.$inferSelect;
export type InsertMeshState = typeof meshState.$inferInsert;
export type SelectMeshMemory = typeof meshMemory.$inferSelect;
export type InsertMeshMemory = typeof meshMemory.$inferInsert;
export const selectMeshFileSchema = createSelectSchema(meshFile);
export const insertMeshFileSchema = createInsertSchema(meshFile);
export const selectMeshFileAccessSchema = createSelectSchema(meshFileAccess);
export const insertMeshFileAccessSchema = createInsertSchema(meshFileAccess);
export type SelectMeshFile = typeof meshFile.$inferSelect;
export type InsertMeshFile = typeof meshFile.$inferInsert;
export type SelectMeshFileAccess = typeof meshFileAccess.$inferSelect;
export type InsertMeshFileAccess = typeof meshFileAccess.$inferInsert;