diff --git a/lib/daemon/__tests__/chat-handlers.test.ts b/lib/daemon/__tests__/chat-handlers.test.ts index 9dadfa60..ac31c76a 100644 --- a/lib/daemon/__tests__/chat-handlers.test.ts +++ b/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" }); diff --git a/lib/daemon/handlers/chat.ts b/lib/daemon/handlers/chat.ts index 29838b1b..67f0209b 100644 --- a/lib/daemon/handlers/chat.ts +++ b/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> => { 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 } }; }, diff --git a/lib/state/chat-store.ts b/lib/state/chat-store.ts index 75f923ad..53ef4f3b 100644 --- a/lib/state/chat-store.ts +++ b/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); diff --git a/lib/state/index.ts b/lib/state/index.ts index d90dc2c5..cabf0246 100644 --- a/lib/state/index.ts +++ b/lib/state/index.ts @@ -107,6 +107,7 @@ export { markRead, unreadWakingCount, listRooms, + roomDefaultWake, listMembers, armMember, touchMember, diff --git a/packages/rt-client/package.json b/packages/rt-client/package.json index 69a661ab..8d262731 100644 --- a/packages/rt-client/package.json +++ b/packages/rt-client/package.json @@ -1,6 +1,6 @@ { "name": "@mattstack/rt-client", - "version": "0.5.0", + "version": "0.6.0", "type": "module", "exports": { ".": { diff --git a/packages/rt-client/src/commands.ts b/packages/rt-client/src/commands.ts index d1de324d..85a0059d 100644 --- a/packages/rt-client/src/commands.ts +++ b/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; } /** diff --git a/packages/rt-client/src/health.ts b/packages/rt-client/src/health.ts new file mode 100644 index 00000000..4bdf83bf --- /dev/null +++ b/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 }; +} diff --git a/packages/rt-client/src/index.ts b/packages/rt-client/src/index.ts index 797ddd79..fb9134ff 100644 --- a/packages/rt-client/src/index.ts +++ b/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) ──────────────────────────────────────────────────────── diff --git a/packages/rt-client/src/relay.ts b/packages/rt-client/src/relay.ts index 04c7bd63..d04b78db 100644 --- a/packages/rt-client/src/relay.ts +++ b/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); +} diff --git a/packages/rt-client/src/transport.ts b/packages/rt-client/src/transport.ts index 82529684..b8243c24 100644 --- a/packages/rt-client/src/transport.ts +++ b/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 diff --git a/packages/rt-client/test/health.test.ts b/packages/rt-client/test/health.test.ts new file mode 100644 index 00000000..acada70e --- /dev/null +++ b/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 }); + }); +}); diff --git a/packages/rt-client/test/relay.test.ts b/packages/rt-client/test/relay.test.ts index 6ea56017..6a6d8617 100644 --- a/packages/rt-client/test/relay.test.ts +++ b/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 | 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((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((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(); + }); + +});