feat(broker,cli): member-key DMs are visible and --fanout goes out per session
A DM addressed to a member key with live sessions is claimed by whichever connection drains first and the member's daemon shows it to every session — the wrong-peer noise. 1.38 CLIs no longer send those (the daemon resolver maps member keys to one session), but older CLIs on the mesh still do. - broker: logs member_target_dm (sender, target, live_sessions) for such DMs; with CLAUDEMESH_MEMBER_DM_POLICY=reject it answers member_target_requires_session instead of queuing. Default is warn, so deploying this changes no delivery — flip after the fleet is on 1.38. - cli: `send --fanout <member-key>` expands client-side to one DM per live session (each sealed to that session), so no member-key DM goes on the wire even on purpose. Spec: .artifacts/specs/2026-09-26-session-identity-and-routing.md §3 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -2501,6 +2501,45 @@ async function handleSend(
|
|||||||
// Soft-fail; the rejection below still fires if no candidate.
|
// Soft-fail; the rejection below still fires if no candidate.
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// 1.38.0 (spec 2026-09-26 §3): a DM addressed to a MEMBER key whose
|
||||||
|
// member has live session connections is claimed by whichever of them
|
||||||
|
// drains first — and the member's daemon shows it to every session.
|
||||||
|
// That is the "messages land on the wrong peers" noise. New CLIs
|
||||||
|
// resolve member keys to one session (or expand --fanout client-side),
|
||||||
|
// so on the wire this only comes from older clients. Policy:
|
||||||
|
// CLAUDEMESH_MEMBER_DM_POLICY=warn (default: log + count) | reject.
|
||||||
|
{
|
||||||
|
const liveSessionsOfTarget = [...connections.values()].filter(
|
||||||
|
(peer) =>
|
||||||
|
peer.meshId === conn.meshId &&
|
||||||
|
peer.memberPubkey === msg.targetSpec &&
|
||||||
|
!!peer.sessionPubkey &&
|
||||||
|
peer.sessionPubkey !== msg.targetSpec,
|
||||||
|
).length;
|
||||||
|
if (liveSessionsOfTarget > 0) {
|
||||||
|
const policy = process.env.CLAUDEMESH_MEMBER_DM_POLICY === "reject" ? "reject" : "warn";
|
||||||
|
log.warn("member_target_dm", {
|
||||||
|
mesh_id: conn.meshId,
|
||||||
|
sender: (conn.sessionPubkey ?? conn.memberPubkey).slice(0, 16),
|
||||||
|
target: msg.targetSpec.slice(0, 16),
|
||||||
|
live_sessions: liveSessionsOfTarget,
|
||||||
|
policy,
|
||||||
|
});
|
||||||
|
if (policy === "reject") {
|
||||||
|
metrics.messagesRejectedTotal.inc({ reason: "member_target_requires_session" });
|
||||||
|
const errAck: WSServerMessage = {
|
||||||
|
type: "ack",
|
||||||
|
id: msg.id ?? "",
|
||||||
|
messageId: "",
|
||||||
|
queued: false,
|
||||||
|
error: `member_target_requires_session: that member has ${liveSessionsOfTarget} live session(s) — address one session pubkey (claudemesh ≥1.38 does this for you)`,
|
||||||
|
};
|
||||||
|
conn.ws.send(JSON.stringify(errAck));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (candidateMemberIds.length === 0) {
|
if (candidateMemberIds.length === 0) {
|
||||||
metrics.messagesRejectedTotal.inc({ reason: "no_recipient" });
|
metrics.messagesRejectedTotal.inc({ reason: "no_recipient" });
|
||||||
const errAck: WSServerMessage = {
|
const errAck: WSServerMessage = {
|
||||||
|
|||||||
@@ -73,8 +73,11 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
|
|||||||
const isFullPubkey = /^[0-9a-f]{64}$/i.test(to);
|
const isFullPubkey = /^[0-9a-f]{64}$/i.test(to);
|
||||||
const isPrefix = /^[0-9a-f]{4,63}$/i.test(to);
|
const isPrefix = /^[0-9a-f]{4,63}$/i.test(to);
|
||||||
const isName = isDirect && !isFullPubkey && !isPrefix;
|
const isName = isDirect && !isFullPubkey && !isPrefix;
|
||||||
|
// --fanout to a member key: expanded here to one DM per live session,
|
||||||
|
// each sealed to that session — no member-key DM goes on the wire.
|
||||||
|
let fanoutTargets: Array<{ pubkey: string; displayName: string }> | null = null;
|
||||||
|
|
||||||
if (isDirect && (isPrefix || isName || (isFullPubkey && !meshSlug))) {
|
if (isDirect && (isPrefix || isName || (isFullPubkey && (!meshSlug || flags.fanout)))) {
|
||||||
const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js");
|
const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js");
|
||||||
const searchSlugs = meshSlug ? [meshSlug] : config.meshes.map((m) => m.slug);
|
const searchSlugs = meshSlug ? [meshSlug] : config.meshes.map((m) => m.slug);
|
||||||
const lower = to.toLowerCase();
|
const lower = to.toLowerCase();
|
||||||
@@ -122,6 +125,7 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
|
|||||||
// A full MEMBER pubkey with several live sessions. Fan-out only on
|
// A full MEMBER pubkey with several live sessions. Fan-out only on
|
||||||
// request; otherwise make the caller pick the session.
|
// request; otherwise make the caller pick the session.
|
||||||
meshSlug = meshesHit[0]!;
|
meshSlug = meshesHit[0]!;
|
||||||
|
if (flags.fanout) fanoutTargets = uniq.map((m) => ({ pubkey: m.pubkey, displayName: m.displayName }));
|
||||||
if (!flags.fanout) {
|
if (!flags.fanout) {
|
||||||
render.err(`${to.slice(0, 16)}… is a member key with ${uniq.length} live sessions — a DM to it reaches all of them.`);
|
render.err(`${to.slice(0, 16)}… is a member key with ${uniq.length} live sessions — a DM to it reaches all of them.`);
|
||||||
render.hint(`sessions: ${uniq.map((m) => `${m.displayName} ${m.pubkey.slice(0, 16)}…`).join(", ")}`);
|
render.hint(`sessions: ${uniq.map((m) => `${m.displayName} ${m.pubkey.slice(0, 16)}…`).join(", ")}`);
|
||||||
@@ -318,6 +322,20 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (fanoutTargets) {
|
||||||
|
const results: Array<{ pubkey: string; ok: boolean; error?: string }> = [];
|
||||||
|
for (const t of fanoutTargets) {
|
||||||
|
const dr = await trySendViaDaemon({ to: t.pubkey, message, priority, expectedMesh: meshSlug ?? undefined });
|
||||||
|
results.push(dr === null ? { pubkey: t.pubkey, ok: false, error: "daemon path unavailable" }
|
||||||
|
: dr.ok ? { pubkey: t.pubkey, ok: true } : { pubkey: t.pubkey, ok: false, error: dr.error });
|
||||||
|
}
|
||||||
|
const okCount = results.filter((r) => r.ok).length;
|
||||||
|
if (flags.json) console.log(JSON.stringify({ ok: okCount === results.length, fanout: results }));
|
||||||
|
else render.ok(`sent to ${okCount}/${results.length} sessions of ${to.slice(0, 16)}… (--fanout)`);
|
||||||
|
if (okCount < results.length) process.exit(1);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
// Daemon path — preferred when a long-lived daemon is local. UDS at
|
// Daemon path — preferred when a long-lived daemon is local. UDS at
|
||||||
// ~/.claudemesh/daemon/daemon.sock; ~1ms round-trip; persists outbox
|
// ~/.claudemesh/daemon/daemon.sock; ~1ms round-trip; persists outbox
|
||||||
// across CLI invocations so a `claudemesh send` survives a daemon
|
// across CLI invocations so a `claudemesh send` survives a daemon
|
||||||
|
|||||||
Reference in New Issue
Block a user