feat: improve collaboration rooms (#1338)

src/backend/hosts/collab/room-hub.ts

@@ -1,4 +1,7 @@
 import type { WebSocket } from "ws";
+import { collabRuntimeStore } from "./runtime-store.js";
+
+const PRESENCE_HEARTBEAT_MS = 15_000;
 
 export interface CollabRoomClient {
   ws: WebSocket;
@@ -7,13 +10,28 @@ export interface CollabRoomClient {
 }
 
 /**
- * In-memory fan-out for collab room events, mirroring the single-instance
- * assumption TerminalSessionManager already makes. REST mutations broadcast
- * through it; the terminal WS server feeds subscribe/unsubscribe.
+ * Local WebSocket fan-out plus optional Redis pub/sub for multi-instance room
+ * events and presence. The terminal WS server feeds subscribe/unsubscribe.
  */
 class CollabRoomHub {
   private rooms = new Map<string, Set<CollabRoomClient>>();
 
+  constructor() {
+    collabRuntimeStore.onEvent((roomId, message) => {
+      if (
+        "type" in message &&
+        typeof message.type === "string" &&
+        message.type.startsWith("collab_internal_")
+      ) {
+        return;
+      }
+      this.broadcastLocal(roomId, message);
+    });
+    setInterval(() => {
+      for (const roomId of this.rooms.keys()) void this.refreshPresence(roomId);
+    }, PRESENCE_HEARTBEAT_MS).unref();
+  }
+
   subscribe(roomId: string, client: CollabRoomClient): void {
     let clients = this.rooms.get(roomId);
     if (!clients) {
@@ -24,7 +42,7 @@ class CollabRoomHub {
       if (existing.ws === client.ws) return;
     }
     clients.add(client);
-    this.broadcastOnline(roomId);
+    void this.refreshPresence(roomId);
   }
 
   /** Drops the socket from one room, or from every room when roomId is omitted. */
@@ -39,11 +57,25 @@ class CollabRoomHub {
         }
       }
       if (clients.size === 0) this.rooms.delete(id);
-      if (removed) this.broadcastOnline(id);
+      if (removed) void this.refreshPresence(id);
     }
   }
 
   broadcast(roomId: string, message: object): void {
+    this.broadcastLocal(roomId, message);
+    void collabRuntimeStore.publish(roomId, message);
+  }
+
+  async onlineUsers(
+    roomId: string,
+  ): Promise<Array<{ userId: string; username: string }>> {
+    return collabRuntimeStore.onlineUsers(
+      roomId,
+      this.localOnlineUsers(roomId),
+    );
+  }
+
+  private broadcastLocal(roomId: string, message: object): void {
     const clients = this.rooms.get(roomId);
     if (!clients) return;
     const payload = JSON.stringify(message);
@@ -57,7 +89,9 @@ class CollabRoomHub {
     }
   }
 
-  onlineUsers(roomId: string): Array<{ userId: string; username: string }> {
+  private localOnlineUsers(
+    roomId: string,
+  ): Array<{ userId: string; username: string }> {
     const clients = this.rooms.get(roomId);
     if (!clients) return [];
     const seen = new Map<string, string>();
@@ -67,11 +101,13 @@ class CollabRoomHub {
     return Array.from(seen, ([userId, username]) => ({ userId, username }));
   }
 
-  private broadcastOnline(roomId: string): void {
+  private async refreshPresence(roomId: string): Promise<void> {
+    const localUsers = this.localOnlineUsers(roomId);
+    await collabRuntimeStore.updatePresence(roomId, localUsers);
     this.broadcast(roomId, {
       type: "collab_online",
       roomId,
-      users: this.onlineUsers(roomId),
+      users: await this.onlineUsers(roomId),
     });
   }
 }

src/backend/hosts/collab/routes.ts

@@ -12,6 +12,10 @@ import { GuacamoleTokenService } from "../guacamole/token-service.js";
 import { collabRoomHub } from "./room-hub.js";
 import { getStageController, setStageController } from "./stage-control.js";
 import { isCollabGuestRateLimited } from "./guest-rate-limit.js";
+import {
+  collabRuntimeStore,
+  type CollabControlRequest,
+} from "./runtime-store.js";
 import { sessionManager } from "../terminal/session-manager.js";
 import {
   isLiveSession,
@@ -29,9 +33,8 @@ import type { CollabRoomRecord } from "../../database/repositories/collab-room-r
 
 /*
  * Known limits, shared with session sharing v1:
- * - Room events and stage control live in this process (room-hub,
- *   stage-control). With more than one backend instance, members connected
- *   to different instances do not see each other's events.
+ * - Redis synchronizes collaboration events and ephemeral state across
+ *   instances, but live session transports still require WebSocket affinity.
  * - Guacamole stages stay read-only because guacamole-lite cannot revoke a
  *   writable viewer without disconnecting the whole shared session.
  */
@@ -44,7 +47,6 @@ const STAGE_SHARE_EXPIRY_HOURS = 12;
 const MAX_INVITE_TARGETS = 200;
 const CONTROL_REQUEST_COOLDOWN_MS = 5000;
 const PROTOCOLS: LiveProtocol[] = ["ssh", "rdp", "vnc", "telnet"];
-const controlRequestTimes = new Map<string, number>();
 
 function isNonEmptyString(value: unknown): value is string {
   return typeof value === "string" && value.trim().length > 0;
@@ -74,6 +76,11 @@ async function revokeStageShare(room: CollabRoomRecord): Promise<void> {
     sessionManager.disconnectShareParticipants(share.sessionId, share.id, {
       reason: "The collaboration stage ended",
     });
+    void collabRuntimeStore.publish(room.id, {
+      type: "collab_internal_stage_revoked",
+      sessionId: share.sessionId,
+      shareId: share.id,
+    });
   }
 }
 
@@ -186,14 +193,20 @@ router.get(
       }
       const members =
         await createCurrentCollabRoomRepository().listMembers(roomId);
+      const mayReviewControlRequests =
+        access.isHost || access.room.presenterUserId === userId;
+      const controlRequests = await collabRuntimeStore.listRequests(roomId);
       res.json({
         room: publicRoom(access.room),
         me: userId,
         isHost: access.isHost,
         members,
-        online: collabRoomHub.onlineUsers(roomId),
+        online: await collabRoomHub.onlineUsers(roomId),
         stage: stagePayload(access.room),
-        controllerUserId: getStageController(roomId),
+        controllerUserId: await getStageController(roomId),
+        controlRequests: mayReviewControlRequests
+          ? controlRequests
+          : controlRequests.filter((request) => request.userId === userId),
       });
     } catch (error) {
       sshLogger.error("Failed to get collab room", error, {
@@ -332,6 +345,7 @@ router.delete(
 
       const repository = createCurrentCollabRoomRepository();
       await repository.removeMember(roomId, targetId);
+      await collabRuntimeStore.removeRequest(roomId, targetId);
 
       if (access.room.stageShareId && access.room.stageProtocol === "ssh") {
         const share =
@@ -350,7 +364,7 @@ router.delete(
         }
       }
 
-      if (getStageController(roomId) === targetId) {
+      if ((await getStageController(roomId)) === targetId) {
         await applyStageControl(access.room, roomId, null);
       }
 
@@ -455,7 +469,6 @@ router.post(
       });
 
       await revokeStageShare(access.room);
-      setStageController(roomId, null);
       const repository = createCurrentCollabRoomRepository();
       const replaced = await repository.replaceStage(
         roomId,
@@ -471,6 +484,8 @@ router.post(
         await shareRepository.revokeAsAdmin(share.id);
         return res.status(409).json({ error: "The stage changed; try again" });
       }
+      await setStageController(roomId, null);
+      await collabRuntimeStore.clearRequests(roomId);
 
       const stage = {
         presenterUserId: userId,
@@ -534,7 +549,8 @@ router.post(
       }
 
       await revokeStageShare(access.room);
-      setStageController(roomId, null);
+      await setStageController(roomId, null);
+      await collabRuntimeStore.clearRequests(roomId);
       await createCurrentCollabRoomRepository().clearStage(roomId);
       collabRoomHub.broadcast(roomId, {
         type: "collab_stage_changed",
@@ -595,7 +611,7 @@ router.get(
       if (!share || !isLiveSession(protocol, share.sessionId)) {
         // The presenter is gone (expired share or dead session): clear the
         // stale stage so the room stops pointing at it.
-        setStageController(roomId, null);
+        await setStageController(roomId, null);
         await createCurrentCollabRoomRepository().clearStage(roomId);
         collabRoomHub.broadcast(roomId, {
           type: "collab_stage_changed",
@@ -605,7 +621,7 @@ router.get(
         return res.json({ stage: null });
       }
 
-      const controllerUserId = getStageController(roomId);
+      const controllerUserId = await getStageController(roomId);
       const stage: Record<string, unknown> = {
         ...stagePayload(room),
         sessionId: share.sessionId,
@@ -632,7 +648,7 @@ async function applyStageControl(
   roomId: string,
   controllerUserId: string | null,
 ): Promise<void> {
-  setStageController(roomId, controllerUserId);
+  await setStageController(roomId, controllerUserId);
   if (room.stageShareId && room.stageProtocol === "ssh") {
     try {
       const share = await createCurrentSessionShareRepository().findActiveById(
@@ -656,6 +672,49 @@ async function applyStageControl(
   });
 }
 
+collabRuntimeStore.onEvent((roomId, message) => {
+  if (!("type" in message) || typeof message.type !== "string") return;
+  if (message.type === "collab_control_changed") {
+    const controllerUserId: string | null =
+      "controllerUserId" in message &&
+      typeof message.controllerUserId === "string"
+        ? message.controllerUserId
+        : null;
+    void createCurrentCollabRoomRepository()
+      .findById(roomId)
+      .then((room) => {
+        if (!room?.stageShareId || room.stageProtocol !== "ssh") return;
+        return createCurrentSessionShareRepository()
+          .findActiveById(room.stageShareId)
+          .then((share) => {
+            if (share) {
+              sessionManager.setRoomShareControl(
+                share.sessionId,
+                share.id,
+                controllerUserId,
+              );
+            }
+          });
+      })
+      .catch(() => {});
+  }
+  if (
+    message.type === "collab_internal_stage_revoked" &&
+    "sessionId" in message &&
+    typeof message.sessionId === "string" &&
+    "shareId" in message &&
+    typeof message.shareId === "string"
+  ) {
+    sessionManager.disconnectShareParticipants(
+      message.sessionId,
+      message.shareId,
+      {
+        reason: "The collaboration stage ended",
+      },
+    );
+  }
+});
+
 /**
  * @openapi
  * /collab/rooms/{id}/control:
@@ -694,7 +753,7 @@ router.post(
       }
 
       const releasingOwnControl =
-        targetId === null && getStageController(roomId) === userId;
+        targetId === null && (await getStageController(roomId)) === userId;
       const mayGrant = access.isHost || access.room.presenterUserId === userId;
       if (!mayGrant && !releasingOwnControl) {
         return res.status(403).json({
@@ -710,6 +769,13 @@ router.post(
       }
 
       await applyStageControl(access.room, roomId, targetId);
+      if (targetId) {
+        await collabRuntimeStore.removeRequest(roomId, targetId);
+        collabRoomHub.broadcast(roomId, {
+          type: "collab_control_requests_changed",
+          roomId,
+        });
+      }
       const { ipAddress, userAgent } = getRequestMeta(req);
       await logAudit({
         userId,
@@ -760,20 +826,26 @@ router.post(
           error: "Remote desktop stages are read-only",
         });
       }
-      const requestKey = `${roomId}:${userId}`;
-      const now = Date.now();
+      const existing = (await collabRuntimeStore.listRequests(roomId)).find(
+        (request) => request.userId === userId,
+      );
       if (
-        now - (controlRequestTimes.get(requestKey) ?? 0) <
-        CONTROL_REQUEST_COOLDOWN_MS
+        existing &&
+        Date.now() - Date.parse(existing.requestedAt) <
+          CONTROL_REQUEST_COOLDOWN_MS
       ) {
         return res.status(429).json({ error: "Control was already requested" });
       }
-      controlRequestTimes.set(requestKey, now);
+      const request: CollabControlRequest = {
+        userId,
+        username: await getAuditUsername(userId),
+        requestedAt: new Date().toISOString(),
+      };
+      await collabRuntimeStore.upsertRequest(roomId, request);
       collabRoomHub.broadcast(roomId, {
         type: "collab_control_requested",
         roomId,
-        userId,
-        username: await getAuditUsername(userId),
+        ...request,
       });
       const { ipAddress, userAgent } = getRequestMeta(req);
       await logAudit({
@@ -787,7 +859,7 @@ router.post(
         userAgent,
         success: true,
       });
-      res.json({ success: true });
+      res.json({ request });
     } catch (error) {
       sshLogger.error("Failed to request collab stage control", error, {
         operation: "collab_control_request_error",
@@ -797,6 +869,77 @@ router.post(
   },
 );
 
+router.get(
+  "/rooms/:id/control/requests",
+  authenticateJWT,
+  async (req: Request, res: Response) => {
+    const userId = (req as AuthenticatedRequest).userId!;
+    const roomId = String(req.params.id);
+    try {
+      const access = await requireRoomMember(roomId, userId);
+      if (!access) return res.status(404).json({ error: "Room not found" });
+      if (!access.isHost && access.room.presenterUserId !== userId) {
+        return res.status(403).json({
+          error: "Only the presenter or host can review control requests",
+        });
+      }
+      res.json({ requests: await collabRuntimeStore.listRequests(roomId) });
+    } catch (error) {
+      sshLogger.error("Failed to list collab control requests", error, {
+        operation: "collab_control_requests_list_error",
+      });
+      res.status(500).json({ error: "Failed to list control requests" });
+    }
+  },
+);
+
+router.delete(
+  "/rooms/:id/control/requests/:userId",
+  authenticateJWT,
+  async (req: Request, res: Response) => {
+    const userId = (req as AuthenticatedRequest).userId!;
+    const roomId = String(req.params.id);
+    const targetId = String(req.params.userId);
+    try {
+      const access = await requireRoomMember(roomId, userId);
+      if (!access) return res.status(404).json({ error: "Room not found" });
+      const mayReview = access.isHost || access.room.presenterUserId === userId;
+      if (!mayReview && targetId !== userId) {
+        return res
+          .status(403)
+          .json({ error: "Not allowed to dismiss request" });
+      }
+      await collabRuntimeStore.removeRequest(roomId, targetId);
+      collabRoomHub.broadcast(roomId, {
+        type: "collab_control_requests_changed",
+        roomId,
+      });
+      const { ipAddress, userAgent } = getRequestMeta(req);
+      await logAudit({
+        userId,
+        username: await getAuditUsername(userId),
+        action:
+          targetId === userId
+            ? "collab_control_request_cancel"
+            : "collab_control_request_dismiss",
+        resourceType: "collab_room",
+        resourceId: roomId,
+        resourceName: access.room.name,
+        details: JSON.stringify({ targetUserId: targetId }),
+        ipAddress,
+        userAgent,
+        success: true,
+      });
+      res.json({ success: true });
+    } catch (error) {
+      sshLogger.error("Failed to dismiss collab control request", error, {
+        operation: "collab_control_request_dismiss_error",
+      });
+      res.status(500).json({ error: "Failed to dismiss control request" });
+    }
+  },
+);
+
 /**
  * @openapi
  * /collab/rooms/{id}/guest-link:
@@ -961,7 +1104,8 @@ router.post(
       }
 
       await revokeStageShare(access.room);
-      setStageController(roomId, null);
+      await setStageController(roomId, null);
+      await collabRuntimeStore.clearRequests(roomId);
       const repository = createCurrentCollabRoomRepository();
       if (access.room.persistent) {
         await repository.setGuestToken(roomId, null);

src/backend/hosts/collab/runtime-store.ts

@@ -0,0 +1,286 @@
+import crypto from "crypto";
+import { createClient } from "redis";
+import { sshLogger } from "../../utils/logger.js";
+
+export interface CollabControlRequest {
+  userId: string;
+  username: string;
+  requestedAt: string;
+}
+
+interface PresenceEntry {
+  instanceId: string;
+  userId: string;
+  username: string;
+}
+
+interface EventEnvelope {
+  source: string;
+  roomId: string;
+  message: object;
+}
+
+const KEY_PREFIX = process.env.TERMIX_REDIS_PREFIX?.trim() || "termix:collab";
+const EVENT_CHANNEL = `${KEY_PREFIX}:events`;
+const PRESENCE_TTL_MS = 45_000;
+const STATE_TTL_SECONDS = 12 * 60 * 60;
+const CONNECT_RETRY_MS = 15_000;
+
+export class CollabRuntimeStore {
+  private readonly instanceId = crypto.randomUUID();
+  private publisher: ReturnType<typeof createClient> | null = null;
+  private subscriber: ReturnType<typeof createClient> | null = null;
+  private connecting: Promise<boolean> | null = null;
+  private nextConnectAttempt = 0;
+  private eventListeners = new Set<(roomId: string, message: object) => void>();
+  private localControllers = new Map<string, string>();
+  private localRequests = new Map<string, Map<string, CollabControlRequest>>();
+  private presenceMembers = new Map<string, Set<string>>();
+
+  onEvent(listener: (roomId: string, message: object) => void): void {
+    this.eventListeners.add(listener);
+    void this.ensureConnected();
+  }
+
+  async publish(roomId: string, message: object): Promise<void> {
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    await this.publisher
+      .publish(
+        EVENT_CHANNEL,
+        JSON.stringify({ source: this.instanceId, roomId, message }),
+      )
+      .catch((error) => this.logRedisFailure("publish", error));
+  }
+
+  async getController(roomId: string): Promise<string | null> {
+    if (await this.ensureConnected()) {
+      const value = await this.publisher
+        ?.get(`${KEY_PREFIX}:controller:${roomId}`)
+        .catch((error) => {
+          this.logRedisFailure("get_controller", error);
+          return undefined;
+        });
+      if (value !== undefined) return value;
+    }
+    return this.localControllers.get(roomId) ?? null;
+  }
+
+  async setController(roomId: string, userId: string | null): Promise<void> {
+    if (userId) this.localControllers.set(roomId, userId);
+    else this.localControllers.delete(roomId);
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    const key = `${KEY_PREFIX}:controller:${roomId}`;
+    const operation = userId
+      ? this.publisher.set(key, userId, { EX: STATE_TTL_SECONDS })
+      : this.publisher.del(key);
+    await operation.catch((error) =>
+      this.logRedisFailure("set_controller", error),
+    );
+  }
+
+  async listRequests(roomId: string): Promise<CollabControlRequest[]> {
+    if (await this.ensureConnected()) {
+      const values = await this.publisher
+        ?.hVals(`${KEY_PREFIX}:requests:${roomId}`)
+        .catch((error) => {
+          this.logRedisFailure("list_requests", error);
+          return null;
+        });
+      if (values) {
+        return values
+          .flatMap((value) => this.parse<CollabControlRequest>(value) ?? [])
+          .sort((a, b) => a.requestedAt.localeCompare(b.requestedAt));
+      }
+    }
+    return Array.from(this.localRequests.get(roomId)?.values() ?? []).sort(
+      (a, b) => a.requestedAt.localeCompare(b.requestedAt),
+    );
+  }
+
+  async upsertRequest(
+    roomId: string,
+    request: CollabControlRequest,
+  ): Promise<void> {
+    let requests = this.localRequests.get(roomId);
+    if (!requests) {
+      requests = new Map();
+      this.localRequests.set(roomId, requests);
+    }
+    requests.set(request.userId, request);
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    const key = `${KEY_PREFIX}:requests:${roomId}`;
+    await this.publisher
+      .multi()
+      .hSet(key, request.userId, JSON.stringify(request))
+      .expire(key, STATE_TTL_SECONDS)
+      .exec()
+      .catch((error) => this.logRedisFailure("upsert_request", error));
+  }
+
+  async removeRequest(roomId: string, userId: string): Promise<void> {
+    const requests = this.localRequests.get(roomId);
+    requests?.delete(userId);
+    if (requests?.size === 0) this.localRequests.delete(roomId);
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    await this.publisher
+      .hDel(`${KEY_PREFIX}:requests:${roomId}`, userId)
+      .catch((error) => this.logRedisFailure("remove_request", error));
+  }
+
+  async clearRequests(roomId: string): Promise<void> {
+    this.localRequests.delete(roomId);
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    await this.publisher
+      .del(`${KEY_PREFIX}:requests:${roomId}`)
+      .catch((error) => this.logRedisFailure("clear_requests", error));
+  }
+
+  async updatePresence(
+    roomId: string,
+    users: Array<{ userId: string; username: string }>,
+  ): Promise<void> {
+    if (!(await this.ensureConnected()) || !this.publisher) return;
+    const key = `${KEY_PREFIX}:presence:${roomId}`;
+    const previous = this.presenceMembers.get(roomId) ?? new Set<string>();
+    const expiresAt = Date.now() + PRESENCE_TTL_MS;
+    const current = new Set(
+      users.map((user) =>
+        JSON.stringify({
+          instanceId: this.instanceId,
+          ...user,
+        } satisfies PresenceEntry),
+      ),
+    );
+    const transaction = this.publisher.multi();
+    if (previous.size > 0) transaction.zRem(key, Array.from(previous));
+    for (const member of current) {
+      transaction.zAdd(key, { score: expiresAt, value: member });
+    }
+    transaction.expire(key, Math.ceil(PRESENCE_TTL_MS / 1000) * 2);
+    await transaction
+      .exec()
+      .then(() => {
+        if (current.size > 0) this.presenceMembers.set(roomId, current);
+        else this.presenceMembers.delete(roomId);
+      })
+      .catch((error) => this.logRedisFailure("update_presence", error));
+  }
+
+  async onlineUsers(
+    roomId: string,
+    localUsers: Array<{ userId: string; username: string }>,
+  ): Promise<Array<{ userId: string; username: string }>> {
+    if (!(await this.ensureConnected()) || !this.publisher) return localUsers;
+    const key = `${KEY_PREFIX}:presence:${roomId}`;
+    const now = Date.now();
+    const transaction = this.publisher
+      .multi()
+      .zRemRangeByScore(key, 0, now)
+      .zRangeByScore(key, now + 1, "+inf");
+    const result = await transaction.exec().catch((error) => {
+      this.logRedisFailure("online_users", error);
+      return null;
+    });
+    const members = result?.[1] as string[] | undefined;
+    if (!members) return localUsers;
+    const seen = new Map<string, string>();
+    for (const member of members) {
+      const entry = this.parse<PresenceEntry>(member);
+      if (entry) seen.set(entry.userId, entry.username);
+    }
+    return Array.from(seen, ([userId, username]) => ({ userId, username }));
+  }
+
+  async close(): Promise<void> {
+    await Promise.allSettled([
+      this.publisher?.isOpen ? this.publisher.quit() : Promise.resolve(),
+      this.subscriber?.isOpen ? this.subscriber.quit() : Promise.resolve(),
+    ]);
+    this.publisher = null;
+    this.subscriber = null;
+  }
+
+  private async ensureConnected(): Promise<boolean> {
+    const url = process.env.REDIS_URL?.trim();
+    if (!url) return false;
+    if (this.publisher?.isReady && this.subscriber?.isReady) return true;
+    if (Date.now() < this.nextConnectAttempt) return false;
+    if (this.connecting) return this.connecting;
+    this.connecting = this.connect(url).finally(() => {
+      this.connecting = null;
+    });
+    return this.connecting;
+  }
+
+  private async connect(url: string): Promise<boolean> {
+    try {
+      this.publisher = createClient({
+        url,
+        socket: { connectTimeout: 1500, reconnectStrategy: false },
+      });
+      this.subscriber = this.publisher.duplicate();
+      this.publisher.on("error", (error) =>
+        this.logRedisFailure("client", error),
+      );
+      this.subscriber.on("error", (error) =>
+        this.logRedisFailure("subscriber", error),
+      );
+      await Promise.all([this.publisher.connect(), this.subscriber.connect()]);
+      await this.subscriber.subscribe(EVENT_CHANNEL, (raw) => {
+        const envelope = this.parse<EventEnvelope>(raw);
+        if (
+          !envelope ||
+          typeof envelope.source !== "string" ||
+          typeof envelope.roomId !== "string" ||
+          !envelope.message ||
+          typeof envelope.message !== "object" ||
+          envelope.source === this.instanceId
+        ) {
+          return;
+        }
+        for (const listener of this.eventListeners) {
+          try {
+            listener(envelope.roomId, envelope.message);
+          } catch (error) {
+            this.logRedisFailure("event_listener", error);
+          }
+        }
+      });
+      sshLogger.info("Collaboration Redis runtime connected", {
+        operation: "collab_redis_connected",
+      });
+      this.nextConnectAttempt = 0;
+      return true;
+    } catch (error) {
+      this.nextConnectAttempt = Date.now() + CONNECT_RETRY_MS;
+      this.logRedisFailure("connect", error);
+      await Promise.allSettled([
+        this.publisher?.disconnect(),
+        this.subscriber?.disconnect(),
+      ]);
+      this.publisher = null;
+      this.subscriber = null;
+      return false;
+    }
+  }
+
+  private parse<T>(value: string): T | null {
+    try {
+      return JSON.parse(value) as T;
+    } catch {
+      return null;
+    }
+  }
+
+  private logRedisFailure(operation: string, error: unknown): void {
+    sshLogger.warn(
+      "Collaboration Redis runtime unavailable; using local state",
+      {
+        operation: `collab_redis_${operation}`,
+        error: error instanceof Error ? error.message : String(error),
+      },
+    );
+  }
+}
+
+export const collabRuntimeStore = new CollabRuntimeStore();

src/backend/hosts/collab/stage-control.ts

@@ -1,20 +1,18 @@
 /**
  * Who besides the presenter may drive the current stage, per room.
  *
- * Deliberately in-memory: control is a property of the live stage, and the
- * stage itself (SSH session / guacd connection) is process-local already.
- * Every stage switch clears it.
+ * Redis-backed when REDIS_URL is configured, with a local fallback for
+ * single-instance deployments. Every stage switch clears it.
  */
-const controllers = new Map<string, string>();
+import { collabRuntimeStore } from "./runtime-store.js";
 
-export function getStageController(roomId: string): string | null {
-  return controllers.get(roomId) ?? null;
+export function getStageController(roomId: string): Promise<string | null> {
+  return collabRuntimeStore.getController(roomId);
 }
 
 export function setStageController(
   roomId: string,
   userId: string | null,
-): void {
-  if (userId) controllers.set(roomId, userId);
-  else controllers.delete(roomId);
+): Promise<void> {
+  return collabRuntimeStore.setController(roomId, userId);
 }

src/backend/tests/hosts/collab/room-hub.test.ts

@@ -1,5 +1,26 @@
 import { describe, expect, it, vi } from "vitest";
 import type { WebSocket } from "ws";
+
+const runtime = vi.hoisted(() => ({
+  listener: null as ((roomId: string, message: object) => void) | null,
+  publish: vi.fn(async () => undefined),
+  updatePresence: vi.fn(async () => undefined),
+}));
+
+vi.mock("../../../hosts/collab/runtime-store.js", () => ({
+  collabRuntimeStore: {
+    onEvent: (listener: (roomId: string, message: object) => void) => {
+      runtime.listener = listener;
+    },
+    publish: runtime.publish,
+    updatePresence: runtime.updatePresence,
+    onlineUsers: async (
+      _roomId: string,
+      users: Array<{ userId: string; username: string }>,
+    ) => users,
+  },
+}));
+
 import { collabRoomHub } from "../../../hosts/collab/room-hub.js";
 
 function fakeWs(open = true): WebSocket {
@@ -11,7 +32,7 @@ function fakeWs(open = true): WebSocket {
 }
 
 describe("collabRoomHub", () => {
-  it("announces the online list on subscribe and unsubscribe, deduplicated per user", () => {
+  it("announces the online list on subscribe and unsubscribe, deduplicated per user", async () => {
     const a1 = fakeWs();
     const a2 = fakeWs();
     const b = fakeWs();
@@ -19,11 +40,12 @@ describe("collabRoomHub", () => {
     collabRoomHub.subscribe("room-1", { ws: a2, userId: "a", username: "A" });
     collabRoomHub.subscribe("room-1", { ws: b, userId: "b", username: "B" });
 
-    expect(collabRoomHub.onlineUsers("room-1")).toEqual([
+    expect(await collabRoomHub.onlineUsers("room-1")).toEqual([
       { userId: "a", username: "A" },
       { userId: "b", username: "B" },
     ]);
 
+    await vi.waitFor(() => expect(b.send).toHaveBeenCalled());
     const last = JSON.parse(
       (b.send as ReturnType<typeof vi.fn>).mock.calls.at(-1)?.[0] as string,
     );
@@ -37,20 +59,22 @@ describe("collabRoomHub", () => {
     });
 
     collabRoomHub.unsubscribe(a1);
-    expect(collabRoomHub.onlineUsers("room-1")).toHaveLength(2);
+    expect(await collabRoomHub.onlineUsers("room-1")).toHaveLength(2);
     collabRoomHub.unsubscribe(a2);
-    expect(collabRoomHub.onlineUsers("room-1")).toEqual([
+    expect(await collabRoomHub.onlineUsers("room-1")).toEqual([
       { userId: "b", username: "B" },
     ]);
     collabRoomHub.unsubscribe(b);
-    expect(collabRoomHub.onlineUsers("room-1")).toEqual([]);
+    expect(await collabRoomHub.onlineUsers("room-1")).toEqual([]);
   });
 
-  it("subscribing the same socket twice keeps one subscription", () => {
+  it("subscribing the same socket twice keeps one subscription", async () => {
     const ws = fakeWs();
     collabRoomHub.subscribe("room-2", { ws, userId: "a", username: "A" });
     collabRoomHub.subscribe("room-2", { ws, userId: "a", username: "A" });
-    expect((ws.send as ReturnType<typeof vi.fn>).mock.calls).toHaveLength(1);
+    await vi.waitFor(() =>
+      expect((ws.send as ReturnType<typeof vi.fn>).mock.calls).toHaveLength(1),
+    );
     collabRoomHub.unsubscribe(ws);
   });
 
@@ -74,4 +98,16 @@ describe("collabRoomHub", () => {
     collabRoomHub.unsubscribe(open);
     collabRoomHub.unsubscribe(closed);
   });
+
+  it("fans out remote Redis events but keeps internal events server-side", () => {
+    const ws = fakeWs();
+    collabRoomHub.subscribe("room-4", { ws, userId: "a", username: "A" });
+    (ws.send as ReturnType<typeof vi.fn>).mockClear();
+
+    runtime.listener?.("room-4", { type: "collab_members_changed" });
+    runtime.listener?.("room-4", { type: "collab_internal_stage_revoked" });
+
+    expect(ws.send).toHaveBeenCalledTimes(1);
+    collabRoomHub.unsubscribe(ws);
+  });
 });

src/backend/tests/hosts/collab/routes.test.ts

@@ -29,6 +29,11 @@ const state = vi.hoisted(() => ({
   liveOwned: new Map<string, string>(), // sessionId -> owner
   broadcasts: [] as Array<Record<string, unknown>>,
   control: [] as Array<unknown[]>,
+  controllers: new Map<string, string>(),
+  requests: new Map<
+    string,
+    Map<string, { userId: string; username: string; requestedAt: string }>
+  >(),
 }));
 
 vi.mock("../../../utils/logger.js", () => ({
@@ -67,6 +72,39 @@ vi.mock("../../../hosts/collab/room-hub.js", () => ({
     onlineUsers: () => [],
   },
 }));
+vi.mock("../../../hosts/collab/runtime-store.js", () => ({
+  collabRuntimeStore: {
+    onEvent: vi.fn(),
+    publish: vi.fn(async () => undefined),
+    getController: async (roomId: string) =>
+      state.controllers.get(roomId) ?? null,
+    setController: async (roomId: string, userId: string | null) => {
+      if (userId) state.controllers.set(roomId, userId);
+      else state.controllers.delete(roomId);
+    },
+    listRequests: async (roomId: string) =>
+      Array.from(state.requests.get(roomId)?.values() ?? []).sort((a, b) =>
+        a.requestedAt.localeCompare(b.requestedAt),
+      ),
+    upsertRequest: async (
+      roomId: string,
+      request: { userId: string; username: string; requestedAt: string },
+    ) => {
+      let requests = state.requests.get(roomId);
+      if (!requests) {
+        requests = new Map();
+        state.requests.set(roomId, requests);
+      }
+      requests.set(request.userId, request);
+    },
+    removeRequest: async (roomId: string, userId: string) => {
+      state.requests.get(roomId)?.delete(userId);
+    },
+    clearRequests: async (roomId: string) => {
+      state.requests.delete(roomId);
+    },
+  },
+}));
 vi.mock("../../../hosts/terminal/session-manager.js", () => ({
   sessionManager: {
     setRoomShareControl: (...args: unknown[]) => {
@@ -302,6 +340,8 @@ describe("collab room routes", () => {
     state.liveOwned.clear();
     state.broadcasts.length = 0;
     state.control.length = 0;
+    state.controllers.clear();
+    state.requests.clear();
     state.sharingEnabled = true;
   });
 
@@ -434,13 +474,13 @@ describe("collab room routes", () => {
       params: { id: roomId },
       body: { userId: "alice" },
     });
-    expect(getStageController(roomId)).toBe("alice");
+    expect(await getStageController(roomId)).toBe("alice");
 
     await as("alice", () => present(roomId, "s2"));
     const room = state.rooms.get(roomId)!;
     expect(room.presenterUserId).toBe("alice");
     expect(state.shares.get(firstShare)!.revokedAt).toBeTruthy();
-    expect(getStageController(roomId)).toBeNull();
+    expect(await getStageController(roomId)).toBeNull();
   });
 
   it("stop is for the presenter or host", async () => {
@@ -533,14 +573,16 @@ describe("collab room routes", () => {
       state.rooms.get(roomId)!.stageShareId,
       "alice",
     ]);
-    expect(state.broadcasts.at(-1)).toMatchObject({
-      type: "collab_control_changed",
-      controllerUserId: "alice",
-    });
+    expect(state.broadcasts).toContainEqual(
+      expect.objectContaining({
+        type: "collab_control_changed",
+        controllerUserId: "alice",
+      }),
+    );
 
     expect((await control("bob", null)).statusCode).toBe(403);
     expect((await control("alice", null)).statusCode).toBe(200);
-    expect(getStageController(roomId)).toBeNull();
+    expect(await getStageController(roomId)).toBeNull();
 
     const asked = await as("bob", () =>
       invoke("post", "/rooms/:id/control/request", { params: { id: roomId } }),
@@ -551,6 +593,31 @@ describe("collab room routes", () => {
       userId: "bob",
       username: "BOB",
     });
+    expect(
+      (
+        await as("bob", () =>
+          invoke("get", "/rooms/:id", { params: { id: roomId } }),
+        )
+      ).jsonBody!.controlRequests,
+    ).toEqual([expect.objectContaining({ userId: "bob" })]);
+
+    const repeated = await as("bob", () =>
+      invoke("post", "/rooms/:id/control/request", {
+        params: { id: roomId },
+      }),
+    );
+    expect((repeated as { statusCode: number }).statusCode).toBe(429);
+
+    const listed = await invoke("get", "/rooms/:id/control/requests", {
+      params: { id: roomId },
+    });
+    expect(listed.jsonBody!.requests).toEqual([
+      expect.objectContaining({ userId: "bob" }),
+    ]);
+    await invoke("delete", "/rooms/:id/control/requests/:userId", {
+      params: { id: roomId, userId: "bob" },
+    });
+    expect(state.requests.get(roomId)?.size ?? 0).toBe(0);
   });
 
   it("guest link: host-only toggle, anonymous resolve follows the stage, rate limited", async () => {

src/backend/tests/hosts/collab/runtime-store.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it } from "vitest";
+import { collabRuntimeStore } from "../../../hosts/collab/runtime-store.js";
+
+describe("collabRuntimeStore local fallback", () => {
+  it("stores stage control and an ordered, deduplicated request queue", async () => {
+    const roomId = `fallback-${crypto.randomUUID()}`;
+
+    await collabRuntimeStore.setController(roomId, "alice");
+    expect(await collabRuntimeStore.getController(roomId)).toBe("alice");
+
+    await collabRuntimeStore.upsertRequest(roomId, {
+      userId: "bob",
+      username: "Bob",
+      requestedAt: "2026-08-25T00:00:02.000Z",
+    });
+    await collabRuntimeStore.upsertRequest(roomId, {
+      userId: "alice",
+      username: "Alice",
+      requestedAt: "2026-08-25T00:00:01.000Z",
+    });
+    await collabRuntimeStore.upsertRequest(roomId, {
+      userId: "bob",
+      username: "Bob",
+      requestedAt: "2026-08-25T00:00:03.000Z",
+    });
+
+    expect(await collabRuntimeStore.listRequests(roomId)).toEqual([
+      expect.objectContaining({ userId: "alice" }),
+      expect.objectContaining({ userId: "bob" }),
+    ]);
+    await collabRuntimeStore.removeRequest(roomId, "alice");
+    expect(await collabRuntimeStore.listRequests(roomId)).toHaveLength(1);
+
+    await collabRuntimeStore.clearRequests(roomId);
+    await collabRuntimeStore.setController(roomId, null);
+    expect(await collabRuntimeStore.listRequests(roomId)).toEqual([]);
+    expect(await collabRuntimeStore.getController(roomId)).toBeNull();
+  });
+});