diff --git a/apps/web/src/app.css b/apps/web/src/app.css index 83428f4f3..0b7d386f6 100644 --- a/apps/web/src/app.css +++ b/apps/web/src/app.css @@ -1121,9 +1121,17 @@ select:disabled, .shell-canvas-pane-actions { display: flex; flex-shrink: 0; + align-items: center; gap: 0.25rem; } +.shell-artifact-presence-status { + padding: 0 0.35rem; + font-size: 0.75rem; + color: var(--muted-foreground); + white-space: nowrap; +} + .shell-profile-pane-hint { margin: 0; font-size: 0.75rem; diff --git a/apps/web/src/presence/use-presence-room.ts b/apps/web/src/presence/use-presence-room.ts index 58ba7871d..f72a17d87 100644 --- a/apps/web/src/presence/use-presence-room.ts +++ b/apps/web/src/presence/use-presence-room.ts @@ -5,7 +5,12 @@ // disconnect lifecycle to reason about per (tenant, surface) pair. import { useCallback, useEffect, useRef, useState } from "react"; import type * as Y from "yjs"; -import { connectPresence, type PresenceHandle } from "@corbits/presence/client"; +import { reportError } from "@corbits/error-sink"; +import { + connectPresence, + type PresenceError, + type PresenceHandle, +} from "@corbits/presence/client"; export interface PresenceRoomMember { readonly principalId: string; @@ -19,8 +24,12 @@ export interface PresenceRoomMember { readonly typing?: boolean; } +/** Healthy until two consecutive transport failures; a single retry stays quiet. */ +export type PresenceConnection = "ok" | "degraded"; + export interface PresenceRoom { readonly members: readonly PresenceRoomMember[]; + readonly connection: PresenceConnection; readonly publishCursor: ( x: number, y: number, @@ -40,6 +49,16 @@ export interface UsePresenceRoomOptions { readonly doc?: Y.Doc; /** Called for every confirmed server-side snapshot — see `formatSaveStateLine` in `@corbits/artifact-ui`. */ readonly onSaved?: (info: { version: number; savedAt: number }) => void; + /** The signed-in principal, carried on `reportError` so a failed join is scoped. */ + readonly principalId?: string; +} + +const TRANSIENT_FAILURE_BUDGET = 1; + +function presenceErrorMessage(error: PresenceError): string { + return error.status === undefined + ? `Presence ${error.operation} failed` + : `Presence ${error.operation} failed (${String(error.status)})`; } /** @@ -56,7 +75,9 @@ export function usePresenceRoom( options?: UsePresenceRoomOptions, ): PresenceRoom { const [members, setMembers] = useState([]); + const [connection, setConnection] = useState("ok"); const handleRef = useRef(null); + const consecutiveFailuresRef = useRef(0); // `onSaved` is read through a ref, not a `connectPresence` dependency: // a caller re-rendering with a fresh inline callback must never tear // down and reconnect the stream (it would re-fetch the whole doc state @@ -64,10 +85,14 @@ export function usePresenceRoom( // different room to sync. const onSavedRef = useRef(options?.onSaved); onSavedRef.current = options?.onSaved; + const principalIdRef = useRef(options?.principalId); + principalIdRef.current = options?.principalId; const doc = options?.doc; useEffect(() => { setMembers([]); + setConnection("ok"); + consecutiveFailuresRef.current = 0; if (tenantId === null || surface === null) { handleRef.current = null; return; @@ -88,8 +113,31 @@ export function usePresenceRoom( const unsubscribe = handle.subscribe((snapshot) => setMembers(snapshot as readonly PresenceRoomMember[]), ); + const unsubscribeErrors = handle.onError((error) => { + const extra: Record = {}; + if (principalIdRef.current !== undefined) { + extra.principalId = principalIdRef.current; + } + if (error.status !== undefined) extra.status = error.status; + reportError(new Error(presenceErrorMessage(error)), { + operation: `presence.${error.operation}`, + tenantId, + roomId: surface, + ...(Object.keys(extra).length > 0 ? { extra } : {}), + }); + consecutiveFailuresRef.current += 1; + if (consecutiveFailuresRef.current > TRANSIENT_FAILURE_BUDGET) { + setConnection("degraded"); + } + }); + const unsubscribeRecovered = handle.onRecovered(() => { + consecutiveFailuresRef.current = 0; + setConnection("ok"); + }); return () => { unsubscribe(); + unsubscribeErrors(); + unsubscribeRecovered(); handle.disconnect(); handleRef.current = null; }; @@ -108,5 +156,5 @@ export function usePresenceRoom( handleRef.current?.publishTyping(typing); }, []); - return { members, publishCursor, publishTyping }; + return { members, connection, publishCursor, publishTyping }; } diff --git a/apps/web/src/shell/app-shell.tsx b/apps/web/src/shell/app-shell.tsx index 7401589ae..05756d4ab 100644 --- a/apps/web/src/shell/app-shell.tsx +++ b/apps/web/src/shell/app-shell.tsx @@ -123,17 +123,20 @@ export function AppShell({ tenantId, canvasArtifact === null ? null : `artifact:${canvasArtifact.id}`, undefined, - artifactDoc === null - ? undefined - : { - doc: artifactDoc, - onSaved: (info) => - setArtifactSaveState({ - kind: "saved", - version: info.version, - savedAt: info.savedAt, - }), - }, + { + ...(viewerPrincipalId === null ? {} : { principalId: viewerPrincipalId }), + ...(artifactDoc === null + ? {} + : { + doc: artifactDoc, + onSaved: (info: { version: number; savedAt: number }) => + setArtifactSaveState({ + kind: "saved", + version: info.version, + savedAt: info.savedAt, + }), + }), + }, ); const editingCoworkers = artifactPresence.members .filter( @@ -213,6 +216,7 @@ export function AppShell({ {...(artifactDoc !== null ? { artifactDoc } : {})} artifactSaveState={artifactSaveStateWithEditors} onArtifactTyping={artifactPresence.publishTyping} + presenceConnection={artifactPresence.connection} /> ) : null} diff --git a/apps/web/src/shell/canvas-column.tsx b/apps/web/src/shell/canvas-column.tsx index 26478a5ce..7733d26cb 100644 --- a/apps/web/src/shell/canvas-column.tsx +++ b/apps/web/src/shell/canvas-column.tsx @@ -52,6 +52,7 @@ import type { ReactNode } from "react"; import type * as Y from "yjs"; import { useBench } from "../bench-context"; +import type { PresenceConnection } from "../presence/use-presence-room"; import { workbenchPath } from "../workbench-path"; import { ensureProfileDm, loadSharedWorkbenches } from "../profile-relations"; import type { @@ -90,6 +91,7 @@ export function CanvasColumn({ artifactDoc, artifactSaveState, onArtifactTyping, + presenceConnection = "ok", }: { readonly open: boolean; readonly profile: ProfileSubject | null; @@ -117,6 +119,8 @@ export function CanvasColumn({ readonly artifactSaveState?: ArtifactSaveState; /** Fired on local typing start/stop in the text editor, for the host to publish through presence's `typing` awareness field. */ readonly onArtifactTyping?: (typing: boolean) => void; + /** Quiet reconnecting caption when presence has failed more than once. */ + readonly presenceConnection?: PresenceConnection; }) { // `inert` rather than `aria-hidden`: a collapsed column has to be out of // both the accessibility tree and the tab order, and `aria-hidden` alone @@ -150,6 +154,7 @@ export function CanvasColumn({ {...(artifactDoc !== undefined ? { artifactDoc } : {})} {...(artifactSaveState !== undefined ? { artifactSaveState } : {})} {...(onArtifactTyping !== undefined ? { onArtifactTyping } : {})} + presenceConnection={presenceConnection} /> ) : routine !== null ? ( @@ -511,6 +516,7 @@ function ArtifactCanvasPane({ artifactDoc, artifactSaveState = { kind: "read-only" }, onArtifactTyping, + presenceConnection = "ok", }: { readonly artifact: CanvasArtifactContent; readonly focus: boolean; @@ -521,6 +527,7 @@ function ArtifactCanvasPane({ readonly artifactDoc?: Y.Doc; readonly artifactSaveState?: ArtifactSaveState; readonly onArtifactTyping?: (typing: boolean) => void; + readonly presenceConnection?: PresenceConnection; }) { const showEditor = showsTextEditor(artifact, artifactDoc); return ( @@ -533,6 +540,18 @@ function ArtifactCanvasPane({ {...(artifact.previewSrc !== undefined ? { previewSrc: artifact.previewSrc } : {})} + {...(presenceConnection === "degraded" + ? { + trailing: ( + + Reconnecting… + + ), + } + : {})} />
{ readonly canEdit?: boolean; }; readonly artifactDoc?: import("yjs").Doc; + readonly presenceConnection?: "ok" | "degraded"; }): Promise { await act(async () => { root.render( @@ -247,6 +248,9 @@ describe("canvas artifact pane: co-editing (CL-5958 phase 2)", () => { {...(props.artifactDoc !== undefined ? { artifactDoc: props.artifactDoc } : {})} + {...(props.presenceConnection !== undefined + ? { presenceConnection: props.presenceConnection } + : {})} /> @@ -359,4 +363,32 @@ describe("canvas artifact pane: co-editing (CL-5958 phase 2)", () => { expect(container.querySelector("textarea")).toBeNull(); expect(container.textContent).toContain("fetched content"); }); + + test("a degraded presence connection shows a quiet reconnecting caption", async () => { + await renderArtifact({ + artifact: { + id: "art_6", + title: "Notes", + rendererKind: "doc", + content: "fetched content", + }, + presenceConnection: "degraded", + }); + + expect(container.textContent).toContain("Reconnecting…"); + }); + + test("a healthy presence connection shows no reconnecting caption", async () => { + await renderArtifact({ + artifact: { + id: "art_7", + title: "Notes", + rendererKind: "doc", + content: "fetched content", + }, + presenceConnection: "ok", + }); + + expect(container.textContent).not.toContain("Reconnecting…"); + }); }); diff --git a/apps/web/test/use-presence-room.test.tsx b/apps/web/test/use-presence-room.test.tsx index dafe9d01e..63cf66b5d 100644 --- a/apps/web/test/use-presence-room.test.tsx +++ b/apps/web/test/use-presence-room.test.tsx @@ -3,21 +3,35 @@ // either changes, and exposes publishCursor/publishTyping as stable // callbacks. Mocks `@corbits/presence/client` directly rather than a real // EventSource/fetch — that transport is `@corbits/presence`'s own test -// responsibility, not apps/web's. +// responsibility, not apps/web's. CL-7228: also wires `onError`/`onRecovered` +// so a degraded connection is visible after more than one retry. -import { afterEach, describe, expect, mock, test } from "bun:test"; +import { afterAll, afterEach, describe, expect, mock, test } from "bun:test"; import { act, createElement, useRef } from "react"; import { createRoot } from "react-dom/client"; import type { Root } from "react-dom/client"; +import type { PresenceError } from "@corbits/presence/client"; +import type { UsePresenceRoomOptions } from "../src/presence/use-presence-room"; interface FakeHandle { roomUrl: string; disconnected: boolean; listener: ((members: unknown[]) => void) | null; + errorListeners: Set<(error: PresenceError) => void>; + recoveredListeners: Set<() => void>; cursorCalls: unknown[]; } const created: FakeHandle[] = []; +const reportErrorCalls: { error: unknown; context: Record }[] = + []; + +mock.module("@corbits/error-sink", () => ({ + reportError: (error: unknown, context: Record) => { + reportErrorCalls.push({ error, context }); + return "presence-error"; + }, +})); mock.module("@corbits/presence/client", () => ({ connectPresence: (options: { roomUrl: string }) => { @@ -25,6 +39,8 @@ mock.module("@corbits/presence/client", () => ({ roomUrl: options.roomUrl, disconnected: false, listener: null, + errorListeners: new Set(), + recoveredListeners: new Set(), cursorCalls: [], }; created.push(handle); @@ -36,6 +52,18 @@ mock.module("@corbits/presence/client", () => ({ handle.listener = null; }; }, + onError: (listener: (error: PresenceError) => void) => { + handle.errorListeners.add(listener); + return () => { + handle.errorListeners.delete(listener); + }; + }, + onRecovered: (listener: () => void) => { + handle.recoveredListeners.add(listener); + return () => { + handle.recoveredListeners.delete(listener); + }; + }, publishCursor: (cursor: unknown) => handle.cursorCalls.push(cursor), publishTyping: () => undefined, disconnect: () => { @@ -49,11 +77,33 @@ const { usePresenceRoom } = await import("../src/presence/use-presence-room"); afterEach(() => { created.length = 0; + reportErrorCalls.length = 0; +}); + +afterAll(() => { + mock.restore(); }); +function fireError(error: PresenceError): void { + const handle = created[created.length - 1]; + if (handle === undefined) throw new Error("no presence handle"); + act(() => { + for (const listener of handle.errorListeners) listener(error); + }); +} + +function fireRecovered(): void { + const handle = created[created.length - 1]; + if (handle === undefined) throw new Error("no presence handle"); + act(() => { + for (const listener of handle.recoveredListeners) listener(); + }); +} + function mountHook( initialTenantId: string | null, initialSurface: string | null, + options?: UsePresenceRoomOptions, ) { const container = document.createElement("div"); document.body.appendChild(container); @@ -71,7 +121,7 @@ function mountHook( tenantId: string | null; surface: string | null; }) { - latest = usePresenceRoom(tenantId, surface); + latest = usePresenceRoom(tenantId, surface, undefined, options); return null; } @@ -141,4 +191,67 @@ describe("usePresenceRoom", () => { harness.unmount(); expect(created[0]?.disconnected).toBe(true); }); + + test("stays ok through a single transient error", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + + expect(harness.result()?.connection).toBe("ok"); + expect(reportErrorCalls).toHaveLength(1); + harness.unmount(); + }); + + test("marks the connection degraded after a second consecutive error", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + fireError({ operation: "join" }); + + expect(harness.result()?.connection).toBe("degraded"); + expect(reportErrorCalls).toHaveLength(2); + expect(reportErrorCalls[1]?.context).toEqual({ + operation: "presence.join", + tenantId: "tnt_1", + roomId: "artifact:art_1", + extra: { principalId: "prn_1" }, + }); + harness.unmount(); + }); + + test("reports every error with room and principal context", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "heartbeat", status: 404 }); + + expect(reportErrorCalls).toHaveLength(1); + const call = reportErrorCalls[0]; + expect(call?.error).toBeInstanceOf(Error); + expect((call?.error as Error).message).toBe( + "Presence heartbeat failed (404)", + ); + expect(call?.context).toEqual({ + operation: "presence.heartbeat", + tenantId: "tnt_1", + roomId: "artifact:art_1", + extra: { principalId: "prn_1", status: 404 }, + }); + harness.unmount(); + }); + + test("clears degraded state on successful rejoin", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + fireError({ operation: "join", status: 500 }); + expect(harness.result()?.connection).toBe("degraded"); + + fireRecovered(); + expect(harness.result()?.connection).toBe("ok"); + harness.unmount(); + }); }); diff --git a/docs/presence.md b/docs/presence.md index 2836b8f76..9b26e40d8 100644 --- a/docs/presence.md +++ b/docs/presence.md @@ -102,20 +102,31 @@ dependency-injectable for testing. `apps/web/src/presence/use-presence-room.ts` is the thin React hook that wraps it — the only place apps/web talks to the presence package directly. Two composition sites use it today: -- `apps/web/src/pages/chat-page.tsx` connects to `workbench:` and - hands the snapshot to `@corbits/chat-ui`'s `ChatWorkspace` as - `presenceMembers`, rendered as a live avatar stack beside the workbench - header's static participant list. +- `apps/web/src/pages/chat-page.tsx` used to connect to `workbench:` + and hand the snapshot to `@corbits/chat-ui`'s `ChatWorkspace` as + `presenceMembers`. That roster now rides the workbench `/stream` + (`chat.presence` / `chat.presence.snapshot`, CL-6328); `@corbits/presence` + no longer backs the header stack. - `apps/web/src/shell/app-shell.tsx` connects to `artifact:` when the canvas has one open, publishes the pointer's fractional position over the artifact pane as `cursor`, and hands the room's snapshot to `CanvasColumn` as `presenceCursors` — colored, labeled dots overlaid on - the read-only artifact renderer. + the artifact renderer. + +The SSE stream opens and stays open regardless of whether join or a later +heartbeat actually succeeded, so a healthy-looking cursor overlay is not +proof the client is in the room. `PresenceHandle.onError` reports every +failed join/heartbeat/leave/update; `onRecovered` fires on a successful +join or heartbeat. `usePresenceRoom` subscribes to both: errors go through +`reportError` with tenant, room (`surface`), and principal context, and the +canvas pane shows a quiet "Reconnecting…" caption only after a second +consecutive failure. A single transient retry stays silent; a successful +rejoin clears the caption. Neither `@corbits/chat-ui` nor `apps/web/src/shell/canvas-column.tsx` imports `@corbits/presence` itself — both only take plain data -(`PresenceMember`, `PresenceCursor`) as props, the same way they take any -other host-supplied data. +(`PresenceMember`, `PresenceCursor`, and a `PresenceConnection` flag) as +props, the same way they take any other host-supplied data. ## Phase 2: co-editing diff --git a/packages/presence/src/client.test.ts b/packages/presence/src/client.test.ts index c864caf4e..2da35e5bb 100644 --- a/packages/presence/src/client.test.ts +++ b/packages/presence/src/client.test.ts @@ -490,6 +490,93 @@ describe("connectPresence: error reporting and rejoin (CL-7202)", () => { handle.disconnect(); }); + test("a successful join is reported through onRecovered", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: fakeFetch([]), + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + + await flushMicrotasks(); + + expect(recovered).toBe(1); + handle.disconnect(); + }); + + test("a failed join is not reported through onRecovered", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: scriptedFetch([], () => 500), + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + + await flushMicrotasks(); + + expect(recovered).toBe(0); + handle.disconnect(); + }); + + test("a successful heartbeat after a failed one is reported through onRecovered", async () => { + let heartbeatStatus = 500; + let recovered = 0; + const fetchImpl: PresenceFetch = (url) => { + const status = url.endsWith("/heartbeat") ? heartbeatStatus : 200; + return Promise.resolve({ + ok: status >= 200 && status < 300, + status, + json: () => Promise.resolve({}), + }); + }; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl, + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + await flushMicrotasks(); + expect(recovered).toBe(1); + + handle.publishTyping(true); + await flushMicrotasks(); + expect(recovered).toBe(1); + + heartbeatStatus = 200; + handle.publishTyping(false); + await flushMicrotasks(); + expect(recovered).toBe(2); + handle.disconnect(); + }); + + test("onRecovered's unsubscribe stops further delivery", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: fakeFetch([]), + openEventSource: () => new FakeEventSource(), + }); + const unsubscribe = handle.onRecovered(() => { + recovered += 1; + }); + await flushMicrotasks(); + expect(recovered).toBe(1); + unsubscribe(); + + handle.publishTyping(true); + await flushMicrotasks(); + expect(recovered).toBe(1); + handle.disconnect(); + }); + test("a client stuck unable to join backs off instead of re-posting on every publish call", async () => { let clock = 0; const calls: string[] = []; diff --git a/packages/presence/src/client.ts b/packages/presence/src/client.ts index 0dbb6ecde..b0026cc66 100644 --- a/packages/presence/src/client.ts +++ b/packages/presence/src/client.ts @@ -112,6 +112,14 @@ export interface PresenceHandle { subscribe(listener: (members: readonly PresenceState[]) => void): () => void; /** Subscribes to every failed join/heartbeat/leave/update request. */ onError(listener: (error: PresenceError) => void): () => void; + /** + * Subscribes to every successful join or heartbeat — the client's only + * signals that the connection is healthy again after failures. A consuming + * UI that showed a reconnecting/degraded caption from `onError` clears it + * here, not from the SSE snapshot (the stream stays open even while join + * is failing). + */ + onRecovered(listener: () => void): () => void; /** Leaves the room and tears down the stream/heartbeat timer. */ disconnect(): void; } @@ -220,6 +228,7 @@ export function connectPresence( const listeners = new Set<(members: readonly PresenceState[]) => void>(); const errorListeners = new Set<(error: PresenceError) => void>(); + const recoveredListeners = new Set<() => void>(); let latestMembers: readonly PresenceState[] = []; let disconnected = false; // Whether the last join attempt is known to have succeeded server-side. @@ -258,6 +267,10 @@ export function connectPresence( for (const listener of errorListeners) listener(error); }; + const notifyRecovered = () => { + for (const listener of recoveredListeners) listener(); + }; + const post = (operation: PresenceOperation, body: unknown) => fetchImpl(`${options.roomUrl}/${operation}`, { method: "POST", @@ -296,6 +309,7 @@ export function connectPresence( } joinFailureCount = 0; joined = true; + notifyRecovered(); flushPendingUpdates(); if (doc === undefined) return; void response.json().then((body) => { @@ -332,6 +346,7 @@ export function connectPresence( } return; } + notifyRecovered(); flushPendingUpdates(); }); } @@ -452,6 +467,13 @@ export function connectPresence( }; }, + onRecovered(listener) { + recoveredListeners.add(listener); + return () => { + recoveredListeners.delete(listener); + }; + }, + disconnect() { if (disconnected) return; disconnected = true; @@ -459,6 +481,7 @@ export function connectPresence( source.close(); listeners.clear(); errorListeners.clear(); + recoveredListeners.clear(); if (doc !== undefined && onLocalDocUpdate !== undefined) { doc.off("update", onLocalDocUpdate); }