chat-viewer: rt-client relay, daemon probe, and RoomSummary.defaultWake

lib/daemon/__tests__/chat-handlers.test.ts

@@ -354,6 +354,20 @@ test("chat:rooms marks a dm and chat:who carries presence statuses", async () =>
   expect(memberA?.status).toBe("idle");
 });
 
+test("chat:rooms carries a room's stamped default wake mode, and leaves it undefined when never stamped", async () => {
+  const h = freshHandlers();
+  await h["chat:join"]({ room: "loud", handle: "a", wakeOn: "all" });
+  await h["chat:join"]({ room: "quiet", handle: "b" });
+
+  const rooms = await h["chat:rooms"]({ handle: "a" });
+  if (!rooms.ok) throw new Error("unreachable");
+  expect(rooms.data.rooms.find((r) => r.room === "loud")).toMatchObject({ defaultWake: "all" });
+
+  const rooms2 = await h["chat:rooms"]({ handle: "b" });
+  if (!rooms2.ok) throw new Error("unreachable");
+  expect(rooms2.data.rooms.find((r) => r.room === "quiet")?.defaultWake).toBeUndefined();
+});
+
 test("chat:who on an agent-agent dm room excludes the silent human row", async () => {
   const h = freshHandlers();
   await h["chat:sign-in"]({ sessionId: "s1", baseHandle: "a" });

lib/daemon/handlers/chat.ts

@@ -16,6 +16,7 @@ import {
   markRead,
   unreadWakingCount,
   listRooms,
+  roomDefaultWake,
   listMembers,
   armMember,
   touchMember,
@@ -176,8 +177,10 @@ export function createChatHandlers(opts: {
 
     "chat:rooms": async (payload: Commands["chat:rooms"]["payload"]): Promise<CommandResult<"chat:rooms">> => {
       const rooms = listRooms(payload.handle, db).map((room) => {
+        const defaultWake = roomDefaultWake(room.room, db);
+        const withDefault = defaultWake ? { ...room, defaultWake } : room;
         const dm = dmParticipants(room.room, db);
-        return dm ? { ...room, kind: "dm" as const, participants: dm } : room;
+        return dm ? { ...withDefault, kind: "dm" as const, participants: dm } : withDefault;
       });
       return { ok: true, data: { rooms } };
     },

lib/state/chat-store.ts

@@ -245,6 +245,12 @@ export function listRooms(handle: string, db: Database = getStateDb()): RoomSumm
   });
 }
 
+/** The wake mode stamped by whichever join created `room`; undefined for a room never stamped (including every DM room — dmRoomFor never stamps one). */
+export function roomDefaultWake(room: string, db: Database = getStateDb()): WakeMode | undefined {
+  const row = db.query(SELECT_ROOM_DEFAULT_WAKE_SQL).get(room) as { wake_on: WakeMode } | null;
+  return row?.wake_on;
+}
+
 export function listMembers(room: string, db: Database = getStateDb()): ChatMember[] {
   const rows = db.query(SELECT_ROOM_MEMBERS_SQL).all(room) as MemberRow[];
   return rows.map(rowToMember);

lib/state/index.ts

@@ -107,6 +107,7 @@ export {
   markRead,
   unreadWakingCount,
   listRooms,
+  roomDefaultWake,
   listMembers,
   armMember,
   touchMember,

packages/rt-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mattstack/rt-client",
-  "version": "0.5.0",
+  "version": "0.6.0",
   "type": "module",
   "exports": {
     ".": {

packages/rt-client/src/commands.ts

@@ -113,6 +113,8 @@ export interface RoomSummary {
   /** Set only by chat:rooms's left join against chat_dms. */
   kind?: "dm";
   participants?: { a: string; b: string };
+  /** Set only by chat:rooms's left join against chat_room_defaults; undefined for a room never stamped a default (every DM room included). */
+  defaultWake?: WakeMode;
 }
 
 /**

packages/rt-client/src/health.ts

@@ -0,0 +1,15 @@
+import { eventsHead } from "./client.ts";
+import type { RtClientOptions } from "./transport.ts";
+
+/**
+ * A daemon-down result is a successful probe, not a failure of this call —
+ * eventsHead already never throws (transport.ts degrades every fetch to
+ * `{ ok: false, error }`), so this only reshapes that envelope for callers
+ * who want a boolean, not `{ ok, data, error }`.
+ */
+export async function daemonHealth(
+  opts: RtClientOptions = {},
+): Promise<{ reachable: boolean; error?: string }> {
+  const res = await eventsHead(opts);
+  return { reachable: res.ok, error: res.ok ? undefined : res.error };
+}

packages/rt-client/src/index.ts

@@ -60,9 +60,11 @@ export type {
   PresenceRow,
 } from "./commands.ts";
 
-export { subscribe, DEFAULT_WS_URL } from "./relay.ts";
+export { subscribe, createRelay, DEFAULT_WS_URL } from "./relay.ts";
 export type { RelayEventType } from "./relay.ts";
 
+export { daemonHealth } from "./health.ts";
+
 export { repoNameForPath } from "./repos.ts";
 
 // ─── Settings (RT-50) ────────────────────────────────────────────────────────

packages/rt-client/src/relay.ts

@@ -59,3 +59,34 @@ export function subscribe(
     try { ws?.close(); } catch { /* already closed */ }
   };
 }
+
+/**
+ * One daemon subscription for the whole process, republished onto a
+ * caller-chosen pub/sub topic. Every subscriber to that topic then shares
+ * one relay connection instead of each opening its own — filtering here
+ * (rather than at each subscriber) is what keeps an unrelated event from
+ * making every subscriber re-render.
+ *
+ * Ported from console's `startRelay` (src/server/ws.ts) with the match
+ * predicate and target topic lifted to arguments.
+ */
+export function createRelay(
+  cfg: { match: (topic: string) => boolean; topic: string; publish: (topic: string, data: string) => void },
+  opts: RtClientOptions = {},
+): () => void {
+  const doSubscribe = opts.subscribeImpl ?? subscribe;
+  return doSubscribe((type, data) => {
+    if (type !== "event") return;
+    const frame = data as { topic?: unknown };
+    if (typeof frame?.topic !== "string" || !cfg.match(frame.topic)) return;
+    // Serializing outside the catch keeps the suppression narrow: only a
+    // publish that rejects its own frame is expected here, and one
+    // subscriber's broken publish must not tear down the shared relay.
+    const payload = JSON.stringify(data);
+    try {
+      cfg.publish(cfg.topic, payload);
+    } catch {
+      /* the subscriber went away */
+    }
+  }, opts);
+}

packages/rt-client/src/transport.ts

@@ -21,6 +21,13 @@ export interface RtClientOptions {
   wsUrl?: string;
   /** Per-call override of rtCommand's own default (15s); chat's pulse wrapper needs an 800ms hook budget. */
   timeoutMs?: number;
+  /**
+   * Test seam for createRelay (relay.ts): swaps the daemon subscription for
+   * a fake without a live WebSocket server. Typed structurally against
+   * relay.ts's `subscribe` rather than importing its RelayEventType, which
+   * would make this module depend on the one that already depends on it.
+   */
+  subscribeImpl?: (onEvent: (type: string, data: unknown) => void, opts?: RtClientOptions) => () => void;
 }
 
 // Duplicates the ~/.mattstack/rt layout: rt-client has no dependency on rt's

packages/rt-client/test/health.test.ts

@@ -0,0 +1,21 @@
+import { describe, expect, test, afterEach } from "bun:test";
+import { daemonHealth } from "../src/health.ts";
+import { fakeDaemon } from "./fake-daemon.ts";
+
+const stops: Array<() => void> = [];
+afterEach(() => { for (const stop of stops) stop(); stops.length = 0; });
+
+describe("daemonHealth", () => {
+  test("maps an unreachable daemon to reachable:false, never a throw", async () => {
+    const res = await daemonHealth({ sockPath: "/nonexistent/rt.sock" });
+    expect(res).toMatchObject({ reachable: false });
+    expect(res.error).toContain("unreachable");
+  });
+
+  test("maps a reachable daemon's events:head to reachable:true with no error", async () => {
+    const { sock, stop } = fakeDaemon({ "events:head": { ok: true, data: { cursor: 5 } } });
+    stops.push(stop);
+    const res = await daemonHealth({ sockPath: sock });
+    expect(res).toEqual({ reachable: true });
+  });
+});

packages/rt-client/test/relay.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, test, afterEach } from "bun:test";
-import { subscribe } from "../src/relay.ts";
+import { subscribe, createRelay } from "../src/relay.ts";
 
 const stops: Array<() => void> = [];
 afterEach(() => { for (const stop of stops) stop(); stops.length = 0; });
@@ -65,3 +65,94 @@ describe("subscribe", () => {
     expect(opens).toBe(2);
   });
 });
+
+describe("createRelay", () => {
+  test("republishes only event frames whose topic matches, onto the configured topic", () => {
+    const published: Array<[string, string]> = [];
+    const cbs: Array<(type: string, data: unknown) => void> = [];
+    const stop = createRelay(
+      { match: (t) => t.startsWith("chat/"), topic: "chat", publish: (t, d) => published.push([t, d]) },
+      { subscribeImpl: (cb) => { cbs.push(cb); return () => {}; } },
+    );
+    cbs[0]!("event", { topic: "chat/build/msg", payload: { id: 1 } });
+    cbs[0]!("event", { topic: "run-updated", payload: {} });
+    cbs[0]!("ports", {});
+    expect(published).toEqual([["chat", JSON.stringify({ topic: "chat/build/msg", payload: { id: 1 } })]]);
+    stop();
+  });
+
+  test("stop() delegates to the underlying subscription's unsubscribe", () => {
+    let stopped = false;
+    const stop = createRelay(
+      { match: () => true, topic: "chat", publish: () => {} },
+      { subscribeImpl: () => () => { stopped = true; } },
+    );
+    stop();
+    expect(stopped).toBe(true);
+  });
+
+  test("a frame missing a string topic is dropped, not thrown on", () => {
+    const published: Array<[string, string]> = [];
+    const cbs: Array<(type: string, data: unknown) => void> = [];
+    const stop = createRelay(
+      { match: () => true, topic: "chat", publish: (t, d) => published.push([t, d]) },
+      { subscribeImpl: (cb) => { cbs.push(cb); return () => {}; } },
+    );
+    cbs[0]!("event", {});
+    cbs[0]!("event", null);
+    expect(published).toEqual([]);
+    stop();
+  });
+
+  test("defaults subscribeImpl to the real subscribe, wiring through wsUrl", async () => {
+    let sock: Bun.ServerWebSocket<unknown> | null = null;
+    const server = Bun.serve({
+      port: 0,
+      fetch(req, srv) { return srv.upgrade(req) ? undefined : new Response("no", { status: 400 }); },
+      websocket: { open(ws) { sock = ws; }, message() {} },
+    });
+    const published: Array<[string, string]> = [];
+    const stop = createRelay(
+      { match: (t) => t === "chat/build/msg", topic: "chat", publish: (t, d) => published.push([t, d]) },
+      { wsUrl: `ws://127.0.0.1:${server.port}/ws` },
+    );
+    await new Promise<void>((resolve, reject) => {
+      const t0 = Date.now();
+      const poll = () => sock ? resolve() : Date.now() - t0 > 3000 ? reject(new Error("no ws connect")) : setTimeout(poll, 10);
+      poll();
+    });
+    sock!.send(JSON.stringify({ type: "event", data: { topic: "chat/build/msg", payload: { id: 1 } }, timestamp: 1 }));
+    await new Promise<void>((resolve, reject) => {
+      const t0 = Date.now();
+      const poll = () => published.length ? resolve() : Date.now() - t0 > 3000 ? reject(new Error("no publish")) : setTimeout(poll, 10);
+      poll();
+    });
+    expect(published).toEqual([["chat", JSON.stringify({ topic: "chat/build/msg", payload: { id: 1 } })]]);
+    stop();
+    server.stop();
+  });
+
+  test("a publish that throws does not tear down the relay", () => {
+    const seen: string[] = [];
+    let cb!: (type: string, data: unknown) => void;
+    const stop = createRelay(
+      {
+        match: (t) => t.startsWith("chat/"),
+        topic: "chat",
+        publish: (_t, d) => {
+          if (seen.length === 0) {
+            seen.push("threw");
+            throw new Error("subscriber went away");
+          }
+          seen.push(d);
+        },
+      },
+      { subscribeImpl: (fn) => { cb = fn; return () => {}; } },
+    );
+    cb("event", { topic: "chat/build/msg", payload: { id: 1 } });
+    cb("event", { topic: "chat/build/msg", payload: { id: 2 } });
+    expect(seen).toEqual(["threw", JSON.stringify({ topic: "chat/build/msg", payload: { id: 2 } })]);
+    stop();
+  });
+
+});