feat(broker+cli): topics — conversation scope within a mesh (v0.2.0)

Adds the third axis of mesh organization: mesh = trust boundary, group = identity tag, topic = conversation scope. Topic-tagged messages filter delivery by topic_member rows and persist to a topic_message history table for back-scroll on reconnect. Schema (additive): - mesh.topic, mesh.topic_member, mesh.topic_message tables - topic_visibility (public|private|dm) and topic_member_role (lead|member|observer) enums - migration 0022_topics.sql, hand-written following project convention (drizzle journal has been drifting since 0011) Broker: - 10 helpers (createTopic, listTopics, findTopicByName, joinTopic, leaveTopic, topicMembers, getMemberTopicIds, appendTopicMessage, topicHistory, markTopicRead) - drainForMember matches "#<topicId>" target_specs via member's topic memberships - 7 WS handlers (topic_create/list/join/leave/members/history/mark_read) + resolveTopicId helper accepting id-or-name - handleSend auto-persists topic-tagged messages to history CLI: - claudemesh topic create/list/join/leave/members/history/read - claudemesh send "#deploys" "..." resolves topic name to id - bundled skill teaches Claude the DM/group/topic decision matrix - policy-classify recognizes topic create/join/leave as writes Spec: .artifacts/specs/2026-05-02-v0.2.0-scope.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-02 01:53:42 +01:00
parent b4f457fceb
commit 1afae7a507
12 changed files with 1741 additions and 196 deletions
--- a/apps/broker/src/broker.ts
+++ b/apps/broker/src/broker.ts
@@ -42,6 +42,9 @@ import {
  meshService,
  meshSkill,
  meshStream,
+  meshTopic,
+  meshTopicMember,
+  meshTopicMessage,
  meshVaultEntry,
  meshTask,
  messageQueue,
@@ -531,6 +534,254 @@ export async function leaveGroup(
  return groups;
 }

+// --- Topics (v0.2.0) ---
+//
+// Conversational primitive within a mesh. Spec:
+// .artifacts/specs/2026-05-02-v0.2.0-scope.md
+//
+// Mesh = trust boundary. Group = identity tag. Topic = conversation scope.
+// Three orthogonal axes; topics complement (don't replace) groups.
+//
+// Routing: topic-tagged messages use targetSpec = "#<topicId>". The drain
+// query joins topic_member to filter delivery, so non-members never see
+// the message. Topic-tagged messages are also persisted to topic_message
+// so humans (and opting-in agents) can fetch history on reconnect.
+
+/** Create a topic in a mesh. Idempotent on (meshId, name). */
+export async function createTopic(args: {
+  meshId: string;
+  name: string;
+  description?: string;
+  visibility?: "public" | "private" | "dm";
+  createdByMemberId?: string;
+}): Promise<{ id: string; created: boolean }> {
+  const existing = await db
+    .select({ id: meshTopic.id })
+    .from(meshTopic)
+    .where(and(eq(meshTopic.meshId, args.meshId), eq(meshTopic.name, args.name)));
+  if (existing[0]) return { id: existing[0].id, created: false };
+
+  const [row] = await db
+    .insert(meshTopic)
+    .values({
+      meshId: args.meshId,
+      name: args.name,
+      description: args.description ?? null,
+      visibility: args.visibility ?? "public",
+      createdByMemberId: args.createdByMemberId ?? null,
+    })
+    .returning({ id: meshTopic.id });
+  if (!row) throw new Error("failed to create topic");
+  return { id: row.id, created: true };
+}
+
+/** List topics in a mesh, with member counts. */
+export async function listTopics(meshId: string): Promise<
+  Array<{
+    id: string;
+    name: string;
+    description: string | null;
+    visibility: "public" | "private" | "dm";
+    memberCount: number;
+    createdAt: Date;
+  }>
+> {
+  const rows = await db
+    .select({
+      id: meshTopic.id,
+      name: meshTopic.name,
+      description: meshTopic.description,
+      visibility: meshTopic.visibility,
+      createdAt: meshTopic.createdAt,
+      memberCount: sql<number>`(SELECT COUNT(*)::int FROM mesh.topic_member WHERE topic_id = ${meshTopic.id})`,
+    })
+    .from(meshTopic)
+    .where(and(eq(meshTopic.meshId, meshId), isNull(meshTopic.archivedAt)))
+    .orderBy(asc(meshTopic.name));
+  return rows;
+}
+
+/** Resolve a topic by name within a mesh. */
+export async function findTopicByName(
+  meshId: string,
+  name: string,
+): Promise<{ id: string; visibility: "public" | "private" | "dm" } | null> {
+  const [row] = await db
+    .select({ id: meshTopic.id, visibility: meshTopic.visibility })
+    .from(meshTopic)
+    .where(
+      and(
+        eq(meshTopic.meshId, meshId),
+        eq(meshTopic.name, name),
+        isNull(meshTopic.archivedAt),
+      ),
+    );
+  return row ?? null;
+}
+
+/** Add a member to a topic. Idempotent. */
+export async function joinTopic(args: {
+  topicId: string;
+  memberId: string;
+  role?: "lead" | "member" | "observer";
+}): Promise<void> {
+  await db
+    .insert(meshTopicMember)
+    .values({
+      topicId: args.topicId,
+      memberId: args.memberId,
+      role: args.role ?? "member",
+    })
+    .onConflictDoNothing();
+}
+
+/** Remove a member from a topic. */
+export async function leaveTopic(args: {
+  topicId: string;
+  memberId: string;
+}): Promise<void> {
+  await db
+    .delete(meshTopicMember)
+    .where(
+      and(
+        eq(meshTopicMember.topicId, args.topicId),
+        eq(meshTopicMember.memberId, args.memberId),
+      ),
+    );
+}
+
+/** List members of a topic with display names. */
+export async function topicMembers(topicId: string): Promise<
+  Array<{
+    memberId: string;
+    pubkey: string;
+    displayName: string;
+    role: "lead" | "member" | "observer";
+    joinedAt: Date;
+    lastReadAt: Date | null;
+  }>
+> {
+  const rows = await db
+    .select({
+      memberId: meshTopicMember.memberId,
+      pubkey: memberTable.peerPubkey,
+      displayName: memberTable.displayName,
+      role: meshTopicMember.role,
+      joinedAt: meshTopicMember.joinedAt,
+      lastReadAt: meshTopicMember.lastReadAt,
+    })
+    .from(meshTopicMember)
+    .innerJoin(memberTable, eq(meshTopicMember.memberId, memberTable.id))
+    .where(eq(meshTopicMember.topicId, topicId))
+    .orderBy(asc(memberTable.displayName));
+  return rows;
+}
+
+/** Return all topic ids a member belongs to (used by message routing). */
+export async function getMemberTopicIds(memberId: string): Promise<string[]> {
+  const rows = await db
+    .select({ id: meshTopicMember.topicId })
+    .from(meshTopicMember)
+    .where(eq(meshTopicMember.memberId, memberId));
+  return rows.map((r) => r.id);
+}
+
+/** Append a topic message to persistent history. */
+export async function appendTopicMessage(args: {
+  topicId: string;
+  senderMemberId: string;
+  senderSessionPubkey?: string;
+  nonce: string;
+  ciphertext: string;
+}): Promise<string> {
+  const [row] = await db
+    .insert(meshTopicMessage)
+    .values({
+      topicId: args.topicId,
+      senderMemberId: args.senderMemberId,
+      senderSessionPubkey: args.senderSessionPubkey ?? null,
+      nonce: args.nonce,
+      ciphertext: args.ciphertext,
+    })
+    .returning({ id: meshTopicMessage.id });
+  if (!row) throw new Error("failed to append topic message");
+  return row.id;
+}
+
+/**
+ * Fetch topic history for a member. Pagination via `before` cursor (id of
+ * an earlier message); pass null for the latest page.
+ */
+export async function topicHistory(args: {
+  topicId: string;
+  limit?: number;
+  beforeId?: string;
+}): Promise<
+  Array<{
+    id: string;
+    senderMemberId: string;
+    senderPubkey: string;
+    nonce: string;
+    ciphertext: string;
+    createdAt: Date;
+  }>
+> {
+  const limit = Math.min(Math.max(args.limit ?? 50, 1), 200);
+  const beforeClause = args.beforeId
+    ? sql`AND tm.created_at < (SELECT created_at FROM mesh.topic_message WHERE id = ${args.beforeId})`
+    : sql``;
+  const result = await db.execute<{
+    id: string;
+    sender_member_id: string;
+    sender_pubkey: string;
+    nonce: string;
+    ciphertext: string;
+    created_at: Date;
+  }>(sql`
+    SELECT tm.id, tm.sender_member_id,
+           COALESCE(tm.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey,
+           tm.nonce, tm.ciphertext, tm.created_at
+    FROM mesh.topic_message tm
+    JOIN mesh.member m ON m.id = tm.sender_member_id
+    WHERE tm.topic_id = ${args.topicId}
+    ${beforeClause}
+    ORDER BY tm.created_at DESC, tm.id DESC
+    LIMIT ${limit}
+  `);
+  const rows = (result.rows ?? result) as Array<{
+    id: string;
+    sender_member_id: string;
+    sender_pubkey: string;
+    nonce: string;
+    ciphertext: string;
+    created_at: Date;
+  }>;
+  return rows.map((r) => ({
+    id: r.id,
+    senderMemberId: r.sender_member_id,
+    senderPubkey: r.sender_pubkey,
+    nonce: r.nonce,
+    ciphertext: r.ciphertext,
+    createdAt: r.created_at instanceof Date ? r.created_at : new Date(r.created_at),
+  }));
+}
+
+/** Update last_read_at for a member's topic subscription. */
+export async function markTopicRead(args: {
+  topicId: string;
+  memberId: string;
+}): Promise<void> {
+  await db
+    .update(meshTopicMember)
+    .set({ lastReadAt: new Date() })
+    .where(
+      and(
+        eq(meshTopicMember.topicId, args.topicId),
+        eq(meshTopicMember.memberId, args.memberId),
+      ),
+    );
+}
+
 // --- Shared state ---

 /**
@@ -1563,7 +1814,7 @@ function deliverablePriorities(status: PeerStatus): Priority[] {
 */
 export async function drainForMember(
  meshId: string,
-  _memberId: string,
+  memberId: string,
  memberPubkey: string,
  status: PeerStatus,
  sessionPubkey?: string,
@@ -1615,6 +1866,17 @@ export async function drainForMember(
    groupTargets.map((t) => `'${t}'`).join(","),
  );

+  // Topic membership targets (v0.2.0). targetSpec for topic-tagged
+  // messages is "#<topicId>". A member receives a topic message iff
+  // they're in topic_member for that topic. We resolve memberships
+  // here and inline the list — same pattern as groups, no schema join
+  // in the hot path.
+  const topicIds = await getMemberTopicIds(memberId);
+  const topicTargetList =
+    topicIds.length > 0
+      ? sql.raw(topicIds.map((id) => `'#${id}'`).join(","))
+      : null;
+
  // Atomic claim with SQL-side ordering. The CTE claims rows via
  // UPDATE...RETURNING; the outer SELECT re-orders by created_at
  // (with id as tiebreaker so equal-timestamp rows stay deterministic).
@@ -1638,7 +1900,7 @@ export async function drainForMember(
        WHERE mesh_id = ${meshId}
          AND delivered_at IS NULL
          AND priority::text IN (${priorityList})
-          AND (target_spec = ${memberPubkey} OR target_spec = '*'${sessionPubkey ? sql` OR target_spec = ${sessionPubkey}` : sql``} OR target_spec IN (${groupTargetList}))
+          AND (target_spec = ${memberPubkey} OR target_spec = '*'${sessionPubkey ? sql` OR target_spec = ${sessionPubkey}` : sql``} OR target_spec IN (${groupTargetList})${topicTargetList ? sql` OR target_spec IN (${topicTargetList})` : sql``})
          ${excludeSenderSessionPubkey ? sql`AND NOT (target_spec IN ('*') AND sender_session_pubkey = ${excludeSenderSessionPubkey})` : sql``}
        ORDER BY created_at ASC, id ASC
        FOR UPDATE SKIP LOCKED
--- a/apps/broker/src/index.ts
+++ b/apps/broker/src/index.ts
@@ -18,7 +18,7 @@ import { WebSocketServer, type WebSocket } from "ws";
 import { and, eq, inArray, isNull, lt, sql } from "drizzle-orm";
 import { env } from "./env";
 import { db } from "./db";
-import { invite as inviteTable, mesh, meshMember, messageQueue, presence, scheduledMessage as scheduledMessageTable, meshWebhook, peerState } from "@turbostarter/db/schema/mesh";
+import { invite as inviteTable, mesh, meshMember, messageQueue, presence, scheduledMessage as scheduledMessageTable, meshWebhook, peerState, meshTopic } from "@turbostarter/db/schema/mesh";
 import { user } from "@turbostarter/db/schema/auth";
 import { handleCliSync, type CliSyncRequest } from "./cli-sync";
 import { generateId } from "@turbostarter/shared/utils";
@@ -84,6 +84,15 @@ import {
  listDbMeshServices,
  deleteService,
  getRunningServices,
+  createTopic,
+  listTopics,
+  findTopicByName,
+  joinTopic,
+  leaveTopic,
+  topicMembers,
+  topicHistory,
+  markTopicRead,
+  appendTopicMessage,
 } from "./broker";
 import * as serviceManager from "./service-manager";
 import { ensureBucket, meshBucketName, minioClient } from "./minio";
@@ -1495,6 +1504,26 @@ function sendError(
  }
 }

+/**
+ * Resolve a topic identifier — accepts either a topic id directly OR a
+ * topic name within the given mesh. Returns the topic id, or null if no
+ * matching topic exists. Used by every topic_* WS handler so callers can
+ * reference topics by human-readable name without an extra round trip.
+ */
+async function resolveTopicId(meshId: string, idOrName: string): Promise<string | null> {
+  // ULID-ish ids are 25-26 chars of base32; names are usually shorter and
+  // human-readable. Try as id first (cheap PK lookup), fall back to name.
+  if (idOrName.length >= 20 && /^[a-z0-9_-]+$/i.test(idOrName)) {
+    const byId = await db
+      .select({ id: meshTopic.id })
+      .from(meshTopic)
+      .where(and(eq(meshTopic.id, idOrName), eq(meshTopic.meshId, meshId)));
+    if (byId[0]) return byId[0].id;
+  }
+  const byName = await findTopicByName(meshId, idOrName);
+  return byName?.id ?? null;
+}
+
 // --- Peer state persistence ---

 async function savePeerState(conn: PeerConn, memberId: string, meshId: string): Promise<void> {
@@ -1901,6 +1930,24 @@ async function handleSend(
    nonce: msg.nonce,
    ciphertext: msg.ciphertext,
  });
+
+  // Topic-tagged messages (targetSpec starts with `#<topicId>`) get
+  // persisted to topic_message in addition to the ephemeral queue, so
+  // humans (and opting-in agents) can fetch history on reconnect.
+  // Spec: .artifacts/specs/2026-05-02-v0.2.0-scope.md
+  if (msg.targetSpec.startsWith("#")) {
+    const topicId = msg.targetSpec.slice(1);
+    void appendTopicMessage({
+      topicId,
+      senderMemberId: conn.memberId,
+      senderSessionPubkey: conn.sessionPubkey ?? undefined,
+      nonce: msg.nonce,
+      ciphertext: msg.ciphertext,
+    }).catch((e) =>
+      log.warn("appendTopicMessage failed", { topic_id: topicId, err: String(e) }),
+    );
+  }
+
  void audit(conn.meshId, "message_sent", conn.memberId, conn.displayName, {
    targetSpec: msg.targetSpec,
    priority: msg.priority,
@@ -2291,6 +2338,125 @@ function handleConnection(ws: WebSocket): void {
          });
          break;
        }
+
+        // ── Topics (v0.2.0) ─────────────────────────────────────────
+        case "topic_create": {
+          const tc = msg as Extract<WSClientMessage, { type: "topic_create" }>;
+          const result = await createTopic({
+            meshId: conn.meshId,
+            name: tc.name,
+            description: tc.description,
+            visibility: tc.visibility,
+            createdByMemberId: conn.memberId,
+          });
+          // Auto-subscribe the creator.
+          await joinTopic({ topicId: result.id, memberId: conn.memberId, role: "lead" });
+          const resp: WSServerMessage = {
+            type: "topic_created",
+            topic: {
+              id: result.id,
+              name: tc.name,
+              visibility: tc.visibility ?? "public",
+            },
+            created: result.created,
+            ...(_reqId ? { _reqId } : {}),
+          };
+          conn.ws.send(JSON.stringify(resp));
+          log.info("ws topic_create", { presence_id: presenceId, topic: tc.name, created: result.created });
+          break;
+        }
+
+        case "topic_list": {
+          const topics = await listTopics(conn.meshId);
+          const resp: WSServerMessage = {
+            type: "topic_list_response",
+            topics: topics.map((t) => ({
+              id: t.id,
+              name: t.name,
+              description: t.description,
+              visibility: t.visibility,
+              memberCount: t.memberCount,
+              createdAt: t.createdAt.toISOString(),
+            })),
+            ...(_reqId ? { _reqId } : {}),
+          };
+          conn.ws.send(JSON.stringify(resp));
+          break;
+        }
+
+        case "topic_join": {
+          const tj = msg as Extract<WSClientMessage, { type: "topic_join" }>;
+          const topicId = await resolveTopicId(conn.meshId, tj.topic);
+          if (!topicId) { sendError(ws, "topic_not_found", `topic "${tj.topic}" not found`, _reqId); break; }
+          await joinTopic({ topicId, memberId: conn.memberId, role: tj.role });
+          log.info("ws topic_join", { presence_id: presenceId, topic: topicId });
+          break;
+        }
+
+        case "topic_leave": {
+          const tl = msg as Extract<WSClientMessage, { type: "topic_leave" }>;
+          const topicId = await resolveTopicId(conn.meshId, tl.topic);
+          if (!topicId) { sendError(ws, "topic_not_found", `topic "${tl.topic}" not found`, _reqId); break; }
+          await leaveTopic({ topicId, memberId: conn.memberId });
+          log.info("ws topic_leave", { presence_id: presenceId, topic: topicId });
+          break;
+        }
+
+        case "topic_members": {
+          const tm = msg as Extract<WSClientMessage, { type: "topic_members" }>;
+          const topicId = await resolveTopicId(conn.meshId, tm.topic);
+          if (!topicId) { sendError(ws, "topic_not_found", `topic "${tm.topic}" not found`, _reqId); break; }
+          const members = await topicMembers(topicId);
+          const resp: WSServerMessage = {
+            type: "topic_members_response",
+            topic: tm.topic,
+            members: members.map((m) => ({
+              memberId: m.memberId,
+              pubkey: m.pubkey,
+              displayName: m.displayName,
+              role: m.role,
+              joinedAt: m.joinedAt.toISOString(),
+              lastReadAt: m.lastReadAt?.toISOString() ?? null,
+            })),
+            ...(_reqId ? { _reqId } : {}),
+          };
+          conn.ws.send(JSON.stringify(resp));
+          break;
+        }
+
+        case "topic_history": {
+          const th = msg as Extract<WSClientMessage, { type: "topic_history" }>;
+          const topicId = await resolveTopicId(conn.meshId, th.topic);
+          if (!topicId) { sendError(ws, "topic_not_found", `topic "${th.topic}" not found`, _reqId); break; }
+          const history = await topicHistory({
+            topicId,
+            limit: th.limit,
+            beforeId: th.beforeId,
+          });
+          const resp: WSServerMessage = {
+            type: "topic_history_response",
+            topic: th.topic,
+            messages: history.map((h) => ({
+              id: h.id,
+              senderPubkey: h.senderPubkey,
+              nonce: h.nonce,
+              ciphertext: h.ciphertext,
+              createdAt: h.createdAt.toISOString(),
+            })),
+            ...(_reqId ? { _reqId } : {}),
+          };
+          conn.ws.send(JSON.stringify(resp));
+          break;
+        }
+
+        case "topic_mark_read": {
+          const tr = msg as Extract<WSClientMessage, { type: "topic_mark_read" }>;
+          const topicId = await resolveTopicId(conn.meshId, tr.topic);
+          if (!topicId) { sendError(ws, "topic_not_found", `topic "${tr.topic}" not found`, _reqId); break; }
+          await markTopicRead({ topicId, memberId: conn.memberId });
+          break;
+        }
+
        case "set_state": {
          const ss = msg as Extract<WSClientMessage, { type: "set_state" }>;
          // Look up the display name for attribution.
--- a/apps/broker/src/types.ts
+++ b/apps/broker/src/types.ts
@@ -179,6 +179,107 @@ export interface WSLeaveGroupMessage {
  name: string;
 }

+// ── Topics (v0.2.0) ─────────────────────────────────────────────────
+// Topics complement groups: groups are identity tags, topics are
+// conversation scopes. targetSpec for topic-tagged messages is
+// "#<topicId>". Spec: .artifacts/specs/2026-05-02-v0.2.0-scope.md
+
+export interface WSTopicCreateMessage {
+  type: "topic_create";
+  name: string;
+  description?: string;
+  visibility?: "public" | "private" | "dm";
+  _reqId?: string;
+}
+
+export interface WSTopicListMessage {
+  type: "topic_list";
+  _reqId?: string;
+}
+
+export interface WSTopicJoinMessage {
+  type: "topic_join";
+  /** Topic id OR name. Server resolves. */
+  topic: string;
+  role?: "lead" | "member" | "observer";
+  _reqId?: string;
+}
+
+export interface WSTopicLeaveMessage {
+  type: "topic_leave";
+  topic: string;
+  _reqId?: string;
+}
+
+export interface WSTopicMembersMessage {
+  type: "topic_members";
+  topic: string;
+  _reqId?: string;
+}
+
+export interface WSTopicHistoryMessage {
+  type: "topic_history";
+  topic: string;
+  limit?: number;
+  beforeId?: string;
+  _reqId?: string;
+}
+
+export interface WSTopicMarkReadMessage {
+  type: "topic_mark_read";
+  topic: string;
+  _reqId?: string;
+}
+
+// Server → client topic responses
+
+export interface WSTopicCreatedMessage {
+  type: "topic_created";
+  topic: { id: string; name: string; visibility: "public" | "private" | "dm" };
+  created: boolean;
+  _reqId?: string;
+}
+
+export interface WSTopicListResponseMessage {
+  type: "topic_list_response";
+  topics: Array<{
+    id: string;
+    name: string;
+    description: string | null;
+    visibility: "public" | "private" | "dm";
+    memberCount: number;
+    createdAt: string;
+  }>;
+  _reqId?: string;
+}
+
+export interface WSTopicMembersResponseMessage {
+  type: "topic_members_response";
+  topic: string;
+  members: Array<{
+    memberId: string;
+    pubkey: string;
+    displayName: string;
+    role: "lead" | "member" | "observer";
+    joinedAt: string;
+    lastReadAt: string | null;
+  }>;
+  _reqId?: string;
+}
+
+export interface WSTopicHistoryResponseMessage {
+  type: "topic_history_response";
+  topic: string;
+  messages: Array<{
+    id: string;
+    senderPubkey: string;
+    nonce: string;
+    ciphertext: string;
+    createdAt: string;
+  }>;
+  _reqId?: string;
+}
+
 /** Client → broker: set a shared state key-value. */
 export interface WSSetStateMessage {
  type: "set_state";
@@ -1145,6 +1246,13 @@ export type WSClientMessage =
  | WSSetProfileMessage
  | WSJoinGroupMessage
  | WSLeaveGroupMessage
+  | WSTopicCreateMessage
+  | WSTopicListMessage
+  | WSTopicJoinMessage
+  | WSTopicLeaveMessage
+  | WSTopicMembersMessage
+  | WSTopicHistoryMessage
+  | WSTopicMarkReadMessage
  | WSSetStateMessage
  | WSGetStateMessage
  | WSListStateMessage
@@ -1313,6 +1421,10 @@ export type WSServerMessage =
  | WSPushMessage
  | WSAckMessage
  | WSPeersListMessage
+  | WSTopicCreatedMessage
+  | WSTopicListResponseMessage
+  | WSTopicMembersResponseMessage
+  | WSTopicHistoryResponseMessage
  | WSStateChangeMessage
  | WSStateResultMessage
  | WSStateListMessage