Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ee585a8370 | ||
|
|
1f078bf0c8 | ||
|
|
2372032a68 | ||
|
|
a70c5fd124 | ||
|
|
5c62d287cf |
58
SPEC.md
58
SPEC.md
@@ -855,7 +855,63 @@ The broker:
|
||||
|
||||
---
|
||||
|
||||
## 13. Encryption
|
||||
## 13. Claude Code Integration — How Push Delivery Works
|
||||
|
||||
Understanding how Claude Code processes channel notifications is critical for claudemesh reliability.
|
||||
|
||||
### The notification pipeline
|
||||
|
||||
```
|
||||
MCP server (claudemesh-cli)
|
||||
└─ server.notification("notifications/claude/channel", { content, meta })
|
||||
└─ writes JSON-RPC to stdout
|
||||
└─ Claude Code reads from MCP process stdout
|
||||
└─ setNotificationHandler fires
|
||||
└─ enqueue({ mode: "prompt", value: wrappedContent, origin: { kind: "channel" } })
|
||||
└─ React useSyncExternalStore triggers re-render
|
||||
└─ useQueueProcessor effect fires
|
||||
└─ processQueueIfReady() → executeInput()
|
||||
└─ Claude sees ← claudemesh: ...
|
||||
```
|
||||
|
||||
### Key requirements (from Claude Code source)
|
||||
|
||||
1. **Feature gate**: `feature('KAIROS') || feature('KAIROS_CHANNELS')` must be true. `KAIROS_CHANNELS` is external (GrowthBook). `--dangerously-load-development-channels` sets `entry.dev = true` which bypasses the allowlist check but still requires the feature gate.
|
||||
|
||||
2. **OAuth auth required**: Channel notifications require `claude.ai` authentication (OAuth tokens). API key users are blocked. This means `claude login --for-claude-ai` must have been run.
|
||||
|
||||
3. **Server name must match**: The MCP server's declared name (`new Server({ name: "claudemesh" })`) must match the channel entry from `--dangerously-load-development-channels server:claudemesh`.
|
||||
|
||||
4. **Meta keys**: Must match `/^[a-zA-Z_][a-zA-Z0-9_]*$/`. No hyphens. All values must be strings.
|
||||
|
||||
5. **Capability declaration**: Server must declare `experimental: { "claude/channel": {} }` in capabilities.
|
||||
|
||||
6. **Queue processing is event-driven**: `enqueue()` triggers a React store update → `useEffect` fires → processes immediately. No polling needed on the Claude Code side. The 1s poll timer in claudemesh is for draining the WS push buffer into notifications — Claude Code handles the rest instantly.
|
||||
|
||||
### Priority gating on the broker
|
||||
|
||||
The broker holds `"next"` and `"low"` priority messages when the peer's status is `"working"`. Only `"now"` messages deliver immediately regardless of status. This is by design — but can cause perceived "push not working" when the hook reports `working` status.
|
||||
|
||||
```
|
||||
Status: idle → delivers: now, next, low
|
||||
Status: working → delivers: now only
|
||||
Status: dnd → delivers: now only
|
||||
```
|
||||
|
||||
If a peer appears to not receive messages, check their status in `list_peers`. A peer stuck in `"working"` (e.g., stale hook) will only receive `"now"` priority messages.
|
||||
|
||||
### Common issues
|
||||
|
||||
| Symptom | Likely cause |
|
||||
|---------|-------------|
|
||||
| Messages never arrive | Session started before CLI update — restart with `claudemesh launch` |
|
||||
| Messages arrive with 5+ minute delay | Peer status stuck on `"working"` — `next` messages held until idle |
|
||||
| `← claudemesh:` never appears in idle session | Feature gate `KAIROS_CHANNELS` not enabled, or not OAuth-authenticated |
|
||||
| Messages arrive only on `check_messages` | Channel handler not registered — check `--dangerously-load-development-channels` flag |
|
||||
|
||||
---
|
||||
|
||||
## 14. Encryption
|
||||
|
||||
### Direct messages
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "claudemesh-cli",
|
||||
"version": "0.5.3",
|
||||
"version": "0.5.7",
|
||||
"description": "Claude Code MCP client for claudemesh — peer mesh messaging between Claude sessions.",
|
||||
"keywords": [
|
||||
"claude-code",
|
||||
|
||||
@@ -445,11 +445,14 @@ Your message mode is "${messageMode}".
|
||||
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})`);
|
||||
try {
|
||||
const fileId = await client.uploadFile(filePath, client.meshId, client.meshSlug, {
|
||||
name: fileName, tags, persistent: true,
|
||||
});
|
||||
return text(`Shared: ${fileName ?? filePath} (${fileId})`);
|
||||
} catch (e) {
|
||||
return text(`share_file: upload failed — ${e instanceof Error ? e.message : String(e)}`, true);
|
||||
}
|
||||
}
|
||||
|
||||
case "get_file": {
|
||||
@@ -707,6 +710,56 @@ Your message mode is "${messageMode}".
|
||||
return text(lines.join("\n"));
|
||||
}
|
||||
|
||||
case "ping_mesh": {
|
||||
const { priorities: pingPriorities } = (args ?? {}) as { priorities?: string[] };
|
||||
const toTest = (pingPriorities ?? ["now", "next"]) as Priority[];
|
||||
const client = allClients()[0];
|
||||
if (!client) return text("ping_mesh: not connected", true);
|
||||
const results: string[] = [];
|
||||
|
||||
// Diagnostics: connection state
|
||||
results.push(`WS status: ${client.status}`);
|
||||
results.push(`Mesh: ${client.meshSlug}`);
|
||||
|
||||
// Check own peer status (explains priority gating)
|
||||
const peers = await client.listPeers();
|
||||
const selfPeer = peers.find(p => p.displayName === myName);
|
||||
results.push(`Your status: ${selfPeer?.status ?? "not found in peer list"}`);
|
||||
results.push(`Peers online: ${peers.length}`);
|
||||
results.push(`Push buffer: ${client.pushHistory.length} buffered`);
|
||||
|
||||
// Test send→ack latency per priority (doesn't need round-trip)
|
||||
for (const prio of toTest) {
|
||||
const sendTime = Date.now();
|
||||
// Send to a peer if one exists, otherwise broadcast
|
||||
const target = peers.find(p => p.displayName !== myName);
|
||||
const sendResult = await client.send(
|
||||
target?.pubkey ?? "*",
|
||||
`__ping__ ${prio} from ${myName} at ${new Date().toISOString()}`,
|
||||
prio,
|
||||
);
|
||||
const ackTime = Date.now();
|
||||
|
||||
if (!sendResult.ok) {
|
||||
results.push(`[${prio}] SEND FAILED: ${sendResult.error}`);
|
||||
} else {
|
||||
results.push(`[${prio}] send→ack: ${ackTime - sendTime}ms (msgId: ${sendResult.messageId?.slice(0, 12)})`);
|
||||
if (prio !== "now" && selfPeer?.status === "working") {
|
||||
results.push(` ⚠ peer status is "working" — broker holds "${prio}" until idle`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check if notification pipeline works
|
||||
results.push("");
|
||||
results.push("Pipeline check:");
|
||||
results.push(` onPush handlers: active`);
|
||||
results.push(` messageMode: ${messageMode}`);
|
||||
results.push(` server.notification: ${messageMode === "off" ? "disabled (mode=off)" : "enabled"}`);
|
||||
|
||||
return text(results.join("\n"));
|
||||
}
|
||||
|
||||
default:
|
||||
return text(`Unknown tool: ${name}`, true);
|
||||
}
|
||||
@@ -722,62 +775,56 @@ Your message mode is "${messageMode}".
|
||||
// any mesh's broker connection becomes a <channel source="claudemesh">
|
||||
// system reminder injected into Claude Code's context.
|
||||
for (const client of allClients()) {
|
||||
// Poll-based push: drain pushBuffer every 1s and emit channel notifications.
|
||||
// This is the proven approach from claude-intercom. The WS onPush handler
|
||||
// fires instantly but server.notification() may not flush stdio reliably
|
||||
// from an async WS callback. Polling on a timer ensures consistent delivery.
|
||||
if (messageMode !== "off") {
|
||||
const pushPollTimer = setInterval(async () => {
|
||||
const buffered = client.drainPushBuffer();
|
||||
if (buffered.length > 0) {
|
||||
process.stderr.write(`[claudemesh] poll: ${buffered.length} message(s) to push\n`);
|
||||
}
|
||||
for (const msg of buffered) {
|
||||
const fromPubkey = msg.senderPubkey || "";
|
||||
const fromName = fromPubkey
|
||||
? await resolvePeerName(client, fromPubkey)
|
||||
: "unknown";
|
||||
// Event-driven push: WS onPush fires immediately when a message arrives.
|
||||
// Claude Code's setNotificationHandler → enqueue → React useEffect pipeline
|
||||
// processes notifications instantly (no polling needed on Claude's side).
|
||||
// The old poll-based approach was an overcorrection — Claude Code source
|
||||
// confirms event-driven notification processing.
|
||||
client.onPush(async (msg) => {
|
||||
if (messageMode === "off") return;
|
||||
|
||||
if (messageMode === "inbox") {
|
||||
try {
|
||||
await server.notification({
|
||||
method: "notifications/claude/channel",
|
||||
params: {
|
||||
content: `[inbox] New message from ${fromName}. Use check_messages to read.`,
|
||||
meta: { kind: "inbox_notification", from_name: fromName },
|
||||
},
|
||||
});
|
||||
} catch { /* best effort */ }
|
||||
continue;
|
||||
}
|
||||
const fromPubkey = msg.senderPubkey || "";
|
||||
const fromName = fromPubkey
|
||||
? await resolvePeerName(client, fromPubkey)
|
||||
: "unknown";
|
||||
|
||||
// push mode — full content
|
||||
const content = msg.plaintext ?? decryptFailedWarning(fromPubkey);
|
||||
try {
|
||||
await server.notification({
|
||||
method: "notifications/claude/channel",
|
||||
params: {
|
||||
content,
|
||||
meta: {
|
||||
from_id: fromPubkey,
|
||||
from_name: fromName,
|
||||
mesh_slug: client.meshSlug,
|
||||
mesh_id: client.meshId,
|
||||
priority: msg.priority,
|
||||
sent_at: msg.createdAt,
|
||||
delivered_at: msg.receivedAt,
|
||||
kind: msg.kind,
|
||||
},
|
||||
},
|
||||
});
|
||||
process.stderr.write(`[claudemesh] pushed: from=${fromName} content=${content.slice(0, 60)}\n`);
|
||||
} catch (pushErr) {
|
||||
process.stderr.write(`[claudemesh] push FAILED: ${pushErr}\n`);
|
||||
}
|
||||
}
|
||||
}, 1_000);
|
||||
pushPollTimer.unref();
|
||||
}
|
||||
if (messageMode === "inbox") {
|
||||
try {
|
||||
await server.notification({
|
||||
method: "notifications/claude/channel",
|
||||
params: {
|
||||
content: `[inbox] New message from ${fromName}. Use check_messages to read.`,
|
||||
meta: { kind: "inbox_notification", from_name: fromName },
|
||||
},
|
||||
});
|
||||
} catch { /* best effort */ }
|
||||
return;
|
||||
}
|
||||
|
||||
// push mode — full content
|
||||
const content = msg.plaintext ?? decryptFailedWarning(fromPubkey);
|
||||
try {
|
||||
await server.notification({
|
||||
method: "notifications/claude/channel",
|
||||
params: {
|
||||
content,
|
||||
meta: {
|
||||
from_id: fromPubkey,
|
||||
from_name: fromName,
|
||||
mesh_slug: client.meshSlug,
|
||||
mesh_id: client.meshId,
|
||||
priority: msg.priority,
|
||||
sent_at: msg.createdAt,
|
||||
delivered_at: msg.receivedAt,
|
||||
kind: msg.kind,
|
||||
},
|
||||
},
|
||||
});
|
||||
process.stderr.write(`[claudemesh] pushed: from=${fromName} content=${content.slice(0, 60)}\n`);
|
||||
} catch (pushErr) {
|
||||
process.stderr.write(`[claudemesh] push FAILED: ${pushErr}\n`);
|
||||
}
|
||||
});
|
||||
|
||||
client.onStreamData(async (evt) => {
|
||||
try {
|
||||
@@ -812,7 +859,21 @@ Your message mode is "${messageMode}".
|
||||
});
|
||||
}
|
||||
|
||||
// Event loop keepalive: Node.js stdout to a pipe is buffered. Without
|
||||
// periodic event loop activity, stdout.write() from WS callbacks may not
|
||||
// flush until the next I/O event. This 1s interval keeps the event loop
|
||||
// ticking so channel notifications flush promptly — same pattern that made
|
||||
// claude-intercom's push delivery reliable (its 1s HTTP poll had this
|
||||
// effect as a side effect). The interval does nothing except prevent the
|
||||
// event loop from settling.
|
||||
const keepalive = setInterval(() => {
|
||||
// Intentionally empty — the interval itself keeps the event loop active.
|
||||
// Do NOT call .unref() — that would defeat the purpose.
|
||||
}, 1_000);
|
||||
void keepalive; // suppress unused warning
|
||||
|
||||
const shutdown = (): void => {
|
||||
clearInterval(keepalive);
|
||||
stopAll();
|
||||
process.exit(0);
|
||||
};
|
||||
|
||||
@@ -555,4 +555,21 @@ export const TOOLS: Tool[] = [
|
||||
"Get a complete overview of the mesh: peers, groups, state, memory, files, tasks, streams, tables. Call on session start for full situational awareness.",
|
||||
inputSchema: { type: "object", properties: {} },
|
||||
},
|
||||
|
||||
// --- Diagnostics ---
|
||||
{
|
||||
name: "ping_mesh",
|
||||
description:
|
||||
"Send test messages through the full pipeline and measure round-trip timing per priority. Diagnoses push delivery issues.",
|
||||
inputSchema: {
|
||||
type: "object",
|
||||
properties: {
|
||||
priorities: {
|
||||
type: "array",
|
||||
items: { type: "string", enum: ["now", "next", "low"] },
|
||||
description: "Priorities to test (default: [\"now\", \"next\"])",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
@@ -526,8 +526,11 @@ export class BrokerClient {
|
||||
body: data,
|
||||
signal: AbortSignal.timeout(30_000),
|
||||
});
|
||||
const body = await res.json() as { ok?: boolean; fileId?: string };
|
||||
return body.fileId ?? null;
|
||||
const body = await res.json() as { ok?: boolean; fileId?: string; error?: string };
|
||||
if (!res.ok || !body.fileId) {
|
||||
throw new Error(body.error ?? `HTTP ${res.status}`);
|
||||
}
|
||||
return body.fileId;
|
||||
}
|
||||
|
||||
// --- Vectors ---
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"build": "next build",
|
||||
"build": "next build --no-turbopack",
|
||||
"clean": "git clean -xdf .cache .next .turbo node_modules",
|
||||
"dev": "next dev",
|
||||
"format": "prettier --check . --ignore-path ../../.gitignore",
|
||||
|
||||
Reference in New Issue
Block a user