feat: add browser owner collab transport

Author: lightsofapollo

.beads/export-state.json

@@ -1 +1 @@
-{"last_dolt_commit":"hj3nlh9ehb039rqg4lcm5vc86eb7h40j","timestamp":"2026-07-11T00:22:01.847764-06:00","issues":297,"memories":13}
+{"last_dolt_commit":"j4dh0qlaf710t1mespf1qt2nes42e5ag","timestamp":"2026-07-11T00:42:18.264734-06:00","issues":297,"memories":13}

.beads/interactions.jsonl

@@ -298,3 +298,5 @@
 {"id":"int-d9dbd71a","kind":"field_change","created_at":"2026-07-11T05:50:42.403172Z","actor":"James Lal","issue_id":"attn-7xl.4.4","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}
 {"id":"int-dfb1fb6a","kind":"field_change","created_at":"2026-07-11T05:57:58.620744Z","actor":"James Lal","issue_id":"attn-7xl.4.4.1","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}
 {"id":"int-39fe275f","kind":"field_change","created_at":"2026-07-11T06:21:54.401105Z","actor":"James Lal","issue_id":"attn-7xl.4.4.1","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Implemented and security-audited epoch-bound browser collab core: atomic prepare/persist/commit/broadcast, authenticated-owner broadcast gating, strict legacy-aware wire parsing, stale catch-up/resync, sealed checkpoint restore, workspace-key AAD, snapshot-base binding, epoch-keyed/version-CASed/lease-fenced persistence, reload-to-controller integration, and visible pause seam. Validation: 61 web test files/0 failures, svelte-check 0/0, production browser build and route bundle gate pass, final adversarial re-audit found no blockers."}}
+{"id":"int-96547502","kind":"field_change","created_at":"2026-07-11T06:22:25.7727Z","actor":"James Lal","issue_id":"attn-7xl.4.4.2","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}
+{"id":"int-ff450871","kind":"field_change","created_at":"2026-07-11T06:42:10.428821Z","actor":"James Lal","issue_id":"attn-7xl.4.4.2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Completed recovered browser-owner GET/WS-only startup, fully validated sealed credentials and authenticated policy/directory, honest owner presence/live-authority state, target-null collab direct+relay exact-envelope delivery with retry-safe dedup, and aggregate same-device relay presence including peer-cap race hardening. Full web/relay tests, static checks, production browser build, route bundle gate, and adversarial re-audit all pass."}}

.beads/issues.jsonl

@@ -3087,21 +3087,24 @@ export class RoomDO extends DurableObject<Env> {
     }
     const participantId = deviceRecord.participantId;
 
-    // Peer cap: count distinct deviceIds currently connected.
+    // Peer cap: count distinct deviceIds currently represented by an
+    // announced socket. Multiple tabs/connections for one registered device
+    // are one presence peer and consume one peer-cap slot.
     const policy = await this.ctx.storage.get<RoomPolicy>(META.policy);
     if (policy === undefined) {
       return errorResponse(500, "ATTN_ROOM_CORRUPT", `room ${roomId} missing policy`);
     }
     const connectedDevices = new Set<string>();
     for (const existing of this.ctx.getWebSockets()) {
       const attached = readAttachment(existing);
-      if (attached !== undefined) connectedDevices.add(attached.deviceId);
+      if (attached?.presenceAnnounced === true) connectedDevices.add(attached.deviceId);
     }
-    // If THIS deviceId is already connected we let it in (the prior socket
-    // will be replaced — the spec doesn't speak to multi-tab per device, so we
-    // treat it as the same peer). New deviceId beyond cap → close 4004.
+    // If THIS deviceId is already connected we let another socket in while
+    // continuing to represent it as one online peer. New deviceId beyond cap
+    // → close 4004.
+    const deviceAlreadyOnline = connectedDevices.has(deviceId);
     const wouldBeOver =
-      !connectedDevices.has(deviceId) && connectedDevices.size >= policy.maxPeers;
+      !deviceAlreadyOnline && connectedDevices.size >= policy.maxPeers;
 
     // Build the upgrade response.
     const pair = new WebSocketPair();
@@ -3119,6 +3122,7 @@ export class RoomDO extends DurableObject<Env> {
       participantId,
       subscribed: false,
       lastPongTs: 0,
+      presenceAnnounced: !wouldBeOver,
     });
 
     if (wouldBeOver) {
@@ -3135,8 +3139,11 @@ export class RoomDO extends DurableObject<Env> {
       });
     }
 
-    // Broadcast presence:join to every OTHER connected peer.
-    this.broadcastPresence({ event: "join", deviceId, participantId }, server);
+    // Presence is per registered device, not per socket. A second tab/socket
+    // for an already-online device must not produce another join.
+    if (!deviceAlreadyOnline) {
+      this.broadcastPresence({ event: "join", deviceId, participantId }, server);
+    }
 
     return new Response(null, {
       status: 101,
@@ -3151,6 +3158,27 @@ export class RoomDO extends DurableObject<Env> {
    * eviction between frames.
    */
   override async webSocketMessage(ws: WebSocket, msg: string | ArrayBuffer): Promise<void> {
+    // A peer-cap rejection uses an accepted-then-closed upgrade so the client
+    // can observe close code 4004. Frames can race that close. Gate on the
+    // attachment before decoding or replying so the rejected socket cannot
+    // subscribe, receive hello/replay/ping, or elicit validation errors.
+    const att = readAttachment(ws);
+    if (att === undefined) {
+      try {
+        ws.close(CLOSE_NORMAL, "missing attachment");
+      } catch {
+        // ignore
+      }
+      return;
+    }
+    if (!att.presenceAnnounced) {
+      try {
+        ws.close(CLOSE_PEER_CAP, "peer cap reached");
+      } catch {
+        // close is already in progress
+      }
+      return;
+    }
     if (typeof msg !== "string") {
       sendError(ws, "ATTN_FRAME_INVALID", "binary frames are reserved");
       return;
@@ -3167,16 +3195,6 @@ export class RoomDO extends DurableObject<Env> {
       sendError(ws, "ATTN_FRAME_INVALID", formatZodError(frame.error));
       return;
     }
-    const att = readAttachment(ws);
-    if (att === undefined) {
-      // Attachment shouldn't be missing — defensively close.
-      try {
-        ws.close(CLOSE_NORMAL, "missing attachment");
-      } catch {
-        // ignore
-      }
-      return;
-    }
     const body = frame.data;
     if (body.type === "subscribe") {
       await this.handleSubscribe(ws, att, body.after);
@@ -3188,19 +3206,20 @@ export class RoomDO extends DurableObject<Env> {
     }
   }
 
-  /** Hibernation entry-point on close. Broadcast presence:leave. */
+  /** Hibernation entry-point on close. Leave only after the final device socket. */
   override async webSocketClose(
     ws: WebSocket,
     _code: number,
     _reason: string,
     _wasClean: boolean,
   ): Promise<void> {
     const att = readAttachment(ws);
-    if (att !== undefined) {
-      this.broadcastPresence(
-        { event: "leave", deviceId: att.deviceId, participantId: att.participantId },
-        ws,
-      );
+    if (att?.presenceAnnounced === true && !this.hasOtherDeviceSocket(ws, att.deviceId)) {
+      this.broadcastPresence({
+        event: "leave",
+        deviceId: att.deviceId,
+        participantId: att.participantId,
+      }, ws);
     }
     // No need to call ws.close — the runtime already did. We just clean up
     // any state we owned (the attachment lives on the socket itself, which
@@ -3210,14 +3229,29 @@ export class RoomDO extends DurableObject<Env> {
   /** Hibernation entry-point on socket-level error. Same handling as close. */
   override async webSocketError(ws: WebSocket, _err: unknown): Promise<void> {
     const att = readAttachment(ws);
-    if (att !== undefined) {
-      this.broadcastPresence(
-        { event: "leave", deviceId: att.deviceId, participantId: att.participantId },
-        ws,
-      );
+    if (att?.presenceAnnounced === true) {
+      // Error may be followed by webSocketClose. Mark this socket first so
+      // that the later callback cannot emit a duplicate leave.
+      writeAttachment(ws, { ...att, presenceAnnounced: false });
+      if (!this.hasOtherDeviceSocket(ws, att.deviceId)) {
+        this.broadcastPresence({
+          event: "leave",
+          deviceId: att.deviceId,
+          participantId: att.participantId,
+        }, ws);
+      }
     }
   }
 
+  /** True when another announced socket still represents `deviceId` online. */
+  private hasOtherDeviceSocket(originator: WebSocket, deviceId: string): boolean {
+    return this.ctx.getWebSockets().some((socket) => {
+      if (socket === originator) return false;
+      const attachment = readAttachment(socket);
+      return attachment?.presenceAnnounced === true && attachment.deviceId === deviceId;
+    });
+  }
+
   /**
    * Handle the client's `subscribe { after }` frame. Sends `hello` plus a
    * replay of every envelope with serverSeq > after, in order. If `after` is
@@ -3276,7 +3310,12 @@ export class RoomDO extends DurableObject<Env> {
     }
     const onlineDeviceIds = [...new Set(
       this.ctx.getWebSockets()
-        .map((socket) => readAttachment(socket)?.deviceId)
+        .map((socket) => {
+          const attachment = readAttachment(socket);
+          return attachment?.presenceAnnounced === true
+            ? attachment.deviceId
+            : undefined;
+        })
         .filter((deviceId): deviceId is string => deviceId !== undefined),
     )].sort();
     const hello: ServerFrame = {
@@ -3336,7 +3375,7 @@ export class RoomDO extends DurableObject<Env> {
       const json = JSON.stringify(frame);
       for (const sock of sockets) {
         const att = readAttachment(sock);
-        if (att === undefined) continue;
+        if (att?.presenceAnnounced !== true) continue;
         if (!att.subscribed) continue;
         if (!deliverableTo(record, att.deviceId)) continue;
         sendRaw(sock, json);
@@ -3362,7 +3401,7 @@ export class RoomDO extends DurableObject<Env> {
     for (const sock of this.ctx.getWebSockets()) {
       if (sock === originator) continue;
       const att = readAttachment(sock);
-      if (att === undefined) continue;
+      if (att?.presenceAnnounced !== true) continue;
       sendRaw(sock, json);
     }
   }
@@ -4458,6 +4497,8 @@ export const WS_CLOSE_CODES = {
 interface WSAttachment {
   deviceId: string;
   participantId: string;
+  /** This socket participates in the device's aggregate online presence. */
+  presenceAnnounced: boolean;
   /** True once the socket has received a successful `subscribe` reply. */
   subscribed: boolean;
   /** Last seen pong timestamp (ms). 0 until the first pong. */
@@ -4587,6 +4628,9 @@ function readAttachment(ws: WebSocket): WSAttachment | undefined {
   return {
     deviceId: r.deviceId,
     participantId: r.participantId,
+    // Attachments written before aggregate same-device presence shipped did
+    // announce presence, so missing legacy state must restore as true.
+    presenceAnnounced: r.presenceAnnounced !== false,
     subscribed: r.subscribed === true,
     lastPongTs: typeof r.lastPongTs === "number" ? r.lastPongTs : 0,
   };

relay/src/room-do.ts

@@ -897,6 +897,65 @@ describe("WS /socket — live broadcast", () => {
 });
 
 describe("WS /socket — presence", () => {
+  it("tracks same-device presence across 0→1 and 1→0 socket transitions", async () => {
+    const roomId = uniqueRoomId("ws-presence-same-device");
+    const owner = await generateEd25519Keypair();
+    const admissionKey = await createRoom({ roomId, ownerKp: owner });
+    await registerDevice({ roomId, admissionKey, deviceId: "observer", participantId: "observer-p" });
+    await registerDevice({ roomId, admissionKey, deviceId: "shared-device", participantId: "shared-p" });
+    await registerDevice({ roomId, admissionKey, deviceId: "isolated-device", participantId: "isolated-p" });
+
+    const observer = await openSocket({ roomId, deviceId: "observer", admissionKey });
+    const queue = new FrameQueue(observer.ws);
+    observer.ws.send(JSON.stringify({ type: "subscribe", after: 0 }));
+    expect(isHello(await queue.next())).toBe(true);
+    expect(isPing(await queue.next())).toBe(true);
+
+    const first = await openSocket({ roomId, deviceId: "shared-device", admissionKey });
+    const firstJoin = await queue.next();
+    expect(firstJoin).toEqual({
+      type: "presence",
+      event: "join",
+      deviceId: "shared-device",
+      participantId: "shared-p",
+    });
+
+    const second = await openSocket({ roomId, deviceId: "shared-device", admissionKey });
+    const isolated = await openSocket({ roomId, deviceId: "isolated-device", admissionKey });
+    // This must be the isolated device's join. If the second shared-device
+    // socket emitted a duplicate join, it would be the next queued frame.
+    expect(await queue.next()).toEqual({
+      type: "presence",
+      event: "join",
+      deviceId: "isolated-device",
+      participantId: "isolated-p",
+    });
+
+    first.ws.close(1000, "first tab closed");
+    isolated.ws.close(1000, "other device closed");
+    // This must be isolated-device's leave. A false leave from the first
+    // shared-device socket would arrive first. Presence is isolated by
+    // deviceId: the remaining shared socket cannot keep this device online.
+    expect(await queue.next()).toEqual({
+      type: "presence",
+      event: "leave",
+      deviceId: "isolated-device",
+      participantId: "isolated-p",
+    });
+
+    second.ws.close(1000, "final tab closed");
+    expect(await queue.next()).toEqual({
+      type: "presence",
+      event: "leave",
+      deviceId: "shared-device",
+      participantId: "shared-p",
+    });
+    // The 1→0 transition emits exactly one leave.
+    expect(await queue.next(250)).toBeUndefined();
+
+    observer.ws.close(1000);
+  });
+
   it("broadcasts presence:join to existing peers when a new socket opens", async () => {
     const roomId = uniqueRoomId("ws-presence-join");
     const owner = await generateEd25519Keypair();
@@ -987,6 +1046,66 @@ describe("WS /socket — ping/pong", () => {
 });
 
 describe("WS /socket — peer cap", () => {
+  it("drops a subscribe racing close on a cap-rejected attachment", async () => {
+    const roomId = uniqueRoomId("ws-peercap-subscribe-race");
+    const owner = await generateEd25519Keypair();
+    const admissionKey = await createRoom({ roomId, ownerKp: owner });
+    await registerDevice({
+      roomId,
+      admissionKey,
+      deviceId: "dev-rejected",
+      participantId: "rejected-p",
+    });
+    // Ensure a successful subscribe would have both hello and replay data to
+    // send, making silence a meaningful assertion rather than an empty room.
+    const envelope = buildEnvelope({
+      envelopeId: "rejected-race-event",
+      authorId: "rejected-p",
+      deviceId: "dev-rejected",
+    });
+    expect(
+      (await postEnvelopes({ roomId, admissionKey, envelopes: [envelope] })).status,
+    ).toBe(201);
+
+    const stub = env.RELAY_ROOMS.get(env.RELAY_ROOMS.idFromName(roomId));
+    const sent: string[] = [];
+    const closes: Array<{ code?: number; reason?: string }> = [];
+    await runInDurableObject(stub, async (instance) => {
+      let attachment: unknown = {
+        deviceId: "dev-rejected",
+        participantId: "rejected-p",
+        presenceAnnounced: false,
+        subscribed: false,
+        lastPongTs: 0,
+      };
+      // Drive the hibernation callback directly with a rejected attachment.
+      // This fixes the race ordering: subscribe is delivered while the
+      // accepted-then-closed socket is still capable of receiving callbacks.
+      const rejectedSocket = {
+        deserializeAttachment: () => attachment,
+        serializeAttachment: (next: unknown) => {
+          attachment = next;
+        },
+        send: (payload: string) => {
+          sent.push(payload);
+        },
+        close: (code?: number, reason?: string) => {
+          closes.push({ code, reason });
+        },
+      } as unknown as WebSocket;
+      const handler = instance as unknown as {
+        webSocketMessage(socket: WebSocket, message: string): Promise<void>;
+      };
+      await handler.webSocketMessage(
+        rejectedSocket,
+        JSON.stringify({ type: "subscribe", after: 0 }),
+      );
+    });
+
+    expect(closes).toEqual([{ code: 4004, reason: "peer cap reached" }]);
+    expect(sent).toEqual([]);
+  });
+
   it("closes the (maxPeers+1)-th distinct device with 4004", async () => {
     const roomId = uniqueRoomId("ws-peercap");
     const owner = await generateEd25519Keypair();
@@ -1004,18 +1123,33 @@ describe("WS /socket — peer cap", () => {
     const qa = new FrameQueue(a.ws);
     a.ws.send(JSON.stringify({ type: "subscribe", after: 0 }));
     expect(isHello(await qa.next())).toBe(true);
+    expect(isPing(await qa.next())).toBe(true);
 
     const b = await openSocket({ roomId, deviceId: "dev-c2", admissionKey });
+    expect(await qa.next()).toEqual({
+      type: "presence",
+      event: "join",
+      deviceId: "dev-c2",
+      participantId: "c2",
+    });
     const qb = new FrameQueue(b.ws);
     b.ws.send(JSON.stringify({ type: "subscribe", after: 0 }));
-    expect(isHello(await qb.next())).toBe(true);
+    const helloB = await qb.next();
+    expect(isHello(helloB)).toBe(true);
+    if (isHello(helloB)) {
+      expect(helloB.onlineDeviceIds).toEqual(["dev-c1", "dev-c2"]);
+    }
+    expect(isPing(await qb.next())).toBe(true);
 
     // Third connect should be rejected with close 4004.
     const c = await openSocket({ roomId, deviceId: "dev-c3", admissionKey });
     const qc = new FrameQueue(c.ws);
     await qc.waitClosed(2000);
     expect(qc.closed).toBe(true);
     expect(qc.closeCode).toBe(4004);
+    // A cap-rejected socket never joins presence and its close cannot emit a
+    // leave during the brief accepted-then-closed transition.
+    expect(await qa.next(250)).toBeUndefined();
 
     a.ws.close(1000);
     b.ws.close(1000);

relay/test/integration/websocket.test.ts

@@ -80,6 +80,9 @@
   // ---------------------------------------------------------------------------
 
   let sessionState = $state<BrowserSessionState>({
+    principal: 'reviewer',
+    ownerOnline: false,
+    liveEditingAvailable: false,
     status: 'idle',
     connection: 'offline',
     directError: null,