improvement(knowledge): resolve the caller's member identities with the rest of the plan

A members-mode document is readable while one of the caller's active members observes
it, freshly, and which members those are is a fact about the caller. Resolving them
with the connectors — one query, one plan — turns each candidate's check into a lookup
on the observation key instead of a join to the member behind it, and lets the vector
planner read the sources the caller belongs to from the same resolution rather than
asking again.

Author: waleedlatif1

apps/sim/lib/knowledge/access/connector-eligibility.ts

@@ -1,7 +1,14 @@
 import { db } from '@sim/db'
-import { knowledgeConnector } from '@sim/db/schema'
+import { knowledgeConnector, knowledgeConnectorMember } from '@sim/db/schema'
 import { and, eq, inArray, isNull, sql } from 'drizzle-orm'
-import type { KnowledgeConnectorEligibility } from '@/lib/knowledge/access/predicate'
+import { SOURCE_ACL_MAX_AGE_MS } from '@/lib/knowledge/access/freshness'
+import {
+  type KnowledgeConnectorEligibility,
+  type KnowledgeMemberObservers,
+  type SearchAccessPlan,
+  textArrayLiteral,
+} from '@/lib/knowledge/access/predicate'
+import type { KnowledgeAccessScope } from '@/lib/knowledge/access/types'
 import { searchIntegrationAccessCondition } from '@/lib/knowledge/search/integration-policy'
 
 /**
@@ -12,7 +19,7 @@ import { searchIntegrationAccessCondition } from '@/lib/knowledge/search/integra
  * facts about a connector. Resolving them once per query — there are tens of connectors against
  * hundreds of thousands of documents — leaves each candidate its own columns to check.
  */
-export async function resolveConnectorEligibility(
+async function resolveConnectorEligibility(
   knowledgeBaseIds: readonly string[]
 ): Promise<KnowledgeConnectorEligibility> {
   const eligibility: {
@@ -52,3 +59,62 @@ export async function resolveConnectorEligibility(
   }
   return eligibility
 }
+
+/**
+ * The caller's own member identities on these connectors, split by whether the member's change
+ * feed is itself current.
+ *
+ * A members-mode document is readable while one of the caller's active members observes it,
+ * freshly — and which members those are is a fact about the caller, not about any document. A
+ * member whose feed drained recently confirms every observation it holds, so its observations need
+ * no age check at all; the rest are checked against the age of the observation itself. Resolved
+ * once, the per-document check becomes one lookup on the observation key, with no join to the
+ * member behind it.
+ */
+async function resolveMemberObservers(
+  access: KnowledgeAccessScope,
+  connectorIds: readonly string[]
+): Promise<{ observers: KnowledgeMemberObservers; memberSources: string[] }> {
+  if (access.kind !== 'user' || connectorIds.length === 0 || access.tokens.length === 0) {
+    return { observers: { confirmed: [], observed: [] }, memberSources: [] }
+  }
+  const rows = await db
+    .select({
+      id: knowledgeConnectorMember.id,
+      connectorId: knowledgeConnectorMember.connectorId,
+      syncedThrough: knowledgeConnectorMember.memberSyncedThrough,
+    })
+    .from(knowledgeConnectorMember)
+    .where(
+      and(
+        inArray(knowledgeConnectorMember.connectorId, [...connectorIds]),
+        eq(knowledgeConnectorMember.status, 'active'),
+        sql`${knowledgeConnectorMember.subjectToken} = ANY(${textArrayLiteral([...access.tokens])})`
+      )
+    )
+  const cutoff = Date.now() - SOURCE_ACL_MAX_AGE_MS
+  const confirmed: string[] = []
+  const observed: string[] = []
+  const memberSources = new Set<string>()
+  for (const row of rows) {
+    if (row.syncedThrough !== null && row.syncedThrough.getTime() > cutoff) confirmed.push(row.id)
+    else observed.push(row.id)
+    memberSources.add(row.connectorId)
+  }
+  return { observers: { confirmed, observed }, memberSources: [...memberSources] }
+}
+
+/**
+ * Everything a search needs to know about its sources and the caller's standing in them, resolved
+ * once: which connectors it may read, the caller's member identities there, and the sources they
+ * are a member of. Each is a fact about a connector or a caller, so deriving them per candidate
+ * document is what made retrieval cost grow with the size of what someone may read.
+ */
+export async function resolveSearchAccessPlan(
+  knowledgeBaseIds: readonly string[],
+  access: KnowledgeAccessScope
+): Promise<SearchAccessPlan> {
+  const connectors = await resolveConnectorEligibility(knowledgeBaseIds)
+  const { observers, memberSources } = await resolveMemberObservers(access, connectors.members)
+  return { connectors, observers, memberSources }
+}

apps/sim/lib/knowledge/access/predicate.postgres.test.ts

@@ -553,8 +553,11 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
       members: ['members'],
       liveProofRequired: [],
     }
+    /** What `resolveSearchAccessPlan` resolves for this caller: their member identity, confirmed. */
+    const observers = { confirmed: ['m-alice'], observed: [] }
     const perRow = knowledgeMetadataCandidateAccessCondition(scope)
-    const perQuery = knowledgeCandidateAccessConditionForConnectors(scope, eligibility)
+    const plan = { connectors: eligibility, observers, memberSources: ['members'] }
+    const perQuery = knowledgeCandidateAccessConditionForConnectors(scope, plan)
     for (const id of [...cases.map(([documentId]) => documentId), 'upload-doc']) {
       expect([id, await admits(perQuery, id)]).toEqual([id, await admits(perRow, id)])
     }
@@ -563,7 +566,7 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
      * caller holds those grants only after authorization, so applying the clause during ranking
      * would drop every candidate of a gated source before it could be proven.
      */
-    const gated = { ...eligibility, liveProofRequired: ['admin'] }
+    const gated = { ...plan, connectors: { ...eligibility, liveProofRequired: ['admin'] } }
     expect(
       await admits(knowledgeCandidateAccessConditionForConnectors(scope, gated), 'admin-current')
     ).toBe(true)
@@ -576,7 +579,10 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
     /** A connector left out of the resolution is refused, however current its documents are. */
     expect(
       await admits(
-        knowledgeCandidateAccessConditionForConnectors(scope, { ...eligibility, admin: [] }),
+        knowledgeCandidateAccessConditionForConnectors(scope, {
+          ...plan,
+          connectors: { ...eligibility, admin: [] },
+        }),
         'admin-current'
       )
     ).toBe(false)

apps/sim/lib/knowledge/access/predicate.ts

@@ -208,6 +208,25 @@ export function liveSourceAccessCondition(scope: KnowledgeAccessScope): SQL {
  * The connectors a search may read from, resolved once per query: their ids grouped by the shape
  * their documents' ACLs take, and separately those whose reader access is proven live per request.
  */
+/**
+ * The caller's active member identities on the connectors a search reads, by what makes their
+ * observations current: `confirmed` members drained their change feed inside the freshness window,
+ * so every observation they hold stands; `observed` members are trusted only where the observation
+ * itself is recent.
+ */
+export interface KnowledgeMemberObservers {
+  confirmed: readonly string[]
+  observed: readonly string[]
+}
+
+/** What a search resolves once about its sources and the caller's standing in them. */
+export interface SearchAccessPlan {
+  connectors: KnowledgeConnectorEligibility
+  observers: KnowledgeMemberObservers
+  /** Connectors the caller is an active member of, whose documents they read broadly. */
+  memberSources: readonly string[]
+}
+
 export interface KnowledgeConnectorEligibility {
   /** Documents carry the workspace ACL. */
   workspace: readonly string[]
@@ -238,9 +257,10 @@ export interface KnowledgeConnectorEligibility {
  */
 export function knowledgeCandidateAccessConditionForConnectors(
   scope: KnowledgeAccessScope | SystemAccessScope,
-  eligibility: KnowledgeConnectorEligibility,
+  plan: SearchAccessPlan,
   liveSourceAccess: SQL = sql`true`
 ): SQL {
+  const eligibility = plan.connectors
   if (scope.kind === 'system') return documentConnectorIsActive()
   if (scope.tokens.length === 0) return sql`false`
   const tokens = textArrayLiteral(scope.tokens)
@@ -270,7 +290,7 @@ export function knowledgeCandidateAccessConditionForConnectors(
       ((${document.connectorId} IS NULL OR ${inConnectors(eligibility.workspace)})
         AND ${document.acl} = ARRAY['ws']::text[])
       OR ${mirrored(eligibility.admin, sql`${document.aclVerifiedAt} > ${cutoff}`)}
-      OR ${mirrored(eligibility.members, memberObservationCondition(tokens, cutoff))}
+      OR ${mirrored(eligibility.members, resolvedObservationCondition(plan.observers, cutoff))}
     )
   )`
 }
@@ -282,6 +302,29 @@ export function knowledgeCandidateAccessConditionForConnectors(
  * inside the access predicate's `OR`, PostgreSQL instead hashes every observation in the table
  * once per statement, a fixed cost paid by every query that carries the predicate.
  */
+/**
+ * The same membership, resolved ahead of the query: each candidate costs one lookup on the
+ * observation key instead of a join to the member behind it. Equivalent by construction — the ids
+ * are the members that join would have matched, and each one's freshness rule is carried over.
+ */
+function resolvedObservationCondition(observers: KnowledgeMemberObservers, cutoff: SQL): SQL {
+  if (observers.confirmed.length === 0 && observers.observed.length === 0) return sql`false`
+  const byMember = (ids: readonly string[]): SQL =>
+    sql`${knowledgeDocumentObservation.memberId} = ANY(${textArrayLiteral([...ids])})`
+  const current =
+    observers.confirmed.length === 0
+      ? sql`${byMember(observers.observed)} AND ${knowledgeDocumentObservation.lastSeenAt} > ${cutoff}`
+      : observers.observed.length === 0
+        ? byMember(observers.confirmed)
+        : sql`(${byMember(observers.confirmed)}
+            OR (${byMember(observers.observed)} AND ${knowledgeDocumentObservation.lastSeenAt} > ${cutoff}))`
+  return sql`EXISTS (
+    SELECT 1 FROM ${knowledgeDocumentObservation}
+    WHERE ${knowledgeDocumentObservation.documentId} = ${document.id}
+      AND ${current}
+  )`
+}
+
 function memberObservationCondition(tokens: SQL, cutoff: SQL): SQL {
   return sql`EXISTS (
     SELECT 1 FROM ${knowledgeDocumentObservation}

apps/sim/lib/knowledge/search/queries.test.ts

@@ -1434,7 +1434,6 @@ describe('permitted-document planner', () => {
   let exactRows: Array<{ id: string }>
   let traversedRows: Array<{ id: string; distance?: number }>
   let rerankRows: Array<ReturnType<typeof hit>>
-  let memberedSources: Array<{ connectorId: string }>
   let indexedSourceRows: Array<{ name: string; connectorId: string }>
   let sourceExactRows: Array<{ id: string; distance: number }>
 
@@ -1445,7 +1444,6 @@ describe('permitted-document planner', () => {
     traversedRows = []
     rerankRows = []
     sourceExactRows = []
-    memberedSources = []
     indexedSourceRows = []
     dbChainMockFns.execute.mockImplementation(async (query) => {
       const statement = render(query).sql
@@ -1501,17 +1499,17 @@ describe('permitted-document planner', () => {
       liveProofRequired: [],
     }
     /** Membership decides the walk; the sliced source contributes enumerated documents. */
-    memberedSources = [{ connectorId: 'member-src' }]
+    const memberSources = ['member-src']
+    const observers = { confirmed: ['m-1'], observed: [] }
     indexedSourceRows = [{ name: 'idx', connectorId: 'member-src' }]
-    queueTableRows(schemaMock.knowledgeConnectorMember, memberedSources)
     sourceExactRows = [{ id: 'sliced-hit', distance: 0.05 }]
     traversedRows = [{ id: 'walked-hit', distance: 0.2 }]
     rerankRows = [hit('sliced-hit', 'sliced-src'), hit('walked-hit', 'member-src')]
     queueTableRows(schemaMock.embedding, rerankRows)
     await handleVectorOnlySearch({
       ...params,
       permitted: { kind: 'unbounded' },
-      connectorEligibility: eligibility,
+      accessPlan: { connectors: eligibility, observers, memberSources },
     })
     const walks = statements().filter((query) => query.sql.includes('AS visible'))
     expect(walks).toHaveLength(1)
@@ -1525,18 +1523,16 @@ describe('permitted-document planner', () => {
   })
 
   it('ranks every source exactly when the caller is a member of none', async () => {
-    memberedSources = []
     sourceExactRows = [{ id: 'sliced-hit', distance: 0.05 }]
     rerankRows = [hit('sliced-hit', 'sliced-src')]
     queueTableRows(schemaMock.embedding, rerankRows)
     await handleVectorOnlySearch({
       ...params,
       permitted: { kind: 'unbounded' },
-      connectorEligibility: {
-        workspace: [],
-        admin: ['sliced-src'],
-        members: [],
-        liveProofRequired: [],
+      accessPlan: {
+        connectors: { workspace: [], admin: ['sliced-src'], members: [], liveProofRequired: [] },
+        observers: { confirmed: [], observed: [] },
+        memberSources: [],
       },
     })
     expect(statements().filter((query) => query.sql.includes('AS visible'))).toHaveLength(0)