fix(file-doc): admit joins on their validated permission and route sweep capability through the per-socket de-dup

Author: waleedlatif1

apps/realtime/src/access-revalidation.test.ts

@@ -13,6 +13,7 @@ import {
   ACCESS_REVALIDATION_SWEEP_INTERVAL_MS,
   startAccessRevalidationSweep,
 } from '@/access-revalidation'
+import { registerRoomPermissionPublisher } from '@/handlers/room-permission'
 import type { IRoomManager, UserPresence } from '@/rooms'
 
 interface FakeSocket {
@@ -131,6 +132,8 @@ describe('access-revalidation sweep', () => {
   })
 
   it('keeps a downgraded reader subscribed and reports the loss of write access', async () => {
+    const publish = vi.fn()
+    registerRoomPermissionPublisher('workspace-file-doc', publish)
     const socket = makeSocket('sock-1', 'user-1', 'workspace-file-doc:file-1')
     socket.rooms.add('table:t-1')
     const manager = makeManager([socket])
@@ -139,20 +142,23 @@ describe('access-revalidation sweep', () => {
     await sweep.runOnce()
     sweep.stop()
     expect(socket.leave).not.toHaveBeenCalled()
-    expect(socket.emit).toHaveBeenCalledWith('file-doc-permission', {
-      fileId: 'file-1',
-      canWrite: false,
-    })
+    expect(publish).toHaveBeenCalledWith(
+      socket,
+      { type: 'workspace-file-doc', id: 'file-1' },
+      'read'
+    )
   })
 
   it('retains a document subscription for retry when authorization is unavailable', async () => {
+    const publish = vi.fn()
+    registerRoomPermissionPublisher('workspace-file-doc', publish)
     const socket = makeSocket('sock-1', 'user-1', 'workspace-file-doc:file-1')
     mockResolveRole.mockRejectedValue(new Error('database unavailable'))
     const sweep = startAccessRevalidationSweep(makeManager([socket]))
     await sweep.runOnce()
     sweep.stop()
     expect(socket.leave).not.toHaveBeenCalled()
-    expect(socket.emit).not.toHaveBeenCalledWith('file-doc-permission', expect.anything())
+    expect(publish).not.toHaveBeenCalled()
   })
 
   it('evicts only the revoked socket, not co-members of the room', async () => {

apps/realtime/src/access-revalidation.ts

@@ -1,11 +1,9 @@
 import { createLogger } from '@sim/logger'
 import { ROOM_MEMBERSHIP_ACTIONS, satisfiesRoomMembership } from '@sim/platform-authz/room-policy'
 import type { AccessRevokedBroadcast } from '@sim/realtime-protocol/events'
-import { FILE_DOC_EVENTS, type FileDocPermission } from '@sim/realtime-protocol/file-doc'
 import {
   isProjectRoom,
   parseRoomName,
-  projectFileDocTarget,
   ROOM_TYPES,
   type RoomRef,
   type RoomType,
@@ -17,6 +15,7 @@ import {
   runRoomEvictionHandler,
   setEvictionCleanupSink,
 } from '@/handlers/room-eviction'
+import { publishRoomPermission } from '@/handlers/room-permission'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import { ROLE_REVALIDATION_TTL_MS, resolveCurrentRoomPermission } from '@/middleware/permissions'
 import type { IRoomManager } from '@/rooms'
@@ -331,20 +330,7 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
         ])
         // {@link SCAN_TIMED_OUT} is the only symbol this race can yield; matching on
         // the type narrows it out of the permission comparison below.
-        if (
-          typeof role !== 'symbol' &&
-          (room.type === ROOM_TYPES.PROJECT_FILE_DOC || room.type === ROOM_TYPES.WORKSPACE_FILE_DOC)
-        ) {
-          const target =
-            room.type === ROOM_TYPES.PROJECT_FILE_DOC
-              ? projectFileDocTarget(room)
-              : { fileId: room.id }
-          if (target)
-            socket.emit(FILE_DOC_EVENTS.PERMISSION, {
-              ...target,
-              canWrite: role === 'write' || role === 'admin',
-            } satisfies FileDocPermission)
-        }
+        if (typeof role !== 'symbol') publishRoomPermission(socket, room, role)
         if (typeof role === 'symbol') {
           logger.warn(
             `Authorization check timed out for user ${userId} on ${name}; skipping this pass`

apps/realtime/src/handlers/file-doc.test.ts

@@ -42,6 +42,7 @@ import {
   setupWorkspaceFileDocHandlers,
 } from '@/handlers/file-doc'
 import { FileDocInvalidatedError, getFileDocStore } from '@/handlers/file-doc-store'
+import { publishRoomPermission } from '@/handlers/room-permission'
 import * as permissions from '@/middleware/permissions'
 import {
   beginRoomPermissionRead,
@@ -686,6 +687,27 @@ describe('setupWorkspaceFileDocHandlers', () => {
     }
   })
 
+  it('publishes a re-validated permission only when it changes what the socket last heard', async () => {
+    const { io } = createIo()
+    const { socket, handlers } = setup('socket-sweep', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 8 })
+    await handlers[FILE_DOC_EVENTS.MESSAGE](awarenessFrame(8, 'Sweep').frame)
+    const room = { type: ROOM_TYPES.WORKSPACE_FILE_DOC, id: 'file-1' } as const
+    const subscriber = socket as unknown as Parameters<typeof publishRoomPermission>[0]
+    const published = () =>
+      socket.emit.mock.calls
+        .filter(([event]) => event === FILE_DOC_EVENTS.PERMISSION)
+        .map(([, payload]) => (payload as { canWrite: boolean }).canWrite)
+
+    publishRoomPermission(subscriber, room, 'write')
+    publishRoomPermission(subscriber, room, 'write')
+    expect(published()).toEqual([true])
+
+    publishRoomPermission(subscriber, room, 'read')
+    publishRoomPermission(subscriber, room, 'read')
+    expect(published()).toEqual([true, false])
+  })
+
   it('applies + fans out an agent-streamed frame (SYNC_NO_PERSIST) but never persists it', async () => {
     mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     const { io, sent } = createIo()

apps/realtime/src/handlers/file-doc.ts

@@ -75,6 +75,7 @@ import {
 } from '@/handlers/file-doc-store'
 import { evictSocketFromRoom, registerRoomEvictionHandler } from '@/handlers/room-eviction'
 import { resolveRoomJoinAuth } from '@/handlers/room-join-auth'
+import { registerRoomPermissionPublisher } from '@/handlers/room-permission'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import { peekRoomPermission, resolveCurrentRoomPermission } from '@/middleware/permissions'
 import type { IRoomManager } from '@/rooms'
@@ -404,7 +405,10 @@ async function persistRoom(name: string, room: FileDocRoom, final: boolean): Pro
     }
     try {
       const snapshot = await store.getStreamSnapshot(name, generation)
-      snapshotEditor = snapshot?.editor ?? null
+      /** Entries without an author (older relays, durable merges) keep workspace attribution local. */
+      snapshotEditor =
+        snapshot?.editor ??
+        (fileDocOwnerAdapter(room.owner).allowLocalSnapshotFallback ? localEditor : null)
       return snapshot?.docState ?? null
     } catch (streamError) {
       if (
@@ -1152,6 +1156,22 @@ function emitJoinError(
   })
 }
 
+/** Tells a socket its write capability only when it differs from what the socket last heard. */
+function publishWriteAccess(
+  socket: AuthenticatedSocket,
+  document: FileDocRoom,
+  permission: string | null
+): void {
+  const canWrite = permission === 'write' || permission === 'admin'
+  if (document.publishedCanWrite.get(socket.id) === canWrite) return
+  document.publishedCanWrite.set(socket.id, canWrite)
+  socket.emit(FILE_DOC_EVENTS.PERMISSION, {
+    fileId: document.fileId,
+    ...fileDocOwnerWireFields(document.owner),
+    canWrite,
+  } satisfies FileDocPermission)
+}
+
 /** Resolve read membership and publish the separate editing capability after authorization. */
 function resolveFileDocPermission(
   socket: AuthenticatedSocket,
@@ -1164,15 +1184,7 @@ function resolveFileDocPermission(
   const room = fileDocRoom(document)
   const accept = (permission: string | null) => {
     if (socketToRoomName.get(socket.id) !== name || fileDocRooms.get(name) !== document) return null
-    const canWrite = permission === 'write' || permission === 'admin'
-    if (document.publishedCanWrite.get(socket.id) !== canWrite) {
-      document.publishedCanWrite.set(socket.id, canWrite)
-      socket.emit(FILE_DOC_EVENTS.PERMISSION, {
-        fileId: document.fileId,
-        ...fileDocOwnerWireFields(document.owner),
-        canWrite,
-      } satisfies FileDocPermission)
-    }
+    publishWriteAccess(socket, document, permission)
     if (permission === null)
       evictSocketFromRoom(socket, room, 'Your access to this document has been revoked', io)
     return permission
@@ -1189,11 +1201,18 @@ function resolveFileDocPermission(
  * revocation and left the Socket.IO room). Scoped to the evicted room so a socket
  * that has since switched documents keeps the one it legitimately holds.
  */
-for (const type of [ROOM_TYPES.WORKSPACE_FILE_DOC, ROOM_TYPES.PROJECT_FILE_DOC])
+for (const type of [ROOM_TYPES.WORKSPACE_FILE_DOC, ROOM_TYPES.PROJECT_FILE_DOC]) {
   registerRoomEvictionHandler(type, (socketId, room, io) => {
     if (socketToRoomName.get(socketId) !== roomName(room)) return
     cleanupFileDocForSocket(socketId, io)
   })
+  registerRoomPermissionPublisher(type, (socket, room, permission) => {
+    const name = roomName(room)
+    const document = fileDocRooms.get(name)
+    if (document && socketToRoomName.get(socket.id) === name)
+      publishWriteAccess(socket, document, permission)
+  })
+}
 
 async function handleMessage(socket: AuthenticatedSocket, io: Server, data: unknown) {
   const name = socketToRoomName.get(socket.id)
@@ -1229,7 +1248,7 @@ async function handleMessage(socket: AuthenticatedSocket, io: Server, data: unkn
     } catch (error) {
       logger.warn('Dropping file-doc frame while its authorization is unavailable', {
         socketId: socket.id,
-        error,
+        error: getErrorMessage(error),
       })
       return
     }
@@ -1673,10 +1692,11 @@ export function setupWorkspaceFileDocHandlers(
         !socket.disconnected &&
         joinGeneration.get(socket.id) === generation &&
         fileDocRooms.get(name) === entry
-      const canRegisterJoin = () => {
-        if (!isCurrentJoin()) return false
+      /** The permission this admission is still current under, or `null` once it is not. */
+      const canRegisterJoin = (): string | null => {
+        if (!isCurrentJoin()) return null
         const permission = peekRoomPermission(userId, room)
-        if (satisfiesRoomMembership(permission ?? null, room.type)) return true
+        if (permission && satisfiesRoomMembership(permission, room.type)) return permission
         emitJoinError(
           socket,
           fileId,
@@ -1685,7 +1705,7 @@ export function setupWorkspaceFileDocHandlers(
           permission === undefined ? 'JOIN_FAILED' : 'ACCESS_DENIED',
           permission === undefined
         )
-        return false
+        return null
       }
       try {
         // A client is attached to a WHOLE document or to nothing. A room assembles itself from the
@@ -1838,7 +1858,8 @@ export function setupWorkspaceFileDocHandlers(
         socketToRoomName.set(socket.id, name)
         registered = true
         await changeMembership(() => socket.leave(`${name}:pending`))
-        if (!canRegisterJoin()) {
+        const admittedPermission = canRegisterJoin()
+        if (!admittedPermission) {
           if (isCurrentJoin() && socketToRoomName.get(socket.id) === name) {
             cleanupFileDocForSocket(socket.id, io)
             await leaveUnclaimedRoom(name)
@@ -1864,9 +1885,7 @@ export function setupWorkspaceFileDocHandlers(
           schemaVersion: FILE_DOC_SCHEMA_VERSION,
           ...fileDocOwnerWireFields(entry.owner),
           acknowledgedUpdates: true as const,
-          canWrite:
-            peekRoomPermission(userId, room) === 'write' ||
-            peekRoomPermission(userId, room) === 'admin',
+          canWrite: admittedPermission === 'write' || admittedPermission === 'admin',
         })
         // Server-authenticated roster → everyone in the room, including this joiner.
         broadcastFileDocPresence(io, name, entry)

apps/realtime/src/handlers/room-permission.ts

@@ -0,0 +1,41 @@
+import { createLogger } from '@sim/logger'
+import type { RoomRef, RoomType } from '@sim/realtime-protocol/rooms'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+
+const logger = createLogger('RoomPermission')
+
+/** Tells a socket, in its room type's own wire form, what a re-validated permission now allows. */
+export type RoomPermissionPublisher = (
+  socket: AuthenticatedSocket,
+  room: RoomRef,
+  permission: string | null
+) => void
+
+const publishers = new Map<RoomType, RoomPermissionPublisher>()
+
+/**
+ * Registers how a room type announces a re-validated permission. A registry for the same reason
+ * as the eviction handlers: the access re-validation sweep stays domain-neutral, and the room's
+ * handler keeps its own record of what each socket was last told.
+ */
+export function registerRoomPermissionPublisher(
+  type: RoomType,
+  publisher: RoomPermissionPublisher
+): void {
+  publishers.set(type, publisher)
+}
+
+/** Publishes a re-validated permission through the room type's publisher, if it has one. */
+export function publishRoomPermission(
+  socket: AuthenticatedSocket,
+  room: RoomRef,
+  permission: string | null
+): void {
+  const publisher = publishers.get(room.type)
+  if (!publisher) return
+  try {
+    publisher(socket, room, permission)
+  } catch (error) {
+    logger.warn(`Room permission publish failed for socket ${socket.id} on ${room.type}`, error)
+  }
+}

apps/sim/app/api/internal/project-file-doc/[projectId]/[fileId]/persist/route.ts

@@ -1,3 +1,4 @@
+import { NextResponse } from 'next/server'
 import { projectFileDocPersistContract } from '@/lib/api/contracts/project-file-doc'
 import { isOffAppHost } from '@/lib/api/mcp/host-routing'
 import {
@@ -7,24 +8,19 @@ import {
 } from '@/lib/api/server/routes'
 import { realtimeProjectFileAuth } from '@/lib/auth/realtime-file-delegation'
 import { MAX_COLLAB_DOC_STATE_BYTES } from '@/lib/collab-doc/state'
-import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { persistProjectFileDoc } from '@/lib/projects/files/application/documents'
 
 /** A base64-encoded maximum snapshot plus 1 MiB for its JSON envelope (17 MiB). */
 const PERSIST_MAX_BODY_BYTES = Math.ceil(MAX_COLLAB_DOC_STATE_BYTES / 3) * 4 + 1024 * 1024
 
-export const POST = defineInternalJsonRoute({
+const persist = defineInternalJsonRoute({
   contract: projectFileDocPersistContract,
   operation: persistProjectFileDoc.operation,
   auth: realtimeProjectFileAuth,
   rateLimit: internalRateLimits.none({
     reason: 'Authenticated realtime relay; socket frame admission bounds collaborative traffic',
   }),
   errorPolicy: internalOrchestrationErrorPolicy,
-  /** Excluded from the proxy matcher, so the route refuses the dedicated MCP host itself. */
-  beforeParse: ({ request }) => {
-    if (isOffAppHost(request)) throw new OrchestrationError('not_found', 'Not found')
-  },
   mapInput: ({ params, body }) => ({
     ...params,
     docState: new Uint8Array(Buffer.from(body.docState, 'base64')),
@@ -33,3 +29,9 @@ export const POST = defineInternalJsonRoute({
   useCase: persistProjectFileDoc,
   parseOptions: { maxBodyBytes: PERSIST_MAX_BODY_BYTES, bypassesProxy: true },
 })
+
+/** Excluded from the proxy matcher, so the route refuses the dedicated MCP host before auth itself. */
+export const POST: typeof persist = (request, context) =>
+  isOffAppHost(request)
+    ? Promise.resolve(new NextResponse(null, { status: 404 }))
+    : persist(request, context)

apps/sim/lib/api/server/validation.ts

@@ -23,10 +23,11 @@ import {
 /**
  * Next.js buffers the client body for the proxy and *silently truncates* anything
  * past `experimental.proxyClientMaxBodySize` (default 10 MB), and `apps/sim/proxy.ts`
- * matches `/api/:path*`. A larger body therefore reaches the handler as a truncated
- * prefix, which fails JSON parsing — so an oversized request has to be rejected on
- * its declared size before it is read, or the truncation gets misreported as a
- * malformed body.
+ * matches every `/api` route except the few its matcher excludes (see
+ * {@link ParseRequestOptions.bypassesProxy}). A larger body therefore reaches the
+ * handler as a truncated prefix, which fails JSON parsing — so an oversized request
+ * has to be rejected on its declared size before it is read, or the truncation gets
+ * misreported as a malformed body.
  */
 const PROXY_CLIENT_MAX_BODY_BYTES = 10 * 1024 * 1024