alezmad/claudemesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: v0.3.0 — State, Memory, message_status, MCP instructions

Browse Source 
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>
This commit is contained in:
Alejandro Gutiérrez
2026-04-06 13:29:45 +01:00
parent 02b1e5695f
commit 37e238e758
12 changed files with 4419 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

552
SPEC.md
View File

@@ -1,10 +1,10 @@
# Claudemesh v0.2 — Specification
# 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.
## Five concepts
## Concepts
```
Organization (billing, auth)
@@ -25,20 +25,22 @@ A peer is a Claude Code session connected to a mesh. Ephemeral — comes and goe
### Identity
Each `claudemesh launch` generates an ephemeral ed25519 keypair (session identity). The member identity (from `claudemesh join`) provides authentication. Session identity provides routing and encryption.
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 across sessions |
|-----------|--------|--------------------------|
| name | `--name` flag or wizard | No |
| role | `--role` flag or wizard | No |
| groups | `--groups` flag or wizard | No |
| status | Hook-driven (idle/working/dnd) | No |
| summary | `set_summary` tool call | No |
| capabilities | Auto-detected from session | No |
| sessionPubkey | Generated on connect | No |
| memberId | From `claudemesh join` | Yes (in config) |
| 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
@@ -46,6 +48,9 @@ Each `claudemesh launch` generates an ephemeral ed25519 keypair (session identit
# 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
@@ -55,7 +60,7 @@ claudemesh launch
### Wizard
Interactive mode when args are missing. Each question is one line. Optional fields accept empty Enter. Only one mesh joined? Skip the mesh picker. Only relevant questions shown.
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
@@ -65,40 +70,66 @@ Interactive mode when args are missing. Each question is one line. Optional fiel
Autonomous mode
Claude will send and receive peer messages without
asking you first. Peers exchange text only.
asking you first. Peers exchange text only — no file
access, no tool calls, no code execution.
Continue? [Y/n]
```
`-y` skips the confirmation. `--quiet` skips the banner. Any arg provided skips its question.
### 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
A group is a named subset of peers. Not a channel — no message history, no persistence. Just a routing label stored on the presence row.
Named subset of peers. No message history, no persistence beyond the session. A routing label stored on the presence row.
### Syntax
`@groupname` in message routing. Declared at launch via `--groups`.
`@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 a free-form string stored as metadata. The broker does not interpret roles — Claude does.
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")
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 whose groups include `frontend`. Sender excluded.
### Built-in groups
- `@all` — every peer in the mesh. Alias for `*` broadcast.
Broker delivers to all peers in the group. Sender excluded.
### Group metadata in list_peers
@@ -106,6 +137,7 @@ Broker delivers to all peers whose groups include `frontend`. Sender excluded.
{
"name": "Alice",
"status": "working",
"role": "dev",
"groups": [
{ "name": "frontend", "role": "lead" },
{ "name": "reviewers", "role": "member" }
@@ -114,209 +146,105 @@ Broker delivers to all peers whose groups include `frontend`. Sender excluded.
}
```
Peers read this metadata and coordinate based on their system prompts. A "lead" gathers input before responding. A "member" sends their take to the lead. An "observer" stays silent unless asked. The broker doesn't enforce these — Claude does.
### Dynamic roles
### Dynamic group management
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: "frontend", role: "member")
leave_group(name: "frontend")
join_group(name: "reviewers", role: "lead") # take over leadership
join_group(name: "reviewers", role: "observer") # step back
```
MCP tools. Update the presence row. Other peers see the change on next `list_peers`.
### 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
A shared key-value store scoped to a mesh. Any peer can read or write. Changes push to subscribed peers.
Shared key-value store scoped to a mesh. Any peer reads or writes. Changes push to all connected peers.
### Why
Peers shouldn't need to message each other to agree on facts. "Is the deploy frozen?" should be a state read, not a conversation.
Replace coordination messages with shared facts. "Is the deploy frozen?" becomes a state read, not a conversation.
### Tools
```
set_state(key: "deploy_frozen", value: true)
get_state(key: "deploy_frozen") → true
list_state() → [{ key, value, updatedBy, updatedAt }]
watch_state(key: "deploy_frozen") → push notification on change
```
### Storage
Broker-side. PostgreSQL table in the mesh schema:
```sql
mesh.state (
id text PK,
mesh_id text FK,
key text NOT NULL,
value jsonb NOT NULL,
updated_by text FK (presence.id),
updated_at timestamp,
UNIQUE(mesh_id, key)
)
```
| 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 a peer calls `set_state`, the broker pushes a notification to all connected peers in the mesh:
When any peer calls `set_state`, the broker pushes to all connected peers:
```json
{ "type": "state_change", "key": "deploy_frozen", "value": true, "updatedBy": "Alice" }
```
The CLI MCP server translates this to a `notifications/claude/channel` push, same as messages.
### Scope
State is mesh-scoped and ephemeral (lives as long as the mesh). Not designed for persistence across mesh restarts — use Memory for that.
---
## 4. Memory
Persistent shared knowledge that survives across sessions. The mesh's institutional memory.
### Why
When a new peer joins the mesh, it has zero context. Memory provides the team's accumulated knowledge: decisions made, bugs found, preferences learned.
### Tools
```
remember(content: "Payments API rate-limits at 100 req/s after the March incident")
recall(query: "payments API") → [{ content, rememberedBy, rememberedAt }]
forget(id: "mem_abc123")
```
Translated to a `notifications/claude/channel` push in the CLI.
### Storage
Broker-side. PostgreSQL table:
```sql
mesh.memory (
id text PK,
mesh_id text FK,
content text NOT NULL,
tags text[],
remembered_by text FK (member.id),
remembered_at timestamp,
forgotten_at timestamp
)
```
### Recall
Full-text search (PostgreSQL `tsvector`). Returns relevant memories ranked by relevance. Peers can call `recall` at session start to load context.
### Memory vs State
| | State | Memory |
|---|---|---|
| Lifetime | Session (ephemeral) | Permanent (until forgotten) |
| Purpose | Operational coordination | Institutional knowledge |
| Example | `deploy_frozen: true` | "Never deploy on Fridays — oncall learned this the hard way" |
| Access | get/set/watch | remember/recall/forget |
---
## 5. MCP Tools (complete surface)
### Messaging
| Tool | Description |
|------|-------------|
| `send_message(to, message, priority?)` | Send to peer name, pubkey, @group, or * |
| `check_messages()` | Drain buffered messages (fallback for non-push) |
### Presence
| Tool | Description |
|------|-------------|
| `list_peers(group?)` | List connected peers, optionally filtered by group |
| `set_summary(summary)` | Set session summary visible to peers |
| `set_status(status)` | Override status: idle, working, dnd |
### Groups
| Tool | Description |
|------|-------------|
| `join_group(name, role?)` | Join a group with optional role |
| `leave_group(name)` | Leave a group |
### State
| Tool | Description |
|------|-------------|
| `get_state(key)` | Read a value |
| `set_state(key, value)` | Write a value (pushes to all peers) |
| `list_state()` | List all state keys and values |
### Memory
| Tool | Description |
|------|-------------|
| `remember(content, tags?)` | Store persistent knowledge |
| `recall(query)` | Search memories by relevance |
| `forget(id)` | Soft-delete a memory |
---
## 6. WS Protocol additions
### Client → Broker
| Type | Fields | Description |
|------|--------|-------------|
| `join_group` | name, role? | Add group to this presence |
| `leave_group` | name | Remove group from this presence |
| `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 |
|------|--------|-------------|
| `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 |
---
## 7. DB schema additions
### mesh.presence (modify existing)
```sql
ADD COLUMN groups jsonb DEFAULT '[]';
-- Format: [{"name": "frontend", "role": "lead"}, ...]
```
### mesh.state (new table)
```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 REFERENCES mesh.presence(id),
updated_by_presence text,
updated_by_name text,
updated_at timestamp DEFAULT NOW(),
UNIQUE(mesh_id, key)
);
```
### mesh.memory (new table)
### 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 (
@@ -333,47 +261,217 @@ CREATE TABLE mesh.memory (
CREATE INDEX memory_search_idx ON mesh.memory USING gin(search_vector);
```
---
### Memory vs State
## 8. Implementation phases
### Phase A: Groups (v0.2.0)
- `--groups` flag in launch + wizard question
- `groups` jsonb column on presence
- `join_group` / `leave_group` WS messages + MCP tools
- `@group` routing in broker's handleSend
- `list_peers` returns group metadata
- Group sender exclusion (don't echo back to sender)
### Phase B: State (v0.3.0)
- `mesh.state` table + migrations
- `set_state` / `get_state` / `list_state` WS messages + MCP tools
- State change push notifications to all mesh peers
- State displayed in dashboard
### Phase C: Memory (v0.4.0)
- `mesh.memory` table with tsvector + gin index
- `remember` / `recall` / `forget` WS messages + MCP tools
- Full-text search via PostgreSQL
- Memory accessible from dashboard
### Phase D: Dashboard (v0.5.0)
- Live peer list with groups, roles, status
- State viewer/editor
- Memory browser
- Message log (opt-in, plaintext only)
| | 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 |
---
## 9. What the broker does NOT do
## 5. AI Context (CLAUDE.md)
- **Interpret roles.** "lead", "member", "observer" are strings. Claude reads them and decides how to behave.
- **Enforce coordination protocols.** Voting, consensus, delegation — all emergent from system prompts + group metadata.
- **Store message history.** Messages are delivered and discarded. The queue holds undelivered messages only.
- **Run agents.** The broker routes messages and stores state. Claude does everything else.
Each `claudemesh install` copies a `CLAUDEMESH.md` file to `~/.claudemesh/CLAUDEMESH.md`. Claude Code discovers it and injects it as context.
The broker is a dumb pipe with a bulletin board. The intelligence lives at the edges.
### 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?)` | Send to peer name, @group, or * |
| `check_messages()` | Drain buffered messages |
| `message_status(id)` | Check if a sent message was delivered |
### 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 |
---
## 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 | Planned | Shared key-value store with push |
| Memory | v0.4.0 | Planned | Persistent knowledge with full-text search |
| AI Context | v0.2.1 | Planned | CLAUDEMESH.md shipped with CLI |
| Dashboard | v0.5.0 | Planned | Live peers, state, memory 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.

206
apps/broker/src/broker.ts
View File

@@ -33,6 +33,8 @@ import {
invite as inviteTable,
mesh,
meshMember as memberTable,
meshMemory,
meshState,
messageQueue,
pendingStatus,
presence,
@@ -489,6 +491,210 @@ export async function leaveGroup(
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),
),
);
}
// --- Message queueing + delivery ---
export interface QueueParams {

254
apps/broker/src/index.ts
View File

@@ -15,22 +15,31 @@
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import type { Duplex } from "node:stream";
import { WebSocketServer, type WebSocket } from "ws";
import { eq } from "drizzle-orm";
import { env } from "./env";
import { db } from "./db";
import { messageQueue } from "@turbostarter/db/schema/mesh";
import {
connectPresence,
disconnectPresence,
drainForMember,
findMemberByPubkey,
forgetMemory,
getState,
handleHookSetStatus,
heartbeat,
joinGroup,
joinMesh,
leaveGroup,
listPeersInMesh,
listState,
queueMessage,
recallMemory,
refreshQueueDepth,
refreshStatusFromJsonl,
rememberMemory,
setSummary,
setState,
startSweepers,
stopSweepers,
writeStatus,
@@ -470,8 +479,6 @@ async function handleSend(
}
// Fan-out over connected peers in the same mesh — skip sender.
// Resolve @group routing: "@all" is alias for "*", "@<name>" matches
// peers whose in-memory groups array contains that group name.
const isGroupTarget = msg.targetSpec.startsWith("@");
const isBroadcast =
msg.targetSpec === "*" ||
@@ -479,6 +486,19 @@ async function handleSend(
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;
@@ -495,7 +515,25 @@ async function handleSend(
&& peer.sessionPubkey !== msg.targetSpec)
continue;
}
void maybePushQueuedMessages(pid, conn.sessionPubkey ?? undefined);
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));
}
}
@@ -593,6 +631,216 @@ function handleConnection(ws: WebSocket): void {
});
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 "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) {
metrics.messagesRejectedTotal.inc({ reason: "parse_or_handler" });

118
apps/broker/src/types.ts
View File

@@ -118,6 +118,43 @@ export interface WSLeaveGroupMessage {
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. */
export interface WSAckMessage {
type: "ack";
@@ -147,6 +184,72 @@ export interface WSPeersListMessage {
}>;
}
/** 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";
}>;
}
/** Broker → client: structured error. */
export interface WSErrorMessage {
type: "error";
@@ -162,11 +265,24 @@ export type WSClientMessage =
| WSListPeersMessage
| WSSetSummaryMessage
| WSJoinGroupMessage
| WSLeaveGroupMessage;
| WSLeaveGroupMessage
| WSSetStateMessage
| WSGetStateMessage
| WSListStateMessage
| WSRememberMessage
| WSRecallMessage
| WSForgetMessage
| WSMessageStatusMessage;
export type WSServerMessage =
| WSHelloAckMessage
| WSPushMessage
| WSAckMessage
| WSPeersListMessage
| WSStateChangeMessage
| WSStateResultMessage
| WSStateListMessage
| WSMemoryStoredMessage
| WSMemoryResultsMessage
| WSMessageStatusResultMessage
| WSErrorMessage;

2
apps/cli/package.json
View File

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

144
apps/cli/src/mcp/server.ts
View File

@@ -129,34 +129,57 @@ function formatPush(p: InboundPush, meshSlug: string): string {
export async function startMcpServer(): Promise<void> {
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(
{ name: "claudemesh", version: "0.1.4" },
{ name: "claudemesh", version: "0.3.0" },
{
capabilities: {
experimental: { "claude/channel": {} },
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 to set to the from_name (display name) of the sender. The \`to\` field can be a peer name, pubkey, @group, or * for broadcast.
## 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. |
Available tools:
- list_peers: see joined meshes + their connection status
- send_message: send to a peer by display name, pubkey, @group, #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)
- join_group: join a @group with optional role
- leave_group: leave a @group
If multiple meshes are joined, prefix \`to\` with \`<mesh-slug>:\` to disambiguate (e.g. \`dev-team:Alice\`).
Message priority:
- "now": delivered immediately regardless of recipient status (use sparingly)
- "next" (default): delivered when recipient is idle
- "low": pull-only (check_messages)
## 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.
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.`,
## 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.
## 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.`,
},
);
@@ -226,6 +249,24 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
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": {
const drained: string[] = [];
for (const c of allClients()) {
@@ -269,6 +310,59 @@ If you have multiple joined meshes, prefix the \`to\` argument of send_message w
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}`);
}
default:
return text(`Unknown tool: ${name}`, true);
}
@@ -312,6 +406,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 */
}
});
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 => {

90
apps/cli/src/mcp/tools.ts
View File

@@ -44,6 +44,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",
description:
@@ -105,4 +120,79 @@ export const TOOLS: Tool[] = [
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"],
},
},
];

155
apps/cli/src/ws/client.ts
View File

@@ -76,6 +76,11 @@ export class BrokerClient {
private pushHandlers = new Set<PushHandler>();
private pushBuffer: InboundPush[] = [];
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;
@@ -325,6 +330,107 @@ export class BrokerClient {
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> = [];
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);
});
}
/** 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 {
this.closed = true;
if (this.helloTimer) clearTimeout(this.helloTimer);
@@ -428,6 +534,55 @@ export class BrokerClient {
})();
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 === "error") {
this.debug(`broker error: ${msg.code} ${msg.message}`);
const id = msg.id ? String(msg.id) : null;

27
packages/db/migrations/0008_add-state-and-memory.sql Normal file
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");

3049
packages/db/migrations/meta/0008_snapshot.json Normal file
View File

File diff suppressed because it is too large Load Diff

7
packages/db/migrations/meta/_journal.json
View File

@@ -57,6 +57,13 @@
"when": 1775476994511,
"tag": "0007_add-presence-groups",
"breakpoints": true
},
{
"idx": 8,
"version": "7",
"when": 1775477883426,
"tag": "0008_add-state-and-memory",
"breakpoints": true
}
]
}

64
packages/db/src/schema/mesh.ts
View File

@@ -5,6 +5,7 @@ import {
pgSchema,
timestamp,
text,
uniqueIndex,
} from "drizzle-orm/pg-core";
import { generateId } from "@turbostarter/shared/utils";
@@ -251,6 +252,43 @@ export const pendingStatus = meshSchema.table("pending_status", {
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(),
});
export const meshRelations = relations(mesh, ({ one, many }) => ({
owner: one(user, {
fields: [mesh.ownerUserId],
@@ -311,6 +349,24 @@ 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 selectMeshSchema = createSelectSchema(mesh);
export const insertMeshSchema = createInsertSchema(mesh);
export const selectMemberSchema = createSelectSchema(meshMember);
@@ -340,3 +396,11 @@ export type SelectMessageQueue = typeof messageQueue.$inferSelect;
export type InsertMessageQueue = typeof messageQueue.$inferInsert;
export type SelectPendingStatus = typeof pendingStatus.$inferSelect;
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;