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

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>

apps/broker/src/broker.ts

@@ -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 {

apps/broker/src/index.ts

@@ -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" });

apps/broker/src/types.ts

@@ -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;