fix(knowledge): close the audit findings on administrator access

Admin mode was unreachable end to end: the queue still refused every mode
but workspace, and neither modal ever sent the mode, so editing an admin
connector saved it as workspace and published the corpus. A switch into
admin mode also flipped before hiding, showing workspace documents under
a mode whose reader expects source ACLs.

- queue: dispatch every content-engine mode; the modals send the chosen
  mode; the access field offers administrator mode on availability, not
  on credential-group support
- switch: hide before flipping into a mode that hides on entry; the
  engine finishes a pending hide before it lists, strips listing caps,
  and hides owned documents the listing did not name
- Drive: domain shares resolve through a synthetic per-domain group whose
  one member is a wildcard the reader matches by their own domain; CUSTOMER
  members and every customer domain are covered
- Confluence: each page falls back to its own space's readers, not the
  union across spaces; ancestors from the v2 collection the API still
  serves; space-role assignments expanded; pagination follows next links;
  per-page containment; retry on every call
- directory sync: one background job per directory instead of a full walk
  inside the scheduler request; runnable statuses only; freshness keyed off
  the latest confirmed group so one unreadable group cannot keep a directory
  due forever
- config edits on an admin connector re-assert the administrator subject;
  a connector whose ACL is derived cannot leave its documents behind
- the access-mode enum lives in the leaf module and the contract derives
  from it; the mirror assertion moves to a leaf; the remaining ad-hoc
  email folds use foldedEmail; the impersonated subject leaves the logs

Author: waleedlatif1

apps/sim/app/api/knowledge/connectors/directory-sync/route.test.ts

@@ -4,41 +4,21 @@
 import { createMockRequest } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const {
-  mockVerifyCronAuth,
-  mockConnectorRows,
-  mockResolveIdentity,
-  mockResolveToken,
-  mockRefreshDirectory,
-} = vi.hoisted(() => ({
+const { mockVerifyCronAuth, mockConnectorRows, mockDispatch } = vi.hoisted(() => ({
   mockVerifyCronAuth: vi.fn(() => null),
   mockConnectorRows: vi.fn(),
-  mockResolveIdentity: vi.fn(),
-  mockResolveToken: vi.fn(),
-  mockRefreshDirectory: vi.fn(),
+  mockDispatch: vi.fn(),
 }))
 
 vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mockVerifyCronAuth }))
-vi.mock('@/lib/credentials/access', () => ({
-  resolveCredentialTokenIdentity: mockResolveIdentity,
-}))
-vi.mock('@/lib/knowledge/connectors/access-token', () => ({
-  resolveConnectorAccessToken: mockResolveToken,
-}))
-vi.mock('@/lib/knowledge/connectors/external-group-sync', () => ({
-  refreshMirroredDirectory: mockRefreshDirectory,
-}))
-vi.mock('@/connectors/registry.server', () => ({
-  CONNECTOR_REGISTRY: {
-    google_drive: { auth: { mode: 'oauth', provider: 'google-drive' }, openDirectory: vi.fn() },
-    notion: { auth: { mode: 'oauth', provider: 'notion' } },
-  },
+vi.mock('@/lib/knowledge/connectors/directory-queue', () => ({
+  dispatchDirectorySync: mockDispatch,
 }))
 vi.mock('@sim/db', () => ({
   db: {
     select: () => ({
       from: () => ({
-        innerJoin: () => ({ where: () => ({ limit: () => mockConnectorRows() }) }),
+        innerJoin: () => ({ where: () => ({ orderBy: () => mockConnectorRows() }) }),
       }),
     }),
   },
@@ -50,11 +30,7 @@ function connector(overrides: Record<string, unknown> = {}) {
   return {
     id: 'connector-1',
     connectorType: 'google_drive',
-    credentialId: 'credential-1',
-    encryptedApiKey: null,
-    sourceConfig: { adminEmail: 'admin@corp.com' },
     workspaceId: 'ws-1',
-    knowledgeBaseOwnerId: 'owner-1',
     ...overrides,
   }
 }
@@ -68,57 +44,51 @@ describe('connector directory sync scheduler', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mockVerifyCronAuth.mockReturnValue(null)
-    mockResolveIdentity.mockResolvedValue({ kind: 'service_account' })
-    mockResolveToken.mockResolvedValue({ accessToken: 'token' })
-    mockRefreshDirectory.mockResolvedValue(undefined)
+    mockDispatch.mockResolvedValue(undefined)
   })
 
-  it('refreshes the directory of every admin-mode connector it finds', async () => {
-    mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
-
-    await expect(run()).resolves.toMatchObject({ considered: 2, refreshed: 2, failed: 0 })
-    expect(mockRefreshDirectory).toHaveBeenCalledTimes(2)
+  it('dispatches one refresh per directory an admin-mode connector mirrors', async () => {
+    mockConnectorRows.mockResolvedValue([
+      connector(),
+      connector({ id: 'connector-2', workspaceId: 'ws-2' }),
+    ])
+
+    await expect(run()).resolves.toMatchObject({
+      considered: 2,
+      directories: 2,
+      dispatched: 2,
+      failed: 0,
+    })
+    expect(mockDispatch).toHaveBeenCalledTimes(2)
   })
 
   /**
-   * The tick refreshes every workspace's directory, so one workspace's lapsed
-   * credential or unreachable source must not stop the others.
+   * Two connectors of one type in one workspace mirror one directory; walking
+   * it twice per tick would double the Admin SDK cost for nothing.
    */
-  it('contains a failure to the connector that caused it', async () => {
+  it('dispatches once for connectors that share a directory', async () => {
     mockConnectorRows.mockResolvedValue([connector(), connector({ id: 'connector-2' })])
-    mockRefreshDirectory.mockRejectedValueOnce(new Error('directory unreachable'))
 
-    await expect(run()).resolves.toMatchObject({ considered: 2, refreshed: 1, failed: 1 })
+    await expect(run()).resolves.toMatchObject({ considered: 2, directories: 1, dispatched: 1 })
+    expect(mockDispatch).toHaveBeenCalledTimes(1)
+    expect(mockDispatch).toHaveBeenCalledWith('connector-1', expect.anything())
   })
 
-  it('reports a connector whose credential no longer resolves rather than failing', async () => {
-    mockConnectorRows.mockResolvedValue([connector()])
-    mockResolveToken.mockResolvedValue(null)
+  it('contains a dispatch failure to the directory that caused it', async () => {
+    mockConnectorRows.mockResolvedValue([
+      connector(),
+      connector({ id: 'connector-2', workspaceId: 'ws-2' }),
+    ])
+    mockDispatch.mockRejectedValueOnce(new Error('queue unreachable'))
 
-    await expect(run()).resolves.toMatchObject({ unusable: 1, refreshed: 0 })
-    expect(mockRefreshDirectory).not.toHaveBeenCalled()
+    await expect(run()).resolves.toMatchObject({ dispatched: 1, failed: 1 })
   })
 
-  it('skips a connector whose source has no directory to read', async () => {
-    mockConnectorRows.mockResolvedValue([connector({ connectorType: 'notion' })])
-
-    await expect(run()).resolves.toMatchObject({ skipped: 1, refreshed: 0 })
-    expect(mockRefreshDirectory).not.toHaveBeenCalled()
-  })
-
-  /**
-   * Token reads are scoped to the credential's own account owner, not the
-   * knowledge base owner, who is routinely a different member.
-   */
-  it('resolves the token as the credential owner for an OAuth credential', async () => {
-    mockConnectorRows.mockResolvedValue([connector()])
-    mockResolveIdentity.mockResolvedValue({ kind: 'oauth', userId: 'credential-owner' })
-
-    await run()
+  it('skips a connector whose knowledge base has no workspace', async () => {
+    mockConnectorRows.mockResolvedValue([connector({ workspaceId: null })])
 
-    expect(mockResolveToken).toHaveBeenCalledWith(
-      expect.objectContaining({ userId: 'credential-owner' })
-    )
+    await expect(run()).resolves.toMatchObject({ directories: 0, dispatched: 0 })
+    expect(mockDispatch).not.toHaveBeenCalled()
   })
 
   it('refuses an unauthenticated tick', async () => {

apps/sim/app/api/knowledge/connectors/directory-sync/route.ts

@@ -2,24 +2,20 @@ import { db } from '@sim/db'
 import { knowledgeBase, knowledgeConnector } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
-import { and, eq, isNull } from 'drizzle-orm'
+import { and, asc, eq, inArray, isNull } from 'drizzle-orm'
 import type { NextRequest } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { resolveCredentialTokenIdentity } from '@/lib/credentials/access'
-import { resolveConnectorAccessToken } from '@/lib/knowledge/connectors/access-token'
-import { refreshMirroredDirectory } from '@/lib/knowledge/connectors/external-group-sync'
-import { CONNECTOR_REGISTRY } from '@/connectors/registry.server'
+import { dispatchDirectorySync } from '@/lib/knowledge/connectors/directory-queue'
+import { RUNNABLE_CONNECTOR_STATUSES } from '@/lib/knowledge/connectors/sync-lock'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('ConnectorDirectorySyncSchedulerAPI')
 
-/** Directories refreshed per tick, and how many at once. */
+/** Directories dispatched per tick. */
 const MAX_DIRECTORIES_PER_TICK = 200
-const REFRESH_CONCURRENCY = 4
 
 /**
  * Refreshes the external directories that admin-mode connectors mirror.
@@ -30,15 +26,15 @@ const REFRESH_CONCURRENCY = 4
  * admin crawl refreshes the directory too — so a crawl can never publish grants
  * against membership nobody has read — but that is a floor, not the cadence.
  *
- * Every eligible connector is offered each tick; `syncExternalDirectoryGroups`
- * decides whether its directory is actually due. Connectors sharing a directory
- * therefore cost one refresh between them: the first brings it up to date and
- * the rest skip. Two ticks overlapping on one directory would both enumerate
- * and write the same rows — wasteful, never wrong, and not worth a lease to
- * prevent, since every write here is idempotent.
+ * Connectors sharing a directory cost one refresh between them: the tick
+ * dispatches one connector per workspace and provider, and
+ * `syncExternalDirectoryGroups` decides whether that directory is actually
+ * due. The walk itself runs in the background, like every other connector
+ * job, because a large domain takes longer than a scheduler request lives.
  */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   const requestId = generateRequestId()
+  const tickAt = new Date()
   logger.info(`[${requestId}] Connector directory sync scheduler triggered`)
 
   const authError = verifyCronAuth(request, 'Connector directory sync scheduler')
@@ -48,83 +44,50 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     .select({
       id: knowledgeConnector.id,
       connectorType: knowledgeConnector.connectorType,
-      credentialId: knowledgeConnector.credentialId,
-      encryptedApiKey: knowledgeConnector.encryptedApiKey,
-      sourceConfig: knowledgeConnector.sourceConfig,
       workspaceId: knowledgeBase.workspaceId,
-      knowledgeBaseOwnerId: knowledgeBase.userId,
     })
     .from(knowledgeConnector)
     .innerJoin(knowledgeBase, eq(knowledgeConnector.knowledgeBaseId, knowledgeBase.id))
     .where(
       and(
         eq(knowledgeConnector.accessMode, 'admin'),
+        inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES),
         isNull(knowledgeConnector.archivedAt),
         isNull(knowledgeConnector.deletedAt),
         isNull(knowledgeBase.deletedAt)
       )
     )
-    .limit(MAX_DIRECTORIES_PER_TICK)
+    .orderBy(asc(knowledgeConnector.createdAt))
 
-  const outcomes = await mapWithConcurrency(connectors, REFRESH_CONCURRENCY, async (connector) => {
-    /**
-     * Every failure is contained here. One workspace whose credential lapsed
-     * must not stop the tick refreshing every other workspace's directory.
-     */
-    try {
-      if (!connector.workspaceId) return 'skipped'
-      const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType]
-      if (!connectorConfig?.openDirectory) return 'skipped'
-
-      /**
-       * The credential's own account owner, not the knowledge base owner —
-       * token reads are scoped to `account.userId`, and a service account
-       * ignores the argument entirely.
-       */
-      let credentialUserId = connector.knowledgeBaseOwnerId
-      if (connector.credentialId) {
-        const identity = await resolveCredentialTokenIdentity(
-          connector.credentialId,
-          connector.workspaceId
-        )
-        if (!identity) return 'unusable'
-        if (identity.kind === 'oauth') credentialUserId = identity.userId
-      }
-
-      const sourceConfig = connector.sourceConfig as Record<string, unknown>
-      const token = await resolveConnectorAccessToken({
-        auth: connectorConfig.auth,
-        connector,
-        userId: credentialUserId,
-        requestId,
-        sourceConfig,
-      })
-      if (!token) return 'unusable'
+  /**
+   * One connector per directory. A connector type implies its provider, and
+   * two connectors of one type in one workspace share a directory by
+   * construction — the first to be created stands for it.
+   */
+  const representatives = new Map<string, string>()
+  for (const connector of connectors) {
+    if (!connector.workspaceId) continue
+    const key = `${connector.workspaceId}:${connector.connectorType}`
+    if (!representatives.has(key)) representatives.set(key, connector.id)
+  }
+  const due = [...representatives.values()].slice(0, MAX_DIRECTORIES_PER_TICK)
 
-      await refreshMirroredDirectory({
-        workspaceId: connector.workspaceId,
-        connectorConfig,
-        sourceConfig,
-        syncContext: {},
-        accessToken: token.accessToken,
-      })
-      return 'refreshed'
+  let dispatched = 0
+  let failed = 0
+  for (const connectorId of due) {
+    try {
+      await dispatchDirectorySync(connectorId, { requestId, tickAt })
+      dispatched += 1
     } catch (error) {
-      logger.error(`[${requestId}] Directory refresh failed for a connector`, {
-        connectorId: connector.id,
+      failed += 1
+      logger.error(`[${requestId}] Failed to dispatch a directory refresh`, {
+        connectorId,
         error: getErrorMessage(error),
       })
-      return 'failed'
     }
-  })
-
-  const summary = {
-    considered: connectors.length,
-    refreshed: outcomes.filter((outcome) => outcome === 'refreshed').length,
-    skipped: outcomes.filter((outcome) => outcome === 'skipped').length,
-    unusable: outcomes.filter((outcome) => outcome === 'unusable').length,
-    failed: outcomes.filter((outcome) => outcome === 'failed').length,
   }
+
+  const summary = { considered: connectors.length, directories: due.length, dispatched, failed }
   logger.info(`[${requestId}] Connector directory sync scheduler finished`, summary)
   return Response.json({ success: true, ...summary })
 })

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/add-connector-modal/add-connector-modal.tsx

@@ -95,6 +95,7 @@ export function AddConnectorModal({
   const { ownerBilling, features } = useWorkspaceHostContext()
   const { canAdmin } = useUserPermissionsContext()
   const memberAccessAvailable = features?.knowledgeMemberAccess === true
+  const mirroredAccessAvailable = features?.knowledgeSourceMirroredAccess === true
   const { mutate: createConnector, isPending: isCreating } = useCreateConnector()
 
   const hasMaxAccess = hasWorkspaceMaxConnectorAccess(ownerBilling)
@@ -247,7 +248,7 @@ export function AddConnectorModal({
                 credentialGroupId: access.credentialGroupId,
                 credentialGroupOptionId: access.credentialGroupOptionId,
               }
-            : { credentialId: effectiveCredentialId! }),
+            : { accessMode: access.accessMode, credentialId: effectiveCredentialId! }),
         sourceConfig: finalSourceConfig,
         syncIntervalMinutes: syncInterval,
       },
@@ -333,13 +334,15 @@ export function AddConnectorModal({
             </div>
           ) : connectorConfig ? (
             <>
-              {!isApiKeyMode && memberAccessAvailable && (
+              {!isApiKeyMode && (memberAccessAvailable || mirroredAccessAvailable) && (
                 <ConnectorAccessField
                   connectorConfig={connectorConfig}
                   value={access}
                   onChange={setAccess}
                   groupOptions={groupOptions}
                   canAdmin={canAdmin}
+                  allowMembers={memberAccessAvailable}
+                  allowAdmin={mirroredAccessAvailable}
                   disabled={isCreating}
                 />
               )}