feat(knowledge): project document ACL and chunk changes asynchronously

.github/workflows/test-build.yml

@@ -282,6 +282,8 @@ jobs:
           lib/knowledge/__integration__/kb-block-search.integration.ts
           lib/knowledge/__integration__/gitlab-workspace.integration.ts
           lib/knowledge/__integration__/unfilled-projection-source.integration.ts
+          lib/knowledge/__integration__/knowledge-projection.integration.ts
+          lib/knowledge/__integration__/async-projection-processing.integration.ts
           lib/knowledge/__integration__/purged-detach-reservation.integration.ts
           lib/core/outbox/service.integration.ts
           lib/knowledge/__integration__/connector-upload.integration.ts

apps/docs/content/docs/platform/self-hosting/background-jobs.mdx

@@ -47,6 +47,7 @@ Point cron at an **internal** address where possible (the in-cluster Service, or
 | Time pause/resume | `/api/resume/poll` | `*/1 * * * *` | Workflows paused on a timer |
 | Outbox processing | `/api/webhooks/outbox/process` | `*/1 * * * *` | Transactional-outbox retries for billing, membership, enterprise issuance, and workflow-deployment side effects |
 | Workspace file search dispatch | `/api/cron/workspace-file-search-dispatch` | `*/1 * * * *` | Dispatches indexing work for workspace file search |
+| Knowledge projection | `/api/cron/knowledge-projection` | `*/1 * * * *` | Brings knowledge base search up to date with document, permission, and chunk changes |
 | Connector sync | `/api/knowledge/connectors/sync` | `*/5 * * * *` | Knowledge base connector syncs |
 | Connector member sync | `/api/knowledge/connectors/member-sync` | `*/5 * * * *` | Per-member access sync for permission-aware connectors |
 | Connector directory sync | `/api/knowledge/connectors/directory-sync` | `*/5 * * * *` | Refreshes the directory groups administrator-mode connectors mirror, so a membership change takes effect without waiting for a content sync |

apps/sim/app/api/cron/knowledge-projection/route.test.ts

@@ -0,0 +1,86 @@
+/**
+ * @vitest-environment node
+ */
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  enqueueSweep: vi.fn(),
+  verifyCronAuth: vi.fn(),
+}))
+
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mocks.verifyCronAuth }))
+vi.mock('@/lib/knowledge/projection/enqueue', () => ({
+  enqueueKnowledgeProjectionSweep: mocks.enqueueSweep,
+}))
+
+import { GET } from '@/app/api/cron/knowledge-projection/route'
+
+function request() {
+  return createMockRequest(
+    'GET',
+    undefined,
+    {},
+    'http://localhost:3000/api/cron/knowledge-projection'
+  )
+}
+
+describe('knowledge projection sweep route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mocks.verifyCronAuth.mockReturnValue(null)
+  })
+
+  it('returns as soon as Trigger.dev accepts the pass', async () => {
+    mocks.enqueueSweep.mockResolvedValue({
+      triggered: true,
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(202)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      triggered: true,
+      backend: 'trigger-dev',
+      jobId: 'run-1',
+    })
+  })
+
+  it('answers 200 without a pass when the projector has nothing to do', async () => {
+    mocks.enqueueSweep.mockResolvedValue({ triggered: false, backend: null, jobId: null })
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      triggered: false,
+      backend: null,
+      jobId: null,
+    })
+  })
+
+  it('returns the cron auth refusal without enqueueing', async () => {
+    mocks.verifyCronAuth.mockReturnValue(new Response(null, { status: 401 }))
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(401)
+    expect(mocks.enqueueSweep).not.toHaveBeenCalled()
+  })
+
+  it('fails closed when Trigger.dev does not accept the pass', async () => {
+    mocks.enqueueSweep.mockRejectedValue(new Error('trigger unavailable'))
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(500)
+    await expect(response.json()).resolves.toEqual({
+      success: false,
+      error: 'Sweep enqueue failed',
+    })
+  })
+})

apps/sim/app/api/cron/knowledge-projection/route.ts

@@ -0,0 +1,29 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { type NextRequest, NextResponse } from 'next/server'
+import { verifyCronAuth } from '@/lib/auth/internal'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { enqueueKnowledgeProjectionSweep } from '@/lib/knowledge/projection/enqueue'
+
+const logger = createLogger('KnowledgeProjectionSweepRoute')
+
+export const dynamic = 'force-dynamic'
+export const maxDuration = 60
+
+/**
+ * The knowledge projector's periodic sweep: enqueues one pass per window while there is work, and
+ * returns once Trigger.dev accepts it. Writers ask for passes as they commit; this converges
+ * whatever those requests missed.
+ */
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  const authError = verifyCronAuth(request, 'Knowledge projection sweep')
+  if (authError) return authError
+
+  try {
+    const result = await enqueueKnowledgeProjectionSweep()
+    return NextResponse.json({ success: true, ...result }, { status: result.triggered ? 202 : 200 })
+  } catch (error) {
+    logger.error('Knowledge projection sweep enqueue failed', { error: getErrorMessage(error) })
+    return NextResponse.json({ success: false, error: 'Sweep enqueue failed' }, { status: 500 })
+  }
+})

apps/sim/background/knowledge-projection.ts

@@ -0,0 +1,45 @@
+import { task } from '@trigger.dev/sdk'
+import {
+  type BackgroundRetryPolicy,
+  backgroundRetryAttemptCeiling,
+  getBackgroundRetryDecision,
+} from '@/lib/core/errors/background-retry'
+import {
+  KNOWLEDGE_PROJECTION_PASS_BUDGET_MS,
+  KNOWLEDGE_PROJECTION_TASK_ID,
+  requestKnowledgeProjection,
+} from '@/lib/knowledge/projection/enqueue'
+import { runKnowledgeProjectionPass } from '@/lib/knowledge/projection/run'
+
+/**
+ * A pass gives a single document up on a lock or statement timeout without failing, so a failed
+ * pass lost its connection or its database. Those back off for minutes; the sweep starts a fresh
+ * pass every minute regardless, so a few attempts are enough.
+ */
+export const KNOWLEDGE_PROJECTION_RETRY_POLICY: BackgroundRetryPolicy = {
+  maxAttempts: 2,
+  database: { maxAttempts: 3, baseDelayMs: 60 * 1000, maxDelayMs: 5 * 60 * 1000 },
+}
+
+/**
+ * Runs one knowledge projector pass. One pass runs at a time and projects several documents at
+ * once itself; the prompt requests and the sweep collapse into whichever pass is queued. A pass
+ * that ran out of budget with marks left asks for the next one. Retry-safe: a pass writes only rows
+ * that differ from their source and removes a mark only on the generation it read.
+ */
+export const knowledgeProjectionTask = task({
+  id: KNOWLEDGE_PROJECTION_TASK_ID,
+  machine: 'small-1x',
+  maxDuration: 15 * 60,
+  retry: { maxAttempts: backgroundRetryAttemptCeiling(KNOWLEDGE_PROJECTION_RETRY_POLICY) },
+  queue: { name: KNOWLEDGE_PROJECTION_TASK_ID, concurrencyLimit: 1 },
+  catchError: async ({ error, ctx }) =>
+    getBackgroundRetryDecision(error, ctx.attempt.number, KNOWLEDGE_PROJECTION_RETRY_POLICY),
+  run: async () => {
+    const result = await runKnowledgeProjectionPass({
+      budgetMs: KNOWLEDGE_PROJECTION_PASS_BUDGET_MS,
+    })
+    if (result.remaining) await requestKnowledgeProjection()
+    return result
+  },
+})

apps/sim/lib/core/config/env.ts

@@ -464,6 +464,7 @@ export const env = createEnv({
     KB_CONFIG_CONCURRENCY_LIMIT:           z.number().optional().default(20),      // Per-tenant concurrent document-processing runs in the interactive lane
     KB_CONFIG_BACKFILL_CONCURRENCY_LIMIT:  z.number().optional().default(20),      // Per-tenant concurrent document-processing runs in the connector-backfill lane
     KB_CONFIG_EMBEDDING_CONCURRENCY:       z.number().optional().default(8),       // Concurrent embedding API requests within one embed call
+    KB_CONFIG_PROJECTION_CONCURRENCY:      z.number().optional().default(8),       // Most documents one knowledge projector pass projects at once, each on its own connection
     /** Deployment operating budgets shared by every caller using the same provider credential. */
     KB_CONFIG_EMBEDDING_REQUESTS_PER_MINUTE: z.number().positive().optional().default(600),
     KB_CONFIG_EMBEDDING_TOKENS_PER_MINUTE:   z.number().positive().optional().default(600000),
@@ -634,6 +635,8 @@ export const env = createEnv({
     CREDENTIAL_GROUPS:                    z.boolean().optional(),                 // Enable enterprise Credential Groups globally
     KNOWLEDGE_MEMBER_ACCESS:              z.boolean().optional(),                 // Enable per-member knowledge connectors and hybrid-by-default retrieval globally
     KNOWLEDGE_TIN_KEYWORD:                z.boolean().optional(),                 // Rank large-scope keyword retrieval through the Tin text index where it exists
+    KNOWLEDGE_ASYNC_PROJECTION:           z.boolean().optional(),                 // Knowledge writers leave search projection rows to the background projector
+    KNOWLEDGE_PROJECTION_FILL:            z.boolean().optional(),                 // The knowledge projector fills projection rows written before they carried a source and ACL
 
     // Organizations - for self-hosted deployments
     ORGANIZATIONS_ENABLED:                 z.boolean().optional(),                 // Enable organizations on self-hosted (bypasses plan requirements)

apps/sim/lib/core/config/feature-flags.test.ts

@@ -10,6 +10,8 @@ const { mockFetch, mockIsPlatformAdmin, envRef } = vi.hoisted(() => ({
   mockIsPlatformAdmin: vi.fn(),
   envRef: {
     APPCONFIG_APPLICATION: 'sim-staging' as string | undefined,
+    KNOWLEDGE_PROJECTION_FILL: undefined as boolean | undefined,
+    KNOWLEDGE_ASYNC_PROJECTION: undefined as boolean | undefined,
     APPCONFIG_ENVIRONMENT: 'staging' as string | undefined,
     TABLES_V2_API: undefined as boolean | undefined,
     TABLE_ROW_TTL: undefined as boolean | undefined,
@@ -148,6 +150,8 @@ describe('isFeatureEnabled', () => {
     envRef.CREDENTIAL_GROUPS = undefined
     envRef.KNOWLEDGE_MEMBER_ACCESS = undefined
     envRef.KNOWLEDGE_TIN_KEYWORD = undefined
+    envRef.KNOWLEDGE_ASYNC_PROJECTION = undefined
+    envRef.KNOWLEDGE_PROJECTION_FILL = undefined
     envRef.SLACK_SEARCH_SHARED_APP = undefined
   })
 
@@ -197,6 +201,32 @@ describe('isFeatureEnabled', () => {
     })
   })
 
+  describe('knowledge-async-projection flag', () => {
+    it('is a global switch', async () => {
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(false)
+      envRef.KNOWLEDGE_ASYNC_PROJECTION = true
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(true)
+    })
+
+    it('follows an AppConfig global rule', async () => {
+      withAppConfig({ 'knowledge-async-projection': { enabled: true } })
+      expect(await isFeatureEnabled('knowledge-async-projection')).toBe(true)
+    })
+  })
+
+  describe('knowledge-projection-fill flag', () => {
+    it('is a global switch', async () => {
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(false)
+      envRef.KNOWLEDGE_PROJECTION_FILL = true
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(true)
+    })
+
+    it('follows an AppConfig global rule', async () => {
+      withAppConfig({ 'knowledge-projection-fill': { enabled: true } })
+      expect(await isFeatureEnabled('knowledge-projection-fill')).toBe(true)
+    })
+  })
+
   describe('knowledge-member-access flag', () => {
     it('uses a global fallback switch off AppConfig', async () => {
       expect(await isFeatureEnabled('knowledge-member-access')).toBe(false)

apps/sim/lib/core/config/feature-flags.ts

@@ -105,6 +105,23 @@ const FEATURE_FLAGS = {
       'invalid. Off-AppConfig falls back to KNOWLEDGE_TIN_KEYWORD.',
     fallback: 'KNOWLEDGE_TIN_KEYWORD',
   },
+  'knowledge-async-projection': {
+    description:
+      'Knowledge writers (document processing and connector ACL writes) leave search projection ' +
+      'rows to the background knowledge projector instead of rewriting them in their own ' +
+      'transaction. Global on/off only; turn it on only once no release older than the ' +
+      'projector serves search. Off-AppConfig falls back to KNOWLEDGE_ASYNC_PROJECTION.',
+    fallback: 'KNOWLEDGE_ASYNC_PROJECTION',
+  },
+  'knowledge-projection-fill': {
+    description:
+      'The knowledge projector also fills search projection rows written before they carried ' +
+      "their document's source and ACL, marking at most 100 documents at once so fresh writes " +
+      'never wait behind much of it. Global on/off only; off pauses the fill, and search keeps ' +
+      'deciding unfilled rows on their document. Off-AppConfig falls back to ' +
+      'KNOWLEDGE_PROJECTION_FILL.',
+    fallback: 'KNOWLEDGE_PROJECTION_FILL',
+  },
 } satisfies Record<string, FeatureFlagDefinition>
 
 /**